多智能体协作的评审边界在大型系统或复杂工作流场景中当出现多个 AI Agent 相互协作、动态分发任务时容易因 Prompt 语义歧义或参数校验缺失引发 Agent 之间的无限循环调用与 Token 费用失控。若 Agent 在接收到非结构化输入时缺乏强类型校验与最大调用深度限制系统可能在短时间内发起数百次无效 HTTP 请求最终触发大模型 API 的 Rate Limit 进而导致服务崩溃。在多 Agent 协作工程实践中单体脚本或线性 Prompt 链的评审标准无法应对 LLM 的非确定性行为。通过引入静态 AST 规则扫描与动态契约防线可以在 Code Review 阶段精准识别潜在的隐性风险。本文梳理多 Agent 协作系统中必须建立的 5 个质量门禁与工程化落地方案。1. Tool 定义未绑定 Pydantic Schema盲目信任 LLM 入参最常见的离谱 Bug就是直接把一个纯 Python 函数暴露给 LLM 作为 Tool函数签名里写着def update_user_status(user_id, status)却没有定义字段校验规则。大模型调用时经常把字符串1024传成整数或者在status字段里凭空捏造一个不存在的枚举值UNKNOWN_STATE。如果在 Code Review 时放过这种不严谨的 Tool 定义线上就会产生大量隐蔽的未处理异常Unhandled Exceptions。在代码审查门禁中强制要求所有暴露给 Agent 的工具必须通过 Pydantic 或 JSON Schema 进行严格的约束。import json import logging from typing import Any, Callable, Dict, Type from pydantic import BaseModel, Field, ValidationError logger logging.getLogger(agent.gatekeeper) class UpdateUserStatusInput(BaseModel): user_id: int Field(..., description目标用户的唯一整数ID, gt0) status: str Field(..., description更新的目标状态必须在许可枚举内, regex^(active|suspended|deactivated)$) reason: str Field(..., description变更状态的操作原因字数在5-100字之间, min_length5, max_length100) def register_agent_tool(schema: Type[BaseModel]): Code Review 门禁装饰器强制所有工具函数必须绑定 Pydantic 输入模型 并在 Agent 运行时进行二段式强类型校验与自动修复提示。 def decorator(func: Callable): def wrapper(raw_json_args: str) - Dict[str, Any]: try: # 1. 强行解析大模型生成的 JSON 字符串 parsed_dict json.loads(raw_json_args) except json.JSONDecodeError as e: logger.error(fAgent 工具调用失败: JSON 格式错误 - {str(e)}) return { success: False, error_type: INVALID_JSON, feedback: f你返回的不是标准 JSON 格式: {str(e)}请重新格式化输入。 } try: # 2. 通过 Pydantic 模型死守类型与范围约束 validated_args schema(**parsed_dict) except ValidationError as e: logger.warning(fAgent 工具调用入参校验未通过: {e.json()}) return { success: False, error_type: SCHEMA_VALIDATION_ERROR, feedback: f参数校验失败请按照以下错误修正后重试: {e.errors()} } # 3. 校验通过执行真正业务逻辑 return {success: True, result: func(validated_args)} wrapper.__tool_schema__ schema return wrapper return decorator # 生产级工具注册示例 register_agent_tool(schemaUpdateUserStatusInput) def update_user_status_tool(args: UpdateUserStatusInput) - str: # 此处为安全执行的真实业务代码 return f用户 {args.user_id} 状态已成功更新为 {args.status}原因{args.reason}在评审代码时只要发现 Tool 注册没有指定对应的schema校验类PR 直接打回。不要在运行时赌 LLM 的输出质量。2. 协作链路缺乏 Max Hops 与 TraceId 传递警惕递归乒乓当 Agent A 遇到无法解决的问题时选择呼叫 Agent BAgent B 思考后又决定求助 Agent A。这种死循环如果缺乏机制干预直到耗尽 Token 或 HTTP 超时前都不会停下。CR 的第二个硬性标准所有 Agent 间调用的 Message Payload 必须携带可透传的 Header且必须包含max_hops递减计数器与全局trace_id。import uuid from typing import List, Optional from pydantic import BaseModel, Field class AgentMessagePayload(BaseModel): trace_id: str Field(default_factorylambda: ftr-{uuid.uuid4().hex[:8]}) sender_agent: str target_agent: str current_hop: int Field(default1) max_hops: int Field(default5, description最大允许跨 Agent 跳转次数) payload: Dict[str, Any] call_stack: List[str] Field(default_factorylist) class AgentRoutingEngine: def __init__(self, agent_name: str): self.agent_name agent_name def dispatch(self, message: AgentMessagePayload, next_agent_func: Callable) - Dict[str, Any]: # 1. 深度检测阻止无节制的 Agent 环形传递 if message.current_hop message.max_hops: logger.error( f[{message.trace_id}] 触发 Agent 协作深度防线! f链路: { - .join(message.call_stack)} - {self.agent_name} ) return { status: CIRCUIT_BROKEN, reason: fAgent 协作递归调用超过最大限制({message.max_hops}次)强行切断链路以防死循环。, trace_id: message.trace_id } # 2. 检查链路中是否已经包含自己防止直接死循环 if self.agent_name in message.call_stack: logger.warning(f[{message.trace_id}] 检测到重复入栈 Agent: {self.agent_name}可能存在乒乓循环) # 3. 更新 Stack 并递增 Hop new_stack list(message.call_stack) [self.agent_name] next_message AgentMessagePayload( trace_idmessage.trace_id, sender_agentself.agent_name, target_agentmessage.target_agent, current_hopmessage.current_hop 1, max_hopsmessage.max_hops, payloadmessage.payload, call_stacknew_stack ) return next_agent_func(next_message)只要评审中看到 Agent 之间的消息传递结构是一个裸字典、没有封装trace_id和递减的max_hops或者没有判断current_hop max_hops的分支就属于存在引发死循环风险的代码。3. 上下文滑窗无硬性 Token 截断内存爆炸与费用失控很多开发者喜欢把与所有的 Agent 历史对话记录messages直接 append 到数组里一路透传给下一个 Agent。运行时间稍微一长单次 Prompt 的 Token 长度直接突破 32k、64k。后果不仅是响应耗时从 1 秒飙升到 15 秒更可怕的是费用呈指数级增长。Code Review 门禁必须检查多 Agent 之间传递的 Context 必须经过滑动窗口裁剪或摘要化压缩Summarization Gate。class ContextWindowGatekeeper: def __init__(self, max_allowed_tokens: int 4096): self.max_allowed_tokens max_allowed_tokens def estimate_tokens(self, messages: List[Dict[str, str]]) - int: # 简单高效的字符数估算预警工程生产环境中可替换为 tiktoken total_chars sum(len(m.get(content, )) for m in messages) return int(total_chars / 3.5) def prune_context(self, messages: List[Dict[str, str]]) - List[Dict[str, str]]: current_tokens self.estimate_tokens(messages) if current_tokens self.max_allowed_tokens: return messages logger.warning(f当前上下文 Token ({current_tokens}) 超过安全阈值 ({self.max_allowed_tokens})触发防线裁剪) # 保留 System Prompt (第 0 条) 与最新的 3 轮对话 system_prompt [messages[0]] if messages and messages[0].get(role) system else [] recent_messages messages[-6:] # 最近 3 轮 user/assistant pruned system_prompt [ {role: system, content: [注意更早期的历史对话已被上下文隔离防护网自动截断以释放内存]} ] recent_messages return pruned在评审代码时只要看到messages.extend(new_messages)后直接调用llm.invoke(messages)而没有经过类似prune_context()的门禁处理一律算作不合格。4. 缺乏降级兜底路径单节点 Agent 宕机拖垮全链路在多 Agent 协作系统中任何一个子 Agent 都可能因为 API 503、模型输出格式错乱或超时而失效。如果主流程对子 Agent 的调用是“同步硬依赖”一旦某个辅助 Agent 卡死整条业务线直接冻结。静态检查的第 4 关必须显式提供降级逻辑Fallback Circuit Breaker。import time class AgentCircuitBreaker: def __init__(self, timeout_seconds: float 5.0, fallback_response: Optional[Dict] None): self.timeout_seconds timeout_seconds self.fallback_response fallback_response or {status: DEGRADED, content: 辅助服务暂不可用已自动切换至兜底响应} def execute_with_protection(self, agent_call_func: Callable, *args, **kwargs) - Dict[str, Any]: start_time time.time() try: # 在生产环境中建议使用 concurrent.futures 或 asyncio.wait_for 实现超时切断 result agent_call_func(*args, **kwargs) if time.time() - start_time self.timeout_seconds: logger.error(fAgent 调用耗时 {time.time() - start_time:.2f}s已超过超时门禁({self.timeout_seconds}s)) return self.fallback_response return result except Exception as exc: logger.error(fAgent 执行抛出未捕获异常: {str(exc)}触发自动降级) return self.fallback_response代码评审时需要关注execute_with_protection是否包裹了对不稳定 Agent 的异步/同步调用降级响应能否让下游业务逻辑继续运行而不是抛出 500 内核错误5. Automated CI AST 代码扫描规则把这些要求写在文档里靠人工看代码很容易漏掉。更靠谱的做法是把它写进 CI/CD 的静态代码扫描脚本里。我们写了一个基于 Pythonast模块的自动化检查脚本作为 Git Commit 和 GitHub Actions 的 PR 强门禁。一旦检测到裸llm.predict()或者无防线 Tool 注册静态检查直接打断 Build。import ast import sys class AgentCodeSanitizer(ast.NodeVisitor): def __init__(self): self.violations [] def visit_FunctionDef(self, node: ast.FunctionDef): # 规则 A: 检查是否有工具函数未挂载 schema 参数 for decorator in node.decorator_list: if isinstance(decorator, ast.Call) and getattr(decorator.func, id, ) register_agent_tool: has_schema any(keyword.arg schema for keyword in decorator.keywords) if not has_schema: self.violations.append( fLine {node.lineno}: 工具定义 register_agent_tool 必须显式声明 schema 校验类 ) self.generic_visit(node) def visit_Call(self, node: ast.Call): # 规则 B: 检查调用 agent_dispatch 时是否传了 max_hops func_name if isinstance(node.func, ast.Name): func_name node.func.id elif isinstance(node.func, ast.Attribute): func_name node.attr if func_name dispatch: has_max_hops any(arg.arg max_hops for arg in node.keywords if isinstance(arg, ast.keyword)) if not has_max_hops: self.violations.append( fLine {node.lineno}: 调用 dispatch 方法时必须显式透传 max_hops 参数防死循环 ) self.generic_visit(node) def run_ci_gate(file_path: str): with open(file_path, r, encodingutf-8) as f: tree ast.parse(f.read(), filenamefile_path) sanitizer AgentCodeSanitizer() sanitizer.visit(tree) if sanitizer.violations: print(f❌ [Agent CI Gate] 在文件 {file_path} 中发现 {len(sanitizer.violations)} 处违规风险:) for v in sanitizer.violations: print(f - {v}) sys.exit(1) else: print(f✅ [Agent CI Gate] {file_path} 安全合规检查通过。) if __name__ __main__: if len(sys.argv) 1: run_ci_gate(sys.argv[1])6. 排障与验收检查表在上线多 Agent 协作系统前除了做单元测试还应该拉着 team 成员对着清单走一遍最后的防护确认。检查维度风险现象静态/动态防线落地方式验收合格标准Tool 入参门禁模型生成非法字段类型拖垮下游 SQLPydantic Schema 二段式强类型校验与提示修复遇到非合法格式不报错崩溃返回 JSON 自愈提示循环调用门禁Agent 之间递归死循环耗尽 TokenHeader 透传trace_id与max_hops递减计数器跳转超过 5 次强行熔断并记录 Error 级 Trace 日志上下文深度门禁内存溢出、耗时长、单次调用花费几美金强制经过 Context Window 滑动窗口裁剪历史 Message 的 Token 总数严格控制在配置阀值内稳定性降级门禁某个 Agent 节点超时导致整个 HTTP 请求挂起显式声明 Timeout 与 Circuit Breaker 兜底数据单节点故障时上游能在 500ms 内收到可预测的降级 Payload治理非确定性的 LLM 行为本质上还是依靠确定性的软件工程手段。把死循环、内存溢出和非法入参拦在 Pull Request 阶段能够有效避免线上因非确定性调用引发的系统崩溃与资源浪费。
