1. 项目概述:当“AI Agent”不再是个模糊标签,而是一套可拆解、可组装、可调试的工程模块
最近在几个技术社区和内部分享会上,我反复听到一个词被高频提起——“AI Agent”。它不像“大模型”那样有明确的技术边界,也不像“RAG”那样有清晰的流程定义,更像是一锅概念浓汤:有人把它等同于“能自动执行任务的ChatGPT”,有人拿它包装成“下一代操作系统入口”,还有人直接用“Agent = Prompt + LLM + 工具调用”一句话草草收场。结果就是,一线开发者想落地时一头雾水:到底该从哪下手?是先搭框架?还是先写工具?要不要上Orchestrator?Memory怎么存才不丢上下文?这些疑问背后,其实不是技术太新,而是“AI Agent”这个说法本身,正在被过度泛化、提前神化,甚至沦为PPT里的万能胶水。
我过去一年带过三个不同方向的AI应用项目:一个是面向客服坐席的实时话术辅助系统,一个是实验室级的科研文献自主综述生成器,还有一个是给中小制造企业做的设备维保工单智能分派引擎。这三个项目底层都用了“Agent”思路,但实现方式、模块粒度、调度逻辑、失败兜底机制,几乎没有一处是重样的。它们共同验证了一件事:AI Agent不是一种技术,而是一种设计范式——它把“目标驱动”“状态感知”“工具协同”“反馈闭环”这四个能力,以不同权重、不同耦合度、不同抽象层级,重新组织进一个运行单元里。所谓“概念串烧”,烧的不是名词堆砌,而是把“目标分解→工具选择→执行调度→结果校验→记忆沉淀→策略迭代”这一整条链路,从黑箱里拎出来,掰开、揉碎、标清接口、定义契约、暴露参数。你不需要一上来就造个“通用Agent OS”,但必须清楚:此刻你手上的这个功能模块,它在哪一环起作用?它的输入契约是什么?失败时谁来兜底?上下文超长了怎么截断?这些细节,才是决定项目能否从Demo跑进产线的关键分水岭。
这篇文章不讲“AI Agent有多伟大”,也不复述论文里的理论框架。我会以一个真实可运行的轻量级Agent系统为蓝本(基于Python + LangChain + 自研调度器,总代码量<800行),带你一层层剥开它的骨架:从最外层的用户指令如何被识别为“可执行目标”,到中间层的工具发现与参数绑定逻辑,再到最内层的状态快照与错误熔断机制。过程中会穿插我在三个项目里踩过的坑——比如某次因未对工具返回做schema校验,导致整个工单分派流程静默错配了37小时;又比如在科研综述项目中,因Memory存储未区分“长期知识”与“本次会话临时状态”,造成模型反复质疑自己前一句结论。所有内容,都指向一个务实目标:让你下次听到“我们做个AI Agent”时,能立刻反问出三个关键问题——目标是否可原子化?工具是否可契约化?失败是否可观测化?这才是“概念串烧”真正该烧出的味道:不是名词的烟火气,而是工程的铁锈味。
2. 核心设计思路拆解:为什么放弃“全能Agent”,选择“可插拔任务流”
2.1 拒绝“银弹思维”:从三个失败案例看过度抽象的代价
在第一个客服话术项目初期,团队曾试图构建一个“万能Agent”:它接收坐席语音转文字后的文本,自动识别用户情绪、提取业务意图、查询知识库、生成三版话术建议、并根据坐席历史偏好排序推荐。听起来很美,但上线后问题频发。最典型的是“意图识别漂移”——当用户说“上次那个退货流程太慢了”,模型把“退货”识别为当前意图,却忽略了“上次”这个关键时间锚点,导致推荐的话术全是标准退货流程,而非针对历史工单的升级处理方案。根因在于,我们把“上下文理解”“意图识别”“知识检索”“话术生成”全塞进一个LLM调用里,没有显式定义各环节的输入输出契约。模型在token压力下,优先保住了“退货”这个高概率词,牺牲了“上次”这个低频但关键的时序标记。
第二个项目是科研文献综述。我们最初采用LangChain的AgentExecutor,配置了“PubMed搜索”“PDF解析”“摘要生成”“参考文献格式化”四个工具。结果跑起来像抽风:有时搜到10篇文献,只解析了前3篇的PDF;有时摘要生成卡在第5篇,后续工具全部阻塞。排查发现,LangChain默认的串行调度器遇到单个工具超时或报错,会直接中断整个链路,且无重试策略。更麻烦的是,所有工具的输入参数都靠LLM自由发挥,没有强制schema约束。某次PubMed搜索工具收到的参数是{"query": "LLM alignment", "max_results": "five"}——字符串“five”传给需要int的参数,工具直接抛异常,而AgentExecutor只记录了“工具调用失败”,没留下任何可追溯的原始请求日志。
第三个制造企业维保项目更典型。我们设计了一个“工单分派Agent”,输入是设备故障描述文本,输出是维修班组ID。为追求“智能”,加入了“历史相似工单匹配”和“班组当前负载查询”两个工具。但实际运行中,90%的工单根本不需要查历史——简单关键词匹配就能分派。而每次调用“班组负载查询”都要访问生产数据库,平均耗时420ms。当系统每秒处理20个工单时,数据库连接池直接被打满。问题根源在于:我们没做“工具调用必要性预判”,把所有工具都当成必选项,而不是按需加载的插件。
这三个案例共同指向一个结论:强行把所有能力塞进一个“全能Agent”外壳,本质是用LLM的不确定性,去掩盖工程设计的确定性缺失。它带来的不是效率提升,而是可观测性崩塌、调试成本飙升、线上故障定位困难。所以,我们最终放弃了“造一个Agent”的思路,转向“定义一套Agent行为契约”,再按需组装具体实例。
2.2 “可插拔任务流”架构:用四层契约替代单体封装
我们最终采用的架构,核心是四层分离设计,每一层都定义了清晰的输入/输出契约和失败处理协议:
目标层(Goal Layer):接收原始用户输入(如“帮我查一下A320发动机的常见故障代码”),输出结构化目标指令。关键不是让LLM直接干活,而是先让它回答三个问题:① 这个请求的核心动词是什么?(查)② 操作对象是什么?(A320发动机故障代码)③ 是否有隐含约束?(“常见”意味着要过滤低频代码)。输出格式强制为JSON Schema:
{"action": "search", "object": "fault_codes", "constraints": ["aircraft: A320", "frequency: common"]}。这一步把模糊语义转化为机器可解析的指令,且为后续工具选择提供确定性依据。工具层(Tool Layer):每个工具都是独立函数,严格遵循统一接口:
def tool_name(input: dict) -> dict。输入字典必须包含tool_id和params两个键;输出字典必须包含status(success/error)、data(成功时的主体内容)、error_msg(失败时的明确原因)。例如“故障代码查询工具”的paramsschema明确定义为{"aircraft_model": "str", "min_frequency": "int"},调用前会做严格类型校验和范围检查,杜绝"five"这种非法值进入。调度层(Orchestration Layer):这是真正的“大脑”,但它不碰LLM。它只做三件事:① 根据目标层输出的
action和object,从注册表中匹配可用工具(支持多工具并行调用);② 按预设策略编排执行顺序(如“先查知识库,若无结果再调用搜索引擎”);③ 监控每个工具调用的耗时、成功率、返回数据质量(如检查PDF解析结果是否为空)。调度逻辑用纯Python配置,而非LLM生成,确保100%可预测。状态层(State Layer):不叫“Memory”,而叫“State Snapshot”。每次任务执行前后,都会保存一个快照,包含:
goal_input(原始目标)、tool_calls(调用的工具列表及参数)、tool_responses(各工具返回的原始数据)、llm_prompts(所有发给LLM的提示词)、final_output(最终返回给用户的内容)。这个快照是JSON格式,可直接存入Elasticsearch供审计,也可用于故障回放——当某次输出错误时,运维人员不用猜“LLM当时想了什么”,而是直接加载快照,复现整个执行链路。
这套设计的最大优势,是把“不可控”部分(LLM的理解与生成)压缩到最小范围,而把“可控”部分(目标解析、工具契约、调度逻辑、状态追踪)做到极致。它不追求“一个Agent解决所有问题”,而是确保“任何一个具体问题,都能被拆解成一组可验证、可替换、可监控的模块”。比如当“故障代码查询工具”需要升级为支持自然语言提问时,只需替换工具层的实现函数,目标层、调度层、状态层完全无需改动。这才是工程化的正道。
2.3 关键取舍:为什么不用AutoGen、CrewAI等热门框架?
AutoGen和CrewAI这类框架,宣传点往往是“多Agent协作”“角色扮演”“自主规划”。听起来很酷,但在我们三个项目实测中,它们带来了三个难以忽视的硬伤:
第一是调试黑洞。以AutoGen的GroupChat为例,当两个Agent(比如“Researcher”和“Writer”)对话出错时,你看到的日志是:“Researcher sent message to Writer: [长文本]”,“Writer replied: [长文本]”。但中间发生了什么?Researcher是如何从用户问题推导出搜索关键词的?Writer又是如何判断哪些信息该保留、哪些该删减的?这些决策过程被框架封装在LLM调用里,无法注入断点、无法修改prompt、无法查看中间状态。有一次,Researcher把“涡轮增压器”误识别为“涡轮风扇”,导致后续所有搜索都偏离方向,而我们花了6小时才定位到问题出在Researcher的system prompt里少了一句“专业术语必须严格匹配航空维修手册缩写”。
第二是性能不可控。CrewAI的Task执行默认开启“agent collaboration”,即一个Task失败后,会自动触发其他Agent介入。这在Demo阶段很炫,但在线上环境就是灾难。某次维保系统中,一个工单分派Task因网络抖动超时,CrewAI自动启动“Backup Dispatcher”Agent,后者又调用了一次数据库查询。结果同一工单在3秒内被处理了两次,维修班组收到了两条冲突指令。而我们的调度层明确禁止自动重试,所有重试必须由运维人员在Dashboard上手动触发,并附带强制填写的“重试原因”。
第三是部署复杂度爆炸。AutoGen要求每个Agent都维护自己的LLM实例和工具集,CrewAI则依赖Docker Compose管理多个服务。在客户现场,我们常遇到内存只有8GB的边缘服务器,根本跑不动这些框架的默认配置。而我们的轻量级架构,核心调度器仅依赖requests和json两个标准库,工具函数可以是本地Python脚本、HTTP API、甚至Shell命令,部署时只需一个requirements.txt,连Docker都不需要。
所以,我们选择“重复造轮子”,不是因为傲慢,而是因为:在工业级落地场景中,可调试性、可预测性、可部署性,永远比“开箱即用的炫技功能”重要十倍。框架的价值在于降低重复劳动,而不是增加新的黑箱。当你能用200行代码把目标解析、工具调用、状态追踪写清楚时,就不该为了省下这200行,去背负一个20000行的框架债。
3. 核心模块实现详解:从目标解析到状态快照的完整链路
3.1 目标解析模块:用结构化Prompt+Schema校验,把“人话”变成“机器指令”
目标解析是整个Agent系统的入口守门员。它的任务不是理解用户,而是把用户输入强制映射到预定义的动作空间里。我们不依赖LLM的自由发挥,而是用“结构化Prompt + JSON Schema校验 + 人工规则兜底”三层保险。
首先,Prompt设计遵循“三段式”原则:
- 角色定义段:明确告诉模型它此刻的身份是“指令翻译官”,不是“问题解答者”。例如:“你是一个严格的指令解析器。你的唯一任务是将用户输入转换为符合以下JSON Schema的目标指令。不要添加任何解释、不要生成额外内容、不要猜测用户未提及的信息。”
- 约束说明段:用自然语言+示例,穷举所有可能的歧义点。例如:“注意:‘查’‘找’‘搜索’都对应action='search';‘总结’‘概括’‘提炼’都对应action='summarize';如果用户提到‘上次’‘之前’‘历史’,必须在constraints中添加'context: historical';如果用户说‘最快的’‘最便宜的’,必须在constraints中添加'sort_by: speed'或'sort_by: cost'。”
- 输出格式段:强制指定输出为纯JSON,且必须包含
action、object、constraints三个字段,constraints必须是字符串列表。最后加一句:“输出必须是合法JSON,不要有任何前缀、后缀或解释文字。”
其次,LLM返回后,立即进行Schema校验。我们用Pydantic定义目标模型:
from pydantic import BaseModel, Field from typing import List, Optional class ParsedGoal(BaseModel): action: str = Field(..., description="操作类型,如search, summarize, compare") object: str = Field(..., description="操作对象,如fault_codes, maintenance_logs") constraints: List[str] = Field(default_factory=list, description="约束条件列表") @validator('action') def validate_action(cls, v): valid_actions = ['search', 'summarize', 'compare', 'generate'] if v not in valid_actions: raise ValueError(f'action must be one of {valid_actions}') return v @validator('object') def validate_object(cls, v): if not v.strip(): raise ValueError('object cannot be empty') return v.strip()调用ParsedGoal.parse_raw(llm_response),若校验失败,会抛出明确异常,如ValueError: action must be one of ['search', 'summarize', ...]。此时不重试,而是直接返回用户:“抱歉,我没理解您的需求,请用更明确的词语,比如‘搜索A320故障代码’或‘总结上次维修报告’。”
最后,对于高频、确定性高的请求,我们设置人工规则兜底。例如,当用户输入包含“故障代码”且机型明确(如“A320”“B737”),直接跳过LLM,用正则匹配生成目标:
import re def rule_based_parse(text: str) -> Optional[ParsedGoal]: # 匹配"机型+故障代码"模式 pattern = r"(A320|B737|B787).*(故障代码|fault code|FDC)" match = re.search(pattern, text, re.IGNORECASE) if match: aircraft = match.group(1) return ParsedGoal( action="search", object="fault_codes", constraints=[f"aircraft: {aircraft}", "frequency: common"] ) return None实测表明,约40%的客服话术请求可通过规则快速解析,响应时间从1.2秒降至80毫秒,且100%准确。剩下的60%才交给LLM处理,大幅降低了LLM的调用压力和出错概率。
提示:规则兜底不是倒退,而是工程智慧。就像TCP协议既有复杂的拥塞控制算法,也有简单的“超时重传”基础机制。在AI系统中,确定性规则是稳定性的压舱石,LLM是处理不确定性的补充手段,二者必须共存。
3.2 工具层实现:契约化接口与熔断保护的双重设计
工具层是Agent能力的物理载体。我们坚持一个原则:每个工具必须是“哑巴”——它只管执行,不管上下文、不管目标、不管失败后怎么办。所有智能决策,都交给上层调度器。
工具接口定义极其简单:
from typing import Dict, Any # 工具注册表,全局单例 TOOLS_REGISTRY = {} def register_tool(name: str): """装饰器:将函数注册为工具""" def decorator(func): TOOLS_REGISTRY[name] = func return func return decorator # 工具执行基类 class ToolResult: def __init__(self, status: str, data: Any = None, error_msg: str = ""): self.status = status # "success" or "error" self.data = data self.error_msg = error_msg def to_dict(self) -> Dict[str, Any]: return { "status": self.status, "data": self.data, "error_msg": self.error_msg } # 工具调用统一入口 def execute_tool(tool_name: str, params: Dict[str, Any], timeout: int = 30) -> ToolResult: if tool_name not in TOOLS_REGISTRY: return ToolResult("error", error_msg=f"Tool '{tool_name}' not found") try: # 熔断器:检查该工具最近1分钟失败率 if _circuit_breaker_is_open(tool_name): return ToolResult("error", error_msg=f"Tool '{tool_name}' is temporarily disabled due to high failure rate") # 执行工具 result = TOOLS_REGISTRY[tool_name](params) return ToolResult("success", data=result) except Exception as e: # 记录详细错误日志,包括params logger.error(f"Tool {tool_name} failed with params {params}: {str(e)}") return ToolResult("error", error_msg=str(e))以“故障代码查询工具”为例,其具体实现如下:
@register_tool("search_fault_codes") def search_fault_codes(params: Dict[str, Any]) -> Dict[str, Any]: # 1. 强制参数校验(工具内部二次校验) required_keys = ["aircraft_model", "min_frequency"] for key in required_keys: if key not in params: raise ValueError(f"Missing required parameter: {key}") # 2. 类型校验 if not isinstance(params["aircraft_model"], str): raise TypeError("aircraft_model must be string") if not isinstance(params["min_frequency"], int) or params["min_frequency"] < 1: raise ValueError("min_frequency must be integer >= 1") # 3. 构建数据库查询 query = """ SELECT code, description, occurrence_rate FROM fault_codes WHERE aircraft_model = %s AND occurrence_rate >= %s ORDER BY occurrence_rate DESC LIMIT 10 """ # 4. 执行查询(此处省略DB连接细节) results = db.execute(query, (params["aircraft_model"], params["min_frequency"])) # 5. 标准化输出 return { "codes": [ {"code": r[0], "description": r[1], "rate": r[2]} for r in results ], "total_count": len(results) }关键点在于:
- 参数校验下沉到工具内部:即使调度层已做过一次校验,工具内部仍要校验,因为工具可能被其他系统直接调用。
- 错误信息必须具体:
raise ValueError("min_frequency must be integer >= 1")比raise Exception("Invalid params")有用十倍,运维人员一眼就知道改哪里。 - 输出结构标准化:无论底层是SQL、API还是文件读取,最终都归一为
{"codes": [...], "total_count": N},上层调度器无需关心实现细节。
熔断器_circuit_breaker_is_open的实现也很务实:用Redis记录每个工具最近60秒的调用次数和失败次数,失败率超过30%且失败数≥5次,则自动熔断5分钟。熔断期间所有调用直接返回错误,避免雪崩。这比依赖LLM的“自我修复”靠谱得多。
3.3 调度层核心逻辑:状态机驱动的确定性编排
调度层是整个系统的“交通指挥中心”。它不依赖LLM做决策,而是用状态机(State Machine)驱动流程。我们定义了五个核心状态:
WAITING_GOAL:等待目标解析完成SELECTING_TOOLS:根据目标选择待调用工具EXECUTING_TOOLS:并发执行工具AGGREGATING_RESULTS:收集工具结果,决定下一步GENERATING_OUTPUT:调用LLM生成最终回复
状态流转由明确的事件触发,例如:
- 事件
GOAL_PARSED→ 状态从WAITING_GOAL→SELECTING_TOOLS - 事件
TOOLS_SELECTED→ 状态从SELECTING_TOOLS→EXECUTING_TOOLS - 事件
ALL_TOOLS_COMPLETED→ 状态从EXECUTING_TOOLS→AGGREGATING_RESULTS
关键逻辑在AGGREGATING_RESULTS状态。这里不靠LLM“思考”,而是用预设规则判断:
def aggregate_results(goal: ParsedGoal, tool_results: List[ToolResult]) -> Dict[str, Any]: # 规则1:如果所有工具都失败,直接返回错误 if all(r.status == "error" for r in tool_results): return {"error": "All tools failed", "details": [r.error_msg for r in tool_results]} # 规则2:如果"search_fault_codes"成功,且返回了至少3个代码,则进入生成环节 fault_code_result = next((r for r in tool_results if r.tool_name == "search_fault_codes" and r.status == "success"), None) if fault_code_result and len(fault_code_result.data.get("codes", [])) >= 3: return {"stage": "generate_output", "input_data": fault_code_result.data} # 规则3:如果"search_fault_codes"失败,但"search_manuals"成功,则用手册内容生成 manual_result = next((r for r in tool_results if r.tool_name == "search_manuals" and r.status == "success"), None) if manual_result: return {"stage": "generate_output", "input_data": manual_result.data, "fallback": True} # 兜底:返回原始错误 return {"error": "No usable data from tools"}这种基于规则的聚合,保证了100%的可预测性。你可以精确计算出:当search_fault_codes返回5个代码时,系统一定会走generate_output分支;当它返回空列表时,系统一定会尝试search_manuals。没有“可能”“大概率”“LLM觉得”,只有“确定会”。
注意:状态机不是银弹,它要求你提前想清楚所有分支。我们在设计初期,用白板画出了所有可能的工具组合和失败路径,共梳理出17种状态流转场景。虽然前期投入大,但换来的是线上零意外——每次发布新工具,只需更新状态流转图,无需担心“LLM会不会突然走错路”。
3.4 状态快照模块:JSON化记录一切,让调试回归“所见即所得”
状态快照(State Snapshot)是我们最自豪的设计。它不叫“Memory”,因为它不参与决策,只负责忠实记录。每次任务生命周期结束,都会生成一个完整的JSON快照,结构如下:
{ "session_id": "sess_abc123", "timestamp": "2024-05-20T14:23:15.892Z", "goal_input": "帮我查一下A320发动机的常见故障代码", "parsed_goal": { "action": "search", "object": "fault_codes", "constraints": ["aircraft: A320", "frequency: common"] }, "tool_selection": ["search_fault_codes"], "tool_calls": [ { "tool_name": "search_fault_codes", "params": {"aircraft_model": "A320", "min_frequency": 5}, "start_time": "2024-05-20T14:23:16.102Z", "end_time": "2024-05-20T14:23:16.523Z", "duration_ms": 421 } ], "tool_responses": [ { "tool_name": "search_fault_codes", "status": "success", "data": { "codes": [ {"code": "ECAM ENG 1 OIL LO PR", "description": "Engine 1 oil low pressure", "rate": 12.3}, {"code": "ECAM ENG 2 OIL LO PR", "description": "Engine 2 oil low pressure", "rate": 11.7}, {"code": "ECAM ENG 1 OIL HI TEMP", "description": "Engine 1 oil high temperature", "rate": 9.2} ], "total_count": 3 } } ], "llm_prompts": [ { "role": "system", "content": "You are a professional aviation maintenance assistant..." }, { "role": "user", "content": "Generate a concise summary of these fault codes for a mechanic..." } ], "final_output": "以下是A320发动机最常见的3个故障代码:1. ECAM ENG 1 OIL LO PR(发动机1滑油低压)...(略)" }这个快照的价值,在于它把整个Agent的“黑箱行为”变成了“白盒日志”。当用户反馈“为什么没显示第4个故障代码?”,运维人员不用去猜LLM的思考过程,而是直接在Elasticsearch里搜索session_id: sess_abc123,打开快照,一眼看到:
tool_calls里min_frequency参数是5,而第4个代码的rate是4.8,被SQL的>=5条件过滤掉了;tool_responses里total_count是3,证明数据库查询本身正确;llm_prompts显示LLM只收到了3个代码,所以它的输出必然只有3个。
所有问题,都在快照里有迹可循。我们甚至开发了一个快照回放工具:上传一个快照JSON,工具会自动重建当时的执行环境,重新跑一遍所有工具调用和LLM生成,100%复现原始结果。这比任何“LLM debug mode”都可靠。
4. 实操部署与避坑指南:从本地测试到生产环境的全流程经验
4.1 本地开发环境搭建:用Docker Compose一键拉起最小可行系统
本地开发的目标是“所见即所得”——你在本地跑通的流程,就是线上会跑的流程。为此,我们用Docker Compose定义了最小依赖栈:
# docker-compose.yml version: '3.8' services: # 模拟生产数据库(PostgreSQL) db: image: postgres:14 environment: POSTGRES_DB: agent_demo POSTGRES_USER: demo POSTGRES_PASSWORD: demo volumes: - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql ports: - "5432:5432" # 模拟外部API服务(如知识库) knowledge_api: image: python:3.11-slim volumes: - ./mock_api:/app working_dir: /app command: python -m http.server 8000 ports: - "8000:8000" # 主应用(Agent调度器) app: build: . environment: DB_URL: postgresql://demo:demo@db:5432/agent_demo KNOWLEDGE_API_URL: http://knowledge_api:8000 depends_on: - db - knowledge_api ports: - "8001:8001"关键点在于:
- 所有外部依赖都模拟:数据库用PostgreSQL官方镜像,知识库用Python内置HTTP Server模拟,确保本地环境与生产环境的网络拓扑一致(如
app服务通过db:5432访问数据库,而非localhost:5432)。 - 初始化数据脚本化:
./sql/init.sql预先插入了A320、B737的故障代码样本数据,避免开发时手动造数据。 - 环境变量统一管理:
DB_URL和KNOWLEDGE_API_URL在app服务中通过环境变量注入,代码里直接读取,不写死地址。
开发时,只需一条命令:docker-compose up --build,30秒内即可拉起完整环境。前端用curl测试:
curl -X POST http://localhost:8001/parse \ -H "Content-Type: application/json" \ -d '{"text": "查A320常见故障代码"}' # 返回结构化目标 {"action":"search","object":"fault_codes","constraints":["aircraft: A320","frequency: common"]}然后调用执行端点:
curl -X POST http://localhost:8001/execute \ -H "Content-Type: application/json" \ -d '{"action":"search","object":"fault_codes","constraints":["aircraft: A320","frequency: common"]}'整个流程完全脱离LLM,纯靠规则和工具,确保基础链路100%稳定。只有当本地所有环节都跑通,才接入真实的LLM API。
4.2 生产环境部署要点:资源隔离、灰度发布与监控告警
生产环境的核心原则是:让Agent系统像一个普通微服务一样被管理,而不是一个神秘的AI黑箱。
资源隔离:Agent调度器与LLM API调用必须分离部署。我们用Kubernetes,将调度器部署在CPU优化型节点(8核16GB),而LLM调用(如调用OpenAI API)放在单独的、带公网出口的节点组。这样,即使LLM API出现网络抖动或限流,调度器的CPU和内存资源依然充足,能继续处理队列中的任务、执行本地工具、生成降级响应。
灰度发布:新工具上线不直接全量。我们设计了“工具版本路由”机制:每个工具注册时,可指定
version和weight(权重)。例如:@register_tool("search_fault_codes", version="v1", weight=0.8) def search_fault_codes_v1(params): ... @register_tool("search_fault_codes", version="v2", weight=0.2) def search_fault_codes_v2(params): ...调度器根据权重随机选择版本。v2上线初期设为weight=0.05,观察一周无异常后,逐步提升至0.5,最后全量。所有版本的调用日志都打上
version标签,便于对比分析。监控告警:我们只监控四个黄金指标,全部接入Prometheus:
agent_goal_parse_success_rate:目标解析成功率,低于95%告警(说明Prompt或规则需优化);agent_tool_call_duration_seconds:工具调用P95耗时,超过2秒告警(可能是DB慢查询或API超时);agent_state_snapshot_size_bytes:快照大小,超过5MB告警(说明LLM返回了过多冗余内容,需优化prompt);agent_circuit_breaker_open_count:熔断器开启次数,1小时内>3次告警(说明某个工具稳定性出问题)。
告警不是发邮件,而是自动创建Jira工单,并@相关负责人。例如,当search_fault_codes的P95耗时超2秒,工单标题自动生成:“【紧急】search_fault_codes工具性能劣化,当前P95=2.3s”,描述里直接嵌入最近10次调用的duration_ms和params,工程师打开就能看到问题模式。
4.3 常见问题速查表:那些文档里不会写的实战教训
| 问题现象 | 根本原因 | 解决方案 | 我的实操心得 |
|---|---|---|---|
| 目标解析偶尔返回非法JSON | LLM在token紧张时,会省略JSON结尾的}或引号 | 在LLM调用后,用正则r'\{.*\}'提取第一个{...}块,再json.loads();若失败,强制重试一次并加大temperature=0.3 | 别信LLM的“100% JSON输出”,永远加一层容错提取。我们线上99.98%的非法JSON都来自这个正则修复 |
| 工具调用返回空数据,但状态是success | 工具函数内部没检查业务逻辑失败(如SQL查不到结果),只捕获了技术异常 | 在每个工具函数末尾,强制检查return_value是否为空或None,若是则raise ValueError("No data returned") | “空结果”不是成功,而是业务失败。必须让工具自己定义什么是“有效返回”,不能依赖上层判断 |
| 快照体积暴涨,单次任务生成10MB日志 | LLM生成的final_output包含大量HTML标签或Base64图片 | 在GENERATING_OUTPUT状态,对LLM返回内容做清洗:re.sub(r'<[^>]+>', '', text)移除HTML,re.sub(r'data:image/[^;]+;base64,[^\"]+', '[IMAGE]', text)替换图片占位符 | 快照是审计用的,不是存档用的。所有非结构化内容(HTML/CSS/JS/图片)必须清洗,否则ES索引会爆掉 |
| 多个相同工单被重复处理 | 调度器在` |