智能旅行规划Agent架构设计与实现

1. 旅行规划Agent架构概述

这个旅行规划Agent项目旨在解决传统旅行规划工具在多约束条件和个性化推荐上的不足。核心思路是通过多智能体协作架构，结合知识图谱和语义检索技术，实现更智能的旅行规划体验。

整个系统采用模块化设计，主要包含以下几个关键组件：

• 对话Agent：负责与用户交互，收集旅行需求和偏好
• 景点推荐Agent：基于用户偏好和地理信息进行景点推荐
• 酒店推荐Agent：根据行程安排推荐合适的住宿
• 记忆系统：由Qdrant向量数据库和Neo4j图数据库组成，用于存储用户偏好和历史数据
• 工作流引擎：使用LangGraph实现的状态机，协调各Agent的执行流程

2. 核心组件实现细节

2.1 状态定义与记忆系统

系统状态使用TypedDict定义，包含以下关键字段：

python复制class TravelState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # 对话历史
    user_id: str  # 用户唯一标识
    user_preferences: dict  # 用户偏好信息
    attractions: list[dict]  # 推荐的景点列表
    hotels: list[dict]  # 推荐的酒店列表
    current_stage: str  # 当前执行阶段

记忆系统由两部分组成：

1. Qdrant向量数据库：存储非结构化的用户偏好数据，支持语义检索
2. Neo4j图数据库：存储结构化的用户-景点-城市关系数据

记忆管理器的核心方法包括：

python复制class MemoryManager:
    async def save_preferences(self, user_id: str, prefs: dict):
        """保存用户偏好到两个数据库"""
        # 实现细节省略...
    
    async def query_graphrag(self, user_id: str, city: str) -> list[dict]:
        """查询符合用户偏好的景点"""
        # 实现细节省略...

2.2 对话Agent实现

对话Agent负责收集用户需求，其核心逻辑是：

1. 通过LLM解析用户输入，提取偏好信息
2. 判断信息是否完整
3. 将确认的偏好存入记忆系统
4. 根据信息完整度决定是否进入下一阶段

关键代码结构：

python复制class DialogueOutput(BaseModel):
    preferences_update: Dict[str, Any]
    completed: bool

async def dialogue_agent(state: TravelState):
    # 加载已有偏好
    current_prefs = memory_mgr.load_preferences(state["user_id"])
    
    # 解析用户输入
    output = await dialogue_chain.ainvoke({
        "messages": state["messages"][-5:],
        "preferences": current_prefs
    })
    
    # 更新并保存偏好
    updated_prefs = {**current_prefs, **output.preferences_update}
    memory_mgr.save_preferences(state["user_id"], updated_prefs)
    
    return {
        "user_preferences": updated_prefs,
        "completed": output.completed,
        "current_stage": "attractions" if output.completed else "collecting"
    }

2.3 景点推荐Agent实现

景点推荐Agent的工作流程：

1. 通过高德MCP接口查询城市热门景点
2. 使用GraphRAG从记忆系统中检索符合用户偏好的景点
3. 综合两方面的结果进行最终推荐

关键实现代码：

python复制async def attractions_agent(state: TravelState):
    city = state["user_preferences"]["city"]
    
    # 1. MCP查询景点
    search_msg = await attractions_llm.ainvoke([
        {"role": "user", "content": f"搜索 {city} 前10热门景点"}
    ])
    raw_attractions = search_msg.tool_calls[0].args["places"]
    
    # 2. GraphRAG查询
    recommended = await memory_mgr.query_graphrag(state["user_id"], city)
    
    # 3. 综合推荐
    final_list = await llm.ainvoke([
        {"role": "user", "content": f"基于偏好 {state['user_preferences']} 推荐景点"}
    ])
    
    return {
        "attractions": final_list.content,
        "current_stage": "hotels"
    }

3. 工作流编排与执行

3.1 LangGraph工作流定义

工作流的核心是状态机，通过条件边实现自动跳转：

python复制workflow = StateGraph(TravelState)

# 添加节点
workflow.add_node("dialogue", dialogue_agent)
workflow.add_node("attractions", attractions_agent)
workflow.add_node("hotels", hotels_agent)

# 定义条件路由
def route_stage(state: TravelState):
    stage = state["current_stage"]
    if stage == "attractions": return "attractions"
    if stage == "hotels": return "hotels"
    return END

# 设置边
workflow.add_conditional_edges(START, lambda s: "dialogue")
workflow.add_conditional_edges("dialogue", route_stage)
workflow.add_conditional_edges("attractions", route_stage)
workflow.add_edge("hotels", END)

app = workflow.compile()

3.2 执行流程示例

典型执行流程如下：

1. 用户输入初始需求（如"计划北京5天文化游，预算5000"）
2. 对话Agent收集补充信息，直到判断需求完整
3. 自动跳转到景点推荐Agent
4. 景点推荐完成后跳转到酒店推荐Agent
5. 所有推荐完成后流程结束

执行代码示例：

python复制async def run_travel_planning(user_id: str, input_msg: str):
    initial_state = {
        "messages": [HumanMessage(content=input_msg)],
        "user_id": user_id,
        "current_stage": "dialogue"
    }
    
    for event in app.stream(initial_state):
        print(f"节点 {event['node']}: {event}")
    
    return app.get_state(...)["values"]

4. 关键技术实现要点

4.1 高德MCP工具集成

MCP工具通过专用适配器集成：

python复制from langchain_mcp_adapters import MultiServerMCPClient

mcp_client = MultiServerMCPClient({
    "gaode": {
        "transport": "streamable_http",
        "url": "https://gaode-mcp.yourserver/mcp",
        "headers": {"Authorization": "Bearer your_gaode_key"}
    }
})
gaode_tools = await mcp_client.get_tools()

4.2 GraphRAG实现细节

GraphRAG结合了两种查询方式：

1. 向量检索：通过Qdrant实现语义相似度查询
2. 图谱查询：通过Neo4j实现关系查询

典型查询示例：

cypher复制MATCH (u:User {id: $user_id})-[:LIKES]->(a:Attraction)-[:IN]->(c:City {name: $city})
RETURN a.name, a.score

4.3 错误处理与重试机制

建议实现的容错机制：

1. 对MCP接口调用添加重试逻辑
2. 设置查询超时时间
3. 对LLM输出添加验证步骤
4. 关键操作添加回滚能力

5. 扩展与优化方向

5.1 性能优化建议

1. 实现记忆系统的缓存层
2. 对频繁查询添加预计算
3. 优化Neo4j查询语句，添加适当索引
4. 批量处理相似请求

5.2 功能扩展思路

1. 添加交通规划模块
2. 支持多人协同规划
3. 集成天气信息考虑
4. 添加实时价格监控
5. 支持行程动态调整

5.3 监控与维护

建议实现的监控指标：

1. 各Agent执行成功率
2. 平均响应时间
3. 用户满意度反馈
4. 推荐结果采纳率
5. 系统资源使用情况