东予薏米1. 项目概述：行程规划智能体的核心价值
作为一名长期从事AI应用开发的工程师，我一直在寻找能够平衡技术深度与实用性的项目案例。这个行程规划智能体项目完美契合了这种需求——它不仅展示了多Agent系统的工程化实现，更提供了一套可复用的开发范式。
1.1 为什么选择行程规划作为切入点？
行程规划是个看似简单实则复杂的领域问题。当用户说"我想去杭州玩三天"时，背后涉及：
- 信息收集（出发地、时间、偏好等）
- 多维度决策（路线、住宿、预算）
- 实时反馈机制
- 结果整合与呈现
传统单体架构很难优雅处理这种多线程任务。而通过Agent化拆分，我们可以：
- 让专业Agent处理专业事务（如酒店推荐Agent专注住宿筛选）
- 实现任务并行执行（路线规划和预算估算同时进行）
- 灵活扩展新能力（后续添加天气、地图等功能不影响现有逻辑）
1.2 技术选型的深层考量
项目采用的技术栈都经过精心挑选：
- FastAPI：异步特性完美支持SSE流式传输
- LangChain：统一不同LLM提供商的接口差异
- Gradio：快速构建可交互演示界面
- MCP协议：抽象外部服务调用，保持核心逻辑纯净
特别值得一提的是SSE（Server-Sent Events）的选择。相比WebSocket，SSE：
- 更轻量级（基于HTTP协议）
- 天然支持断线重连
- 后端实现更简单
- 完全满足单向通知场景需求
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 系统架构深度解析
2.1 分层架构设计
系统采用经典的三层架构，但每层都有独特设计：
code复制前端展示层 (Gradio)
│
├─ 实时事件渲染
├─ 多轮对话界面
└─ 结果可视化
│
API网关层 (FastAPI)
│
├─ 会话管理
├─ 请求路由
└─ SSE事件分发
│
业务逻辑层 (Agent集群)
│
├─ 信息收集Agent
├─ 路线规划Agent
├─ 酒店推荐Agent
├─ 预算估算Agent
└─ 结果聚合Agent
这种架构的关键优势在于：
- 职责分离：每层只关注自己的核心功能
- 弹性扩展：新增Agent不影响现有流程
- 故障隔离：单个Agent故障不会导致系统崩溃
2.2 Agent通信机制
Agent之间通过两种方式交互：
- 直接调用：Supervisor显式调用各Agent
- 共享状态：通过Session Store传递数据
为避免循环依赖，我们制定了严格的调用规则：
- 上游Agent不能直接调用下游Agent
- 跨Agent通信必须通过Supervisor
- 所有状态变更必须经由Session Store
2.3 容错设计三原则
系统遵循以下容错原则：
- 超时控制：每个Agent设置40秒超时
- 优雅降级：关键路径必须有fallback方案
- 故障隔离：单个Agent失败不影响整体流程
例如酒店推荐Agent的实现：
python复制async def recommend_hotels(destination):
try:
# 尝试主逻辑
result = await primary_hotel_api(destination)
if not result:
# 主逻辑无结果时尝试备用方案
result = await fallback_hotel_search(destination)
return result
except Exception as e:
# 记录错误但返回空结果让流程继续
logger.error(f"Hotel error: {str(e)}")
return []
3. 核心实现细节
3.1 槽位收集的工程实践
槽位收集看似简单，实则暗藏玄机。我们的实现包含多个优化点：
多策略抽取：
- LLM主抽取：用prompt工程指导信息提取
- 正则兜底：针对常见模式（如日期）编写正则
- 上下文补全：利用对话历史推断缺失信息
优先级队列设计：
python复制SLOT_PRIORITY = [
Slot(name="destination", prompt="您想去哪个城市？"),
Slot(name="days", prompt="计划游玩几天？"),
Slot(name="departure", prompt="从哪里出发？"),
# ...其他槽位
]
这种设计确保：
- 每次只追问最高优先级的缺失槽位
- 用户不会被同时询问多个问题
- 系统可以灵活调整追问顺序
3.2 多Agent协作的实现
Supervisor是系统的指挥中心，其核心逻辑如下：
python复制class Supervisor:
async def execute_plan(self, session):
# 阶段1：信息收集
if not self._validate_slots(session):
return await self.request_missing_slots(session)
# 阶段2：并行执行
tasks = [
self.route_planner.execute(session),
self.hotel_agent.execute(session),
self.budget_agent.execute(session)
]
route, hotels, budget = await asyncio.gather(*tasks)
# 阶段3：结果聚合
return await self.aggregator.merge_results(
route=route,
hotels=hotels,
budget=budget
)
关键设计点：
- 使用asyncio.gather实现真正并行
- 每个Agent返回标准化结构体
- 聚合器处理数据格式转换
3.3 SSE实时流的工程细节
SSE实现中最容易忽略的是连接管理。我们的解决方案：
-
心跳机制：每30秒发送注释行保持连接
python复制async def heartbeat(): while True: await asyncio.sleep(30) yield ":keepalive\n\n" -
连接池管理：使用WeakValueDictionary跟踪活跃连接
python复制class ConnectionManager: def __init__(self): self._connections = WeakValueDictionary() -
事件缓冲：防止快速连续事件导致客户端丢失消息
python复制async def event_buffer(queue): last_event = None while True: event = await queue.get() if event != last_event: yield event last_event = event
4. 部署与调优指南
4.1 性能优化实践
经过实测，以下几个优化效果显著：
-
LLM调用批处理：
python复制# 优化前：串行调用 for item in items: response = await llm.call(item) # 优化后：批量调用 batch_responses = await llm.batch_call(items) -
缓存热点数据：
python复制@lru_cache(maxsize=100) def get_city_landmarks(city): return query_landmarks(city) -
连接池复用：
python复制async with httpx.AsyncClient(timeout=60) as client: await client.get(url)
4.2 监控指标设计
完善的监控是稳定运行的保障。我们收集以下指标：
| 指标名称 | 类型 | 说明 |
|---|---|---|
| slot_fill_time | 耗时 | 槽位收集阶段耗时 |
| agent_exec_time | 耗时 | 各Agent执行时间 |
| sse_clients | 计数 | 当前活跃SSE连接数 |
| fallback_used | 计数 | 降级逻辑触发次数 |
通过Prometheus+Grafana实现可视化监控。
5. 典型问题排查手册
5.1 SSE连接不稳定
现象：前端频繁断开连接
排查步骤：
- 检查网络延迟：
ping backend_host - 验证防火墙设置：
telnet backend_host 8000 - 调整心跳间隔：
SSE_HEARTBEAT=15
5.2 Agent执行超时
现象：日志中出现TimeoutError
解决方案：
- 优化慢查询：
python复制# 在数据库查询中添加超时 await database.execute(timeout=10, query=...) - 增加重试逻辑：
python复制@retry(times=3, delay=1) async def call_external_api(): ...
5.3 内存泄漏排查
工具：
tracemalloc：跟踪内存分配objgraph：分析对象引用
常见原因：
- 未关闭的数据库连接
- 全局变量累积数据
- 循环引用
6. 扩展开发指南
6.1 添加天气模块
完整实现步骤：
-
创建天气连接器：
python复制class WeatherConnector: async def get_forecast(self, location, date): params = {"key": API_KEY, "location": location, "date": date} async with httpx.AsyncClient() as client: resp = await client.get(WEATHER_URL, params=params) return resp.json() -
修改路线规划Agent：
python复制async def plan_route(session): weather = await weather_connector.get_forecast( session.destination, session.departure_date ) if weather["will_rain"]: return adjust_for_rainy_day(session) return default_plan(session)
6.2 集成地图服务
高德地图集成示例：
-
前端添加地图组件：
javascript复制const map = new AMap.Map('map-container', { zoom: 13, center: [116.397428, 39.90923] }); -
后端返回POI坐标：
python复制{ "attractions": [ { "name": "西湖", "location": { "lng": 120.15507, "lat": 30.274085 } } ] }
7. 项目演进路线
7.1 短期优化
-
对话记忆：实现多轮对话上下文保持
python复制class DialogMemory: def __init__(self, max_turns=5): self.history = deque(maxlen=max_turns) -
个性化推荐：基于用户历史偏好优化推荐
7.2 长期规划
- 多模态输出：支持语音、图片等富媒体结果
- 离线模式：集成本地小模型减少API依赖
- 自动优化：通过用户反馈持续改进推荐质量
这个项目的魅力在于它像乐高积木一样，可以不断添加新模块而不会破坏原有结构。我在实际开发中最深的体会是：好的架构设计应该像城市道路规划，既要满足当前需求，又要为未来发展留出空间。