如果你今天还在用给LLM加几个工具调用来描述你的Agent那我们需要认真谈谈了。2026年的AI工程现实是绝大多数Agent项目死在了从Demo到生产的路上。不是因为模型不够强而是因为工程没跟上。本文会系统梳理Agent工程化的核心路径从基础架构到可观测性从错误处理到成本控制给你一份可直接参考的生产级指南。为什么Demo能跑、生产就崩先说痛点。一个典型的Agent Demo通常长这样一个System Prompt几个工具函数一个while循环完事。在测试集上跑得飞起一到真实用户手里就各种翻车。原因其实很明确输入分布偏移。测试时你知道用户会问什么生产环境用户问的是你完全没预料到的东西。Agent的规划能力在面对奇怪输入时会快速退化。工具调用失败的传导效应。一个工具返回了意外格式Agent不知道怎么处理然后开始幻觉然后调用下一个错误的工具雪球越滚越大。上下文窗口的隐形炸弹。多轮对话跑久了工具调用的结果积累在上下文里Token越来越贵模型注意力越来越分散最后答非所问。缺少可观测性。你不知道Agent在哪个步骤出问题不知道为什么它做了那个决策出了问题只能瞪着日志发呆。生产级Agent的架构基础明确区分协调层和执行层这是最关键的架构决策。协调层Orchestrator负责规划、决策、工具选择执行层Executor负责具体的工具调用和结果处理。两者不要混在一起。classAgentOrchestrator:协调层只负责规划和决策def__init__(self,llm_client,tool_registry):self.llmllm_client self.toolstool_registry self.plan_history[]asyncdefplan_next_action(self,goal:str,context:dict)-Action:根据目标和当前上下文规划下一步行动promptself._build_planning_prompt(goal,context)responseawaitself.llm.complete(prompt)returnself._parse_action(response)def_build_planning_prompt(self,goal:str,context:dict)-str:# 包含目标、已完成步骤、可用工具、当前状态available_toolsself.tools.list_tools()completed_stepscontext.get(completed_steps,[])returnf 你是一个任务规划器。你的目标是{goal}已完成的步骤{self._format_steps(completed_steps)}可用工具{self._format_tools(available_tools)}基于当前状态请规划下一步行动。如果任务已完成返回 DONE。 以JSON格式返回{{action: tool_name, params: {{...}}, reasoning: ...}} classToolExecutor:执行层只负责工具调用和错误处理def__init__(self,tools:dict):self.toolstools self.retry_configRetryConfig(max_retries3,backoff_factor2)asyncdefexecute(self,action:Action)-ToolResult:toolself.tools.get(action.name)ifnottool:returnToolResult.error(fUnknown tool:{action.name})forattemptinrange(self.retry_config.max_retries):try:resultawaittool.call(action.params)returnToolResult.success(result)exceptToolTimeout:ifattemptself.retry_config.max_retries-1:returnToolResult.error(Tool timeout after retries)awaitasyncio.sleep(self.retry_config.backoff_factor**attempt)exceptToolErrorase:returnToolResult.error(str(e))# 不重试业务错误 这种分离让你可以独立优化每一层也让测试变得更简单。### 状态管理不只是记录对话历史生产级Agent需要显式的状态管理而不是把所有东西都塞进对话历史。 pythondataclassclassAgentState:session_id:strgoal:strstatus:Literal[planning,executing,waiting,done,failed]# 执行进度completed_steps:list[StepResult]field(default_factorylist)current_step:Optional[Step]None# 上下文严格控制大小working_memory:dictfield(default_factorydict)# 当前任务相关的临时数据# 统计token_usage:int0tool_call_count:int0start_time:floatfield(default_factorytime.time)propertydefelapsed_seconds(self)-float:returntime.time()-self.start_timedefadd_step_result(self,result:StepResult):self.completed_steps.append(result)# 自动摘要只保留最近N步的完整结果更早的压缩成摘要iflen(self.completed_steps)10:self._compress_early_steps()def_compress_early_steps(self):把早期步骤压缩成摘要避免上下文无限增长early_stepsself.completed_steps[:5]summaryf已完成{len(early_steps)}个早期步骤.join(s.summaryforsinearly_steps)self.completed_steps[StepResult.summary(summary)]self.completed_steps[5:]### 工具契约防御性设计每个工具都应该有明确的输入输出契约并且在边界处做好防御 pythonfrompydanticimportBaseModel,validatorclassSearchToolInput(BaseModel):query:strmax_results:int5validator(query)defquery_not_empty(cls,v):ifnotv.strip():raiseValueError(搜索词不能为空)returnv.strip()[:500]# 截断过长的查询validator(max_results)defresults_in_range(cls,v):returnmax(1,min(v,20))# 强制限制范围classSearchToolOutput(BaseModel):results:list[SearchResult]total_found:intsearch_took_ms:int# 提供给Agent的结构化摘要defto_agent_context(self)-str:ifnotself.results:return搜索未找到相关结果returnf找到{self.total_found}条结果以下是前{len(self.results)}条\n\\n.join(f-{r.title}:{r.snippet}forrinself.results)## 可观测性让Agent不再是黑盒没有可观测性你就是在盲飞。生产级Agent必须能回答这几个问题-这次任务Agent做了哪些决策理由是什么--哪个步骤花了最多时间和Token--失败是在哪里发生的### Trace设计pythonimportuuidfromcontextlibimportasynccontextmanagerclassAgentTracer:def__init__(self,backend):# backend可以是Langfuse、自建系统等self.backendbackendasynccontextmanagerasyncdeftrace_session(self,session_id:str,goal:str):traceTrace(idsession_id,goalgoal,start_timetime.time())try:yieldtrace trace.statussuccessexceptExceptionase:trace.statusfailedtrace.errorstr(e)raisefinally:trace.end_timetime.time()awaitself.backend.save(trace)asynccontextmanagerasyncdeftrace_step(self,trace:Trace,step_name:str,**metadata):stepStep(idstr(uuid.uuid4()),namestep_name,metadatametadata,start_timetime.time())try:yieldstep step.statussuccessexceptExceptionase:step.statusfailedstep.errorstr(e)raisefinally:step.end_timetime.time()trace.add_step(step) 实际使用时每个规划决策和工具调用都包在trace里 pythonasyncwithtracer.trace_session(session_id,goal)astrace:whilenotdone:asyncwithtracer.trace_step(trace,planning,context_sizelen(state.completed_steps))asstep:actionawaitorchestrator.plan_next_action(goal,state)step.record_llm_call(tokensaction.tokens_used,modelaction.model)asyncwithtracer.trace_step(trace,ftool:{action.name},paramsaction.params)asstep:resultawaitexecutor.execute(action)step.record_result(result)## 成本控制让Agent可持续Token费用是Agent上生产后的第一个噩梦。几个实用策略**上下文压缩策略**。不要把每次工具调用的完整响应都放进上下文。设计一个摘要函数把长结果压缩成关键信息 pythonclassContextManager:MAX_CONTEXT_TOKENS8000# 为规划留出足够空间defbuild_context(self,state:AgentState)-str:context_parts[]# 目标始终保留context_parts.append(f目标{state.goal})# 已完成步骤最近3步保留完整更早的只保留摘要recentstate.completed_steps[-3:]earlierstate.completed_steps[:-3]ifearlier:summaries[s.summaryforsinearlier]context_parts.append(f早期步骤已摘要{; .join(summaries)})forstepinrecent:context_parts.append(f步骤{step.name}{step.result_text})context\n\n.join(context_parts)# Token超限时进一步压缩ifself._estimate_tokens(context)self.MAX_CONTEXT_TOKENS:contextself._emergency_compress(context)returncontext **工具调用缓存**。同样的工具调用不要重复执行 pythonclassCachedToolExecutor:def__init__(self,executor:ToolExecutor,cache_ttl:int300):self.executorexecutor self.cache{}self.cache_ttlcache_ttlasyncdefexecute(self,action:Action)-ToolResult:# 只缓存幂等的工具搜索、查询等不缓存写操作ifnotaction.is_cacheable:returnawaitself.executor.execute(action)cache_keyf{action.name}:{json.dumps(action.params,sort_keysTrue)}ifcache_keyinself.cache:entryself.cache[cache_key]iftime.time()-entry[time]self.cache_ttl:returnentry[result]resultawaitself.executor.execute(action)self.cache[cache_key]{result:result,time:time.time()}returnresult ## 错误处理给Agent一个降级策略Agent在生产环境会遇到各种意外情况必须为每种失败模式设计明确的处理策略 pythonclassAgentRunner:asyncdefrun(self,goal:str,session_id:str)-AgentResult:stateAgentState(session_idsession_id,goalgoal)asyncwithself.tracer.trace_session(session_id,goal):whileTrue:# 安全边界检查ifstate.tool_call_count50:returnAgentResult.failed(reasonexceeded_tool_limit,partial_resultself._extract_partial_result(state))ifstate.elapsed_seconds300:# 5分钟超时returnAgentResult.failed(reasontimeout,partial_resultself._extract_partial_result(state))try:actionawaitself.orchestrator.plan_next_action(goal,state)exceptLLMErrorase:# LLM调用失败等待后重试最多3次ifstate.llm_errors3:state.llm_errors1awaitasyncio.sleep(5)continuereturnAgentResult.failed(reasonllm_unavailable)ifaction.is_done:returnAgentResult.success(state)resultawaitself.executor.execute(action)# 工具失败通知Agent让它决策如何继续ifnotresult.ok:state.add_step_result(StepResult(stepaction,resultf工具调用失败{result.error}。请考虑替代方案。))else:state.add_step_result(StepResult(stepaction,resultresult.data))## 从这里开始的实践建议1.**先跑通最简单的任务建立基线**。不要一上来就构建通用Agent先找一个具体场景把它做到生产可靠。2.**从第一天就接入可观测性**。Langfuse是个不错的选择开源可自建。不要等到出问题再加。3.**设置硬性的安全边界**。工具调用上限、时间上限、Token上限每一个都要有不能让Agent无限跑。4.**测试要包含对抗性输入**。专门设计一批会让Agent迷惑的输入纳入你的回归测试集。5.**记录每一次生产失败**。建立一个失败案例库每次新的异常都要分析根因更新你的错误处理策略。 Agent工程化不是一天就能完成的事但架构方向对了后面每一步优化都会积累成真正的竞争壁垒。---*本文关键词Agent工程、生产级AI、LLM工具调用、可观测性、成本控制*