Modern travel planning requires synthesizing real-time geospatial data, personalized preferences, and domain-specific knowledge. The platform architecture centers on a three-tier fusion model: an LLM-based decision engine, a geospatial service abstraction layer, and a retrieval-augmented generation (RAG) pipeline. This design enables conversational itinerary generation, dynamic map interactions, and context-aware recommendations.

LLM Orchestration and API Gateway

The intelligence layer relies on a dedicated orchestration module that manages multi-turn conversations and external tool routing. Instead of direct model invocation, a gateway service handles authentication, parameter tuning, and response parsing. The orchestrator defines a strict tool schema, allowing the model to delegate tasks like location searches or knowledge retrieval. For complex itinerary generation, a high-capacity model variant processes structured prompts to output standardized markdown guides.

import httpx
from typing import List, Dict, Any, Optional

class ModelGateway:
    def __init__(self, endpoint: str, auth_token: str):
        self.endpoint = endpoint
        self.headers = {"Authorization": f"Bearer {auth_token}", "Content-Type": "application/json"}

    async def generate_text(self, prompt_history: List[Dict[str, str]], tools: Optional[List[Dict]] = None) -> Dict[str, Any]:
        payload = {
            "model": "qwen-plus",
            "input": {"messages": prompt_history},
            "parameters": {"temperature": 0.7, "top_p": 0.9, "max_tokens": 4096}
        }
        if tools:
            payload["parameters"]["tools"] = tools

        async with httpx.AsyncClient(timeout=45.0) as client:
            resp = await client.post(self.endpoint, json=payload, headers=self.headers)
            resp.raise_for_status()
            return resp.json().get("output", {}).get("choices", [{}])[0].get("message", {})

async def orchestrate_conversation(user_input: str, session_history: List[Dict], gateway: ModelGateway, tool_registry: Dict):
    session_history.append({"role": "user", "content": user_input})
    response_msg = await gateway.generate_text(session_history, tools=list(tool_registry.values()))

    if response_msg.get("tool_calls"):
        for call in response_msg["tool_calls"]:
            func_name = call["function"]["name"]
            args = call["function"]["arguments"]
            if func_name in tool_registry:
                exec_result = await tool_registry[func_name](**args)
                session_history.append({"role": "tool", "content": str(exec_result), "name": func_name})
        return await gateway.generate_text(session_history)
    return response_msg

Geospatial Service Abstraction and Tool Binding

Spatial data operations are encapsulated within a dedicated service class that normalizes external mapping APIs. The abstraction handles coordinate transformation, point-of-interest (POI) discovery, multi-modal routing, and meteorological queries. By decoupling the raw HTTP calls from the business logic, the system maintains resilience against API version changes. The LLM orchestrator binds these methods to function-calling schemas, translating natural language intents into precise geographic queries.

import httpx
from urllib.parse import urlencode

class GeoSpatialProvider:
    def __init__(self, api_credential: str, base_uri: str = "https://restapi.amap.com/v3"):
        self.credential = api_credential
        self.base_uri = base_uri

    async def _fetch(self, path: str, query_params: dict) -> dict:
        query_params.update({"key": self.credential})
        url = f"{self.base_uri}/{path}?{urlencode(query_params)}"
        async with httpx.AsyncClient() as client:
            res = await client.get(url)
            return res.json()

    async def resolve_coordinates(self, location_name: str) -> dict:
        return await self._fetch("geocode/geo", {"address": location_name})

    async def query_pois(self, term: str, region: str = "全国", limit: int = 15) -> dict:
        return await self._fetch("place/text", {
            "keywords": term, "city": region, "offset": limit, "extensions": "all"
        })

    async def compute_path(self, start: str, end: str, travel_mode: str = "auto") -> dict:
        mode_map = {"auto": "driving", "foot": "walking", "transit": "bus"}
        endpoint = f"direction/{mode_map.get(travel_mode, 'driving')}"
        return await self._fetch(endpoint, {"origin": start, "destination": end})

    async def fetch_climate(self, region_code: str) -> dict:
        return await self._fetch("weather/weatherInfo", {"city": region_code, "extensions": "base"})

GEO_TOOLS = {
    "locate_place": {
        "type": "function",
        "function": {
            "name": "locate_place",
            "description": "Resolve a textual address to geographic coordinates",
            "parameters": {"type": "object", "properties": {"location_name": {"type": "string"}}, "required": ["location_name"]}
        }
    }
}

Retrieval-Augmented Generation Pipeline

To ground the LLM's outputs in verified travel data, a local vector store pipeline processes unstructured documents into searchable embeddings. The ingestion workflow chunks raw text, generates dense vectors via a dedicated embedding model, and persists them in a lightweight database. During inference, the system performs semantic similarity searches against user queries. Retrieved snippets are injected into the prompt context or exposed as callable tools, ensuring recommendations reference curated guidelines rather than hallucinated data.

import chromadb
from chromadb import EmbeddingFunction, Documents, Embeddings
import httpx
from typing import List

class VectorEmbedder(EmbeddingFunction):
    def __init__(self, model_endpoint: str, auth_key: str):
        self.endpoint = model_endpoint
        self.headers = {"Authorization": f"Bearer {auth_key}"}

    def __call__(self, input: Documents) -> Embeddings:
        embeddings = []
        for doc in input:
            payload = {"model": "text-embedding-v3", "input": {"texts": [doc]}}
            resp = httpx.post(self.endpoint, json=payload, headers=self.headers, timeout=30)
            vec = resp.json().get("output", {}).get("embeddings", [{}])[0].get("embedding", [])
            embeddings.append(vec)
        return embeddings

class KnowledgeRetriever:
    def __init__(self, storage_dir: str = "./vector_store", collection_id: str = "travel_guides"):
        self.client = chromadb.PersistentClient(path=storage_dir)
        self.embedder = VectorEmbedder("https://dashscope.aliyuncs.com/api/v1/services/embeddings/text-embedding/text-embedding", "YOUR_KEY")
        self.store = self.client.get_or_create_collection(name=collection_id, embedding_function=self.embedder)

    def load_documents(self, file_path: str):
        with open(file_path, "r", encoding="utf-8") as f:
            raw_text = f.read()
        segments = [seg.strip() for seg in raw_text.split("\n\n") if len(seg.strip()) > 50]
        ids = [f"seg_{idx}" for idx in range(len(segments))]
        self.store.upsert(documents=segments, ids=ids, metadatas=[{"origin": file_path}] * len(segments))

    def query_context(self, question: str, top_k: int = 4) -> List[str]:
        hits = self.store.query(query_texts=[question], n_results=top_k)
        return hits.get("documents", [[]])[0]

The RAG module integrates into the planning workflow through dual mechanisms. During itinerary compilation, the system proactively fetches destination-specific advisories and appends them to the system prompt. Concurrently, a query_travel_guide tool is registered within the LLM's function schema, enabling on-demand retrieval during conversational exchanges. This hybrid approach balances proactive context loading with reactive fact-checking, ensuring the generated travel plans remain both creative and factually grounded.