智能体原型的交付边界
演示环境中的图片、网络和提示词通常经过控制,不能代表真实流量。多模态 Agent 进入灰度后,应重点验证解析失败、尾部延迟和工具调用的终止条件;文中的数值仅适合作为压测场景示例。
1. 演示会上完美的 Agent,上线首日丢包率和解析失败率达到了 18%
现场演示通常使用低延迟网络、固定分辨率图片和标准提示词;移动端真实流量则会引入网络波动、输入大小和内容质量的变化。
用户上传的照片千奇百怪。有人直接传了 12MB 的无压缩原图,造成网络传输层直接超时。有人上传了严重模糊的照片,引发大模型在识别阶段开始瞎猜,返回的 Tool Calling 参数里充斥着不合法的特殊字符,直接打爆了下游的 JSON 解析器。
更严重的问题发生在决策循环中。由于缺乏确定性的状态终结条件,当 API 返回“未找到记录”时,Agent 没有选择停止,而是开始在工具调用链路里无限循环。
2. 多模态输入流裁剪:把 10MB 的原始图片处理成确定性的特征图
多模态 Agent 性能崩塌的第一关在图片预处理阶段。直接把原始 Image 编码成 Base64 丢给 Vision Model 是典型的避重就轻做法。这不仅极度浪费带宽高,而且会拖慢模型的注意力机制计算速度。
我们在入口处加了一道多模态预处理流水线。任何传入的图片先经过轻量级的 OpenCV 算子,完成自动旋转纠偏、灰度化归一化以及动态尺寸缩放。
如果图片尺寸超过 1024x1024,流水线会自动执行长边等比例缩放,并提取图像的 ROI(感兴趣区域)。经过这一层拦截,单张图片的 Token 占用从平均 1600 降到了 350,模型理解的速度提升了接近三倍。
3. Tool Calling 格式修复代理:用 JSON Schema 正则防线截断非标输出
虽然最新的模型都宣称原生支持 Function Calling,但在生产高并发场景下,LLM 依然有概率输出破坏格式的 Markdown 代码块或带有尾随逗号的非法 JSON。
解决问题的思路不能寄希望于在 Prompt 里反复强调“请严格输出 JSON”,而应该在 Agent 内部挂载一个确定性的 JSON 校验与修复代理(Repair Proxy)。
当 LLM 吐出返回文本后,校验代理先执行快速正则抽取。如果json.loads失败,立刻自动抹除尾随逗号、补全缺失的双引号或剥离外层的 Markdown 标记。如果修复后依然无法通过 Schema 校验,直接触发自动重试,向 LLM 注入极简的错误反馈提示词。
4. 面向生产环境的 Agent 状态机与容错调度管理器代码
下面是一个完整的 Agent 状态机管理器,内置了最大轮次管制、JSON 自动修复代理以及工具调用的幂等性拦截。
import json import re from typing import Dict, Any, Callable, Tuple class AgentExecutionEngine: def __init__(self, max_steps: int = 4): self.max_steps = max_steps self.tools: Dict[str, Callable] = {} def register_tool(self, name: str, func: Callable): self.tools[name] = func def _sanitize_and_parse_json(self, raw_response: str) -> Tuple[bool, Dict[str, Any], str]: """ 自动剥离 Markdown 并修复常见 JSON 语法错误 """ if not raw_response: return False, {}, "Empty response" cleaned = raw_response.strip() # 剥离 ```json ... ``` 代码块 if "```" in cleaned: match = re.search(r'```(?:json)?\s*(.*?)\s*```', cleaned, re.DOTALL) if match: cleaned = match.group(1) # 剥离非 JSON 尾随字符 cleaned = re.sub(r',\s*([\]}])', r'\1', cleaned) try: data = json.loads(cleaned) return True, data, "" except json.JSONDecodeError as err: return False, {}, f"JSONParseError: {str(err)} near position {err.pos}" def run_step(self, step_idx: int, mock_llm_output: str) -> Tuple[bool, Any, str]: if step_idx >= self.max_steps: return True, None, "MAX_STEPS_EXCEEDED_FORCED_STOP" success, parsed_call, error_msg = self._sanitize_and_parse_json(mock_llm_output) if not success: return False, None, f"SCHEMA_REPAIR_FAILED: {error_msg}" tool_name = parsed_call.get("tool") arguments = parsed_call.get("args", {}) if not tool_name or tool_name not in self.tools: return False, None, f"UNKNOWN_TOOL: {tool_name}" try: # 执行注册的工具 result = self.tools[tool_name](**arguments) is_final = parsed_call.get("is_final", False) return is_final, result, "SUCCESS" except Exception as e: return False, None, f"TOOL_EXECUTION_ERROR: {str(e)}" # 生产仿真测试 def query_inventory(item_id: str, region: str = "cn-east") -> str: if not item_id: raise ValueError("item_id cannot be empty") return f"Item {item_id} in {region}: Stock = 42" if __name__ == "__main__": engine = AgentExecutionEngine(max_steps=3) engine.register_tool("query_inventory", query_inventory) # 模拟 LLM 吐出带非标格式的响应 bad_llm_output = """ ```json { "tool": "query_inventory", "args": { "item_id": "SKU-9921", "region": "cn-east", }, "is_final": true } ``` """ for current_step in range(1, 5): done, output, log = engine.run_step(current_step, bad_llm_output) print(f"Step {current_step} - 状态: {log} | 结果: {output}") if done: print("Agent 状态机正常终止退出。") break5. 真实上线后的防护兜底:何时切回静态规则引擎
在完成输入流裁剪和格式修复代理后,Agent 的解析成功率从 82% 提升至 99.4%。但在极致高并发和网络波动的生产环境下,没有任何大模型能保证 100% 可用。
上线必须配备熔断兜底开关。当监测到 LLM API 连续超时超过 3 次,或者工具执行失败率超过 5% 时,调度网关会自动触发断路保护。
系统会在毫秒级切回传统的基于状态树与规则引擎的保底逻辑。虽然静态规则无法像大模型那样泛化理解复杂意图,但它能保证核心业务流程(如商品下单、基础查询)不中断。把原型变成可用功能,核心从来不是大模型有多聪明,而是外围的工程防线有多坚固。