1. 为什么我要用智能体重构旅行规划这件事做旅行规划这件事我前前后后折腾了快三年。最早是纯手工Excel 拉表格、浏览器开二十个标签页比价、备忘录里记航班号一趟出行规划下来少说四五个小时遇上旺季票价波动前面查的全白费。后来用脚本爬数据能省点力气但脚本脆得很页面结构一变就崩维护成本比手工还高。再后来大模型起来了我第一反应就是拿它来干这个活但一开始也只是把它当个高级搜索框用问一句答一句体验很割裂。真正的转折点是我意识到旅行规划本质上不是一个问答任务而是一个多步骤、有状态、需要外部实时数据支撑的工作流。你得先理解用户模糊的需求我想找个暖和的地方躺几天再把它翻译成结构化的约束预算、天数、出发地、偏好然后去查实时航班和票价比对酒店最后组装成一份能直接执行的行程。这里面每一步的输入输出格式都不一样中间还会因为票价变化、余票不足而反复调整。用一个大 Prompt 硬塞给模型它要么漏步骤要么编数据要么在长上下文里把前面的约束忘了。所以我决定用AI 智能体Agent的思路来重构整个流程核心是两件事一是把规划逻辑拆成可组合的 Prompt 工程模块让每个环节职责单一、可测试二是用FastAPI搭一个实时票务查询服务让智能体能真正拿到此刻的航班和价格数据而不是靠模型记忆里的过期信息瞎编。这套东西跑通之后我规划一趟国内往返的行程从输入需求到拿到可执行方案稳定在 40 秒以内而且票价是真实的、可下单的。这篇文章我会把这套工作流从设计思路到代码落地完整拆一遍包括 Prompt 怎么分层、FastAPI 服务怎么设计接口、智能体怎么编排工具调用、以及我在实测中踩过的那些坑。适合已经会用大模型 API、想往智能体方向深入的后端或全栈开发者也适合做旅游类产品的同学参考架构。不需要你是算法专家但得能看懂 Python 和基本的 HTTP 接口。2. 整体架构设计与技术选型思路2.1 为什么是智能体 工具服务而不是一个大模型先说清楚一个概念上的分界。很多人把用大模型做应用和做智能体混为一谈其实差别很大。前者是你把问题整理好、喂给模型、拿回答案模型是被动的后者是模型自己决定我现在该干什么它会主动选择调用哪个工具、传什么参数、拿到结果后下一步做什么。旅行规划这个场景天然适合后者因为它的决策链条长且依赖外部状态。我举个具体的例子你就明白了。用户说下个月想去成都吃火锅三天预算三千。如果是一个大 Prompt模型会直接给你一份行程但里面的航班时间、票价全是它编的因为它没有实时数据。而智能体的做法是它先解析出出发地未知、目的地成都、时长三天、预算三千、主题美食发现出发地缺失于是主动追问拿到出发地后它调用票务查询工具查下个月的低价航班拿到真实结果后再排行程。这个发现缺失、主动追问、调用工具、整合结果的循环就是智能体的核心价值。我选这个架构的另一个原因是可维护性。旅行规划涉及航司、酒店、天气、签证、汇率等一堆外部依赖如果全塞进一个 Prompt任何一处数据源变化都要重写整个 Prompt。拆成智能体加工具服务之后每个工具是独立的 APIPrompt 只管决策逻辑数据源换了只改对应的服务互不影响。2.2 FastAPI 在这个架构里扮演什么角色FastAPI 是我给智能体准备的手和脚。智能体本身只会思考和决策它要拿到真实数据必须通过工具调用而工具的背后就是 FastAPI 提供的 HTTP 接口。我选 FastAPI 而不是 Flask 或 Django主要看中三点。第一是异步原生支持。票务查询往往要并发请求多个数据源比如同时查几家航司的接口FastAPI 基于 Starlette 的 async 能力让并发写起来很自然不用自己折腾线程池。第二是自动生成接口文档。智能体在决定调用哪个工具时需要知道工具的参数格式FastAPI 自动生成的 OpenAPI schema 可以直接喂给模型做 function calling 的定义省了我手写工具描述的工作。第三是Pydantic 的数据校验。票务查询的入参出参格式必须严格Pydantic 模型能在入口就把脏数据挡掉避免模型传了奇怪的参数导致下游报错。整个架构的数据流是这样的用户在前端输入需求请求打到智能体编排层编排层调用大模型做意图理解和任务分解模型决定调用哪个工具编排层通过 HTTP 请求 FastAPI 服务FastAPI 去查真实数据源并返回结构化结果编排层把结果回灌给模型模型继续下一步决策直到生成最终行程。2.3 工作流的分层设计我把整个工作流拆成了四层每层职责清晰层与层之间通过明确定义的数据结构通信。层级职责关键技术输出物意图理解层解析用户自然语言提取结构化约束Prompt 工程 JSON Schema结构化需求对象决策编排层决定调用哪些工具、处理缺失信息智能体循环 function calling工具调用序列工具服务层提供实时票务、酒店等数据FastAPI 异步请求结构化数据行程生成层整合所有数据生成可执行行程Prompt 工程 模板Markdown 行程单这么分层的好处是每一层都可以单独测试和替换。比如我想换一个更便宜的大模型做意图理解只改第一层的 Prompt 和模型配置其他层完全不动。我想加一个新的数据源比如高铁票只在工具服务层加一个接口决策编排层注册一下就行。3. Prompt 工程的分层拆解与实操要点3.1 意图理解 Prompt 怎么写才不跑偏意图理解是整个工作流的入口这一步错了后面全错。我一开始的写法是让模型提取用户需求中的所有信息结果它经常自作主张补全缺失字段比如用户没说出发地它默认成北京导致后面查出来的航班全是错的。后来我改成强制模型区分已知和未知明确要求缺失字段必须返回 null并且列出需要追问的问题。我的意图理解 Prompt 核心结构是这样的先给模型一个严格的 JSON Schema 定义规定输出必须包含哪些字段、每个字段的类型和取值范围然后给几个 few-shot 示例覆盖信息完整和信息缺失两种情况最后加一条硬约束禁止模型编造用户没提到的信息。实测下来加了 few-shot 之后字段提取的准确率从大概七成提到了九成以上。这里有个细节值得说日期处理是意图理解里最容易翻车的部分。用户说下个月、五一前后、这周末模型需要结合当前日期换算成具体日期。我的做法是在 Prompt 里注入当前日期并明确要求所有相对时间必须换算成 YYYY-MM-DD 格式无法确定的返回 null 并追问。这个改动之后日期相关的错误基本消失了。3.2 决策编排 Prompt 的设计逻辑决策编排层是智能体的大脑它要决定在当前状态下下一步该做什么。这里的 Prompt 设计核心是把可用工具和当前状态清晰地告诉模型让它做选择题而不是填空题。我的做法是把所有可用工具的函数签名、参数说明、返回格式整理成一段结构化文本放在 system prompt 里。然后在每轮对话中把已经收集到的信息、已经调用过的工具、工具返回的结果都作为上下文传进去。模型的任务就是判断信息够不够不够的话该调用哪个工具够了的话是不是可以生成行程了这里我踩过一个坑工具描述写得太模糊模型会乱调用。比如我一开始把票务查询工具描述成查询航班信息结果模型在用户还没确定日期的时候就调用了传了个空日期进去。后来我把描述改成查询指定出发地、目的地、出发日期的航班列表和价格缺少任一参数时不要调用模型的行为就规矩多了。工具描述本质上也是 Prompt 工程的一部分得当成产品文案来打磨。3.3 行程生成 Prompt 的模板化技巧行程生成是最后一层目标是把前面收集的所有数据组装成一份人类可读的行程单。这一层的 Prompt 相对简单但有个关键技巧用模板约束输出结构。我会在 Prompt 里给出一个 Markdown 模板规定行程单必须包含概览、每日安排、交通详情、预算明细、注意事项几个部分模型只需要往模板里填内容。这么做的好处是输出格式稳定前端可以直接解析渲染。如果不给模板模型每次生成的格式都不一样有时候用表格有时候用列表前端处理起来很痛苦。另外我要求模型在生成行程时必须引用工具返回的真实数据比如航班号、起降时间、价格都要来自票务查询的结果禁止自己编。为了强化这一点我会在 Prompt 里明确说以下数据来自实时查询请直接使用不要修改。3.4 Prompt 版本管理与测试Prompt 是要迭代的我强烈建议从一开始就做版本管理。我的做法是把每个 Prompt 存成独立的文本文件用 Git 管理每次修改都写清楚改了什么、为什么改。同时我建了一个小型的测试集包含二十来个典型的用户输入每次改完 Prompt 就跑一遍看输出是否符合预期。这个测试集帮我避免了好几次改好一个场景、弄坏三个场景的悲剧。比如我有次为了提升日期解析准确率在 Prompt 里加了一堆日期格式的说明结果模型开始把三天这种时长也当成日期处理。跑测试集的时候立刻发现了赶紧回滚调整。没有测试集的话这种回归问题可能要等线上用户反馈才发现。4. FastAPI 实时票务查询服务的落地实现4.1 项目目录结构怎么组织FastAPI 项目的目录结构我试过好几种最后稳定在这套组织方式上兼顾了清晰和可扩展。ticket-service/ ├── app/ │ ├── main.py # 应用入口注册路由 │ ├── config.py # 配置管理读环境变量 │ ├── models/ │ │ ├── request.py # 请求体 Pydantic 模型 │ │ └── response.py # 响应体 Pydantic 模型 │ ├── routers/ │ │ ├── flight.py # 航班查询路由 │ │ └── hotel.py # 酒店查询路由 │ ├── services/ │ │ ├── flight_service.py # 航班查询业务逻辑 │ │ └── cache.py # 缓存逻辑 │ └── clients/ │ └── data_source.py # 外部数据源客户端 ├── tests/ ├── requirements.txt └── .env这么分的原因很简单路由层只管 HTTP 协议相关的事业务逻辑放 service外部依赖放 client。这样我想换数据源只改 client 层想加缓存只改 service 层路由层几乎不用动。很多新手把所有逻辑堆在路由函数里项目一大就变成几百行的巨型函数改起来要命。4.2 票务查询接口的设计与参数校验票务查询接口是整个服务的核心我设计成 POST 而不是 GET因为查询参数比较多而且未来可能扩展成批量查询。请求体用 Pydantic 模型定义把校验规则写死在模型里。from pydantic import BaseModel, Field, field_validator from datetime import date class FlightQueryRequest(BaseModel): origin: str Field(..., min_length2, max_length3, description出发地城市代码) destination: str Field(..., min_length2, max_length3, description目的地城市代码) depart_date: date Field(..., description出发日期) return_date: date | None Field(None, description返程日期单程为空) adults: int Field(1, ge1, le9, description成人数量) max_price: float | None Field(None, gt0, description最高可接受价格) field_validator(return_date) classmethod def check_return_after_depart(cls, v, info): if v and depart_date in info.data and v info.data[depart_date]: raise ValueError(返程日期不能早于出发日期) return v这里有几个设计考量。城市代码限制长度是为了防止模型传进来一长串自然语言虽然模型一般不会这么干但校验一下更保险。返程日期用可选字段单程和往返共用一个接口减少接口数量。自定义校验器检查日期顺序这个逻辑放在模型层比放在业务层好因为校验失败 FastAPI 会自动返回 422 错误模型能直接看到错误信息并修正参数。4.3 异步并发查询多个数据源真实场景下一个航班查询往往要问好几个数据源串行查太慢。我用 asyncio.gather 做并发把多个数据源的查询同时发出去谁先回来先处理。import asyncio import httpx async def query_all_sources(req: FlightQueryRequest): async with httpx.AsyncClient(timeout8.0) as client: tasks [ query_source_a(client, req), query_source_b(client, req), query_source_c(client, req), ] results await asyncio.gather(*tasks, return_exceptionsTrue) flights [] for r in results: if isinstance(r, Exception): continue # 单个数据源失败不影响整体 flights.extend(r) return sorted(flights, keylambda x: x[price])这段代码有两个关键点。return_exceptionsTrue让某个数据源超时或报错时不会拖垮整个请求其他数据源的结果照常返回。超时设成 8 秒是我实测下来的平衡点太短会漏掉慢但有效的数据源太长会让用户等得不耐烦。另外结果统一按价格排序方便模型直接取最便宜的选项。4.4 缓存策略与限流保护票务数据变化快但也不是每秒都变。我加了一层短时缓存同一个查询条件在 5 分钟内直接返回缓存结果减少对上游数据源的压力。缓存用内存字典实现简单够用如果要多实例部署再换 Redis。import time from typing import Any _cache: dict[str, tuple[float, Any]] {} CACHE_TTL 300 # 5分钟 def get_cached(key: str): if key in _cache: ts, value _cache[key] if time.time() - ts CACHE_TTL: return value del _cache[key] return None def set_cache(key: str, value: Any): _cache[key] (time.time(), value)限流这块我用的是简单的令牌桶每个 IP 每分钟最多 30 次查询。为什么要限流因为智能体在调试阶段可能会疯狂调用接口没有限流的话上游数据源可能把你封了。限流阈值我设得比较宽松正常使用完全够异常调用能挡住。注意缓存 key 一定要包含所有影响结果的参数我一开始漏了成人数量导致两个人查出来的价格和一个人一样闹了笑话。后来我把请求体的关键字段序列化成 key确保不同查询不会串。5. 智能体编排与工具调用的完整流程5.1 智能体主循环的实现智能体的核心是一个循环把当前状态发给模型模型返回要么是工具调用请求要么是最终答案如果是工具调用就执行工具、把结果加回状态、继续循环直到模型给出最终答案或达到最大轮数。async def run_agent(user_input: str, max_turns: int 8): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] for turn in range(max_turns): response await call_llm(messages, toolsTOOL_DEFINITIONS) if response.tool_calls: for call in response.tool_calls: result await execute_tool(call.name, call.arguments) messages.append({role: assistant, tool_calls: [call]}) messages.append({role: tool, content: result}) else: return response.content return 抱歉规划过程超出预期步数请简化需求后重试max_turns 设成 8是我反复调出来的。设太小复杂需求比如多城市联程规划不完设太大万一模型陷入死循环会浪费大量 token。8 轮基本能覆盖绝大多数场景超了就说明需求太复杂或者模型跑偏了直接返回提示让用户简化。5.2 工具定义与 function calling 对接工具定义要跟 FastAPI 的接口严格对应我写了个小脚本从 OpenAPI schema 自动生成工具定义避免手写不一致。工具定义的核心是 name、description、parameters 三部分其中 description 最重要模型就是靠它判断什么时候该用这个工具。TOOL_DEFINITIONS [ { type: function, function: { name: query_flights, description: 查询指定出发地、目的地、出发日期的航班列表和价格。缺少任一必填参数时不要调用应先向用户追问。, parameters: { type: object, properties: { origin: {type: string, description: 出发城市三字码如 PEK}, destination: {type: string, description: 目的地城市三字码如 CTU}, depart_date: {type: string, description: 出发日期格式 YYYY-MM-DD}, return_date: {type: string, description: 返程日期单程不传}, }, required: [origin, destination, depart_date], }, }, }, ]5.3 多轮追问与状态管理智能体最实用的能力是主动追问。当用户信息不全时它不应该瞎猜而应该问清楚。我在 system prompt 里明确要求必填参数缺失时必须追问一次最多问两个问题避免把用户问烦。状态管理我用的是消息列表累积的方式每一轮的工具调用和结果都追加到 messages 里模型能看到完整的历史。这里要注意上下文长度控制工具返回的航班列表可能很长我会在返回给模型之前做一次精简只保留航班号、时间、价格、航司这几个关键字段把冗余信息砍掉既省 token 又让模型更容易抓重点。5.4 异常处理与降级方案智能体跑起来之后异常处理是绕不开的。我遇到过的异常主要有三类模型返回格式错误、工具调用失败、超时。针对每一类我都做了处理。模型返回格式错误时我会把错误信息作为一条 system 消息塞回去让模型重新生成最多重试两次。工具调用失败时我把失败原因返回给模型让它决定是换个参数重试还是告诉用户暂时查不到。超时的话如果已经拿到了部分数据就让模型基于部分数据生成行程并标注哪些信息未获取到。实操心得给模型返回工具错误信息时一定要用自然语言描述清楚比如查询失败出发地代码 PEK 无效请使用标准三字码而不是直接抛 Python 异常堆栈。模型看不懂堆栈但看得懂自然语言能据此修正参数。6. 实测中的常见问题与排查技巧6.1 模型编造数据怎么破这是最常见也最危险的问题。模型在没有调用工具的情况下凭记忆编出航班号和价格用户拿去下单发现根本不存在。我的解决办法有三层第一层是在 system prompt 里反复强调所有航班数据必须来自工具调用禁止编造第二层是在行程生成时做校验检查行程里的航班号是否出现在工具返回结果中不在就标记为可疑第三层是在最终输出里明确标注数据来源和查询时间让用户知道这是实时数据还是模型推测。实测下来加了第二层校验之后编造数据的情况基本杜绝了。校验逻辑很简单就是把工具返回的所有航班号收集成一个集合生成行程后逐个比对发现不在集合里的就触发重新生成。6.2 工具调用参数错误的排查模型传错参数是家常便饭常见的有日期格式不对、城市代码用了中文、成人数量传成字符串。排查这类问题的关键是把错误信息完整地反馈给模型。我一开始只返回参数错误模型不知道该改什么反复犯同样的错。后来我把 Pydantic 的校验错误信息原样返回模型看到depart_date 必须是 YYYY-MM-DD 格式就知道怎么改了。下面这张表是我整理的常见参数错误和对应处理方式可以直接抄。错误类型典型表现处理方式日期格式错误传下个月而非具体日期返回格式要求让模型重新换算城市代码错误传北京而非PEK返回三字码要求附常见城市对照数值类型错误成人数量传2字符串Pydantic 自动转换失败则报错必填参数缺失没传出发日期返回缺失字段让模型追问用户日期逻辑错误返程早于出发返回校验错误让模型修正6.3 响应太慢的优化思路智能体规划一趟行程涉及多轮模型调用和工具调用响应慢是必然的。我实测下来不做优化的话一趟要一分半优化后压到 40 秒左右。主要的优化手段有三个。并行化工具调用。如果模型在一轮里请求了多个互不依赖的工具比如同时查航班和酒店我会并发执行而不是串行。这一项就能省下十几秒。精简上下文。工具返回的数据只保留必要字段历史消息超过一定长度就做摘要压缩减少模型处理时间。流式输出。最终行程生成时用流式返回用户能边看边等感知上的等待时间大幅缩短。6.4 成本控制的实战经验智能体跑起来 token 消耗很快尤其是多轮循环加上工具返回的长文本。我做了几件事来控制成本。意图理解用便宜的小模型这个任务简单小模型完全够用只有决策编排和行程生成才用大模型。工具返回结果做截断航班列表最多返回 10 条按价格排序取前 10避免把上百条结果全塞给模型。缓存重复查询同一个用户短时间内重复问类似问题直接命中缓存。我算过一笔账优化前规划一趟行程大概消耗 15000 token优化后降到 5000 左右成本降了三分之二而用户体验几乎没受影响。7. 这套工作流还能怎么扩展跑通基础版本之后我陆续加了一些扩展这里挑几个实用的说说。多城市联程规划。用户说北京到成都再到昆明最后回北京这涉及三段航班智能体需要分别查询再拼接。我的做法是让意图理解层把这种需求拆成多个航段决策层对每个航段分别调用工具最后统一组装。这里要注意航段之间的时间衔接我加了一个校验确保下一段的出发时间晚于上一段的到达时间加两小时。预算约束下的自动比价。用户给了预算上限智能体在拿到航班列表后会自动筛选出预算内的选项如果全都超预算它会提示用户并给出最接近的选项。这个逻辑我放在行程生成层用简单的数值比较实现不需要模型参与更快更准。行程的二次修改。用户拿到行程后说第二天太赶了能不能轻松点智能体需要理解这是对已有行程的修改请求而不是全新规划。我的做法是在状态里保留上一版行程修改请求进来时把旧行程作为上下文一起传给模型让它做增量修改而不是重新生成。接入更多数据源。除了航班我还接了天气和酒店。天气接口用来在行程里提示第三天有雨建议室内活动酒店接口用来补充住宿推荐。每接一个新数据源就是在 FastAPI 里加一个路由在工具定义里加一条工作量很小。这套东西我陆陆续续迭代了小半年现在自己出门基本都靠它规划。最大的体会是智能体的价值不在于模型多聪明而在于你把工作流拆得多清楚。Prompt 分层、工具解耦、状态管理这些工程上的功夫才是决定体验的关键。模型能力会一直进步但好的架构设计能让你在换模型的时候几乎零成本迁移。如果你也在做类似的东西建议先把意图理解和工具服务这两层做扎实上层编排反而没那么难。
AI智能体重构旅行规划：Prompt分层+FastAPI实时票务查询实战
1. 为什么我要用智能体重构旅行规划这件事做旅行规划这件事我前前后后折腾了快三年。最早是纯手工Excel 拉表格、浏览器开二十个标签页比价、备忘录里记航班号一趟出行规划下来少说四五个小时遇上旺季票价波动前面查的全白费。后来用脚本爬数据能省点力气但脚本脆得很页面结构一变就崩维护成本比手工还高。再后来大模型起来了我第一反应就是拿它来干这个活但一开始也只是把它当个高级搜索框用问一句答一句体验很割裂。真正的转折点是我意识到旅行规划本质上不是一个问答任务而是一个多步骤、有状态、需要外部实时数据支撑的工作流。你得先理解用户模糊的需求我想找个暖和的地方躺几天再把它翻译成结构化的约束预算、天数、出发地、偏好然后去查实时航班和票价比对酒店最后组装成一份能直接执行的行程。这里面每一步的输入输出格式都不一样中间还会因为票价变化、余票不足而反复调整。用一个大 Prompt 硬塞给模型它要么漏步骤要么编数据要么在长上下文里把前面的约束忘了。所以我决定用AI 智能体Agent的思路来重构整个流程核心是两件事一是把规划逻辑拆成可组合的 Prompt 工程模块让每个环节职责单一、可测试二是用FastAPI搭一个实时票务查询服务让智能体能真正拿到此刻的航班和价格数据而不是靠模型记忆里的过期信息瞎编。这套东西跑通之后我规划一趟国内往返的行程从输入需求到拿到可执行方案稳定在 40 秒以内而且票价是真实的、可下单的。这篇文章我会把这套工作流从设计思路到代码落地完整拆一遍包括 Prompt 怎么分层、FastAPI 服务怎么设计接口、智能体怎么编排工具调用、以及我在实测中踩过的那些坑。适合已经会用大模型 API、想往智能体方向深入的后端或全栈开发者也适合做旅游类产品的同学参考架构。不需要你是算法专家但得能看懂 Python 和基本的 HTTP 接口。2. 整体架构设计与技术选型思路2.1 为什么是智能体 工具服务而不是一个大模型先说清楚一个概念上的分界。很多人把用大模型做应用和做智能体混为一谈其实差别很大。前者是你把问题整理好、喂给模型、拿回答案模型是被动的后者是模型自己决定我现在该干什么它会主动选择调用哪个工具、传什么参数、拿到结果后下一步做什么。旅行规划这个场景天然适合后者因为它的决策链条长且依赖外部状态。我举个具体的例子你就明白了。用户说下个月想去成都吃火锅三天预算三千。如果是一个大 Prompt模型会直接给你一份行程但里面的航班时间、票价全是它编的因为它没有实时数据。而智能体的做法是它先解析出出发地未知、目的地成都、时长三天、预算三千、主题美食发现出发地缺失于是主动追问拿到出发地后它调用票务查询工具查下个月的低价航班拿到真实结果后再排行程。这个发现缺失、主动追问、调用工具、整合结果的循环就是智能体的核心价值。我选这个架构的另一个原因是可维护性。旅行规划涉及航司、酒店、天气、签证、汇率等一堆外部依赖如果全塞进一个 Prompt任何一处数据源变化都要重写整个 Prompt。拆成智能体加工具服务之后每个工具是独立的 APIPrompt 只管决策逻辑数据源换了只改对应的服务互不影响。2.2 FastAPI 在这个架构里扮演什么角色FastAPI 是我给智能体准备的手和脚。智能体本身只会思考和决策它要拿到真实数据必须通过工具调用而工具的背后就是 FastAPI 提供的 HTTP 接口。我选 FastAPI 而不是 Flask 或 Django主要看中三点。第一是异步原生支持。票务查询往往要并发请求多个数据源比如同时查几家航司的接口FastAPI 基于 Starlette 的 async 能力让并发写起来很自然不用自己折腾线程池。第二是自动生成接口文档。智能体在决定调用哪个工具时需要知道工具的参数格式FastAPI 自动生成的 OpenAPI schema 可以直接喂给模型做 function calling 的定义省了我手写工具描述的工作。第三是Pydantic 的数据校验。票务查询的入参出参格式必须严格Pydantic 模型能在入口就把脏数据挡掉避免模型传了奇怪的参数导致下游报错。整个架构的数据流是这样的用户在前端输入需求请求打到智能体编排层编排层调用大模型做意图理解和任务分解模型决定调用哪个工具编排层通过 HTTP 请求 FastAPI 服务FastAPI 去查真实数据源并返回结构化结果编排层把结果回灌给模型模型继续下一步决策直到生成最终行程。2.3 工作流的分层设计我把整个工作流拆成了四层每层职责清晰层与层之间通过明确定义的数据结构通信。层级职责关键技术输出物意图理解层解析用户自然语言提取结构化约束Prompt 工程 JSON Schema结构化需求对象决策编排层决定调用哪些工具、处理缺失信息智能体循环 function calling工具调用序列工具服务层提供实时票务、酒店等数据FastAPI 异步请求结构化数据行程生成层整合所有数据生成可执行行程Prompt 工程 模板Markdown 行程单这么分层的好处是每一层都可以单独测试和替换。比如我想换一个更便宜的大模型做意图理解只改第一层的 Prompt 和模型配置其他层完全不动。我想加一个新的数据源比如高铁票只在工具服务层加一个接口决策编排层注册一下就行。3. Prompt 工程的分层拆解与实操要点3.1 意图理解 Prompt 怎么写才不跑偏意图理解是整个工作流的入口这一步错了后面全错。我一开始的写法是让模型提取用户需求中的所有信息结果它经常自作主张补全缺失字段比如用户没说出发地它默认成北京导致后面查出来的航班全是错的。后来我改成强制模型区分已知和未知明确要求缺失字段必须返回 null并且列出需要追问的问题。我的意图理解 Prompt 核心结构是这样的先给模型一个严格的 JSON Schema 定义规定输出必须包含哪些字段、每个字段的类型和取值范围然后给几个 few-shot 示例覆盖信息完整和信息缺失两种情况最后加一条硬约束禁止模型编造用户没提到的信息。实测下来加了 few-shot 之后字段提取的准确率从大概七成提到了九成以上。这里有个细节值得说日期处理是意图理解里最容易翻车的部分。用户说下个月、五一前后、这周末模型需要结合当前日期换算成具体日期。我的做法是在 Prompt 里注入当前日期并明确要求所有相对时间必须换算成 YYYY-MM-DD 格式无法确定的返回 null 并追问。这个改动之后日期相关的错误基本消失了。3.2 决策编排 Prompt 的设计逻辑决策编排层是智能体的大脑它要决定在当前状态下下一步该做什么。这里的 Prompt 设计核心是把可用工具和当前状态清晰地告诉模型让它做选择题而不是填空题。我的做法是把所有可用工具的函数签名、参数说明、返回格式整理成一段结构化文本放在 system prompt 里。然后在每轮对话中把已经收集到的信息、已经调用过的工具、工具返回的结果都作为上下文传进去。模型的任务就是判断信息够不够不够的话该调用哪个工具够了的话是不是可以生成行程了这里我踩过一个坑工具描述写得太模糊模型会乱调用。比如我一开始把票务查询工具描述成查询航班信息结果模型在用户还没确定日期的时候就调用了传了个空日期进去。后来我把描述改成查询指定出发地、目的地、出发日期的航班列表和价格缺少任一参数时不要调用模型的行为就规矩多了。工具描述本质上也是 Prompt 工程的一部分得当成产品文案来打磨。3.3 行程生成 Prompt 的模板化技巧行程生成是最后一层目标是把前面收集的所有数据组装成一份人类可读的行程单。这一层的 Prompt 相对简单但有个关键技巧用模板约束输出结构。我会在 Prompt 里给出一个 Markdown 模板规定行程单必须包含概览、每日安排、交通详情、预算明细、注意事项几个部分模型只需要往模板里填内容。这么做的好处是输出格式稳定前端可以直接解析渲染。如果不给模板模型每次生成的格式都不一样有时候用表格有时候用列表前端处理起来很痛苦。另外我要求模型在生成行程时必须引用工具返回的真实数据比如航班号、起降时间、价格都要来自票务查询的结果禁止自己编。为了强化这一点我会在 Prompt 里明确说以下数据来自实时查询请直接使用不要修改。3.4 Prompt 版本管理与测试Prompt 是要迭代的我强烈建议从一开始就做版本管理。我的做法是把每个 Prompt 存成独立的文本文件用 Git 管理每次修改都写清楚改了什么、为什么改。同时我建了一个小型的测试集包含二十来个典型的用户输入每次改完 Prompt 就跑一遍看输出是否符合预期。这个测试集帮我避免了好几次改好一个场景、弄坏三个场景的悲剧。比如我有次为了提升日期解析准确率在 Prompt 里加了一堆日期格式的说明结果模型开始把三天这种时长也当成日期处理。跑测试集的时候立刻发现了赶紧回滚调整。没有测试集的话这种回归问题可能要等线上用户反馈才发现。4. FastAPI 实时票务查询服务的落地实现4.1 项目目录结构怎么组织FastAPI 项目的目录结构我试过好几种最后稳定在这套组织方式上兼顾了清晰和可扩展。ticket-service/ ├── app/ │ ├── main.py # 应用入口注册路由 │ ├── config.py # 配置管理读环境变量 │ ├── models/ │ │ ├── request.py # 请求体 Pydantic 模型 │ │ └── response.py # 响应体 Pydantic 模型 │ ├── routers/ │ │ ├── flight.py # 航班查询路由 │ │ └── hotel.py # 酒店查询路由 │ ├── services/ │ │ ├── flight_service.py # 航班查询业务逻辑 │ │ └── cache.py # 缓存逻辑 │ └── clients/ │ └── data_source.py # 外部数据源客户端 ├── tests/ ├── requirements.txt └── .env这么分的原因很简单路由层只管 HTTP 协议相关的事业务逻辑放 service外部依赖放 client。这样我想换数据源只改 client 层想加缓存只改 service 层路由层几乎不用动。很多新手把所有逻辑堆在路由函数里项目一大就变成几百行的巨型函数改起来要命。4.2 票务查询接口的设计与参数校验票务查询接口是整个服务的核心我设计成 POST 而不是 GET因为查询参数比较多而且未来可能扩展成批量查询。请求体用 Pydantic 模型定义把校验规则写死在模型里。from pydantic import BaseModel, Field, field_validator from datetime import date class FlightQueryRequest(BaseModel): origin: str Field(..., min_length2, max_length3, description出发地城市代码) destination: str Field(..., min_length2, max_length3, description目的地城市代码) depart_date: date Field(..., description出发日期) return_date: date | None Field(None, description返程日期单程为空) adults: int Field(1, ge1, le9, description成人数量) max_price: float | None Field(None, gt0, description最高可接受价格) field_validator(return_date) classmethod def check_return_after_depart(cls, v, info): if v and depart_date in info.data and v info.data[depart_date]: raise ValueError(返程日期不能早于出发日期) return v这里有几个设计考量。城市代码限制长度是为了防止模型传进来一长串自然语言虽然模型一般不会这么干但校验一下更保险。返程日期用可选字段单程和往返共用一个接口减少接口数量。自定义校验器检查日期顺序这个逻辑放在模型层比放在业务层好因为校验失败 FastAPI 会自动返回 422 错误模型能直接看到错误信息并修正参数。4.3 异步并发查询多个数据源真实场景下一个航班查询往往要问好几个数据源串行查太慢。我用 asyncio.gather 做并发把多个数据源的查询同时发出去谁先回来先处理。import asyncio import httpx async def query_all_sources(req: FlightQueryRequest): async with httpx.AsyncClient(timeout8.0) as client: tasks [ query_source_a(client, req), query_source_b(client, req), query_source_c(client, req), ] results await asyncio.gather(*tasks, return_exceptionsTrue) flights [] for r in results: if isinstance(r, Exception): continue # 单个数据源失败不影响整体 flights.extend(r) return sorted(flights, keylambda x: x[price])这段代码有两个关键点。return_exceptionsTrue让某个数据源超时或报错时不会拖垮整个请求其他数据源的结果照常返回。超时设成 8 秒是我实测下来的平衡点太短会漏掉慢但有效的数据源太长会让用户等得不耐烦。另外结果统一按价格排序方便模型直接取最便宜的选项。4.4 缓存策略与限流保护票务数据变化快但也不是每秒都变。我加了一层短时缓存同一个查询条件在 5 分钟内直接返回缓存结果减少对上游数据源的压力。缓存用内存字典实现简单够用如果要多实例部署再换 Redis。import time from typing import Any _cache: dict[str, tuple[float, Any]] {} CACHE_TTL 300 # 5分钟 def get_cached(key: str): if key in _cache: ts, value _cache[key] if time.time() - ts CACHE_TTL: return value del _cache[key] return None def set_cache(key: str, value: Any): _cache[key] (time.time(), value)限流这块我用的是简单的令牌桶每个 IP 每分钟最多 30 次查询。为什么要限流因为智能体在调试阶段可能会疯狂调用接口没有限流的话上游数据源可能把你封了。限流阈值我设得比较宽松正常使用完全够异常调用能挡住。注意缓存 key 一定要包含所有影响结果的参数我一开始漏了成人数量导致两个人查出来的价格和一个人一样闹了笑话。后来我把请求体的关键字段序列化成 key确保不同查询不会串。5. 智能体编排与工具调用的完整流程5.1 智能体主循环的实现智能体的核心是一个循环把当前状态发给模型模型返回要么是工具调用请求要么是最终答案如果是工具调用就执行工具、把结果加回状态、继续循环直到模型给出最终答案或达到最大轮数。async def run_agent(user_input: str, max_turns: int 8): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_input}, ] for turn in range(max_turns): response await call_llm(messages, toolsTOOL_DEFINITIONS) if response.tool_calls: for call in response.tool_calls: result await execute_tool(call.name, call.arguments) messages.append({role: assistant, tool_calls: [call]}) messages.append({role: tool, content: result}) else: return response.content return 抱歉规划过程超出预期步数请简化需求后重试max_turns 设成 8是我反复调出来的。设太小复杂需求比如多城市联程规划不完设太大万一模型陷入死循环会浪费大量 token。8 轮基本能覆盖绝大多数场景超了就说明需求太复杂或者模型跑偏了直接返回提示让用户简化。5.2 工具定义与 function calling 对接工具定义要跟 FastAPI 的接口严格对应我写了个小脚本从 OpenAPI schema 自动生成工具定义避免手写不一致。工具定义的核心是 name、description、parameters 三部分其中 description 最重要模型就是靠它判断什么时候该用这个工具。TOOL_DEFINITIONS [ { type: function, function: { name: query_flights, description: 查询指定出发地、目的地、出发日期的航班列表和价格。缺少任一必填参数时不要调用应先向用户追问。, parameters: { type: object, properties: { origin: {type: string, description: 出发城市三字码如 PEK}, destination: {type: string, description: 目的地城市三字码如 CTU}, depart_date: {type: string, description: 出发日期格式 YYYY-MM-DD}, return_date: {type: string, description: 返程日期单程不传}, }, required: [origin, destination, depart_date], }, }, }, ]5.3 多轮追问与状态管理智能体最实用的能力是主动追问。当用户信息不全时它不应该瞎猜而应该问清楚。我在 system prompt 里明确要求必填参数缺失时必须追问一次最多问两个问题避免把用户问烦。状态管理我用的是消息列表累积的方式每一轮的工具调用和结果都追加到 messages 里模型能看到完整的历史。这里要注意上下文长度控制工具返回的航班列表可能很长我会在返回给模型之前做一次精简只保留航班号、时间、价格、航司这几个关键字段把冗余信息砍掉既省 token 又让模型更容易抓重点。5.4 异常处理与降级方案智能体跑起来之后异常处理是绕不开的。我遇到过的异常主要有三类模型返回格式错误、工具调用失败、超时。针对每一类我都做了处理。模型返回格式错误时我会把错误信息作为一条 system 消息塞回去让模型重新生成最多重试两次。工具调用失败时我把失败原因返回给模型让它决定是换个参数重试还是告诉用户暂时查不到。超时的话如果已经拿到了部分数据就让模型基于部分数据生成行程并标注哪些信息未获取到。实操心得给模型返回工具错误信息时一定要用自然语言描述清楚比如查询失败出发地代码 PEK 无效请使用标准三字码而不是直接抛 Python 异常堆栈。模型看不懂堆栈但看得懂自然语言能据此修正参数。6. 实测中的常见问题与排查技巧6.1 模型编造数据怎么破这是最常见也最危险的问题。模型在没有调用工具的情况下凭记忆编出航班号和价格用户拿去下单发现根本不存在。我的解决办法有三层第一层是在 system prompt 里反复强调所有航班数据必须来自工具调用禁止编造第二层是在行程生成时做校验检查行程里的航班号是否出现在工具返回结果中不在就标记为可疑第三层是在最终输出里明确标注数据来源和查询时间让用户知道这是实时数据还是模型推测。实测下来加了第二层校验之后编造数据的情况基本杜绝了。校验逻辑很简单就是把工具返回的所有航班号收集成一个集合生成行程后逐个比对发现不在集合里的就触发重新生成。6.2 工具调用参数错误的排查模型传错参数是家常便饭常见的有日期格式不对、城市代码用了中文、成人数量传成字符串。排查这类问题的关键是把错误信息完整地反馈给模型。我一开始只返回参数错误模型不知道该改什么反复犯同样的错。后来我把 Pydantic 的校验错误信息原样返回模型看到depart_date 必须是 YYYY-MM-DD 格式就知道怎么改了。下面这张表是我整理的常见参数错误和对应处理方式可以直接抄。错误类型典型表现处理方式日期格式错误传下个月而非具体日期返回格式要求让模型重新换算城市代码错误传北京而非PEK返回三字码要求附常见城市对照数值类型错误成人数量传2字符串Pydantic 自动转换失败则报错必填参数缺失没传出发日期返回缺失字段让模型追问用户日期逻辑错误返程早于出发返回校验错误让模型修正6.3 响应太慢的优化思路智能体规划一趟行程涉及多轮模型调用和工具调用响应慢是必然的。我实测下来不做优化的话一趟要一分半优化后压到 40 秒左右。主要的优化手段有三个。并行化工具调用。如果模型在一轮里请求了多个互不依赖的工具比如同时查航班和酒店我会并发执行而不是串行。这一项就能省下十几秒。精简上下文。工具返回的数据只保留必要字段历史消息超过一定长度就做摘要压缩减少模型处理时间。流式输出。最终行程生成时用流式返回用户能边看边等感知上的等待时间大幅缩短。6.4 成本控制的实战经验智能体跑起来 token 消耗很快尤其是多轮循环加上工具返回的长文本。我做了几件事来控制成本。意图理解用便宜的小模型这个任务简单小模型完全够用只有决策编排和行程生成才用大模型。工具返回结果做截断航班列表最多返回 10 条按价格排序取前 10避免把上百条结果全塞给模型。缓存重复查询同一个用户短时间内重复问类似问题直接命中缓存。我算过一笔账优化前规划一趟行程大概消耗 15000 token优化后降到 5000 左右成本降了三分之二而用户体验几乎没受影响。7. 这套工作流还能怎么扩展跑通基础版本之后我陆续加了一些扩展这里挑几个实用的说说。多城市联程规划。用户说北京到成都再到昆明最后回北京这涉及三段航班智能体需要分别查询再拼接。我的做法是让意图理解层把这种需求拆成多个航段决策层对每个航段分别调用工具最后统一组装。这里要注意航段之间的时间衔接我加了一个校验确保下一段的出发时间晚于上一段的到达时间加两小时。预算约束下的自动比价。用户给了预算上限智能体在拿到航班列表后会自动筛选出预算内的选项如果全都超预算它会提示用户并给出最接近的选项。这个逻辑我放在行程生成层用简单的数值比较实现不需要模型参与更快更准。行程的二次修改。用户拿到行程后说第二天太赶了能不能轻松点智能体需要理解这是对已有行程的修改请求而不是全新规划。我的做法是在状态里保留上一版行程修改请求进来时把旧行程作为上下文一起传给模型让它做增量修改而不是重新生成。接入更多数据源。除了航班我还接了天气和酒店。天气接口用来在行程里提示第三天有雨建议室内活动酒店接口用来补充住宿推荐。每接一个新数据源就是在 FastAPI 里加一个路由在工具定义里加一条工作量很小。这套东西我陆陆续续迭代了小半年现在自己出门基本都靠它规划。最大的体会是智能体的价值不在于模型多聪明而在于你把工作流拆得多清楚。Prompt 分层、工具解耦、状态管理这些工程上的功夫才是决定体验的关键。模型能力会一直进步但好的架构设计能让你在换模型的时候几乎零成本迁移。如果你也在做类似的东西建议先把意图理解和工具服务这两层做扎实上层编排反而没那么难。
