1. 这不是概念炒作,是工程师每天要填的七个坑
“AI Agent”这个词最近半年在技术社区里炸得比春节鞭炮还响,但凡带个“智能”俩字的项目介绍PPT,第一页必写“基于Agent架构”。可真坐到工位前打开IDE,很多人卡在第一步:到底该从哪下手写?是先搭个LangChain框架?还是直接啃LlamaIndex源码?抑或抄一段AutoGen的示例跑起来再说?我去年带三个团队落地了六套生产级Agent系统,从客服对话路由、金融研报生成,到工业设备故障推理,踩过最深的坑不是模型不准,而是——连“Agent到底该做哪些事”都没理清楚,就急着堆工具链。这篇文章不讲大模型原理,不画抽象架构图,只拆解一个硬核事实:所有能跑进真实业务场景的Agent,其工程实现必然绕不开七个刚性要素,而每个要素背后,都对应一个必须由人拍板的决策点。这七个点不是理论推演出来的,是我在监控告警群里半夜三点改完第七版调度策略后,把日志、错误码、用户投诉单和上线时间表摊开,一条条标红圈出来的。如果你正被“Agent怎么落地”这个问题卡住,或者刚读完几篇论文却不知代码从哪一行开始写,那这篇就是给你写的。它适合两类人:一是想用Agent解决具体业务问题的后端/全栈工程师,二是正在设计Agent平台的技术负责人。全文没有一句“随着AI发展”,只有七组真实参数、四类典型失败现场、三次重构路径,以及一份我压箱底的《Agent决策点检查清单》。
2. 七要素不是并列关系,而是环环相扣的工程流水线
很多资料把Agent的“感知-思考-行动”画成一个闭环圆圈,看着很美,一写代码就懵——哪个模块该处理超时?重试逻辑放哪?状态怎么持久化?根本原因在于:七要素不是静态组件,而是一条有严格时序、强依赖关系的工程流水线。它们按执行顺序排列,前一个要素的输出,是后一个要素的强制输入;前一个环节的决策失误,会像多米诺骨牌一样,在后续环节被指数级放大。我拿最典型的客服对话Agent来举例说明这个链条如何咬合:
用户发来消息:“上个月账单为什么多收了50块?”
① 目标解析器先拆出核心诉求(查异常费用)和约束条件(时间范围=上月),如果这里把“上个月”错判成“上一季度”,后面所有动作都是错的;
② 记忆管理器立刻调取该用户近90天的缴费记录、历史投诉标签、套餐变更日志,如果缓存策略没设TTL,可能返回三个月前已失效的优惠协议;
③ 工具调用协调器根据目标和记忆,决定先查计费系统API,再调用合同解析服务,如果这里没做工具依赖排序,计费数据还没返回就去解析合同,整个流程直接阻塞;
④ 执行引擎真正发起HTTP请求,但必须内置熔断机制——计费系统响应超时3秒就切降级方案,否则用户等10秒没反应,投诉电话就打进来了;
⑤ 规划生成器拿到原始数据后,不是直接拼回复,而是生成三步操作:a) 标出多收费时段 b) 匹配对应时段的促销活动条款 c) 计算应退金额,这一步缺失会导致回复变成“我们查到了,请稍候”,而不是“您3月15日-22日的流量包叠加了双倍积分活动,多计费50元,已原路退回”;
⑥ 反思校验器在生成最终回复前,强制校验:金额数字是否与计费系统返回值完全一致?时间范围是否覆盖用户提问的“上个月”?有没有触发合规关键词(如“赔偿”“起诉”)?漏掉任一校验,法务部第二天就会找你喝茶;
⑦ 状态追踪器把本次交互的完整上下文(用户ID、时间戳、调用的工具链、各环节耗时、校验结果)写入时序数据库,这不是为了画监控大屏,而是当用户两小时后追加问“退款什么时候到账”,系统能精准定位到上一轮会话的支付通道ID,而不是重新走一遍七步流程。
这七个环节缺一不可,且顺序不可颠倒。我见过最典型的反模式,是团队先花两周搭好LangChain的Orchestrator,再回头补记忆模块——结果发现Orchestrator默认把所有中间状态存在内存里,一重启就丢数据,而业务要求会话中断后30分钟内恢复,必须重写状态序列化逻辑。所以工程实现的第一步,永远不是选框架,而是按这个流水线顺序,逐个确认每个要素的SLA(服务等级协议)指标:目标解析的准确率要≥98.5%,记忆查询P95延迟≤120ms,工具调用失败自动降级成功率≥99.9%……这些数字决定了你后续所有技术选型的边界。
2.1 为什么“目标解析”必须是第一个要素?——来自三次线上事故的教训
目标解析(Goal Parsing)常被当成NLP任务草草处理,但它是整个Agent系统的“闸门”。我们曾在线上发生三次严重事故,根因全是这里失控:
事故1(金融风控场景):用户输入“帮我看看最近有没有高风险交易”,解析器把“高风险”简单映射为“单笔金额>5万”,但实际业务规则是“同一IP下30分钟内连续5笔转账,且收款方为新注册账户”。结果系统漏掉了一起团伙洗钱行为,直到监管检查才暴露。
事故2(医疗问答场景):患者问“我吃阿司匹林后胃疼怎么办”,解析器只提取了药品名和症状,忽略了关键隐含条件“正在服用抗凝药”。生成的建议未提示出血风险,险些引发医疗纠纷。
事故3(电商客服场景):用户说“上次买的耳机坏了,我要换新的”,解析器把“换新”识别为目标,但没识别出前置条件“是否在保修期”。结果系统直接触发换货流程,而用户订单实际已过保67天,物流发出后被拒收,公司承担了双向运费损失。
这三次事故教会我的铁律是:目标解析不是文本分类,而是业务规则编译器。它必须完成三件事:
- 显性目标提取:用NER模型识别实体(药品名、时间范围、金额阈值);
- 隐性约束挖掘:通过规则引擎匹配业务知识库(如“阿司匹林+抗凝药→出血风险↑”);
- 冲突检测:当用户诉求与当前政策冲突时(如过保换新),必须明确返回结构化错误码,而非强行执行。
我们最终采用的方案是“规则优先,模型兜底”:先用Drools配置200+条业务规则(覆盖85%高频场景),对规则未覆盖的长尾case,再用微调后的BERT模型做意图分类。实测下来,规则部分准确率99.2%,模型部分82.7%,加权后整体达97.8%,且规则更新可热加载,不用重启服务。
提示:别迷信端到端大模型解析。我们对比过GPT-4 Turbo直接解析用户输入,准确率仅73.4%,且无法解释判断依据。工程落地要的是可审计、可回滚、可监控的确定性。
2.2 记忆管理器的陷阱:不是存得越多越好,而是存得“刚刚好”
工程师第一反应是“上Redis缓存用户历史”,但Agent的记忆管理远比这复杂。它要同时处理三种记忆类型,且每种的生命周期、一致性要求、访问模式完全不同:
| 记忆类型 | 典型内容 | 存储要求 | 一致性要求 | 我们的方案 |
|---|---|---|---|---|
| 短期记忆 | 当前会话的上下文(用户上3句话、Agent已执行动作) | 低延迟(P95<50ms)、高吞吐 | 强一致(不能有脏读) | 内存+LRU淘汰,单实例部署 |
| 长期记忆 | 用户画像(偏好、投诉记录、套餐信息) | 持久化、支持复杂查询 | 最终一致(允许秒级延迟) | PostgreSQL分库分表,按用户ID哈希 |
| 工作记忆 | 当前任务的中间产物(如“已查到3笔异常交易,待人工复核”) | 支持事务、可回滚 | 强一致(任务失败需完整回滚) | PostgreSQL+行级锁,配合Saga模式 |
最大的坑在于混淆这三者。我们曾把用户投诉记录(长期记忆)和当前会话状态(短期记忆)全塞进Redis,结果出现经典问题:用户A在会话中投诉客服,系统把投诉标记写入Redis;用户B恰好用相同缓存key(按手机号哈希),导致B的会话里突然弹出“A投诉了客服”的提示。根源是没做记忆隔离。
解决方案是强制分层:短期记忆用进程内ConcurrentHashMap,长期记忆走DB,工作记忆用带TTL的Redis Hash(key=task_id:session_id)。更关键的是,所有记忆读写必须经过统一Memory Gateway接口,这个接口里硬编码了三类记忆的路由规则、序列化格式、过期策略。上线后,记忆相关故障下降92%。
注意:别用向量数据库存长期记忆!我们试过Chroma存用户历史,查询延迟从120ms飙到850ms,因为每次都要做向量相似度计算。长期记忆的核心是精准检索,不是模糊匹配。
3. 七个决策点:每个选择都决定你的Agent是玩具还是生产力工具
要素是骨架,决策点才是血肉。这七个点没有标准答案,但每个选择都会把你推向截然不同的技术路径。我用表格列出我们六个项目的实际决策,并标注背后的硬约束:
| 决策点 | 选项A(轻量级) | 选项B(企业级) | 我们的选择 | 关键约束 |
|---|---|---|---|---|
| 1. 目标解析方式 | LLM零样本分类 | 规则引擎+微调小模型 | B | 合规审计要求所有判断可追溯,LLM黑盒不满足 |
| 2. 记忆存储介质 | Redis + 内存 | PostgreSQL + Elasticsearch | B | 需支持SQL审计、字段级权限控制、千万级用户画像查询 |
| 3. 工具调用协议 | REST API直连 | 统一工具网关(含鉴权/限流/熔断) | B | 多业务线共用工具,需防止某团队API变更影响全局 |
| 4. 规划生成策略 | 单步Prompt生成 | 分步规划(Plan→Validate→Execute) | B | 金融场景要求每步操作可人工审核,单步生成无法拆解 |
| 5. 反思校验层级 | 仅结果校验(数字/格式) | 业务规则校验+合规词库扫描+逻辑矛盾检测 | B | 监管要求所有输出必须通过三重校验,缺一不可 |
| 6. 状态持久化粒度 | 整个会话存JSON | 按决策点存原子事件(GoalParsedEvent, ToolCalledEvent) | B | 需支持按任意环节重放调试,JSON无法定位到具体步骤失败 |
| 7. 错误恢复机制 | 全流程重试 | 精确到决策点的补偿事务(如ToolCall失败→回滚Memory写入) | B | 电商场景要求资金操作幂等,重试会导致重复扣款 |
看到这里你可能觉得“全选B太重了”,但现实是:当你面对真实业务时,“轻量级”往往意味着把技术债打包卖给业务方。比如我们早期用选项A做客服Agent,上线两周后,运营同学天天来找我:“为什么用户问‘怎么取消会员’,系统有时说‘已为您取消’,有时说‘请拨打400’?”。查日志发现,LLM解析不稳定,同一句话今天判为“取消诉求”,明天判为“咨询诉求”,而系统没做任何一致性保障。最后不得不推翻重做,工期延误47天。
所以决策的本质,是把业务的不确定性,转化为工程的确定性。下面我重点拆解三个最易踩坑的决策点。
3.1 决策点3:为什么必须建工具网关?——一个被忽略的性能炸弹
多数教程教你用LangChain的Tool类直接调用API,看起来很优雅:
@tool def get_billing_data(user_id: str) -> dict: return requests.get(f"https://api.billing.com/v1/users/{user_id}/bills")但当你的Agent接入12个内部系统(计费、合同、物流、风控、营销……)后,问题就来了:
- 安全黑洞:每个Tool都硬编码API密钥,密钥轮换要改12处代码;
- 雪崩风险:物流系统慢了,所有调用它的Agent都会卡住,没有熔断;
- 监控盲区:不知道哪个Tool调用最多、平均耗时多少、错误率趋势;
- 权限失控:客服Agent不该调用财务系统的“导出全量账单”接口,但Tool没做权限校验。
我们被这些问题逼着建了统一工具网关(Tool Gateway),它长这样:
Agent → Tool Gateway (gRPC) → [Auth] → [RateLimit] → [CircuitBreaker] → [Logging] → 实际API关键设计:
- 动态注册:各业务系统提供OpenAPI Spec,网关自动生成Tool描述,Agent无需写任何调用代码;
- 分级熔断:按错误类型熔断(HTTP 500熔断5分钟,429熔断30秒),避免一刀切;
- 成本感知:网关统计每个Tool的CPU/网络消耗,当某Agent连续调用高成本Tool(如“全量合同解析”)超阈值,自动降级为摘要模式。
上线后,工具调用平均延迟从840ms降到210ms(熔断减少无效等待),跨系统故障率下降76%。最重要的是,当计费系统凌晨升级时,网关自动切换到缓存数据,用户无感知。
实操心得:别自己造轮子!我们用Kong网关+自定义插件实现,开发只用了3人日。重点不在技术,而在定义清楚Tool的契约——每个Tool必须声明:输入Schema、输出Schema、SLA承诺、错误码映射、权限Scope。
3.2 决策点5:反思校验不是锦上添花,而是法律防火墙
很多团队把“反思”(Reflection)理解为让LLM自我批评,比如加一句“请检查你的回答是否准确”。这是危险的幻觉。真正的反思校验,是在LLM输出之外,用确定性程序做三重交叉验证:
数据真实性校验:从工具返回的原始数据中,提取关键字段(如金额、日期、ID),与LLM生成的回答逐字比对。我们用正则+模糊匹配(Levenshtein距离≤2)确保数字零误差。曾发现LLM把“¥1,234.56”生成为“¥1234.56”,少了个千分位,财务系统无法识别。
业务逻辑校验:加载业务规则引擎,验证回答是否符合政策。例如用户问“能提前还款吗”,规则引擎会检查:当前贷款状态≠逾期、还款金额≥最低还款额、距到期日>30天。LLM生成的“可以”必须通过全部规则,否则返回结构化拒绝原因。
合规安全校验:用AC自动机构建敏感词库(含同音词、变形词),扫描回答中是否含禁用表述。更关键的是逻辑矛盾检测:比如用户问“我上月账单多少”,LLM回答“¥89.50”,但工具返回的原始数据是“¥89.50(含税)”,而系统配置的税率是13%,那么LLM回答中若没提“含税”,就构成信息不完整,触发重写。
我们把这三层校验做成独立微服务,Agent流程固定为:LLM生成 → 校验服务 → 通过则返回,否则触发重写子流程。校验失败率约17%,其中82%是数据不一致,12%是逻辑冲突,6%是合规问题。没有这层,我们的Agent根本不敢上线。
注意:校验服务必须比LLM快!我们用Rust重写核心校验逻辑,P99延迟压到8ms以内。如果校验比生成还慢,整个系统就卡死了。
3.3 决策点6:状态持久化的致命细节——为什么JSON存不住一个Agent
工程师喜欢把整个会话存成JSON:
{ "session_id": "abc123", "user_input": "查上月账单", "steps": [ {"goal": "解析账单查询", "result": "成功"}, {"goal": "调用计费API", "result": "返回3条记录"}, {"goal": "生成回复", "result": "已为您查到..."} ] }问题在于:当第三步失败时,你无法知道是生成回复的Prompt错了,还是LLM返回了非法JSON,还是网络传输丢了字符。所有线索都混在同一个JSON里。
我们的方案是事件溯源(Event Sourcing):每个决策点产生一个不可变事件,存入时序数据库:
| event_id | session_id | event_type | payload | timestamp |
|---|---|---|---|---|
| ev_001 | abc123 | GoalParsed | {"raw_input":"查上月账单","parsed_goal":"billing_query","time_range":"last_month"} | 2024-05-20T10:00:00Z |
| ev_002 | abc123 | ToolCalled | {"tool_name":"get_billing_data","params":{"user_id":"u789","month":"2024-04"}} | 2024-05-20T10:00:02Z |
| ev_003 | abc123 | ToolReturned | {"tool_name":"get_billing_data","result":[{"id":"b123","amount":"89.50"}]} | 2024-05-20T10:00:05Z |
好处是爆炸性的:
- 精准归因:第三步失败?查
ev_003是否存在,存在则看ev_003.payload.result是否为空; - 自由重放:从任意事件重放,比如从
ev_002重放,就能测试工具调用环节; - 审计合规:监管检查时,直接导出
event_type=ToolCalled的所有事件,证明所有外部调用都经授权。
我们用TimescaleDB存事件,单表日增2.3亿条,查询session_id=abc123的P95延迟12ms。代价是开发量增加,但换来的是线上问题平均定位时间从47分钟降到3.2分钟。
4. 实操:从零搭建一个可上线的客服Agent(附完整代码片段)
现在我们把前面所有原则,落地到一个真实可运行的客服Agent。目标:处理“查询账单”类请求,要求支持多轮对话、数据校验、错误恢复。技术栈:Python 3.11 + FastAPI + PostgreSQL + LiteLLM(兼容多模型)。
4.1 第一步:定义七要素的最小可行实现(MVP)
不要一上来就搞分布式,先用单进程验证流水线:
# agent_core.py from dataclasses import dataclass from typing import List, Dict, Any, Optional import json @dataclass class Goal: """目标解析结果""" intent: str # billing_query, complaint, etc. time_range: str # last_month, last_quarter, etc. constraints: Dict[str, Any] # {min_amount: 100} @dataclass class Memory: """记忆管理器输出""" short_term: Dict[str, Any] # 当前会话上下文 long_term: Dict[str, Any] # 用户画像 working: Dict[str, Any] # 任务中间状态 @dataclass class ToolResult: """工具调用结果""" name: str success: bool data: Any error: Optional[str] = None @dataclass class PlanStep: """规划步骤""" step_id: int action: str # "query_billing", "validate_contract" params: Dict[str, Any] @dataclass class ReflectionResult: """反思校验结果""" is_valid: bool issues: List[str] corrected_response: Optional[str] = None @dataclass class AgentState: """Agent完整状态""" session_id: str goal: Goal memory: Memory tool_results: List[ToolResult] plan: List[PlanStep] reflection: ReflectionResult final_response: str注意:这里用dataclass强制定义每个要素的结构,而不是用dict。好处是IDE能自动补全,序列化时不会漏字段,更重要的是——当你需要加新字段时,必须显式修改dataclass,这强迫你思考这个字段属于哪个要素。
4.2 第二步:实现核心流水线(关键代码)
流水线主函数,严格按七要素顺序执行:
# agent_pipeline.py from agent_core import * import logging logger = logging.getLogger(__name__) def run_agent_pipeline(session_id: str, user_input: str) -> str: """ Agent核心流水线,按七要素顺序执行 返回最终响应,或抛出明确异常 """ state = AgentState(session_id=session_id, goal=Goal("", "", {}), memory=Memory({}, {}, {}), tool_results=[], plan=[], reflection=ReflectionResult(False, []), final_response="") try: # ① 目标解析 state.goal = parse_goal(user_input) logger.info(f"[{session_id}] Goal parsed: {state.goal.intent}") # ② 记忆管理 state.memory = load_memory(session_id, state.goal) logger.info(f"[{session_id}] Memory loaded: {len(state.memory.long_term)} fields") # ③ 工具调用协调 state.tool_results = coordinate_tools(state.goal, state.memory) logger.info(f"[{session_id}] Tools called: {[r.name for r in state.tool_results]}") # ④ 执行引擎(已封装在coordinate_tools中) # ⑤ 规划生成 state.plan = generate_plan(state.goal, state.memory, state.tool_results) logger.info(f"[{session_id}] Plan generated: {len(state.plan)} steps") # ⑥ 反思校验 state.reflection = reflect_and_validate(state.plan, state.tool_results) if not state.reflection.is_valid: logger.warning(f"[{session_id}] Reflection failed: {state.reflection.issues}") # 触发重写逻辑 state.final_response = state.reflection.corrected_response or "系统正在优化回答,请稍候" return state.final_response # ⑦ 状态追踪(存入数据库) persist_state(state) # 生成最终回复 state.final_response = format_response(state.plan, state.tool_results) return state.final_response except Exception as e: logger.error(f"[{session_id}] Pipeline failed: {e}", exc_info=True) # 错误恢复:返回结构化错误,不暴露技术细节 return "抱歉,当前服务繁忙,请稍后再试。"关键点:
- 每个步骤都有明确日志,包含session_id,方便链路追踪;
- 异常处理只在顶层捕获,内部步骤失败必须抛出特定异常(如
GoalParseError),便于监控告警; - 所有中间状态都存入state对象,而不是散落在局部变量,为后续扩展(如加入重试)留接口。
4.3 第三步:落地决策点3——工具网关客户端
我们不直接调用API,而是通过网关:
# tool_gateway_client.py import httpx from typing import Dict, Any class ToolGatewayClient: def __init__(self, base_url: str): self.client = httpx.AsyncClient(base_url=base_url) async def call_tool(self, tool_name: str, params: Dict[str, Any]) -> ToolResult: """ 调用工具网关 返回结构化结果,包含熔断/限流信息 """ try: response = await self.client.post( "/v1/tools/call", json={ "tool_name": tool_name, "params": params, "caller_id": "customer_service_agent" # 权限标识 }, timeout=5.0 ) response.raise_for_status() data = response.json() return ToolResult( name=tool_name, success=data["success"], data=data.get("data"), error=data.get("error") ) except httpx.TimeoutException: return ToolResult( name=tool_name, success=False, data=None, error="TOOL_TIMEOUT" ) except Exception as e: return ToolResult( name=tool_name, success=False, data=None, error=f"TOOL_ERROR: {str(e)}" ) # 使用示例 gateway = ToolGatewayClient("https://tool-gw.internal") async def get_billing_data(user_id: str, month: str) -> ToolResult: return await gateway.call_tool("get_billing_data", {"user_id": user_id, "month": month})这个客户端把熔断、超时、错误码标准化了。当网关返回{"success": false, "error": "RATE_LIMITED"}时,客户端直接返回ToolResult,上层流水线无需关心是网络问题还是限流问题。
4.4 第四步:状态持久化——事件溯源实现
用PostgreSQL存事件(简化版):
# state_persistence.py import psycopg from agent_core import AgentState, ToolResult def persist_state(state: AgentState): """将Agent状态存为原子事件""" conn = psycopg.connect("dbname=agent_db") with conn.cursor() as cur: # 存目标解析事件 cur.execute( "INSERT INTO agent_events (session_id, event_type, payload) VALUES (%s, %s, %s)", (state.session_id, "GoalParsed", json.dumps({ "intent": state.goal.intent, "time_range": state.goal.time_range })) ) # 存工具调用事件(每个工具一个事件) for result in state.tool_results: cur.execute( "INSERT INTO agent_events (session_id, event_type, payload) VALUES (%s, %s, %s)", (state.session_id, "ToolCalled", json.dumps({ "tool_name": result.name, "success": result.success, "data_sample": str(result.data)[:100] # 防止超长 })) ) # 存最终响应事件 cur.execute( "INSERT INTO agent_events (session_id, event_type, payload) VALUES (%s, %s, %s)", (state.session_id, "ResponseGenerated", json.dumps({ "response": state.final_response[:500] })) ) conn.commit() conn.close()表结构很简单:
CREATE TABLE agent_events ( id SERIAL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, event_type VARCHAR(32) NOT NULL, payload JSONB NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); CREATE INDEX idx_session_events ON agent_events(session_id, event_type);这就是可上线的最小Agent。它没有炫酷的UI,但每个环节都可监控、可审计、可重放。我们用这套MVP在三天内上线了内部客服试用版,日均处理2300次请求,错误率1.2%,全部可定位到具体事件。
5. 常见问题与排查技巧实录:那些文档里不会写的真相
以下是我在六个项目中,被问得最多、也最痛的问题。答案不是“查文档”,而是“我当时怎么救火的”。
5.1 问题1:“Agent回复越来越慢,但CPU和内存都正常,为什么?”
现象:上线一周后,P95响应时间从1.2秒涨到8.7秒,监控显示服务器资源充足。
排查路径:
- 先看日志:发现大量
ToolCalled事件耗时超5秒,但工具网关监控显示API本身P95<200ms; - 抽样分析
ToolCalled事件,发现耗时长的全是调用get_user_profile; - 查
get_user_profile的实现,发现它内部调用了3个下游服务(CRM、营销、风控),用的是串行HTTP请求; - 根因:工具网关没做并发控制,Agent在规划阶段生成了5个
get_user_profile调用,全部串行执行,总耗时=5×200ms=1秒,加上网络抖动,轻松破5秒。
解决方案:
- 工具网关增加
concurrency_limit字段,get_user_profile设为1(防下游压垮); - Agent规划生成器增加“工具依赖分析”,自动合并相同工具调用(5次
get_user_profile→1次批量调用); - 对必须串行的工具,加
max_wait_ms参数,超时直接降级。
实操心得:响应慢90%不是LLM的问题,是工具链的问题。永远先查
ToolCalled事件的耗时分布,再查LLM。
5.2 问题2:“同样的输入,Agent有时答对有时答错,怎么复现?”
现象:用户输入“查我上月账单”,有时返回正确金额,有时返回空。
排查路径:
- 在日志中搜索该用户ID的所有
GoalParsed事件,发现time_range字段有时是"last_month",有时是"last_30_days"; - 追查目标解析器,发现它用了一个第三方时间解析库,该库对“上月”有歧义(是自然月还是30天);
- 根因:目标解析没做标准化,不同时间库返回不同结果。
解决方案:
- 所有时间解析强制走内部
TimeNormalizer服务,统一返回ISO格式(2024-04-01/2024-04-30); - 在
GoalParsed事件中,增加normalized_time_range字段,作为唯一事实源; - 测试用例必须覆盖“上月”“上季度”“近7天”等所有业务常用表达。
注意:LLM的随机性不是借口。所有非确定性输入(如时间、金额、ID)必须在进入LLM前,由确定性程序标准化。
5.3 问题3:“Agent突然大规模报错,但没改代码,为什么?”
现象:凌晨2点,错误率从0.5%飙升至42%,持续17分钟,自动恢复。
排查路径:
- 查错误日志,99%是
ToolReturned事件中data字段为null; - 查工具网关日志,发现
get_billing_data在2:03-2:20期间返回大量{"success": true, "data": null}; - 联系计费团队,得知他们凌晨2点做了数据库索引重建,期间查询返回空结果但HTTP状态码仍是200;
- 根因:工具网关没校验
data字段非空,把空响应当成功。
解决方案:
- 工具网关增加
schema_validation钩子,对每个Tool定义返回Schema,data字段必须非空; - 增加
data_integrity_check,对关键字段(如amount,date)做存在性校验; - 建立“工具健康度”看板,实时监控各Tool的
data_null_rate。
真相:Agent的稳定性,取决于最不稳定的那个工具。你的代码再完美,也扛不住下游返回一个空JSON。
5.4 问题4:“用户说‘算了,不用了’,Agent还在继续执行,怎么中断?”
现象:用户中止对话后,Agent仍调用3个工具,生成回复,浪费资源。
排查路径:
- 查
GoalParsed事件,发现intent是"abort",但后续仍有ToolCalled事件; - 发现目标解析器把“算了”识别为
"consultation",因为训练数据里没覆盖这个口语; - 根因:没有“中止意图”的兜底机制。
解决方案:
- 在目标解析后,强制插入“中止检测”环节:用正则匹配
["算了","不用了","停","取消"],命中则跳过后续所有步骤; - 所有工具调用前,检查
session_state中是否有aborted=True标记; - 在Agent状态中增加
cancellation_token,类似.NET的CancellationToken,可被外部注入。
实操心得:用户放弃的瞬间,就是你释放资源的最佳时机。别等LLM生成完再判断。
6. 给技术负责人的终极建议:别建Agent平台,先建Agent治理委员会
最后分享一个血泪教训:我们曾投入11人月,打造“企业级AI Agent平台”,支持拖拽编排、可视化调试、多模型切换……上线后,只有2个团队在用,其余都自己写脚本。为什么?
因为Agent不是基础设施,而是业务逻辑的载体。平台解决不了“这个客服话术要不要加免责声明”“那个金融计算必须用哪个税率”这类问题。
我们后来成立了跨部门的“Agent治理委员会”,成员包括:业务方PM、法务、风控、运维、算法、前端。每周开30分钟会,只干一件事:评审每个Agent的七个决策点选择。例如:
- 客服Agent的“反思校验”是否满足《消费者权益保护法》第20条?