如果你最近在认真构建 AI Agent大概率经历过这种场景本地带着一两个示例反复验证的时候Agent 按部就班地规划、调工具、给出结果顺畅得让人产生“已经稳了”的错觉。一旦接到真实流量面对用户随手输入的脏数据、上游接口的偶发超时、两条并发请求互相踩踏它就立刻变成另一个人——工具参数乱传、状态脏读、重复执行、死循环空转严重时直接把上游系统打挂。我今年大部分时间都在跟这类问题搏斗。项目从单个 Demo 演进到要扛真实并发、要能追溯每一步决策的服务,最大的认知转变是Agent 做得“聪明”和做得“稳定”完全是两个维度。前者是模型能力和提示词的事后者是工程治理的事——也就是圈子里越来越多人提到的Harness 工程。这篇文章不聊概念就聊我在 FastAPI LangChain LangGraph 这套技术栈里从一次次线上事故中总结出来的机制设计和实践思路。1. 为什么 Agent“能演示不能上线”先看清不稳定长什么样1.1 先搞清楚“不稳定”三个字到底指什么很多人一听到“Agent 不稳定”第一反应是换更强的模型。但根据我自己的项目经验模型能力偏弱确实会让下限变低可真正把服务拖垮的往往不是模型本身而是 Agent 特有的工程问题在作祟。Agent 和传统后端服务最大的差别在于非确定性。传统接口你给入参、它走固定代码分支、返回固定结构只要依赖不挂行为是可预期的。Agent 不一样同样的用户问题在不同时间进来模型可能会规划出完全不同的工具调用路径即使路径一样每一步工具返回的内容不同后续推理也会跟着变。这种非确定性本身不是罪过问题是如果你没有在外围做约束非确定性就会放大成随机性参数偶尔漏传、格式偶尔漂移、状态偶尔错乱。我把实际线上遇到的现象归纳为三类调用链路不稳定Agent 的一次回答往往要经历多次模型调用规划、调用工具、总结链路越长任何一步的超时、限流、返回异常都会让整个请求失败。传统接口你只需要处理一次外部依赖Agent 要处理五六次。状态管理混乱Agent 必须记住“用户要什么、我已经查了什么、下一步该做什么”。这个状态如果落在进程内存里重启就丢如果和其他请求共享一个存储又会串。多数半成品 Agent 项目的崩溃源头都在这里。工具调用失控Agent 把用户意图转换成工具入参时什么奇葩参数都可能出现。我见过 Agent 把日期传成负数、把金额枚举传成自由文本、用 GET 请求去跑批量删除。模型只顾着让参数“看起来对”不会管业务约束。1.2 从崩溃现场反推出的三层防线在动手做任何优化之前我建议先把你自己的 Agent 跑一轮“故障演练”人为给上游接口加延迟、加错误返回、让两个请求同时跑同一个任务观察它在什么环节最先崩。我第一次做这个演练时10 个请求里大概有 6 个失败而且失败点各不相同。那种感觉不是“让 AI 干活”是“我在给随机事件写作文”。拆开这些失败现场我发现它们其实都落在三层问题上执行层工具调用没有超时、没有重试外部一抖动就整体失败。状态层Agent 的“记忆”没有和请求生命周期绑定也没有持久化的保存和恢复。接口层没有限流、没有幂等、没有对模型输出的结构性校验。后续所有稳定性改造本质都是围绕这三层做文章。这也就是 Harness 工程要覆盖的地盘。2. 把“套上马具”讲清楚Harness 的每一层到底在管什么2.1 用马具来理解你无法控制马的想法但可以控制马的行动边界“Harness”这个词原意是马具——缰绳、挽具、嚼子那一套东西。你控制不了一匹马此刻想往哪跑但通过马具你至少能限制它的路线、控制它的速度、在它失控时把它拉停。Agent 工程里的 Harness 就是干这个的模型内部怎么推理、怎么“想”你控制不了也不该控制但模型外部怎么约束输入、怎么解析输出、怎么限制工具权限、怎么恢复现场这些纯粹是工程题。这和传统的“Prompt 工程”是互补的。Prompt 是告诉模型“你该怎么做”Harness 是强制“它只能这么做”。提示词写得再好也只是概率上的引导Harness 里的 schema 校验、白名单、幂等控制则是确定性的兜底。我见过很多团队只在提示词上卷把“请你务必只返回 JSON”写满了屏幕结果模型偶发输出 Markdown 包裹整个分支就崩了。正确做法是在提示词之外加一层输出解析和强制重试让非确定性永远触碰不到核心业务逻辑。2.2 Harness 的职责边界该拿掉的、该加固的对一个实际可用的 Agent 来说Harness 至少包含这四块输入治理Input Control清洗用户消息超长截断敏感词过滤把显式的指令注入和提示词注入挡在进入模型之前。输出治理Output Control强制结构化输出校验字段、类型、取值范围解析失败时自动重试或向用户澄清而不是把模型原始输出直接丢给下游。工具治理Tool Control给每个工具定义清晰的 JSON Schema限制工具调用的目标范围做好限流、超时、重试必要时做人工审批闸口。状态与兜底State RecoveryAgent 执行到一半崩了能从最近的 checkpoint 恢复重复发起的请求有幂等键兜底执行步数超过阈值就强制中止。这四块听起来像常识但实际项目里最容易漏的是第五件事——可观测性。没有观测你就没法知道 Agent 每一步决策的依据是什么出了事只能抓瞎。这块我会在后面单独展开。2.3 一个可以被直接落地的 Harness 分层我目前项目中用到的分层结构大致是这样的┌───────────────────────────────────┐ │ 接入层FastAPI 路由 / WebSocket │ ├───────────────────────────────────┤ │ 编排层LangGraph 状态机 │ ├───────────────────────────────────┤ │ 治理层输出解析 / 工具调用护栏 │ ├───────────────────────────────────┤ │ 模型层OpenAI 兼容接口 │ └───────────────────────────────────┘治理层是后来加上去的也是改动最频繁的地方。实际开发中你会发现模型推理链路可以保持不变但治理规则会随着业务上线不断加码某个工具调用失败率高了就补一个重试策略某个字段频繁被模型填错就加一步校验某类请求容易死循环就给一个最大步数限制。Harness 不是一次性做出来的是随着事故记录一层一层加固出来的。3. 状态设计是稳定性的地基LangGraph 里最容易被忽视的细节3.1 State 的追加式更新与消息合并如果你用 LangGraph 编排 Agent状态State设计是第一道也是最容易被忽略的防线。LangGraph 的核心模型是状态图每个节点接收 State、修改 State、传给下一个节点。State 的设计方式直接决定了你的 Agent 会不会出现覆盖丢失。一个常见错误是把用户消息、中间推理、工具结果全塞在一个字段里然后用赋值方式整体替换。这会导致并发分支执行时后完成的节点覆盖先完成的节点结果。正确做法是使用 LangGraph 内置的合并策略。我目前的写法大致是from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): messages: Annotated[list, add_messages] # 消息自动追加 input: str plan: list[str] tool_results: Annotated[dict, operator.setitem] # 工具结果按键写入 completed: bool注意messages用add_messages它会在每次节点返回时把新消息追加到列表而不是覆盖旧消息。tool_results则用operator.setitem相当于在同一个字典上做按键更新。这两个合并器分别解决了“对话历史的增量累积”和“工具结果的并发写入”问题。如果你在 LangGraph 里写过日志一定会遇到多次节点返回同一批消息导致重复追加的情况这就需要节点内部做幂等或者让模型调用的入参显式截取最新的一部分消息。3.2 不要让 Agent 亲手管理自己的进度Checkpoint 与恢复Agent 执行到第三步挂了是重头开始跑还是从第三步继续在很多入门项目里答案是前者——因为压根没做持久化。但在生产环境长任务重跑一次的成本极高尤其是已经调用了付费 API 或者外部系统已经执行了不可逆操作的情况。LangGraph 的原生Checkpointer值得好好利用。它的作用简单说就是把每一步执行完的 State 快照存到持久化存储我用的 PostgreSQL请求中断后可以从最近的一个快照恢复执行。from langgraph.checkpoint.postgres import PostgresSaver checkpoint PostgresSaver.from_conn_string(dsn) graph workflow.compile(checkpointercheckpoint) # 每次请求带着同一个 thread_idLangGraph 会自动恢复该会话的最近状态 config {configurable: {thread_id: request_id}}这里有个关键点thread_id就是 Agent 的“会话身份”。所有针对同一个会话的请求必须串行处理不同thread_id之间可以随意并行。如果这个约束没守住两条并发请求同时执行同一个会话checkpoint 会互相覆盖出现比“不持久化”更可怕的混乱。所以我在接入层做了一件事同一个thread_id的请求走同一个队列靠 Redis 分布式锁保证同一时刻只有一个执行体在写这个线程的状态。3.3 分支合并时的“幽灵状态”LangGraph 支持并行分支执行比如同时查天气和查机票。这里有个特别隐蔽的坑分支合并时如果两个分支都对同一个 State 字段做赋值型更新后完成的那个分支会覆盖先完成的。你说“我两个分支写的是不同字段啊”但由于你用的是整体赋值后执行的节点会把整个 State 换成它自己看到的那一份另一个分支的写入就凭空消失了。这就是为什么上面那段示例里我把tool_results设计成按 key 更新的字典。但即使这样分支合并时依然要在父节点里做显式 mergedef merge_results(state: AgentState) - AgentState: # 将子分支返回的 partial state 合并进主 state for key, value in state.get(sub_results, {}).items(): state[tool_results][key] value return state我吃过这个亏之后总结了一条经验State 更新尽量设计成“增量写入”而不是“整体替换”。这不仅是 LangGraph 的最佳实践也是所有有状态 Agent 系统的通用原则。4. 扛并发不是只加容器链路长、模型延迟大调度层才是关键4.1 Agent 和普通接口的并发差异一次请求 N 次回源很多团队看 Agent 服务吞吐不行第一反应是横向扩容。扩完之后发现数据库连接打满了、下游 API 被限流了、模型服务排队了而 CPU 可能还在闲置。原因很简单Agent 的一次用户请求内部可能包含 4 到 7 次模型调用每次模型调用耗时 1 到 5 秒不等。这和传统接口“一次请求一次响应”完全不是一回事。所以接并发之前先算一笔账。假设你的模型调用平均 3 秒单 Agent 任务需要 4 次调用那么完成一个任务需要 12 秒模型占用时间。一台部署了 8 个 worker 的实例理论并发上限大约就是8 * (12 / 平均用户思考时间)。如果不做任何隔离几个慢任务就能把所有 worker 占满后面所有请求开始排队。4.2 同一个会话串行不同会话并行并发设计的分界线在会话级别。前面提到同一个thread_id的上下文是有状态且互相依赖的两条消息同时进来后一条的推理不能基于前一条的最新状态的话就会出逻辑混乱。所以跨thread_id随意并行这是吞吐量的来源。同thread_id严格串行必要时用分布式锁拦在入口。我在 FastAPI 里给会话加锁的方式很简单async def handle_message(thread_id: str, message: str): lock_key flock:thread:{thread_id} async with redis.lock(lock_key, timeout60): result await agent_executor(thread_id, message) return result这里有个注意点锁的 timeout 必须大于单条消息的最长可接受执行时间否则锁自动过期后第二个请求进来了状态就打架。我一开始把 timeout 设为 15 秒结果有一次 Agent 跑了一个 40 秒的查询任务锁提前释放第二个请求进来直接读到半截状态事故现场的日志那叫一个精彩。4.3 限流、幂等、重试顺序和位置比数量更重要关于限流我踩过的一个坑是在 Agent 入口统一限流。看起来公平但实际上慢任务占用的资源远大于快任务只看请求数限流是隔靴搔痒。后来我改成两段限流入口限流按thread_id数量限防止一个用户开几十个会话把服务打爆。模型调用限流在调用模型的地方用滑动窗口限流保护你接的模型服务或 API Key 不被触发限流上限。重试的位置同样关键。一个常见的错误是把重试放在 Agent 整体调用上——一旦失败就整体重跑。这在无状态接口里没问题但 Agent 已经调用过的外部工具可能会被重复执行重复扣费、重复发消息、重复改配置。正确做法是对工具调用做幂等重试给每个工具调用加request_id参数下游靠这个 ID 去重。对模型调用做简单重试模型层的超时重试是安全的它本身没有副作用。对整体流程做 Checkpoint 恢复整体失败可以从最后成功的一步恢复而不是从头再来。4.4 缓存比你想的更有价值同样的用户问题Agent 是不是每次都傻乎乎地重新查一遍我在项目里加了一层“语义缓存”把用户问题的向量表示存起来相似度达到一定阈值时直接返回上次结果。这个做法不仅能省模型调用费还能显著降低下游系统压力。对稳定性来说每次省掉的模型调用都意味着少一次失败风险。不过语义缓存也有坑它对动态数据不友好。比如用户问“我的订单发货了吗”如果直接命中缓存返回昨天的结果问题就大了。我的处理方式是只对“知识查询型”工具开启缓存对“业务状态查询型”工具一律绕过。5. 可观测性与可复现出了问题你能在十分钟内定位5.1 没有 trace 的 Agent 等于盲飞Agent 比传统服务更需要可观测性因为它每走一步都牵扯一次模型决策。用户抱怨“你给我的答案是错的”时你需要能回答模型当时看到了什么、它调了哪些工具、每个工具返回了什么、它最后依据什么生成结论。这五个问题任何一个答不上来排查就是大海捞针。我最开始做 Agent 时只在出问题的地方打了一条 error 日志。后来发现毫无用处——模型调用本身大概率没报错它就是顺着错误信息推理到了错误结论。这种“过程错误”必须通过完整链路回放才能发现。5.2 该打点的地方和该记录的内容我目前项目里强制记录的明细包括每轮模型调用的入参messages 长度、system prompt 版本模型返回的完整内容包括 token 用量每次工具调用的名称、入参、出参每个节点执行耗时State 变更增量用户请求的request_id贯穿全程落到实现上就是在 LangGraph 的每个节点包装一层打点中间件把所有明细写到日志系统再在日志系统里启用 trace 关联。你看到的理想结果是这样一条链request_id8f3a2... user_message帮我查下北京明天的天气 request_id8f3a2... nodeplanner model_call1 usage1280 tokens cost0.0032 request_id8f3a2... nodetool_call toolweather_api args{city:北京,date:2025-06-01} request_id8f3a2... nodeweather_api result{status:200, temp:22} request_id8f3a2... noderesponder model_call2 usage856 tokens有了这条链大部分问题可以一眼定位是模型规划错了、工具参数错了、还是上游接口返回错了。我建议不管项目多小这一步都不要省。它花的时间不多省下的排查时间却是几十倍。5.3 把坏案例变成回归集Eval 是稳定性的远期护城河可观测性的最终目的不只是排障而是积累“坏案例集”。我现在的做法是凡是线上出了事故的对话都用脚本扒下来脱敏后存成测试用例。每次改完模型配置、提示词、Harness 规则都在这个集合上跑一遍回归。谁也不敢保证改动一定提升稳定性但至少能证明没有让以前修好的问题复发。这个集合不需要很大质量取胜。我跑了大概 400 个来自线上事故的用例每次回归成本大概十几分钟换来的是改代码时的安全感。有了回归集之后我可以放心地做 A/B 实验同一个 Agent 用旧 Harness 规则和新 Harness 规则各跑一遍对比工具调用成功率、重试次数、回复的字段完整度用数据说话而不是靠感觉。6. 我在 FastAPI LangGraph 落地时真正踩过的坑6.1 长任务与短请求的冲突别让 HTTP 连接等到底Agent 任务有时会跑得非常久而 HTTP 客户端通常有 30 秒、60 秒的读取超时。等到客户端断开连接Agent 任务还在后台执行用户看到的是“请求失败”但操作其实已经发生了——这是最尴尬的状态。我的处理方式是把 API 拆成两类短路径走同步返回长任务直接用任务系统启动再给用户一个轮询接口。FastAPI 本身支持BackgroundTasks但更稳妥的方案是把任务提交到队列系统让 worker 执行。LangGraph 的 checkpoint 在这种架构下成了刚需——任务在任意时刻被中断都能恢复。6.2 工具调用并发的超时控制LangGraph 里可以配置多个工具并行执行但每个工具的超时如果不单独设置一个慢工具就能拖住整个分支。我在封装工具时统一加上了超时和结果兜底from langchain_core.tools import tool tool def query_weather(city: str, date: str) - dict: 查询指定城市指定日期的天气信息。 try: resp requests.get(..., timeout3) return resp.json() except Exception: return {_error: weather service timeout}注意我把异常兜底成一个包含_error字段的 dict 返回而不是直接抛出异常。LangGraph 的执行不太喜欢节点抛异常一旦抛出整个图可能直接走到错误分支。兜底返回值的好处是让模型自己决定拿到错误信息之后是重试、还是换一个工具、还是如实告诉用户“服务暂时不可用”。这比硬邦邦的异常中断要自然得多。6.3 结构化输出返回“合法但无效”内容时的兜底用模型输出 JSON 再解析是 Agent 最常见的翻车点之一。即使你用了with_structured_output它也只保证“格式合法”不保证“内容合理”。我遇到过模型把日期填成 2024-02-30、把数字金额填成负数、把状态填成一个不在枚举值里的字符串。这些都能通过 JSON Schema 校验但业务上完全不可用。所以我在 Harness 里加了一层业务校验器Validator专门写业务规则def validate_tool_args(tool_name: str, args: dict): errors [] if tool_name create_order: if args.get(amount, 0) 0: errors.append(amount must be positive) if not re.match(r\d{4}-\d{2}-\d{2}, args.get(delivery_date, )): errors.append(delivery_date format error) if errors: # 把错误反馈给模型让它重新生成参数 return {acceptable: False, feedback: errors} return {acceptable: True, feedback: None}这个 Validator 的反馈信息是接回模型做二次修正的告诉模型“你的参数不合法因为 xxx请重新生成”。这样做之后工具调用失败率从接近 20% 降到了 2% 以下。有个细节很重要二次修正时给模型看用户原始诉求 第一次的坏参数 错误原因不要只给错误原因否则模型容易“忘记”用户真实意图。6.4 流式输出时不要丢掉结构化信息很多 Agent 应用为了体验好用流式输出逐字吐出模型生成的内容。这会导致日志里只有碎片文本看不到完整工具调用链。我的做法是流式输出只用于展示模型对用户说话的部分工具调用的元数据走旁路日志两部分靠request_id关联。这样用户看到的体验是流式的后端排障时依然能拿到结构化全链路。6.5 LangChain 调用模型时的 token 管理长对话场景下消息列表如果不截断迟早会爆掉上下文窗口。LangChain 有内置的trim_messages但直接用它容易把系统提示词剪掉。我后来自己写了一个精简函数固定保留 system 提示和最近两轮对话把更早的消息压缩成一两句话的摘要。这算是个土办法但非常有效——既保证上下文不膨胀又尽量保留关键信息。总 token 下降后响应延迟和失败率都跟着下降了。7. 把稳定性做成一种开发习惯而不是救火动作到这里你会发现Harness 工程的所有手段——输出校验、checkpoint、幂等、限流、可观测性、回归集——单独拿出来都不是什么新技术组合起来却会产生质变。它不会让 Agent 变得更“聪明”但会让 Agent 变得“可合作”该做成的事一定能做成不该做的事一定拦得住出错了十分钟内能找出原因。我个人在实际项目里最深的体会是稳定性的建设应该跟功能建设同步进行而不是等出了问题再回头补。每加一个新工具顺手就加超时、鉴权、入参校验每写一个新节点顺手就加打点每次事故结束后顺手补一条回归用例。这些“顺手”看起来拖慢了开发速度但在线上事故和半夜排查面前都是性价比极高的投资。如果你现在正卡在“Demo 跑得通、上线就崩”的阶段我建议你先别急着调提示词按优先级做三件事第一给所有外部调用加上超时和兜底返回值第二把 Agent 的完整执行状态落进持久化存储让任务可断点恢复第三把每次请求的全链路 trace 打出来。这三件事做完你大概率会发现原来 80% 的“不稳定”根本不是模型不行而是工程上少了约束。最后说一句Agent 开发这件事聪明会给你上限稳定才保你下限。Harness 工程的方法论会随着技术栈演进而变化但“把非确定性关进笼子里”这个思路值得贯穿始终。