旅行规划这件事表面上看是选个目的地、订张票、排个行程三步走但真正动手做过完整系统的人都知道这里面藏着一堆琐碎又容易出错的环节用户需求模糊、车次数据实时变化、行程冲突检测、预算动态调整。我前阵子花了大概三周时间把一套原本靠人工反复查票、手动排行程的流程改造成了由 AI 智能体驱动的自动化工作流核心链路是Prompt 工程做意图理解 FastAPI 做实时票务查询 智能体做决策编排。整套东西跑下来最直观的感受是真正难的不是调模型而是把模型输出稳定地接进一个能查真实数据、能做真实判断的后端系统里。这篇文章我会把整个重构过程拆开讲包括 Prompt 怎么设计才能让模型稳定输出结构化意图、FastAPI 后端怎么组织目录和接口、实时票务查询怎么处理并发和缓存、智能体的决策循环怎么避免死循环和幻觉。适合已经有一点 Python 后端基础、想动手做一个真实可用的 AI 智能体项目的朋友。如果你只是想让模型帮你写个行程文案那这篇可能偏重了但如果你想做一个能真正查票、能根据实时数据调整方案的旅行规划系统那接下来的内容应该能帮你少走不少弯路。1. 为什么旅行规划值得用智能体重构1.1 传统旅行规划工作流的三个断点先说清楚我为什么要做这件事。原来的流程大概是这样的用户说我想下个月去成都玩三天预算两千左右然后人工去查车次、查酒店、排景点、算预算。这个流程有三个明显的断点。第一个断点是意图到结构化参数的转换全靠人脑。下个月到底是哪几天预算两千含不含往返交通玩三天是整三天还是两个晚上这些模糊表达在人工场景下靠经验补全但一旦要自动化就必须有一个稳定的意图解析层。我试过直接用正则去匹配结果发现用户表达方式太多样维护成本极高最后还是回到用大模型做意图抽取。第二个断点是实时数据查询和决策是割裂的。传统做法是先查一堆车次存下来再基于静态数据排行程。但票务数据是实时变化的你十分钟前查到的余票十分钟后可能就没了。这就导致排出来的行程经常看起来合理但订不到票。智能体的价值在于它可以把查询和决策放在一个循环里查一次、判断一次、不行就换方案再查。第三个断点是异常处理没有闭环。比如用户指定的日期没有直达车传统流程要么报错要么人工介入。而智能体可以自己触发换中转方案或者调整日期的分支逻辑。这三个断点基本就是我用智能体重构的切入点。1.2 智能体在这个场景里到底承担什么角色很多人一提到智能体就想到自主规划、自主执行但在旅行规划这个具体场景里我对智能体的定位其实很克制它不是一个全知全能的规划师而是一个意图解析器 决策调度器。具体来说它负责三件事。第一把自然语言需求转成结构化查询参数。这部分靠 Prompt 工程完成输出必须是严格的 JSON字段包括出发地、目的地、日期范围、预算上限、偏好标签等。第二根据查询结果做分支决策。比如查到有票就进入行程编排没票就触发备选方案生成。第三在多个候选方案之间做权衡排序。这部分我会给它明确的评分维度而不是让它自由发挥。我特意没有让智能体去自由生成整个行程因为实测下来让模型自由发挥的行程往往在时间衔接上出问题比如两个景点之间通勤时间不够。所以我的设计是智能体负责决策和调度具体的行程编排用规则引擎兜底。这个分工在后面几节会详细展开。1.3 技术选型的取舍逻辑选 FastAPI 而不是 Flask 或 Django理由很直接这个项目需要大量异步 IO尤其是票务查询环节要并发请求多个数据源FastAPI 的原生 async 支持省了很多事。而且它自带 Pydantic 校验正好可以用来约束智能体输出的结构化数据一举两得。选智能体框架的时候我对比过几种方案。重型框架功能全但抽象层太厚调试的时候很难定位问题到底出在 Prompt 还是框架调度轻量方案灵活但很多轮子要自己造。最后我的选择是用一个轻量的智能体循环自己实现调度逻辑只借用框架的工具调用Tool Calling能力。原因是旅行规划的决策逻辑其实不复杂核心就是解析意图 → 调用工具 → 判断结果 → 决定下一步自己写循环反而更可控出问题也好排查。提示如果你团队里没有人熟悉智能体框架不要一上来就选抽象层最厚的那个。先用最朴素的方式把模型输出 → 工具调用 → 结果回填这条链路跑通再考虑引入框架。2. Prompt 工程让模型稳定吐出结构化意图2.1 意图解析 Prompt 的字段设计这部分是整个系统的入口也是最容易翻车的地方。我最初版本的 Prompt 很随意就是请从下面这段话里提取出发地、目的地、日期、预算结果模型输出的格式五花八门有时候是 JSON有时候是自然语言有时候字段名还给我换成同义词。后来我做了三件事才把它稳定下来。第一件事是用 JSON Schema 显式约束输出结构。我把所有字段的定义、类型、是否必填都写进 Prompt并且明确要求只输出 JSON不要任何解释文字。第二件事是给每个字段加边界说明和示例。比如日期字段我要求输出 ISO 格式并且明确如果用户说的是相对时间如下个月请基于当前日期 {current_date} 计算。第三件事是加兜底字段。对于模型无法确定的字段我要求它输出 null 而不是瞎猜这样后端可以触发追问逻辑。下面是我最终用的意图解析 Prompt 的核心结构你可以直接参考你是一个旅行需求解析器。请从用户输入中提取以下字段只输出 JSON不要任何额外文字。 字段定义 - origin: 出发城市字符串无法确定时输出 null - destination: 目的地城市字符串无法确定时输出 null - date_start: 出发日期ISO 格式 YYYY-MM-DD相对时间基于 {current_date} 计算 - date_end: 返回日期ISO 格式无法确定时输出 null - budget: 预算上限整数单位元无法确定时输出 null - preferences: 偏好标签数组可选值 [自然风光,历史人文,美食,亲子,购物] - confidence: 你对本次解析的置信度0 到 1 之间的小数 当前日期{current_date} 用户输入{user_input}这个 Prompt 我迭代了大概七八版。中间踩过一个坑一开始我没写只输出 JSON模型总喜欢在 JSON 前面加一句好的我来帮你解析导致后端解析失败。加上这句约束后就干净了。另一个坑是相对时间处理模型对下个月的理解经常出错后来我把当前日期作为变量注入并要求它显式计算准确率才上来。2.2 少样本示例怎么放才有效光有字段定义还不够模型对某些字段的理解还是会飘。比如预算两千左右里的左右模型有时候输出 2000有时候输出 2200。我的做法是加少样本示例Few-shot但示例不是越多越好我实测下来三到五个高质量示例的效果最好再多反而会让模型过度拟合示例格式。示例的选择也有讲究。我覆盖了四类典型输入完整信息型出发地目的地日期预算都有、模糊时间型只说下个月、缺字段型没说预算、多意图型既想玩又想顺便办事。每类给一个示例模型基本就能泛化到其他类似输入。这里有个细节值得说示例里的输出我全部用了标准 JSON并且字段顺序保持一致。实测发现字段顺序一致能略微提升模型输出的稳定性虽然理论上 JSON 顺序无所谓但模型确实会模仿示例的格式。2.3 处理模型输出不合规的兜底策略即使 Prompt 写得再好模型偶尔还是会输出不合规的内容。我的兜底策略分三层。第一层是解析层兜底。用 Pydantic 定义好数据模型解析失败就捕获异常。第二层是重试层兜底。解析失败后我会把错误信息拼回 Prompt 里让模型重新输出一次最多重试两次。第三层是降级层兜底。如果重试还是失败就返回一个需要用户补充信息的响应把已经解析出来的字段带上让用户确认或补充。from pydantic import BaseModel, Field, ValidationError from typing import Optional, List class TravelIntent(BaseModel): origin: Optional[str] None destination: Optional[str] None date_start: Optional[str] None date_end: Optional[str] None budget: Optional[int] Field(defaultNone, ge0) preferences: List[str] [] confidence: float Field(default0.0, ge0.0, le1.0) def parse_intent(raw_output: str) - TravelIntent: try: return TravelIntent.model_validate_json(raw_output) except ValidationError as e: # 触发重试逻辑把 e 的错误信息回填给模型 raise IntentParseError(str(e)) from e这套三层兜底跑下来意图解析的成功率从最初的七成左右提到了九成五以上。剩下那不到百分之五基本都是用户输入本身太模糊这时候追问反而是正确的产品行为。3. FastAPI 后端的目录组织与接口设计3.1 项目目录结构怎么划分才不混乱FastAPI 项目最容易犯的错就是所有东西堆在 main.py 里写到后面自己都找不到代码。我在这个项目里用的目录结构是这样的按职责分层每层只干一件事travel_agent/ ├── app/ │ ├── main.py # 应用入口注册路由和中间件 │ ├── api/ │ │ ├── routes_intent.py # 意图解析相关接口 │ │ ├── routes_ticket.py # 票务查询相关接口 │ │ └── routes_plan.py # 行程规划相关接口 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── logging.py # 日志配置 │ ├── models/ │ │ ├── intent.py # 意图相关 Pydantic 模型 │ │ └── ticket.py # 票务相关 Pydantic 模型 │ ├── services/ │ │ ├── llm_service.py # 大模型调用封装 │ │ ├── ticket_service.py # 票务查询业务逻辑 │ │ └── agent_service.py # 智能体调度逻辑 │ └── utils/ │ └── cache.py # 缓存工具 ├── tests/ └── requirements.txt这个结构的关键在于api 层只做参数校验和响应组装业务逻辑全部下沉到 services 层。这样做的好处是智能体的调度逻辑agent_service可以独立于 HTTP 接口进行单元测试不用起服务就能跑。我一开始没分层后来想给智能体逻辑写测试的时候发现根本没法测因为所有逻辑都耦合在路由函数里只能重构。3.2 接口设计意图解析与票务查询怎么拆接口拆分我遵循一个原则一个接口只做一件可独立测试的事。所以意图解析和票务查询是两个独立接口而不是一个大接口里串起来做。这样做的好处是意图解析的结果可以被缓存和复用用户如果只是想改日期不用重新解析一遍全部意图。意图解析接口接收自然语言返回结构化意图from fastapi import APIRouter, HTTPException from app.models.intent import IntentRequest, IntentResponse from app.services.llm_service import parse_travel_intent router APIRouter(prefix/api/intent, tags[intent]) router.post(/parse, response_modelIntentResponse) async def parse_intent(req: IntentRequest): try: intent await parse_travel_intent(req.text) return IntentResponse(successTrue, intentintent) except IntentParseError as e: raise HTTPException(status_code422, detailstr(e))票务查询接口接收结构化参数返回车次列表。这里有个设计细节我把查询参数设计成出发地、目的地、日期三个必填项其余都是可选项。因为票务数据源对查询参数很敏感参数太多反而容易查不到结果。3.3 用 Pydantic 做请求响应校验的实战细节Pydantic 在这个项目里承担了两个角色一是校验 HTTP 请求响应二是校验模型输出。这两个场景的模型定义我特意分开了因为它们的约束强度不一样。HTTP 请求的校验可以严格一些字段类型不对直接返回 422而模型输出的校验要宽松一些因为模型偶尔会输出一些边界值太严格会导致大量重试。举个例子预算字段在 HTTP 请求里我要求必须是正整数但在模型输出校验里我允许它是 null 或者 0因为用户可能压根没提预算。这种同一字段在不同场景下约束不同的处理是实际项目里很常见但文档里很少提的细节。注意Pydantic v2 和 v1 的 API 差异很大尤其是 model_validate_json 和 parse_raw 这类方法。如果你参考的教程比较老注意确认版本不然会踩一堆方法不存在的坑。4. 实时票务查询并发、缓存与容错4.1 查询链路为什么要做并发票务查询这个环节性能是核心矛盾。用户输入一个查询后端可能需要同时查多个数据源、多个日期、多个车次类型。如果串行查一个查询可能要好几秒用户体验很差。我的做法是用 asyncio.gather 做并发查询把独立的查询任务并行化。import asyncio from app.services.ticket_service import query_single_source async def query_tickets(origin: str, destination: str, date: str): sources [source_a, source_b, source_c] tasks [query_single_source(s, origin, destination, date) for s in sources] results await asyncio.gather(*tasks, return_exceptionsTrue) valid [r for r in results if not isinstance(r, Exception)] return merge_results(valid)这里有个关键点return_exceptionsTrue 必须加。不加的话任何一个数据源查询失败都会导致整个 gather 抛异常其他成功的查询结果也拿不到。加上之后失败的源会被过滤掉成功的源照常返回。这个细节我在第一次写的时候漏了结果一个源不稳定就导致整个查询挂掉排查了半天才发现。4.2 缓存策略哪些数据能缓存缓存多久票务数据是实时变化的所以缓存策略要非常小心。我的原则是查询结果可以短暂缓存但缓存时间必须短且要能主动失效。具体来说我用了两级缓存。第一级是内存缓存缓存时间 30 秒。这主要是为了应对用户在短时间内重复查询同一路线的情况比如反复切换日期看余票。30 秒的窗口足够覆盖这种交互又不至于让数据太旧。第二级是 Redis 缓存缓存时间 5 分钟主要用于跨请求共享一些变化不那么频繁的数据比如车站列表、车次类型字典这类。import time from app.utils.cache import memory_cache CACHE_TTL 30 async def get_tickets_cached(origin, destination, date): key fticket:{origin}:{destination}:{date} cached memory_cache.get(key) if cached and time.time() - cached[ts] CACHE_TTL: return cached[data] data await query_tickets(origin, destination, date) memory_cache.set(key, {data: data, ts: time.time()}) return data这里要特别提醒余票数量这种字段不要缓存太久。我一开始把缓存设成了 5 分钟结果用户看到有票点进去发现没了投诉了好几次。后来改成 30 秒并且在下单前强制刷新一次问题才解决。4.3 数据源不稳定时的降级方案票务数据源不稳定是常态尤其是高峰期。我的降级方案分三档。第一档是重试对单个数据源的失败请求重试两次间隔用指数退避。第二档是降级到缓存如果所有数据源都失败就返回最近一次的缓存结果并在响应里标记数据可能不是最新。第三档是降级到提示如果连缓存都没有就返回一个明确的提示告诉用户当前查询不可用建议稍后重试。这三档降级的关键是要让用户知道数据的可信度。我在响应里加了一个 data_freshness 字段值可以是 realtime、cached、unavailable。前端根据这个字段决定怎么展示比如 cached 状态就加一个数据可能有延迟的提示。这种透明化处理比默默返回旧数据要好得多。5. 智能体决策循环从查票到行程编排5.1 决策循环的状态机设计智能体的决策循环我用状态机的方式来实现而不是让模型自由决定下一步。原因是自由决策太容易跑偏比如模型可能陷入查票 → 觉得不合适 → 再查票的死循环。状态机把整个流程拆成几个明确的状态每个状态有明确的退出条件。状态包括PARSE_INTENT解析意图、QUERY_TICKETS查询票务、EVALUATE_OPTIONS评估选项、GENERATE_PLAN生成行程、FALLBACK降级处理、DONE完成。每个状态执行完根据结果决定下一个状态。比如QUERY_TICKETS查到票就进EVALUATE_OPTIONS查不到就进FALLBACK。class AgentState(Enum): PARSE_INTENT parse_intent QUERY_TICKETS query_tickets EVALUATE_OPTIONS evaluate_options GENERATE_PLAN generate_plan FALLBACK fallback DONE done async def run_agent(user_input: str, max_steps: int 10): state AgentState.PARSE_INTENT context {user_input: user_input} steps 0 while state ! AgentState.DONE and steps max_steps: state await step(state, context) steps 1 return contextmax_steps 这个限制非常重要。我一开始没加结果有一次模型在某个状态下反复横跳请求一直不返回最后超时。加上步数限制后即使逻辑出问题也能保证请求在有限步内结束。5.2 工具调用让智能体真正能查数据智能体要能查真实数据靠的是工具调用。我把票务查询封装成一个工具智能体在需要的时候调用它。工具的定义要清晰包括名称、描述、参数 schema。描述尤其重要模型是根据描述来判断什么时候该调用这个工具的。ticket_tool { name: query_tickets, description: 查询指定日期从出发地到目的地的可用车次。当用户需要查询票务信息时调用。, parameters: { type: object, properties: { origin: {type: string, description: 出发城市}, destination: {type: string, description: 目的地城市}, date: {type: string, description: 出发日期ISO 格式} }, required: [origin, destination, date] } }实测下来工具描述里明确写出什么时候调用比只写这个工具做什么效果更好。我最初的描述只写了功能模型经常在该调用的时候不调用。加上当用户需要查询票务信息时调用这句后调用准确率明显提升。5.3 避免幻觉让智能体只基于真实数据决策智能体最大的风险是幻觉也就是编造不存在的数据。在旅行规划场景里幻觉的典型表现是编造车次号、编造余票数量、编造价格。我的应对策略是把决策和生成严格分开。决策阶段智能体只能基于工具返回的真实数据做判断不能自己编数据。生成阶段行程编排用的是规则引擎输入是真实数据输出是结构化行程模型不参与具体数字的生成。这样就把幻觉的空间压缩到了最小。具体实现上我在 Prompt 里明确要求所有车次信息必须来自工具返回结果不得自行编造并且在代码层面做了校验生成行程时如果发现某个车次号不在查询结果里直接拒绝并触发重新生成。这层校验虽然简单但拦住了不少潜在问题。6. 实测中踩过的坑与调优经验6.1 Prompt 输出格式漂移的排查过程前面提到过模型输出格式漂移的问题这里展开讲讲排查过程。现象是同样的 Prompt大部分时候输出标准 JSON但偶尔会在 JSON 外面包一层 markdown 代码块或者加一句解释。我一开始以为是 Prompt 不够明确反复加强约束但效果不稳定。后来我做了个对比实验把同一段输入跑 100 次统计格式不合规的比例。发现不合规的比例大概在 5% 左右而且和输入内容有关——输入越模糊模型越倾向于加解释。找到原因后我的解决方案是在解析层做预处理先用正则把 markdown 代码块剥掉再尝试解析 JSON。这个预处理加上之后格式问题基本解决了。import re import json def extract_json(raw: str) - str: # 剥掉 markdown 代码块 raw re.sub(r^(?:json)?\s*, , raw.strip()) raw re.sub(r\s*$, , raw) # 尝试找到第一个完整的 JSON 对象 match re.search(r\{.*\}, raw, re.DOTALL) return match.group(0) if match else raw这个函数看起来简单但省了我大量重试的开销。经验就是不要指望 Prompt 能 100% 约束模型输出解析层一定要有容错。6.2 并发查询把数据源打挂的教训这个坑比较惨痛。我一开始做并发查询的时候没做限流用户一多并发请求直接把某个数据源打挂了导致整个查询功能不可用。后来我加了两层保护。第一层是信号量限流用 asyncio.Semaphore 限制同时进行的查询数量。第二层是请求间隔控制对同一个数据源两次请求之间强制间隔一定时间。这两层加上之后再没出现过打挂数据源的情况。import asyncio semaphore asyncio.Semaphore(5) async def query_with_limit(source, origin, destination, date): async with semaphore: return await query_single_source(source, origin, destination, date)提示并发不是越多越好。对第三方数据源一定要假设它很脆弱主动限流。我现在的经验值是单个数据源并发不超过 5请求间隔不低于 200 毫秒。6.3 智能体循环超时的定位方法智能体循环超时这个问题定位起来比较麻烦因为日志里只看到请求没返回不知道卡在哪一步。我的做法是给每个状态加详细日志记录进入状态的时间、退出状态的时间、状态内的关键决策。这样一旦超时看日志就能知道卡在哪个状态。有一次超时日志显示卡在EVALUATE_OPTIONS状态。进一步排查发现是因为某个查询结果里有个字段是 null评估逻辑没处理 null 导致抛异常异常又被上层吞掉了所以看起来像卡住。修复方式是在评估逻辑里加 null 检查。这个问题的教训是智能体的每个状态都要有异常处理不能让异常静默吞掉。6.4 缓存与实时性的平衡点怎么找缓存时间设多长这个平衡点我调了好几轮。设太长数据不新鲜设太短缓存没意义。最后我根据实际用户行为数据定了个策略查询结果缓存 30 秒余票字段缓存 10 秒下单前强制刷新。这个策略的依据是用户从看到结果到点击下单平均间隔在 20 秒左右30 秒的缓存能覆盖大部分场景又不会让数据太旧。另外我还加了个机制当缓存命中时后台异步刷新一次数据。这样用户拿到的是缓存结果快但缓存本身也在更新下次请求就能拿到更新的数据。这个机制叫缓存预热或者后台刷新实现起来不复杂但对体验提升很明显。7. 这套架构还能怎么扩展7.1 接入更多数据源的注意事项现在这套架构只接了票务数据源如果要接入酒店、景点门票等更多数据源架构上不用大改因为 services 层已经是按数据源拆分的。但有几个注意事项。第一不同数据源的查询语义不一样。票务查询是出发地 目的地 日期酒店查询是城市 入住日期 离店日期字段不完全一样。所以工具定义要分开不能用一个通用工具糊弄。第二不同数据源的限流策略要独立配置不能共用一个信号量否则一个慢数据源会拖累其他数据源。第三结果的合并逻辑要能处理异构数据票务和酒店的数据结构不同合并时要先归一化。7.2 多轮对话场景下的上下文管理现在的实现是单轮为主用户输入一次系统返回一次。如果要支持多轮对话比如用户说换一个日期试试系统要能理解这是在修改之前的查询就需要上下文管理。我的思路是把意图解析的结果作为会话状态存起来每轮对话时把历史意图和当前输入一起给模型让它输出更新后的意图。这里的关键是要能识别修改和新增两种意图。用户说预算改成三千是修改说再加一个目的地是新增。这两种操作对状态的影响不同Prompt 里要明确区分。我目前的做法是在意图模型里加一个 action 字段值可以是 create、update、append让模型来判断。7.3 从单智能体到多智能体的演进路径如果业务再复杂一些比如要同时处理票务、酒店、行程、预算四个维度单智能体可能会力不从心。这时候可以考虑多智能体架构每个智能体负责一个维度由一个协调者智能体来调度。但我要提醒的是多智能体不是银弹它带来的复杂度是成倍增加的。调试难度、状态同步、错误传播都会变复杂。我的建议是先用单智能体把业务跑通等到确实遇到单智能体处理不了的场景再考虑拆分。过早引入多智能体很可能是在给自己挖坑。7.4 监控与可观测性建设最后说说监控。这套系统上线后我加了几个关键指标的监控意图解析成功率、票务查询平均耗时、智能体平均步数、缓存命中率、降级触发次数。这几个指标基本能反映系统的健康度。其中智能体平均步数这个指标特别有用。正常情况下一次完整的规划大概在 4 到 6 步。如果平均步数突然涨到 8 步以上说明某个环节出了问题可能是查询失败导致反复重试也可能是评估逻辑卡住了。这个指标帮我提前发现了好几次潜在问题。监控数据我建议至少保留 7 天方便做趋势对比。另外关键路径的日志要打全尤其是智能体每一步的输入输出出问题的时候这些日志就是救命稻草。我现在的做法是把每一步的输入输出都结构化记录虽然日志量大一些但排查问题时真的省事。这套系统从最初的想法到跑通前后大概三周中间踩的坑基本都写在上面了。如果让我给一个最重要的建议那就是先把数据链路跑通再优化智能体的决策逻辑。我一开始花了很多时间调 Prompt结果发现真正卡住流程的是数据源不稳定和缓存策略不合理。数据链路稳了智能体的价值才能真正发挥出来。
