大部分 Agent 教程停在能跑通这一步定义一个工具写个 while 循环调用大模型解析输出结束。本地跑几次没问题就以为可以上线了。真放到生产环境问题会一个接一个冒出来模型偶尔返回一段无法解析的文本循环卡住不动某个工具超时把整个请求拖死用户重试导致同一个操作执行两次上下文越堆越长最后超出模型窗口账单在某个下午突然翻了好几倍。这些问题不是模型能力问题是工程设计问题。下面这 8 个机制是我认为从 Demo 跨到生产必须补齐的部分。代码用 Python 写尽量不依赖特定框架方便你移植到自己的技术栈里。涉及具体库版本的地方我会写清楚不确定的地方会明确说明。一、状态持久化别把 Agent 的命交给内存Demo 里最常见的写法是把对话历史和中间步骤放在一个 list 里进程一重启就全没了。生产环境里用户可能等 30 秒才回来继续对话也可能同时开三个会话甚至服务本身会滚动重启。所以第一步是把 Agent 的状态外置。状态的粒度可以粗可以细最粗的是只存消息历史最细的是把每一步的思考、工具调用、观察结果都存下来。我倾向于后者因为出问题时能回放。一个最简的状态结构大概是这样# Python 3.10fromdataclassesimportdataclass,field,asdictfromtypingimportAnyimportjsonimporttimedataclassclassAgentStep:step_id:inttype:str# thought | tool_call | tool_result | finalcontent:Any created_at:floatfield(default_factorytime.time)dataclassclassAgentState:session_id:strsteps:list[AgentStep]field(default_factorylist)status:strrunning# running | done | failed | cancelledcreated_at:floatfield(default_factorytime.time)defto_json(self)-str:returnjson.dumps(asdict(self),ensure_asciiFalse)classmethoddeffrom_json(cls,raw:str)-AgentState:datajson.loads(raw)data[steps][AgentStep(**s)forsindata[steps]]returncls(**data)存储层可以是 Redis、Postgres甚至本地 SQLite。选哪个取决于你的并发量和一致性要求。Redis 快但需要自己处理持久化策略Postgres 稳但写入延迟高一些。这点没有银弹按业务量来。【关键结论】状态外置之后Agent 就变成了无状态服务可以水平扩容也可以做断点续跑。这一步不做后面所有机制都会打折扣。二、循环与步数控制防止 Agent 陷入死循环Agent 的核心是一个循环模型输出 → 解析 → 如果有工具调用就执行 → 把结果塞回上下文 → 再问模型。这个循环必须有硬性退出条件否则模型可能反复调用同一个工具或者在一段推理里绕圈。我一般会同时设三层限制最大步数max_steps比如 15 步单次请求的总超时wall-clock timeout比如 90 秒重复调用检测同一个工具 同一组参数连续出现两次直接判定为循环前两个好理解第三个容易被忽略。举个场景模型调用search(天气)返回空它可能原封不动再调一次。加一个简单的哈希去重就能拦住importhashlibdeftool_call_signature(tool_name:str,args:dict)-str:rawf{tool_name}:{json.dumps(args,sort_keysTrue)}returnhashlib.md5(raw.encode()).hexdigest()classLoopGuard:def__init__(self,max_steps:int15,max_repeat:int2):self.max_stepsmax_steps self.max_repeatmax_repeat self.seen:dict[str,int]{}self.steps0defcheck_step(self):self.steps1ifself.stepsself.max_steps:raiseRuntimeError(f超过最大步数{self.max_steps})defcheck_tool_call(self,name:str,args:dict):sigtool_call_signature(name,args)self.seen[sig]self.seen.get(sig,0)1ifself.seen[sig]self.max_repeat:raiseRuntimeError(f检测到重复工具调用{name})这里max_repeat2的意思是允许模型重试一次第二次再出现相同调用就中断。设成 1 会太激进因为工具本身可能因为网络抖动失败一次重试是合理的。【踩坑提醒】步数上限不要设得太小。设成 5 步稍微复杂一点的任务就完不成设成 50 步出问题时你要等很久。经验值在 10~20 之间具体看任务复杂度。这个数字我没有严谨的基准测试支撑属于工程经验判断。三、工具调用容错模型输出的东西不一定合法模型返回的 JSON 经常有各种小毛病多一句解释、少一个引号、把null写成None、参数类型不对。如果直接json.loads一个字符错误就让整个请求失败。处理方式分两层。第一层是解析容错尽量从文本里抠出 JSON第二层是参数校验用 schema 检查类型和必填项。importjsonimportrefromtypingimportCallable JSON_PATTERNre.compile(r\{.*\},re.DOTALL)defsafe_parse_tool_call(text:str)-dict|None:从模型输出里尽量抠出工具调用 JSONtry:returnjson.loads(text)exceptjson.JSONDecodeError:passmatchJSON_PATTERN.search(text)ifnotmatch:returnNonetry:returnjson.loads(match.group())exceptjson.JSONDecodeError:returnNone参数校验用pydantic会比较省事v2 版本当前主流是 2.x的写法frompydanticimportBaseModel,ValidationErrorclassSearchArgs(BaseModel):query:strtop_k:int5defvalidate_args(raw:dict,schema:type[BaseModel])-BaseModel|None:try:returnschema(**raw)exceptValidationErrorase:# 把错误信息回传给模型让它自己改returnNone校验失败时不要直接报错终止。更好的做法是把错误信息作为一条 observation 塞回上下文让模型自己修正。这通常比抛异常更有效因为模型看到query 字段缺失往往能自己补上。【注意】不要无限让模型重试修正。一般给两次机会两次还不行就降级到兜底逻辑或直接返回失败。四、超时与取消一个慢工具能拖垮整条链路生产环境里任何一个外部调用都可能变慢搜索 API 抖动、数据库锁等待、第三方服务限流。如果 Agent 没有超时机制一个请求可能挂几分钟占着 worker 不放。Python 里做超时asyncio.wait_for是最直接的方式importasyncioasyncdefcall_tool_with_timeout(tool_fn:Callable,args:dict,timeout:float10.0,):try:returnawaitasyncio.wait_for(tool_fn(**args),timeouttimeout)exceptasyncio.TimeoutError:return{error:f工具执行超时{timeout}s}超时之后返回一个结构化的错误而不是抛异常这样 Agent 循环可以继续模型有机会换一个策略。取消cancellation是另一个维度。用户点了停止或者上游请求断开Agent 应该能收到信号并停止后续步骤。在 asyncio 里取消一个 task 会抛出CancelledError需要确保工具函数里没有忽略它的地方否则取消会失效。【关键结论】超时和取消是两个不同的东西。超时是等太久了主动放弃取消是外部要求停止。两者都要处理不能只做一个。五、可观测性出问题时你得知道发生了什么Agent 出问题最难受的地方是不知道它为什么这么做。它调用了三次搜索改了两次参数最后给了一个莫名其妙的答案。如果没有日志和追踪你只能靠猜。最小可用的可观测性包括三件事每一步的结构化日志步骤号、类型、耗时、输入输出摘要会话级别的 trace_id串起所有步骤关键指标步数分布、工具调用成功率、平均耗时、token 消耗importloggingimportuuid loggerlogging.getLogger(agent)deflog_step(trace_id:str,step:AgentStep,duration:float):logger.info(agent_step,extra{trace_id:trace_id,step_id:step.step_id,type:step.type,duration_ms:int(duration*1000),content_preview:str(step.content)[:200],},)content 只打前 200 个字符避免日志爆炸。完整的输入输出可以存到对象存储或专门的 trace 系统里。如果团队已经在用 OpenTelemetry可以直接把 Agent 的每一步做成 span和现有的后端链路打通。这点我没有在超大流量场景下验证过但在中小规模下是可行的。【踩坑提醒】不要在日志里打印完整的 prompt 和模型输出尤其是涉及用户数据的时候。脱敏和截断是必须的。六、幂等与重试用户重试不等于要执行两次用户看到请求超时很自然会点重试。如果 Agent 的操作有副作用下单、发邮件、写数据库重试就可能导致重复执行。解决思路是幂等键。客户端每次请求带一个request_id服务端记录已处理的 request_id 和对应结果。重试时如果发现这个 id 处理过直接返回缓存结果不再执行。classIdempotencyStore:def__init__(self):self._cache:dict[str,Any]{}defget(self,request_id:str)-Any|None:returnself._cache.get(request_id)defset(self,request_id:str,result:Any):self._cache[request_id]result生产环境里_cache要换成 Redis 之类的共享存储并且设置合理的过期时间。太短起不到作用太长会占内存。重试本身也要有策略指数退避、最大次数、只对可重试的错误重试。模型返回格式错误可以重试但用户余额不足这种业务错误重试没意义。七、上下文管理窗口不是无限的Agent 跑十几步之后上下文会变得很长历史消息、工具定义、每一步的观察结果全堆在里面。超出模型窗口就直接报错没超出也会让推理变慢、变贵。处理方式有几种我一般组合使用方案优点缺点适用场景滑动窗口实现简单可能丢关键早期信息短对话摘要压缩保留语义摘要本身有信息损失长对话关键信息提取精准保留需要额外规则或模型调用结构化任务向量检索召回容量大引入检索延迟和误差知识密集型对于 Agent 场景我倾向于保留系统提示 最近 N 步 对早期步骤做摘要。因为 Agent 的早期步骤往往是探索性的语义价值不如最近几步。defcompress_history(steps:list[AgentStep],keep_recent:int6)-list[AgentStep]:iflen(steps)keep_recent:returnsteps earlysteps[:-keep_recent]recentsteps[-keep_recent:]summaryAgentStep(step_id-1,typesummary,contentf前{len(early)}步的摘要...# 实际调用模型生成)return[summary]recent摘要生成本身要调用模型所以要么异步做要么在步骤数超过阈值时才触发避免每步都调。【关键结论】上下文管理不是删掉旧的这么简单关键是判断哪些信息对当前决策还有用。这一步做不好Agent 会在长任务里逐渐失忆。八、成本与权限边界别让 Agent 花光你的预算最后一个机制经常被放到最后才想但它其实应该在最开始就设计。成本方面需要限制的维度包括单次会话的最大 token 消耗、单个用户单位时间的调用次数、单个工具的调用频率。超出阈值时降级或拒绝而不是继续烧钱。权限方面Agent 能调用的工具应该是最小集合。一个只做问答的 Agent 不应该有写数据库的权限。工具的参数也要校验比如文件路径不能带..SQL 不能是DROP。classBudgetGuard:def__init__(self,max_tokens_per_session:int):self.max_tokensmax_tokens_per_session self.used0defconsume(self,tokens:int):self.usedtokensifself.usedself.max_tokens:raiseRuntimeError(会话 token 预算已耗尽)权限校验最好放在工具执行的入口而不是散落在各个工具函数里。这样新增工具时不容易漏掉。【注意】预算和权限的阈值不要拍脑袋定。先跑一段时间收集真实数据再根据 P95、P99 来设。设太紧会误伤正常请求设太松等于没设。这些机制怎么组合起来上面 8 个机制不是并列的它们之间有依赖关系。状态持久化是基础没有它取消、幂等、断点续跑都做不了。可观测性贯穿始终没有它其他机制出问题时你很难定位。成本和权限是边界决定了 Agent 能走多远。一个实际的 Agent 执行循环大概长这样asyncdefrun_agent(state:AgentState,guard:LoopGuard,budget:BudgetGuard):whilestate.statusrunning:guard.check_step()# 1. 压缩上下文contextcompress_history(state.steps)# 2. 调用模型responseawaitcall_llm(context)budget.consume(response.usage.total_tokens)# 3. 解析工具调用tool_callsafe_parse_tool_call(response.content)iftool_callisNone:state.statusdonebreak# 4. 重复检测guard.check_tool_call(tool_call[name],tool_call[args])# 5. 执行工具带超时resultawaitcall_tool_with_timeout(TOOLS[tool_call[name]],tool_call[args],timeout10)# 6. 记录状态state.steps.append(AgentStep(len(state.steps),tool_result,result))# 7. 持久化awaitsave_state(state)returnstate真实系统里还要加上异常处理、重试、权限校验、trace 上报但骨架就是这样。写在最后这 8 个机制没有哪个是高级技巧都是后端工程里的老思路超时、重试、幂等、限流、日志、状态外置。只是 Agent 这个形态把它们的必要性放大了因为循环是模型驱动的、输出是不确定的、工具是外部依赖。我的建议是不要一次全上。先做状态持久化和步数控制这两个投入小、收益直接。然后补可观测性因为后面调优全靠它。成本和权限放在上线前做避免出事故。剩下几个根据你的实际痛点逐步加。有一个问题我到现在也没有特别好的答案Agent 的失败该怎么定义。是没给出答案算失败还是给了错误答案算失败还是答案对了但过程花了 20 步也算失败这个定义直接影响你怎么设阈值、怎么评估。如果你有更清晰的思路欢迎交流。