1. 玩具级Agent与企业级Agent的分水岭在哪
这两年我手上过的Agent项目不少,最早自己拿开源模型加一个ReAct循环就拼出“能查天气、能算数学题”的Demo,发给朋友看大家都觉得挺酷。但真到了要把它放进业务系统里,让非技术同事每天使用,甚至要对接公司内部的数据和权限体系时,问题一个接一个地冒出来。核心痛点其实不在模型本身,而在工程化程度。
1.1 先定义清楚:什么才算玩具级
玩具级Agent有几个典型特征:
- 只有一个主循环,把大模型的输出直接当成可执行指令,没有结构化的工具协议
- 上下文处理靠简单拼接,把历史消息一股脑塞进Prompt,没有分级记忆,没有裁剪策略
- 工具调用靠正则或者字符串匹配,模型一旦换个说法就识别不了
- 没有统一的错误处理,外部接口超时、返回非JSON、参数缺失这类高频问题全靠运气
- 没有状态持久化,服务一重启,会话就断了,用户要重新描述一遍需求
- 没有观测手段,模型为什么这么决策、调用了哪些工具、耗时多少,全都不透明
这些问题的共性,是“能用”和“能稳定地用”之间的巨大差距。我见过一个Demo在演示时表现完美,一旦并发上来,或者用户问题表述变长,就开始随机报错,最后只能靠重启恢复。这类项目拿去面对真实用户,基本属于给自己挖坑。
1.2 企业级Agent的本质变化是什么
企业级与玩具级的区别,我认为可以用一句话概括:从“让模型自由发挥”变成“让模型在规则内自由发挥”。前者追求上限,后者追求下限。业务系统最不能接受的是行为不可控,所以企业级Agent一定要解决以下能力:
- 确定性:同一个问题在相同条件下,行为要可预期,不能今天这样明天那样
- 可维护性:代码结构清晰,换一个模型、加一个新工具,不需要推倒重来
- 可观测性:每一次决策、每一步工具调用都有日志和链路追踪,出问题能回溯
- 安全性:权限校验、数据隔离、敏感信息过滤必须在前置层解决,而不是依赖模型自觉
- 成本可控:Token消耗要能被计算、被限制、被优化,否则一个月账单会吓到财务
我在做架构拆分时,会刻意把“模型能力”和“工程能力”剥离开。模型负责理解和生成,工程负责流程、约束和兜底。这样即使日后换更强的模型,或者同时接入多家模型,对业务层来说只是配置切换,代码不用大改。
2. 代码架构的分层设计与核心模块拆解
一个能跑到生产环境的Agent,代码组织上我建议至少拆成五层:接入层、编排层、工具层、记忆层、观测层。每一层职责单一,层与层之间通过接口通信,这样后续扩展和替换成本都低。
2.1 基础骨架:从一个可扩展的Agent类说起
先看一个核心骨架,这个骨架几乎适用于所有类型的Agent应用。我用了抽象基类来定义统一协议,后续任何具体Agent都基于它扩展。
from abc import ABC, abstractmethod from typing import Any, AsyncIterator class BaseAgent(ABC): """Agent 抽象基类,定义统一的执行协议。""" def __init__(self, model_client, memory, tool_registry, observer): self.model_client = model_client self.memory = memory self.tool_registry = tool_registry self.observer = observer @abstractmethod async def run(self, user_input: str, **kwargs) -> Any: """执行一次完整的 Agent 任务。""" pass @abstractmethod async def stream_run(self, user_input: str, **kwargs) -> AsyncIterator[str]: """流式执行,适合需要实时输出的场景。""" pass这个骨架看起来简单,但有几个设计点值得展开。
第一,把所有依赖都通过构造函数注入,而不是在内部直接实例化。这样写单元测试时非常舒服,每个组件都可以替换成Mock。第二,run和stream_run分成两个方法,是因为生产环境里很多场景要边生成边输出,比如客服对话、报告生成,用户等不了完整结果。
第三,observer这个参数很多初学者会忽视。它负责收集指标和日志,几乎不在业务逻辑里起作用,但线上排查问题时,没有它就像蒙着眼睛修机器。
2.2 工具层设计:让模型安全地调用外部能力
工具层是整个Agent架构里最容易出问题的地方。玩具级实现通常是这样:
if "天气" in user_input: result = get_weather(city) prompt += f"天气结果是:{result}"这个方案在演示时没问题,但模型一旦回答“帮我看看北京明天会不会下雨”,这种硬编码匹配就失效了。正确做法是先定义一套JSON Schema协议,模型通过结构化参数来请求工具调用。
下面是我常用的工具注册接口示例:
from pydantic import BaseModel from typing import Dict, Any class ToolSpec(BaseModel): name: str description: str parameters: Dict[str, Any] class BaseTool(ABC): spec: ToolSpec @abstractmethod async def execute(self, **kwargs) -> Any: pass实际项目中,每个工具都要实现execute方法,并声明自己的spec。比如一个获取用户信息的工具,它的spec里会说明需要user_id参数,参数类型是字符串,并写清楚“根据用户ID查询用户基本信息,包含姓名、部门、职级”。模型读到这个描述后,才知道什么时候该调用它,以及该传什么参数。
工具层的核心原则是:参数的校验和转换必须在工具层完成,不要信任模型的输出。模型返回{"user_id": 123},但业务系统里用户ID可能是字符串,这一层要做类型转换、边界检查、权限校验。如果直接把模型给的参数透传给内部服务,很容易出安全问题。
2.3 编排层:从单步调用到多步任务规划
编排层是Agent的大脑,决定了“先做什么、再做什么、做不了怎么办”。玩具级通常是单轮的“思考→调用→回答”,企业级则需要支持多轮规划、中途改计划、任务拆分子任务。
一个比较成熟的模式是Plan-and-Execute模式,分两个阶段:
- Planner阶段:根据用户目标拆解出任务清单
- Executor阶段:逐个执行任务,并根据执行结果动态调整后续步骤
我用一个简单例子说明这个流程。假设用户提出“帮我整理上季度销售数据,并生成一份PPT大纲”,Planner会拆解出两步:第一步,查询上季度销售数据;第二步,基于数据生成大纲。Executor先调用数据查询工具,拿到数据后再去生成大纲。如果第一步查不到数据,Executor会把错误信息反馈给Planner,由Planner重新规划,比如改为“提示用户补充查询条件”。
这个流程里的关键点是“任务状态管理”。我习惯用状态机来管理每一步的状态:pending、running、succeeded、failed、skipped。每一步执行完成后,状态都会持久化,这样系统中断后可以断点续跑,而不是从头再来。
3. 从“能跑”到“稳跑”:记忆、上下文与状态管理
很多Agent在演示时表现良好,一旦进入多轮对话就“失忆”,本质是上下文管理策略不到位。企业级Agent必须区分短期工作记忆和长期持久记忆,并对Token消耗做精细控制。
3.1 上下文窗口不是无限大的:记忆三级策略
我建议把记忆拆成三个级别:
- 短期记忆:当前会话内的消息,直接拼进上下文
- 工作记忆:当前任务相关的中间结果和关键结论
- 长期记忆:跨会话的用户偏好、历史摘要、重要事实
短期记忆很好理解,就是普通的多轮对话历史。工作记忆稍微复杂一些,比如用户中途问了一句“上次那个方案报价是多少”,Agent需要能从之前的中间状态里找回答案,而不是重新推演一遍。长期记忆则需要在会话结束时做一次“总结压缩”,把本次对话的关键信息提炼成结构化摘要存入向量库或数据库。
我举一个处理长期记忆的简化实现思路:
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage def summarize_conversation(messages: list) -> str: # 这里调用模型对本次会话做压缩总结 pass def build_prompt_with_memory(user_input: str, short_memory: list, long_memory: str) -> list: system_msg = SystemMessage( content=( "你是企业智能助手。请结合用户长期偏好和当前会话上下文回答问题。\n" f"用户长期画像:{long_memory}" ) ) return [system_msg] + short_memory + [HumanMessage(content=user_input)]关键设计在于:长期记忆的容量和更新频率都要控制,不能每轮都写入,否则很快就会覆盖掉早期真正重要的信息。我一般只在两类时机触发写入:一是用户明确表达了偏好(比如“以后汇报都用表格”),二是任务完成且产生了稳定的结论(比如“客户最终报价是15万”)。
3.2 状态持久化:会话中断也能恢复
生产环境里,服务重启是常态,用户网络断开也是常态。如果Agent的状态只放在进程内存里,会话一断就全部丢失,这在企业场景里是不可接受的。
我常用的做法是把会话状态序列化后存到Redis,键名设计为agent:{session_id}:state,值是一个包含以下内容的JSON:
{ "history": [], "current_task_plan": [], "tool_results_cache": {}, "retry_count": 0, "last_updated_at": 1699000000, "user_profile_snapshot": "" }每次Agent执行一步,就把整个状态对象刷新到Redis,同时设置一个合理的过期时间,比如30分钟。这样即使进程崩溃,新实例启动后也能通过session_id把状态拉回来,继续执行未完成的任务。
同时这里要提一个容易被忽略的问题:状态恢复时,上下文历史会原样带回给模型,但模型上下文窗口有限,如果历史已经很长,恢复时要做截断或摘要。我通常的做法是,保留最近5轮完整消息,再往前的历史全部被摘要替代。这个策略基本不会影响对话质量,但Token消耗能降一半以上。
4. 企业级Agent的可靠性保障:容错、重试与可观测性
架构搭完了,接下来聊生产环境中最磨人的部分。模型输出有随机性,外部接口不稳定,这些都是常态。只有把容错和观测做扎实了,Agent才能7x24小时稳定服务。
4.1 模型输出不可控,怎么兜底
大模型偶尔会输出非严格JSON的格式,这是Agent工具调用场景里最大的坑之一。明明给了JSON Schema,模型就是会在JSON前后加个```json标记,或者把布尔值写成True而不是true,工具层一解析就报错。
我的兜底策略分三层:
- 第一层:正常解析,直接
json.loads - 第二层:修复常见格式错误,比如去掉Markdown代码块标记,修正单引号为双引号
- 第三层:调用模型自身做“格式修复”,把原始输出和解析错误信息交回给模型,让它重新生成
需要说明的是,第三层会额外消耗Token,所以必须加一个最大重试次数限制。我通常设为2次,如果2次后仍然解析失败,就终止本轮执行,并给用户一个“暂时无法处理,请稍后再试”的统一兜底话术。宁可直接失败,也不能让错误结果进入业务数据流。
4.2 外部工具超时与重试策略
外部API超时是个躲不掉的坑。我在某个项目中对接内部系统接口时,对方服务偶发5秒以上延迟,Agent在等待期间完全卡死,用户体验非常差。
我的解决方案分两步。第一步是给每个工具调用设置超时时间,所有工具必须支持asyncio.wait_for,超过阈值就直接报错。
import asyncio async def execute_with_timeout(tool: BaseTool, timeout_seconds: int = 20, **kwargs): try: return await asyncio.wait_for(tool.execute(**kwargs), timeout=timeout_seconds) except asyncio.TimeoutError: raise ToolTimeoutError(f"工具 {tool.spec.name} 执行超时")第二步是区分“可重试错误”和“不可重试错误”。超时、服务暂时不可用属于可重试;参数错误、没有权限属于不可重试。对可重试错误采用指数退避策略:第一次失败等2秒重试,第二次等4秒,第三次等8秒,最多3次。超过重试次数后,整个任务的规划层要能根据这个失败信息重新调整任务路径。
我还习惯在工具执行前做一次“预检”,快速判断参数是否大致合理。比如查询日期范围,如果开始日期晚于结束日期,直接在工具层拦截,返回结构性错误,而不去实际调用业务接口。这样既省了API开销,又降低了错误传播链的长度。
4.3 给Agent装上仪表盘:日志、链路追踪与指标
企业级系统如果没有观测能力,出了问题根本没法定位。我给Agent设计了三类观测数据:
- 日志:记录每一次用户输入、每一次模型响应、每一次工具调用的入参和出参
- 链路追踪:串联一个任务的完整执行链路,从接收请求到最终生成回答,每个环节都有耗时采集
- 指标:任务成功率、平均响应时间、Token消耗量、工具调用次数分布
其中一个核心指标我建议单独监控:模型输出的JSON解析失败率。这个指标能直观反映当前Prompt模板和模型能力的匹配度,如果某个版本上线后这个数字突然升高,基本可以定位是Prompt改动或者模型被换掉了。
可观测性的落地,我推荐“结构化工件”的思路,每次决策和调用都输出标准化JSON事件流,写入统一日志平台。原始日志长这样:
{ "event": "tool_call_start", "agent_id": "agent-sales-01", "session_id": "abc123", "tool_name": "query_sales_data", "input_args": {"start_date": "2024-01-01", "end_date": "2024-03-31"}, "timestamp": 1710000000 }之后再用日志平台的查询能力做聚合分析和告警。所有这类操作,日常排查全靠这里的数据,项目上线前一定先把这个基础打好。
5. 常见问题与排查技巧实录
最后整理一下我在实战中遇到最多的坑,以及对应的排查思路。这些内容很零散,但每一条几乎都是从线上故障和反复调试中换来的,值得先收藏。
5.1 上下文爆炸
现象是响应越来越慢,Token消耗越来越高,用户问了一句很早就问过的事情,Agent却“忘了”。原因是短期记忆没有裁剪,所有历史提示词都重新传给模型。
排查方法是查看每次请求的Token用量日志,重点关注input_tokens的增长曲线。如果每轮都是线性增长,基本就是上下文管理出了问题。解决方案是引入摘要压缩策略,只保留最近几轮原始消息,更早的历史转为摘要保存。
这里有一个容易踩的误区:摘要不能每次对话都重新生成,否则成本会翻倍。我建议只在短期记忆累计达到一定阈值(比如超过10轮)时触发一次摘要生成,然后清空已被摘要覆盖的原始消息。
5.2 工具调用无限循环
有些场景下,Agent在解决一个任务时会反复调用同一个工具,比如查询登录用户信息,每次拿到结果都不会推进步骤,反而继续调用。本质上是在规划阶段死循环了。
我给出的排查思路是:必查工具调用记录里有没有频繁重复的调用序列。如果是,就要在编排层加上“同工具同参数调用次数限制”逻辑,比如一个工具在同一个任务里最多调用3次,超出就强制终止并进入兜底流程。
同时,可以在Prompt里明确约束:若某个步骤连续执行失败,立即向用户说明情况,不要自行重试。模型其实很容易被“不能回答就换个方式再试”的指令带偏,Promot里的约束语句要具体到动作级别。
5.3 输出内容“幻觉”成真实数据
模型在总结业务数据时,偶尔会编造一些看起来合理但没有依据的数字。这在企业级场景里是致命的,尤其在财务、经营分析类Agent里。我的做法是强制“引用溯源”,在最终的输出模板里加约束:所有数值类回答必须附上数据来源工具名和查询参数。做不到就直接拒绝回答,也不要硬编。
实操层面,在工序里增加一个“可信校验”步骤:模型生成回答后,写一个校验器扫描关键数字,并和工具返回的原始数据做比对,不一致就重新生成一次。这个校验器不调用模型,只做规则匹配,成本极低,但效果很明显。
5.4 流式输出与工具调用的用户感知冲突
用户在对话时希望打字就出字,但Agent在调用工具时,好几秒没有文字输出,体验上像“卡住了”。我在项目里是这样处理的:启用流式输出,但是在流式接口的实现里,工具调用阶段会同步推送一个自定义的中间事件,前端收到这个事件后显示“正在查询数据...”之类的提示,让用户知道任务在推进中,而不是死机。
实现思路是在stream_run方法里区分两类事件类型:
{ "type": "text_delta", "content": "正在生成回答..." }{ "type": "tool_call", "tool_name": "query_sales_data", "status": "running" }前端按事件类型分别处理。这一改动需要的代码量不大,但对整体体验的提升非常明显。
5.5 多模型切换时的行为不一致
同一个Agent在模型A上表现良好,换到模型B后工具调用格式就变了,甚至完全不按Prompt要求走。这是目前最多团队遇到且最头疼的兼容性问题。排查的核心思路是:对模型的依赖不能建立在“隐形默契”上。
所有关键指令必须在系统级完成约束,而不是靠模型“理解能力”。工具调用的JSON格式解析、参数校验、错误重试,这些逻辑都要放在工具层做适配,模型输出最多只能作为“候选结果”。不同模型之间确实存在能力差异,这也是我推荐把模型接入做成可配置开关的原因。切模型不是改一行API密钥的事,要做一次全量回归测试。
6. 写在最后的几条实操心得
这套从玩具级到企业级的改造方案,本质上不是某一个惊天动地的设计,而是把软件工程的基本原则老老实实搬进了Agent开发里。对我个人而言,最大的改变是心态上的:以前看到模型输出一个惊艳的结果会兴奋,现在更关注它在100次调用里有几次不稳定,以及不稳定的那几次能不能被系统自动兜住。
如果你正在做自己的Agent项目,我的建议是不要上来就追求复杂架构,先把最简单的主循环跑通,然后按这个清单逐步加能力:工具协议标准化、错误处理兜底、上下文管理、状态持久化、观测日志。每加一层,就用真实业务压力测试一轮,不要等到全堆完了再回头查问题,那将会是一场灾难。
另外,代码架构再完善,也不要忽略Prompt本身的调校。很多诡异的模型行为,最后排查下来都是Prompt表述有歧义。好的Agent提示词要尽量短,但要覆盖足够多的边界情况,这需要反复打磨。建议专门留一个调试集,里面放大量真实用户问题,每次改Prompt或改架构,都拿这套测试集跑一遍回归,这样模型升级或代码重构时才不会慌。