多智能体协作的评审边界-平芜编程栈

多智能体协作的评审边界

在大型系统或复杂工作流场景中，当出现多个 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", gt=0) status: str = Field(..., description="更新的目标状态，必须在许可枚举内", regex="^(active|suspended|deactivated)$") reason: str = Field(..., description="变更状态的操作原因，字数在5-100字之间", min_length=5, max_length=100) 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(f"Agent 工具调用失败: 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(f"Agent 工具调用入参校验未通过: {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(schema=UpdateUserStatusInput) 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 B，Agent 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_factory=lambda: f"tr-{uuid.uuid4().hex[:8]}") sender_agent: str target_agent: str current_hop: int = Field(default=1) max_hops: int = Field(default=5, description="最大允许跨 Agent 跳转次数") payload: Dict[str, Any] call_stack: List[str] = Field(default_factory=list) 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": f"Agent 协作递归调用超过最大限制({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_id=message.trace_id, sender_agent=self.agent_name, target_agent=message.target_agent, current_hop=message.current_hop + 1, max_hops=message.max_hops, payload=message.payload, call_stack=new_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(f"Agent 调用耗时 {time.time() - start_time:.2f}s，已超过超时门禁({self.timeout_seconds}s)") return self.fallback_response return result except Exception as exc: logger.error(f"Agent 执行抛出未捕获异常: {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( f"Line {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( f"Line {node.lineno}: 调用 dispatch 方法时必须显式透传 max_hops 参数防死循环！" ) self.generic_visit(node) def run_ci_gate(file_path: str): with open(file_path, "r", encoding="utf-8") as f: tree = ast.parse(f.read(), filename=file_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 入参门禁	模型生成非法字段类型拖垮下游 SQL	Pydantic Schema 二段式强类型校验与提示修复	遇到非合法格式不报错崩溃，返回 JSON 自愈提示
循环调用门禁	Agent 之间递归死循环耗尽 Token	Header 透传`trace_id`与`max_hops`递减计数器	跳转超过 5 次强行熔断并记录 Error 级 Trace 日志
上下文深度门禁	内存溢出、耗时长、单次调用花费几美金	强制经过 Context Window 滑动窗口裁剪	历史 Message 的 Token 总数严格控制在配置阀值内
稳定性降级门禁	某个 Agent 节点超时导致整个 HTTP 请求挂起	显式声明 Timeout 与 Circuit Breaker 兜底数据	单节点故障时，上游能在 500ms 内收到可预测的降级 Payload