12 · 实战手敲二:多智能体旅行规划系统
手敲系列第二篇(第一篇是第 11 篇的 Mini Agent)。这次你要亲手实现项目一(智能旅行助手)的同架构自研版:多个专家 Agent 并行工作、规划 Agent 汇总产出结构化 JSON 行程、Pydantic 校验 + 自修复循环、网页展示行程卡片。
全部 8 个文件、约 420 行,每一行都在下面给出。做完后再回头精读官方项目一(第 05 篇),你会发现是完全相同的骨架。
预计 2~3 天。核心新技能:让 LLM 稳定输出程序能用的结构化数据——这是 Agent 开发中最值钱、也最容易翻车的技能。
0. 成品长什么样
浏览器打开 http://127.0.0.1:8000,填"城市=北京、天数=2、偏好=喜欢历史古迹",点生成:
- 后端并行派出景点专家、美食专家(LLM 角色)和天气查询(直接调 API,不浪费 token)
- 规划专家把三份材料整合成 JSON 行程,Pydantic 逐字段校验,不合法就把报错喂回去让模型自己改
- 页面渲染出逐天的行程卡片:时间线活动、天气、总预算、小贴士
1. 架构设计(先想清楚再动手)
浏览器表单 ──POST /api/trip/plan──▶ main.py (FastAPI)
│
▼ make_plan()
┌── 并行(ThreadPoolExecutor) ──┐
▼ ▼ ▼
景点专家Agent 美食专家Agent get_weather()
(LLM角色) (LLM角色) (直接调wttr.in)
└─────────────┴────────────────┘
▼
规划专家Agent (LLM)
输出JSON → Pydantic校验 → 失败回喂重试
▼
TripPlan对象 → 前端渲染
三个设计决策,面试常被问到:
- 为什么天气不用 LLM? 代码一行能查到的确定性数据,交给 LLM 只会更慢、更贵、还可能编造。原则:LLM 只做需要"理解和生成"的事。
- 为什么并行? 三个信息源互不依赖,串行 30 秒的活并行只要 12 秒(第 10 篇第 5 节的延迟优化)。线程池(ThreadPoolExecutor)= 预先雇好几名"工人线程",
submit把任务丢进去就有人认领,result()等着收货;线程是什么见第 06 篇的名词补课。 - 为什么要 Pydantic 校验 + 自修复? LLM 输出 JSON 的"最后一公里"是 Agent 系统最常见的崩溃点:多个逗号、字段名拼错、数字带引号……校验失败时把具体报错文本喂回去重试,成功率能从 ~85% 拉到 ~99%。
| 文件 | 职责 |
|---|---|
requirements.txt / .env |
依赖与密钥 |
llm.py |
chat() + 新增 chat_json()(带解析重试) |
schemas.py |
Pydantic 行程数据模型 |
agents.py |
三个专家 + 并行汇集 + 自修复规划循环 |
main.py |
FastAPI 接口 |
static/index.html |
表单 + 行程卡片渲染 |
2. 建项目(10 分钟)
PyCharm New Project → trip-planner,虚拟环境同第 11 篇。
requirements.txt:
openai>=1.40.0
fastapi>=0.115.0
uvicorn[standard]>=0.32.0
python-dotenv>=1.0.0
httpx>=0.27.0
pydantic>=2.8.0
pip install -r requirements.txt,.env 与第 11 篇相同(三件套)。新建 static 目录。
3. llm.py —— 新增"要 JSON 就必须给 JSON"的能力(1 小时)
比第 11 篇多了 _extract_json 和 chat_json,这两个函数值得背下来:
"""LLM调用封装: chat=自由文本, chat_json=强制结构化输出(带解析重试)"""
import json
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
_client = OpenAI(
api_key=os.getenv("LLM_API_KEY"),
base_url=os.getenv("LLM_BASE_URL"),
)
MODEL = os.getenv("LLM_MODEL_ID")
def chat(messages: list[dict], temperature: float = 0.3) -> str:
resp = _client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=temperature,
)
return resp.choices[0].message.content or ""
def _extract_json(text: str) -> str:
"""模型经常用```json包裹,或前后夹带解释文字 → 抠出第一个{到最后一个}"""
start, end = text.find("{"), text.rfind("}")
if start == -1 or end == -1:
raise ValueError(f"回复里没有JSON: {text[:120]}")
return text[start:end + 1]
def chat_json(messages: list[dict], temperature: float = 0.2, max_retries: int = 2) -> dict:
"""要求模型输出JSON并解析成dict;解析失败就把报错喂回去让它自己改"""
msgs = list(messages) # 复制一份,不污染调用方的列表
for attempt in range(max_retries + 1):
text = chat(msgs, temperature)
try:
return json.loads(_extract_json(text))
except (ValueError, json.JSONDecodeError) as e:
if attempt == max_retries:
raise
msgs.append({"role": "assistant", "content": text})
msgs.append({"role": "user", "content": f"上面的输出不是合法JSON({e}),请只输出修正后的JSON,不要任何解释文字"})
if __name__ == "__main__":
print(chat_json([{"role": "user", "content": '返回JSON: {"city": "北京", "score": 95}'}]))
✅ Checkpoint 1:Run,打印出一个真正的 Python 字典(注意不是字符串)。
4. schemas.py —— 用类型"锁死"行程的形状(30 分钟)
第 04 篇讲过 Pydantic 校验请求体,这里用它校验 LLM 的输出——同一个工具,第二种用法:
"""行程数据模型: LLM输出的JSON必须能通过这些校验才算合格"""
from pydantic import BaseModel, Field
class Activity(BaseModel):
time: str # "09:00"
name: str
description: str = ""
cost: float = 0 # 模型给字符串"60"时Pydantic会自动转成60.0
class DayPlan(BaseModel):
day: int = Field(ge=1) # ge=1: 天数必须≥1,模型给0或-1直接校验失败
theme: str = ""
activities: list[Activity]
class TripPlan(BaseModel):
city: str
days: int = Field(ge=1, le=30)
weather: str = ""
daily_plans: list[DayPlan]
total_budget: float = 0
tips: list[str] = []
if __name__ == "__main__":
ok = TripPlan.model_validate({
"city": "北京", "days": 1,
"daily_plans": [{"day": 1, "activities": [{"time": "09:00", "name": "故宫", "cost": "60"}]}],
})
print("校验通过, cost自动转成了:", type(ok.daily_plans[0].activities[0].cost))
try:
TripPlan.model_validate({"city": "北京", "days": 0, "daily_plans": []})
except Exception as e:
print("非法数据被拦住了:", str(e)[:80])
✅ Checkpoint 2:Run,第一条通过且 cost 变成 <class 'float'>,第二条(days=0)被拦截。
5. agents.py —— 多智能体协作核心(3 小时,本篇灵魂)
"""多智能体: 信息类专家并行工作,规划专家汇总产出结构化行程(带自修复)"""
import json
from concurrent.futures import ThreadPoolExecutor
import httpx
from pydantic import ValidationError
from llm import chat, chat_json
from schemas import TripPlan
def get_weather(city: str, days: int) -> str:
"""确定性数据直接调API,不劳驾LLM"""
try:
resp = httpx.get(f"https://wttr.in/{city}", params={"format": "j1"}, timeout=10)
data = resp.json()
lines = [
f"{d['date']}: {d['mintempC']}~{d['maxtempC']}°C"
for d in data["weather"][:days]
]
return "\n".join(lines)
except Exception as e:
return f"(天气获取失败: {e},请按当季常规天气安排)"
class RoleAgent:
"""通用角色Agent: 一个系统提示词 + 一个问题 = 一份专业答案"""
def __init__(self, name: str, system_prompt: str):
self.name = name
self.system_prompt = system_prompt
def ask(self, question: str) -> str:
return chat([
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": question},
])
attraction_agent = RoleAgent(
"景点专家",
"你是资深旅行景点专家。列出目的地最值得去的景点,每个给出:名称、一句话特色、建议游玩时长、大致门票价格。",
)
food_agent = RoleAgent(
"美食专家",
"你是本地美食专家。推荐目的地必吃美食和对应的餐厅类型,标注人均价格区间。",
)
PLANNER_PROMPT = """你是行程规划专家。根据用户需求和各位专家提供的材料,输出一份JSON格式的行程。
只输出JSON,不要输出任何其他文字。格式示例:
{
"city": "北京",
"days": 2,
"weather": "两天多云,15~22度,适合户外",
"daily_plans": [
{"day": 1, "theme": "当天主题", "activities": [
{"time": "09:00", "name": "活动名", "description": "一句话说明", "cost": 60}
]}
],
"total_budget": 800,
"tips": ["建议1", "建议2"]
}
要求: 行程松紧合理,含三餐,cost是人均预估花费(数字),total_budget是全程人均总预算。"""
def gather_info(city: str, days: int, preferences: str) -> dict:
"""三个信息源互不依赖 → 并行执行,总耗时≈最慢的那一个"""
with ThreadPoolExecutor(max_workers=3) as pool:
f_attr = pool.submit(attraction_agent.ask, f"目的地:{city},玩{days}天,用户偏好:{preferences}")
f_food = pool.submit(food_agent.ask, f"目的地:{city},用户偏好:{preferences}")
f_weather = pool.submit(get_weather, city, days)
return {
"attractions": f_attr.result(),
"food": f_food.result(),
"weather": f_weather.result(),
}
def make_plan(city: str, days: int, preferences: str) -> TripPlan:
info = gather_info(city, days, preferences)
user_msg = (
f"用户需求: 去{city}玩{days}天,偏好: {preferences}\n\n"
f"[景点专家的材料]\n{info['attractions']}\n\n"
f"[美食专家的材料]\n{info['food']}\n\n"
f"[未来天气]\n{info['weather']}"
)
messages = [
{"role": "system", "content": PLANNER_PROMPT},
{"role": "user", "content": user_msg},
]
# 自修复循环: 校验失败时把Pydantic的报错原文喂回去,让模型自己修
for _ in range(3):
data = chat_json(messages)
try:
return TripPlan.model_validate(data)
except ValidationError as e:
messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)})
messages.append({"role": "user", "content": f"JSON字段不符合要求,错误如下,请修正后重新输出完整JSON:\n{e}"})
raise RuntimeError("规划Agent连续3次未能产出合法行程")
if __name__ == "__main__":
plan = make_plan("北京", 1, "喜欢历史古迹")
print(f"{plan.city} {plan.days}天, 预算{plan.total_budget}元")
for d in plan.daily_plans:
print(f"Day{d.day} {d.theme}")
for a in d.activities:
print(f" {a.time} {a.name} ¥{a.cost}")
✅ Checkpoint 3:Run(要花 30~60 秒:三个并行调用 + 一次规划),打印出一份真实的一日行程。如果模型第一次 JSON 就合格,自修复循环一次就返回了——但它必须在:线上跑一千次,总会遇到不合格的那几次。
6. main.py —— 包一层 API(30 分钟)
"""FastAPI入口: 行程规划接口 + 静态页面"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from agents import make_plan
app = FastAPI(title="Trip Planner")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
class PlanRequest(BaseModel):
city: str
days: int = Field(default=2, ge=1, le=15)
preferences: str = "无特殊偏好"
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/api/trip/plan")
def plan_trip(req: PlanRequest):
"""统一包装结构: 前端只看success字段判断成败(官方项目一同款约定)"""
try:
plan = make_plan(req.city, req.days, req.preferences)
return {"success": True, "data": plan.model_dump()}
except Exception as e:
return {"success": False, "message": str(e)}
@app.get("/")
def index():
return FileResponse("static/index.html")
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
✅ Checkpoint 4:Run main.py,在 /docs 里调 POST /api/trip/plan,拿到 {"success": true, "data": {...}}。
7. static/index.html —— 行程卡片渲染(1.5 小时)
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="utf-8">
<title>旅行规划助手</title>
<style>
body { font-family: system-ui, sans-serif; max-width: 720px; margin: 32px auto; padding: 0 16px; background: #f6f7f9; }
form { display: flex; gap: 8px; flex-wrap: wrap; margin-bottom: 16px; }
input, button { padding: 8px 10px; border: 1px solid #ccc; border-radius: 6px; }
button { background: #0a58ca; color: #fff; border: none; cursor: pointer; }
.card { background: #fff; border-radius: 10px; padding: 14px 16px; margin: 10px 0; box-shadow: 0 1px 3px rgba(0,0,0,.08); }
.meta { color: #555; font-size: 14px; }
.act { display: flex; gap: 10px; padding: 6px 0; border-bottom: 1px dashed #eee; font-size: 14px; }
.act .t { color: #0a58ca; min-width: 52px; font-weight: 600; }
.act .c { margin-left: auto; color: #d63384; white-space: nowrap; }
#status { color: #888; margin: 8px 0; }
</style>
</head>
<body>
<h3>多智能体旅行规划</h3>
<form id="f">
<input id="city" placeholder="城市,如: 北京" required>
<input id="days" type="number" value="2" min="1" max="15" style="width:70px">
<input id="prefs" placeholder="偏好,如: 喜欢历史古迹" style="flex:1">
<button>生成行程</button>
</form>
<div id="status"></div>
<div id="result"></div>
<script>
const $ = (id) => document.getElementById(id);
$("f").onsubmit = async (e) => {
e.preventDefault();
$("result").innerHTML = "";
$("status").textContent = "多个Agent正在并行工作,约需30~60秒…";
try {
const resp = await fetch("/api/trip/plan", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
city: $("city").value,
days: Number($("days").value),
preferences: $("prefs").value || "无特殊偏好",
}),
});
const body = await resp.json();
if (!body.success) { $("status").textContent = "失败: " + body.message; return; }
$("status").textContent = "";
render(body.data);
} catch (err) {
$("status").textContent = "请求出错: " + err;
}
};
function render(plan) {
const r = $("result");
r.innerHTML = "";
const head = document.createElement("div");
head.className = "card";
head.innerHTML = `<b>${plan.city} · ${plan.days}天</b>
<div class="meta">天气: ${plan.weather || "—"}</div>
<div class="meta">人均总预算: ¥${plan.total_budget}</div>`;
r.appendChild(head);
for (const day of plan.daily_plans) {
const card = document.createElement("div");
card.className = "card";
card.innerHTML = `<b>Day ${day.day}</b> <span class="meta">${day.theme || ""}</span>`;
for (const a of day.activities) {
const row = document.createElement("div");
row.className = "act";
row.innerHTML = `<span class="t">${a.time}</span>
<span>${a.name}<div class="meta">${a.description || ""}</div></span>
<span class="c">¥${a.cost}</span>`;
card.appendChild(row);
}
r.appendChild(card);
}
if (plan.tips && plan.tips.length) {
const tips = document.createElement("div");
tips.className = "card";
tips.innerHTML = "<b>小贴士</b>" + plan.tips.map(t => `<div class="meta">· ${t}</div>`).join("");
r.appendChild(tips);
}
}
</script>
</body>
</html>
✅ 最终验收:
- [ ] 首页表单提交后,等待提示出现,30~60 秒后渲染出逐天行程卡片
- [ ] 换城市和偏好("亲子游""穷游")再生成,行程内容明显跟着偏好变化
- [ ] 把
.env里的 Key 改错一位再请求 → 页面显示失败: ...而不是整个崩掉(错误被{"success": false}兜住了) - [ ] 对照第 05 篇官方项目一:同样的"多专家并行 + 规划汇总 + success 包装",你现在能读懂它的每个设计
8. 进阶练习
- 加一个"预算控制 Agent":拿到行程后检查
total_budget是否超过用户给的上限,超了就把行程和上限一起回喂给规划专家重排(这就是第 03 篇讲的 Reflection 模式) - 把
gather_info改成串行,用秒表对比总耗时,把提速数字写进你的简历项目描述 - 给
/api/trip/plan加一个简单的结果缓存:同样的 (city, days, preferences) 十分钟内直接返回上次结果(第 10 篇第 5 节)
9. 面试怎么讲
"我实现了一个多智能体系统:信息类 Agent 用线程池并行,确定性数据(天气)直接走 API 不消耗 LLM;汇总层的核心难点是结构化输出的可靠性,我用 Pydantic 逐字段校验 + 把 ValidationError 原文回喂的自修复循环,把解析成功率做到接近 100%。"
下一篇:13 · 实战手敲三:深度研究智能体 | 返回目录:README