适合人群想入门 Agent 项目开发的新手有基础 Python 能力想做一个能跑起来、能演示、能扩展的 AI 项目希望同时理解为什么这么设计与怎么一步步复现一、为什么做这个项目最近很多同学在做 AI 应用时会遇到同一个困境Demo 很炫但无法稳定运行—— 换个输入就崩一报错就卡死只会调用一个模型—— 流程稍复杂就兜不住不知道怎么拆任务没有过程可视化—— 用户点一下等半天不知道系统在干嘛我这次做的是一个MVP 行程规划智能体目标不是全能而是三个关键词关键词含义可演示输入旅行需求就能稳定给出结构化结果可扩展后续加天气、地图、票务能力时不用推倒重来可观察前端能看到后端每一步在干什么SSE 实时流1.1 应用场景想象一下这个场景用户说我五一想带家人去杭州玩 3 天喜欢自然风景。一个好的行程规划系统应该追问缺失信息—— 是从哪出发要不要自驾有没有预算限制多路并行规划—— 路线规划、酒店推荐、预算估算同时跑实时反馈进度—— 用户看到正在规划路线而不是干等给出可执行方案—— 不是泛泛的杭州很好玩而是有每天的具体安排这听起来简单但要稳定做到需要一套工程化设计。1.2 MVP 思维这个项目叫 MVPMinimum Viable Product意思是最小可行产品。不是说功能少就low而是先做核心闭环用户输入 → 槽位收集 → 多 Agent 协作 → 结果聚合 → 用户拿到方案这个闭环跑通之后再加天气、加地图、加存储都是锦上添花而不是推倒重来。github地址:qq1065507891/ItineraryPlanningAgent二、项目架构全景先看一下整体架构避免一开始就陷入细节。2.1 系统分层┌─────────────────────────────────────────────────────┐ │ Frontend (Gradio) │ │ 对话 UI / SSE 事件渲染 / 结果展示 │ └──────────────────────┬──────────────────────────────┘ │ HTTP SSE ┌──────────────────────▼──────────────────────────────┐ │ Backend (FastAPI) │ │ API 网关 / 会话管理 / Supervisor 编排 / 多 Agent 协同│ └──────────────────────┬──────────────────────────────┘ │ 并行调用 ┌────────┬────────┼────────┬────────┐ ▼ ▼ ▼ ▼ ▼ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ Info │ │Dest. │ │Route │ │Hotel │ │Budget│ │Collec│ │Reser.│ │Planer│ │Agent │ │Agent │ └──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘ │ │ │ │ │ └────────┴────────┼────────┴────────┘ ▼ ┌──────────────┐ │ Aggregator │ │ 汇总输出 │ └──────────────┘2.2 技术栈层级技术选型说明前端Gradio轻量级 Python UI 框架快速搭建交互界面后端FastAPI高性能 async API 框架内置 SSE 支持Agent 编排Supervisor协调多 Agent 协作、超时控制、容错降级LLM 调用LangChain统一接入不同 ProviderOpenAI/Kimi/本地模型外部能力MCPModel Context Protocol高德路线/POI、酒店推荐会话存储内存存储可扩展至 Redis维护 slot 状态支持多轮对话2.3 项目结构行程规划智能体/ ├─ backend/ │ ├─ app/ │ │ ├─ api/ │ │ │ └─ routes_chat.py # SSE 流式接口 │ │ ├─ orchestration/ │ │ │ └─ supervisor.py # Agent 编排器 │ │ ├─ agents/ # 各业务 Agent │ │ │ ├─ info_collector_agent.py # 槽位收集 │ │ │ ├─ destination_research/ # 目的地研究 │ │ │ ├─ route_planner/ # 路线规划 │ │ │ ├─ hotel/ # 酒店推荐 │ │ │ ├─ budget/ # 预算估算 │ │ │ └─ aggregator/ # 结果聚合 │ │ ├─ llm/ │ │ │ ├─ factory.py # LLM 工厂 │ │ │ └─ config_map.py # 配置映射 │ │ ├─ mcp_connectors/ # MCP 连接器封装 │ │ ├─ storage/ │ │ │ └─ memory_store.py # 内存会话存储 │ │ └─ domain/ │ │ └─ models.py # 领域模型 │ ├─ main.py # FastAPI 入口 │ └─ requirements.txt ├─ frontend/ │ ├─ app.py # Gradio 入口与事件渲染 │ └─ requirements.txt ├─ .env.example # 环境变量模板 └─ README.md三、核心机制讲解这一部分是重点我会拆解三个核心机制槽位收集、多 Agent 协作、SSE 实时流。3.1 槽位收集Slot Filling什么是槽位槽位就是规划行程所需的原材料。在 MVP 版本中定义了 6 个必填槽位槽位含义示例destination目的地杭州、成都、三亚days天数3、5、7departure_place出发地上海、南京departure_time出发时间2026-05-01 08:00interests兴趣偏好[“自然风光”, “美食”]self_drive是否自驾true / false为什么要先收集槽位想象你要帮朋友规划行程但你连他要去哪、玩几天都不知道你只能追问。槽位收集就是这个追问机制# 伪代码示例if槽位不完整:返回追问问题给用户 暂停后续规划else:启动多 Agent 并行规划InfoCollectorAgent 的实现这个 Agent 做了三件事LLM 抽取—— 用 LLM 从用户输入中提取结构化槽位正则兜底—— LLM 不可信时用正则表达式补充动态事实收集—— 除了必填槽位还收集额外信息预算、人数、酒店偏好等# info_collector_agent.py 核心逻辑defvalidate_and_fill_slots(self,message:str,existing_slots:dict):# 1. LLM 抽取llm_resultself._extract_with_llm(message)slots.update(llm_result)# 2. 正则兜底防止 LLM 幻觉fallback_slotsself._extract_with_regex(message)forkey,valueinfallback_slots.items():ifslots.get(key)isNone:slots[key]value# 3. 检查缺失槽位missing[kforkinREQUIRED_SLOT_PRIORITYifslots.get(k.value)in(None,,[])]returnslots,missing追问策略当槽位缺失时系统不是一次列出所有缺失项而是按优先级逐个追问# 缺失槽位优先级REQUIRED_SLOT_PRIORITY[DESTINATION,# 优先级1目的地DAYS,# 优先级2天数DEPARTURE_PLACE,# 优先级3出发地DEPARTURE_TIME,# 优先级4出发时间INTERESTS,# 优先级5兴趣SELF_DRIVE,# 优先级6是否自驾]这样做的好处是用户体验友好一次只问一件事。3.2 多 Agent 协作为什么需要多个 Agent行程规划本质上是一个复合任务拆成多个单一职责的 Agent 更稳定Agent职责特点InfoCollectorAgent槽位收集与追问优先执行缺失时阻断DestinationResearchAgent目的地网络信息研究可开关Feature FlagRoutePlannerAgent每日路线规划 跨城交通核心输出HotelAgent酒店推荐依赖目的地BudgetAgent预算估算依赖天数和人数AggregatorAgent结果汇总与结构化最后执行Supervisor 编排器Supervisor 是整个系统的导演负责顺序执行 InfoCollector—— 必须先拿到完整槽位并行执行规划 Agent—— 路线、酒店、预算同时跑超时控制与容错降级—— 某个 Agent 挂了不影响全链路汇总输出—— Aggregator 整合所有结果# supervisor.py 核心编排逻辑asyncdefhandle_turn(self,session_id,message,slots):# Step 1: 槽位收集顺序updated_slots,missing_slotsself.info_collector.validate_and_fill_slots(message,slots)ifmissing_slots:return{need_followup:True,...}# 追问模式# Step 2: 并行执行规划 Agentasyncio.gatherroute_pack,hotel_pack,budget_packawaitasyncio.gather(run_route(),run_hotel(),run_budget())# Step 3: 汇总聚合final_planawaitself.aggregator.run(slotsupdated_slots,route_resultroute_result,hotel_resulthotel_result,budget_resultbudget_result,)return{need_followup:False,final_plan:final_plan,...}容错降级机制每个 Agent 都有 40 秒超时超时或异常时自动降级asyncdef_run_agent_with_guard(self,agent_name,coro,fallback):try:resultawaitasyncio.wait_for(coro,timeout40)returnresult,Noneexceptasyncio.TimeoutError:# 超时降级返回兜底数据 记录 warningreturnfallback,{agent:agent_name,code:AGENT_TIMEOUT,message:...}这样设计的好处是单个 Agent 挂了不影响全局用户仍然能拿到部分结果。Agent 独立 LLM 配置项目支持每个 Agent 独立配置 LLM# .env 示例INFO_COLLECTOR_API_KEYsk-xxxINFO_COLLECTOR_MODEL_NAMEgpt-4o-miniROUTE_PLANNER_API_KEYsk-yyyROUTE_PLANNER_MODEL_NAMEgpt-4oHOTEL_API_KEYsk-zzzHOTEL_MODEL_NAMEgpt-4o-mini这是很实用的工程思维简单任务用便宜模型如槽位抽取复杂任务用强推理模型如路线规划成本与效果平衡3.3 SSE 实时流Server-Sent Events为什么用 SSE 而不是轮询传统方案是前端轮询处理完了没这有以下问题延迟高轮询间隔不可能太短资源浪费大量无效请求体验差用户不知道进度SSE 是后端推送机制前端建立一次连接后端持续推送事件前端 后端 │ │ │──── GET /chat/stream ──────►│ │ │ │◄─── session_started ───────│ │◄─── agent_progress 25% ────│ │◄─── agent_progress 50% ────│ │◄─── agent_result ──────────│ │◄─── agent_result ──────────│ │◄─── final_plan ───────────│ │◄─── session_finished ──────│SSE 事件类型事件类型含义session_started会话启动agent_progressAgent 进度0-100%agent_result单个 Agent 结果followup_question追问问题warning子任务降级警告deepsearch_result深度搜索结果final_plan最终方案session_finished会话结束FastAPI SSE 实现# routes_chat.py SSE 核心代码router.post(/stream)asyncdefchat_stream(req:ChatRequest,request:Request):asyncdefevent_generator():queueasyncio.Queue()asyncdefemit(event,data):payloadbuild_payload(event,req.session_id,data)awaitqueue.put(to_sse(event,payload))# Worker 异步执行主逻辑通过 emit 推送事件asyncio.create_task(worker())# 前端断开时自动停止whilenotawaitrequest.is_disconnected():itemawaitqueue.get()ifitemisNone:breakyielditemreturnStreamingResponse(event_generator(),media_typetext/event-stream)前端 Gradio 渲染# frontend/app.py 事件处理forevent_name,event_payloadinparse_sse_lines(resp.iter_lines()):ifevent_nameagent_progress:# 更新进度显示statusf{event_payload[agent]}:{event_payload[message]}ifevent_namefinal_plan:# 渲染最终方案final_plan_mdrender_final_plan_markdown(event_payload)四、手把手复现这一节让你从零把项目跑起来。4.1 环境准备Python 版本3.10克隆项目gitclonerepo-urlcd行程规划智能体4.2 安装依赖# 后端依赖cdbackend pipinstall-rrequirements.txt# 前端依赖cd../frontend pipinstall-rrequirements.txt后端 requirements.txtfastapi0.115.0 uvicorn[standard]0.30.0 pydantic2.8.0 httpx0.27.0 langchain0.3.0 langchain-openai0.2.0 python-dotenv1.0.1 beautifulsoup44.12.3前端 requirements.txtgradio4.0.0 requests2.31.04.3 配置环境变量cp.env.example .env编辑.env文件配置你的 API Key# 全局配置 APP_ENVdevLOG_LEVELINFO# MCP 服务可选AMAP_MCP_URLhttp://localhost:8080/mcpHOTEL_MCP_URLhttp://localhost:8081/mcp# Feature Flags ENABLE_DESTINATION_RESEARCHtrue# Agent LLM 配置 INFO_COLLECTOR_API_KEYsk-your-keyINFO_COLLECTOR_MODEL_NAMEgpt-4o-miniINFO_COLLECTOR_URLhttps://api.openai.com/v1DESTINATION_RESEARCH_API_KEYsk-your-keyDESTINATION_RESEARCH_MODEL_NAMEgpt-4o-miniDESTINATION_RESEARCH_URLhttps://api.openai.com/v1ROUTE_PLANNER_API_KEYsk-your-keyROUTE_PLANNER_MODEL_NAMEgpt-4o-miniROUTE_PLANNER_URLhttps://api.openai.com/v1HOTEL_API_KEYsk-your-keyHOTEL_MODEL_NAMEgpt-4o-miniHOTEL_URLhttps://api.openai.com/v1BUDGET_API_KEYsk-your-keyBUDGET_MODEL_NAMEgpt-4o-miniBUDGET_URLhttps://api.openai.com/v1AGGREGATOR_API_KEYsk-your-keyAGGREGATOR_MODEL_NAMEgpt-4o-miniAGGREGATOR_URLhttps://api.openai.com/v1提示MCP 服务如果暂时没有可以先留空或使用本地 mock。系统有容错降级不会阻断主流程。4.4 启动服务终端 1 - 启动后端cdbackend uvicorn main:app--reload--host127.0.0.1--port8000终端 2 - 启动前端cdfrontend python app.py4.5 联调验证启动后访问服务地址前端 Gradio UIhttp://127.0.0.1:7860后端 APIhttp://127.0.0.1:8000健康检查http://127.0.0.1:8000/healthz健康检查返回{status:ok}即表示后端正常。五、一次完整运行示例这一节展示一次完整的多轮对话流程。5.1 第一轮信息收集用户输入我五一想去杭州玩三天喜欢自然风景系统响应[agent_progress] supervisor: 正在进行槽位抽取与完整性校验 (25%) [agent_result] destination_research: statusunknown, enabledFalse, sourceduckduckgo, articles0 [agent_result] route_planner: days0, intercity_statusunknown, intercity_availableFalse [followup_question] 你从哪里出发例如上海、南京、北京。系统发现缺少出发地信息触发追问。5.2 第二轮补充信息用户输入从上海出发系统响应[agent_progress] supervisor: 槽位完整启动并行子任务 (45%) [agent_progress] destination_research: 开始检索目的地网页信息 (50%) [agent_progress] route_planner: 开始规划每日路线与跨城交通 (60%) [agent_progress] hotel: 开始检索酒店 (55%) [agent_progress] budget: 开始估算预算 (55%) [agent_progress] destination_research: 目的地网页信息检索完成 (58%) [agent_progress] hotel: 酒店推荐完成 (80%) [agent_progress] route_planner: 路线规划完成 (82%) [agent_progress] budget: 预算估算完成 (80%) [agent_progress] aggregator: 最终结果已生成 (98%) [final_plan] 已生成最终行程方案 [session_finished] 会话完成5.3 输出结果示例最终方案包含出发地到目的地交通建议行程上海 → 杭州出发时间2026-05-01 08:00推荐结论高铁约45分钟二等座票价¥73-100每日路线Day 1: 西湖深度游 - 主题自然风光 - 1. 苏堤散步停留约2小时 - 2. 花港观鱼赏景停留约1.5小时 - 3. 雷峰塔登塔停留约1.5小时 - 晚间建议河坊街步行街 Day 2: 宋城一日游 - 主题文化体验 - 1. 宋城主题公园 - 2. 《宋城千古情》演出 Day 3: 灵隐寺 龙井茶园 - 主题人文 自然 - 1. 灵隐寺 - 2. 龙井村品茶酒店推荐Top 3杭州西湖希尔顿酒店西湖区约¥680/晚- 交通便利杭州君悦酒店湖滨路约¥850/晚- 近西湖桔子水晶杭州西湖店约¥350/晚- 性价比高预算摘要总预算区间¥2500-3500/人交通¥200-300住宿¥1200-2000餐饮¥600-800门票¥200-300六、常见报错与排查清单6.1 前端提示后端调用失败排查步骤检查后端是否已在127.0.0.1:8000启动检查frontend/app.py中的BACKEND_STREAM_URL是否正确查看后端日志是否有RequestValidationError# 测试后端是否可达curlhttp://127.0.0.1:8000/healthz6.2 一直处于追问阶段无法出最终方案排查步骤检查是否已补全所有必填槽位查看后端日志中的missing_slots字段确认用户输入包含完整的 6 个槽位信息6.3 路线/酒店结果为空或质量低排查步骤检查 MCP URL 是否可用检查对应 Agent 的 LLM 配置是否正确查看后端日志中的warning/error/trace_id6.4 SSE 中断排查步骤检查网络与代理配置重试同一session_id继续对话增加 SSE 心跳配置SSE_HEARTBEAT_SECONDS36.5 LLM API 调用失败排查步骤确认 API Key 正确且有余额检查 API URL 是否可访问如有代理需求查看.env中的*_URL配置6.6 日志分析后端日志包含session_id和trace_id便于定位问题# 健康会话日志 {session_id: s_abc123, trace_id: xyz789, agent_name: api_gateway, status: success} # 异常日志 {session_id: s_abc123, trace_id: xyz789, agent_name: route_planner, status: error, error: ...}七、如何二次开发MVP 跑通后你可以扩展以下能力。7.1 加天气能力方案接入天气 API如和风天气在路线规划时考虑天气因素步骤申请天气 API Key创建mcp_connectors/weather/connector.py在RoutePlannerAgent中调用天气 API根据天气调整景点顺序或交通方式# 示例天气降级逻辑asyncdefget_weather_recommendation(destination,date):weatherawaitweather_mcp.get(destination,date)ifweather.get(rain_probability)0.7:return建议准备雨具或调整室外景点顺序returnNone7.2 加地图可视化方案集成高德/百度地图 API生成路线图步骤申请地图 JS API Key在前端final_plan渲染时调用地图组件使用 Polyline 绑点绘制路线# 后端输出 POI 坐标{daily_plan:[{day:1,pois:[{name:苏堤,lat:30.242,lng:120.142},{name:花港观鱼,lat:30.229,lng:120.136},]}]}7.3 加会话存储Redis当前问题会话存在内存中服务重启会丢失方案迁移至 Redis步骤安装 Redis修改backend/app/storage/memory_store.py实现 Redis 客户端连接# 示例Redis 存储importredisclassRedisSessionStore:def__init__(self):self.clientredis.Redis(hostlocalhost,port6379)defsave(self,session_id,state):self.client.setex(fsession:{session_id},3600,json.dumps(state))defget(self,session_id):dataself.client.get(fsession:{session_id})returnjson.loads(data)ifdataelseNone7.4 加节假日动态重排方案检测用户行程是否涉及节假日自动调整预算和景点开放时间步骤接入节假日 API在BudgetAgent中增加节假日溢价系数在RoutePlannerAgent中提醒景点限流八、总结这个项目麻雀虽小五脏俱全槽位收集体现了输入质量优先的设计思维多 Agent 协作展示了如何拆解复杂任务SSE 实时流解决了 AI 应用无反馈的体验问题容错降级保证了系统的稳定性你可以把它当作 Agent 开发的练手项目也可以基于它扩展出更复杂的应用。关键是要理解为什么这么设计而不是单纯复制代码。祝复现顺利