1. 为什么我选择用 LangGraph 而不是 LangChain 来做多 Agent 旅游规划1.1 从单链到图编排一次踩坑后的架构反思去年年底我接了一个旅游路线规划的项目需求方想要一个能根据用户偏好自动生成行程的系统。一开始我用的 LangChain 的 SequentialChain把解析需求→搜索景点→规划路线→生成文案串成一条链。跑通 Demo 只用了半天但上线测试第一天就崩了。问题出在旅游规划本质上不是一个线性流程。用户说我想去成都玩三天喜欢美食和自然风光预算三千这里面至少涉及几个需要反复迭代的子任务景点筛选要根据预算动态调整路线要兼顾地理距离和开放时间餐饮推荐要匹配用户口味偏好。用链式结构做每一步的输出都是下一步的输入中间任何一步出问题整条链就断了而且没法回溯。后来我换成了 LangGraph核心变化是把整个流程从链改成了图。图的好处是每个节点可以独立决策节点之间可以有条件边、循环边Agent 之间可以互相调用和反馈。举个实际例子路线规划 Agent 发现某天安排的景点距离太远它可以主动触发一个重新筛选的信号让景点推荐 Agent 重新出一批候选而不是硬着头皮往下走。LangGraph 和 LangChain 的区别用一句话概括就是LangChain 管的是调用顺序LangGraph 管的是状态流转。LangChain 适合做确定性的流水线LangGraph 适合做需要多轮协商、动态决策的场景。旅游规划恰好是后者。1.2 多 Agent 协作编排的核心设计思路这个系统我一共设计了四个 Agent每个 Agent 职责单一但互相配合需求解析 Agent负责把用户的自然语言拆解成结构化参数目的地、天数、预算、偏好标签、出行人数等景点推荐 Agent根据结构化参数检索和筛选景点输出候选池路线规划 Agent从候选池中选取景点并编排每日行程考虑地理距离、开放时间、游玩时长文案生成 Agent把最终路线包装成可读性强的行程单包含推荐理由和实用贴士这四个 Agent 不是简单的串行关系。需求解析完成后景点推荐和路线规划之间会有一个协商循环路线规划 Agent 如果发现候选景点无法满足约束比如预算超了、时间不够会反馈给景点推荐 Agent 要求重新筛选。这个循环最多执行三次三次后如果还不满足就降级处理给用户一个近似最优的方案并说明原因。用 LangGraph 实现这个逻辑核心是定义好State状态和Conditional Edge条件边。State 是一个贯穿全图的字典每个 Agent 读写自己关心的字段。条件边则决定了下一步走哪个节点比如路线规划完成后根据is_feasible字段决定是进入文案生成还是回到景点推荐。实操心得State 的设计是整个项目最容易翻车的地方。我一开始把所有字段都塞进一个扁平字典结果 Agent 之间互相覆盖数据调试了两天才发现。后来改成按 Agent 分组的嵌套结构每个 Agent 只写自己的命名空间问题就解决了。2. 四个 Agent 的详细拆解与 DeepSeek 接入实操2.1 需求解析 Agent把大白话变成结构化参数这个 Agent 是整个系统的入口它的输出质量直接决定了后续所有环节的上限。我用的 DeepSeek 的 chat 模型通过 API 调用Prompt 设计上花了最多心思。核心思路是用 Few-shot JSON Schema 约束输出格式。DeepSeek 对中文的理解很好但如果不加约束它有时候会输出一段解释性文字而不是纯 JSON。我的做法是在 System Prompt 里明确要求只输出 JSON不要任何额外文字然后给两个完整的输入输出示例。import json from openai import OpenAI client OpenAI( api_keyyour_deepseek_api_key, base_urlhttps://api.deepseek.com ) def parse_requirement(user_input: str) - dict: system_prompt 你是一个旅游需求解析助手。请将用户的自然语言输入解析为JSON格式。 输出格式 { destination: 目的地城市, days: 天数(整数), budget: 预算(整数,单位元), preferences: [偏好标签1, 偏好标签2], travelers: 出行人数(整数), special_needs: 特殊需求描述 } 只输出JSON不要任何其他文字。 response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: system_prompt}, {role: user, content: user_input} ], temperature0.1 ) return json.loads(response.choices[0].message.content)这里有几个关键参数需要说明。temperature0.1是为了让输出尽可能稳定解析任务不需要创造性。DeepSeek 的 API 兼容 OpenAI 的 SDK所以直接用 openai 库就能调只需要改base_url。踩过的坑DeepSeek 有时候会在 JSON 外面包一层 json 的代码块标记导致json.loads报错。我的处理方式是加一个清洗函数把首尾的代码块标记去掉再解析。另外如果用户输入特别模糊比如随便玩玩解析出来的字段可能缺失这时候需要有一个默认值填充逻辑。2.2 景点推荐 Agent检索增强生成的实际应用景点推荐不能全靠大模型编那样出来的结果不可靠。我的方案是本地维护一个景点数据库 大模型做语义匹配和排序。景点数据库我用 SQLite 存字段包括名称、城市、类型标签、门票价格、建议游玩时长、开放时间、经纬度、简介。数据来源是公开的旅游数据集加上我自己整理的补充数据大概覆盖了国内 50 个热门旅游城市每个城市 30-80 个景点。推荐流程分两步。第一步是硬过滤根据目的地、预算上限、偏好标签做 SQL 查询筛出一个候选集。第二步是软排序把候选集的景点简介和用户偏好一起喂给 DeepSeek让它按匹配度打分并返回 Top N。def recommend_attractions(params: dict, top_n: int 15) - list: # 第一步硬过滤 conn sqlite3.connect(attractions.db) cursor conn.cursor() query SELECT * FROM attractions WHERE city ? AND price ? cursor.execute(query, (params[destination], params[budget] * 0.3)) candidates cursor.fetchall() # 第二步软排序 candidate_text \n.join([f{c[1]}: {c[7]} for c in candidates]) prompt f用户偏好{, .join(params[preferences])} 以下是候选景点请按匹配度排序返回前{top_n}个景点名称每行一个 {candidate_text} response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], temperature0.3 ) return response.choices[0].message.content.strip().split(\n)注意预算的 30% 用于门票是一个经验值。实际测试下来大部分用户的门票支出占总预算的 20%-35%取 30% 作为硬过滤上限比较合理既不会漏掉好景点也不会推荐明显超预算的。2.3 路线规划 Agent带约束的路径优化这是整个系统技术含量最高的部分。路线规划本质上是一个带时间窗和预算约束的旅行商问题TSP精确求解是 NP-hard 的所以我在工程上做了简化。我的方案是贪心 局部优化。先把候选景点按地理距离聚类每天安排一个区域内的景点然后用贪心算法在区域内排序最后做一次 2-opt 局部优化来减少总移动距离。import math def haversine(lat1, lon1, lat2, lon2): R 6371 dlat math.radians(lat2 - lat1) dlon math.radians(lon2 - lon1) a math.sin(dlat/2)**2 math.cos(math.radians(lat1)) * \ math.cos(math.radians(lat2)) * math.sin(dlon/2)**2 return R * 2 * math.asin(math.sqrt(a)) def plan_route(attractions: list, days: int) - list: # 按地理距离聚类分成 days 个簇 # 每个簇内用贪心排序 # 返回每日行程列表 ...路线规划 Agent 还有一个重要职责是可行性检查。它会计算每天的游玩总时长景点游玩时间 交通时间如果超过 10 小时就标记为不可行触发重新规划。交通时间用直线距离乘以一个系数估算城市内一般取 1.5跨区取 2.0。2.4 文案生成 Agent让行程单有人情味前三个 Agent 的输出都是结构化的数据文案生成 Agent 负责把它们变成用户能直接看的行程单。这个环节我特意把temperature调到了 0.7让输出更有变化和温度。Prompt 里我会给一个示例行程单的格式包括每日标题、时间轴、每个景点的推荐理由和实用贴士。推荐理由不是简单复述景点简介而是结合用户偏好来写。比如用户喜欢美食推荐宽窄巷子时就会侧重写周边的小吃。3. LangGraph 图编排的完整实现与可视化大屏对接3.1 State 设计与条件边配置LangGraph 的核心概念是 StateGraph。我先定义 State 的结构from typing import TypedDict, List, Optional class TravelState(TypedDict): user_input: str parsed_params: Optional[dict] candidate_attractions: Optional[List[dict]] planned_route: Optional[List[dict]] final_itinerary: Optional[str] retry_count: int is_feasible: bool error_message: Optional[str]然后定义节点和边from langgraph.graph import StateGraph, END workflow StateGraph(TravelState) workflow.add_node(parse, parse_node) workflow.add_node(recommend, recommend_node) workflow.add_node(plan, plan_node) workflow.add_node(generate, generate_node) workflow.set_entry_point(parse) workflow.add_edge(parse, recommend) workflow.add_edge(recommend, plan) workflow.add_conditional_edges( plan, lambda state: generate if state[is_feasible] else recommend, {generate: generate, recommend: recommend} ) workflow.add_edge(generate, END)条件边的判断函数是关键。如果is_feasible为 True 或者retry_count超过 3就进入文案生成否则回到景点推荐重新筛选。这里要注意防止无限循环所以retry_count每次回到 recommend 节点时都要加一。3.2 可视化大屏的数据对接方案可视化大屏用的是 ECharts 一个开源的大屏模板。数据对接分两种方式实时推送和定时轮询。实时推送用 WebSocket当 LangGraph 的某个节点完成时后端主动推一条消息给前端前端更新对应的图表。比如需求解析完成后推送解析出的参数前端更新用户画像面板路线规划完成后推送每日行程数据前端更新地图上的路线和景点标记。定时轮询用于一些非关键数据的刷新比如系统运行状态、Agent 调用次数统计等每 30 秒拉一次。大屏上我放了几个核心模块左侧是用户需求解析结果和偏好雷达图中间是地图展示每日路线右侧是 Agent 协作流程图实时高亮当前执行的节点和行程详情列表。Agent 协作流程图那个模块效果特别好答辩的时候老师一眼就能看出多 Agent 的协作关系。实操心得WebSocket 推送的数据格式一定要和前端约定好我一开始用嵌套很深的 JSON前端解析经常出错。后来改成扁平结构每个消息带一个type字段标识数据类型前端根据 type 分发处理稳定多了。3.3 多 Agent 协作的日志与可观测性多 Agent 系统最头疼的问题之一是调试。四个 Agent 互相调用出了问题很难定位是哪个环节的锅。我的做法是在每个节点的入口和出口都打日志记录输入 State、输出 State 和耗时。import time import logging def with_logging(node_func, node_name): def wrapper(state): start time.time() logging.info(f[{node_name}] 输入: {state}) result node_func(state) elapsed time.time() - start logging.info(f[{node_name}] 输出: {result}, 耗时: {elapsed:.2f}s) return result return wrapper这些日志不仅用于调试还直接喂给了可视化大屏的Agent 运行状态面板。每个 Agent 的调用次数、平均耗时、成功率都实时展示答辩的时候这个面板是加分项。4. 常见问题排查与性能优化实录4.1 DeepSeek API 调用的稳定性处理DeepSeek 的 API 在高峰期偶尔会出现超时或返回空结果。我的处理策略是重试 降级。重试用指数退避第一次等 1 秒第二次等 2 秒第三次等 4 秒。如果三次都失败就降级到本地缓存的结果或者返回一个默认值。import time def call_with_retry(func, max_retries3): for i in range(max_retries): try: return func() except Exception as e: if i max_retries - 1: raise time.sleep(2 ** i)另外DeepSeek 的 API 有并发限制我在代码里加了一个简单的令牌桶限流器保证不会因为并发过高被限流。4.2 路线规划不可行时的降级策略实际测试中大概有 15% 的情况会出现三次重试后仍然不可行。这时候不能直接报错要给用户一个可用的结果。我的降级策略是放宽约束 明确告知。具体做法是如果预算约束导致景点太少就适当提高预算上限 10%-20%如果时间约束导致行程太赶就减少每天安排的景点数量。然后在最终行程单里加一段说明由于您的预算/时间限制以下行程做了适当调整供您参考。4.3 常见问题速查表问题现象可能原因排查方法解决方案JSON 解析失败模型输出带代码块标记打印原始输出加清洗函数去除标记路线规划死循环retry_count 未递增检查条件边逻辑确保每次回退时计数加一景点推荐为空硬过滤条件太严打印 SQL 查询结果放宽预算或标签匹配大屏数据不更新WebSocket 连接断开查看浏览器控制台加心跳重连机制API 调用超时网络波动或限流查看错误码重试 降级4.4 性能优化从 30 秒到 8 秒的响应时间最初的版本从用户输入到生成完整行程要 30 秒左右体验很差。我做了几项优化第一并行化。景点推荐和路线规划之间的协商循环是串行的但候选景点的详情获取可以并行。我用concurrent.futures把多个景点的详情查询并行执行节省了大概 5 秒。第二缓存。同一个城市的景点数据在短时间内不会变化我用 Redis 做了缓存第二次查询同一个城市直接从缓存读节省了 3-4 秒。第三Prompt 精简。最初的 Prompt 写得很长包含了大量示例和说明导致模型处理时间增加。我精简了 Prompt只保留必要的约束和一个示例模型响应时间从平均 6 秒降到了 3 秒左右。第四流式输出。文案生成环节改成流式输出用户可以看到文字逐字出现虽然总时间没变但感知上的等待时间短了很多。优化后整体响应时间稳定在 8-10 秒对于这种复杂度的规划任务来说已经可以接受了。4.5 几个容易忽略的细节时区问题如果用户输入的目的地和服务器时区不同开放时间的判断会出错。我在景点数据里统一存了当地时间计算时做了转换。特殊日期节假日景点开放时间会变我在数据库里加了一个special_dates字段存储特殊日期的开放时间调整。这个功能在五一、十一期间特别有用。多用户并发LangGraph 的 State 是每个请求独立的但数据库连接和 API 调用需要做并发控制。我用连接池管理数据库连接用信号量控制 API 并发数。错误信息的友好化技术错误直接抛给用户很不友好。我加了一层错误转换把API timeout转成系统繁忙请稍后重试把no attractions found转成没有找到符合条件的景点请尝试放宽预算或更换偏好标签。这个项目从立项到答辩大概花了六周时间其中 LangGraph 的图编排和 Agent 之间的协商逻辑占了将近一半的调试时间。但跑通之后的效果确实比单链方案好很多尤其是路线规划 Agent 和景点推荐 Agent 之间的反馈循环让最终方案的合理性提升了一个档次。如果让我重新做一遍我会在 State 设计上花更多时间把每个字段的读写权限提前规划清楚这样能省下不少调试功夫。