智能体代码评审该看哪些细节评审常规业务代码时重点通常放在 SQL 性能、空指针防护和接口设计上。但当团队开始提交 AI Agent 系统的 Pull Request (PR) 时原有的 CR 标准会瞬间失效。Agent 代码最大的特征就是非确定性逻辑与自动化循环调用的结合。如果评审人缺乏对 Agent 运行机制的深度理解只看代码格式优雅就给 Pass上线后大概率会遇到 Tool Calling 无限死循环、上下文无上限膨胀导致 OOM、甚至工具调用未经校验直接注入删除数据库的灾难事故。Agent 的 CR 必须有一套针对性极强的审查抓手。1. Agent 代码提交上来了看似优雅的递归调用成了线上死循环上周评审了一段自动分析多模态图表并生成报告的 Agent 代码。代码结构写得很高级使用了优雅的面向对象抽象与异步递归。# 看起来很美优雅但在生产环境极度危险的 Agent 递归代码 class UntrustedAgentRunner: async def step(self, user_msg: str): response await self.llm.call(user_msg) if response.has_tool_call(): # 危险递归调用自身没有设置任何最大递归深度Max Steps控制 result await self.execute_tool(response.tool) return await self.step(fTool output: {result}) return response.content提交人在本地测试了 3 个简单用例运行完美。但在代码评审时我们发现了两个致命的死穴第一完全没有设置max_steps步数限制。一旦 LLM 返回了无法解析的工具格式或者工具返回了格式错乱的错误堆栈 Agent 会在死循环里无限递归调用下去直到把 API Token 额度全部烧光。第二上下文消息列表history是一个无界的list每一次 Tool 调用的中间结果都往里面 append。在多模态交互下几轮图片分析做下来内存和 Token 开销呈指数级暴增。2. 代码评审CR必须盯死 4 个防线限流、隔离、幂等与超时评审 Agent 代码必须像审查高危基础设施代码一样严苛。重点盯死 4 道确定性防线。第一是硬性步数闸门与预算限流Max Steps Token Budget。必须在代码中显式注入单次 Task 的最大循环次数与 Token 开销硬上限。第二是工具调用的沙箱隔离与权限最小化Sandbox Permission。 Agent 能调用的工具参数输入必须经过 Schema 强校验严禁直接把模型输出拼接成 Shell 命令或 SQL 语句。第三是状态机的幂等性控制Idempotency Key。在多轮交互中如果模型因为网络抖动发起了重复的工具调用如“下单”或“扣款”工具层必须根据 Request Hash 进行强幂等拦截。第四是异步 Callback 的超时与异常吞没防护Timeout Exception Safety。 Agent 的 Tool Calling 必须挂载独立的超时控制不能因为某个第三方工具挂起而拉垮整个 Agent 运行时。CR 检查维度典型错误模式隐患危害审查看守标准循环控制while True或无界递归陷入无限死循环Token 账单爆表强制检查max_steps硬编码断言上下文管理历史消息list.append无清理内存 OOM长尾延迟暴增检查滑窗裁剪Sliding Window逻辑工具安全直接eval()或拼 SQL触发 Prompt 注入与任意代码执行强制要求 Schema 校验与输入参数清洗异步处理await tool_func()无 Timeout线程/协程永久挂起死锁强制检查asyncio.wait_for(timeout...)3. Agent 状态演进与安全性审查决策树为了提升团队代码评审的效率可以建立一套专属于 Agent 代码的安全性审查决策树。凡是没有通过这一链条校验的 PR必须打回修改。4. 面向生产环境的 Agent 核心框架代码带有沙箱隔离与上下文滑窗的 Runner以下 Python 代码示范了一个符合面向生产环境的 CR 标准的 Agent Runner 框架实现。import hashlib import asyncio import logging from typing import List, Dict, Any, Optional from pydantic import BaseModel, ValidationError logger logging.getLogger(AgentCRGuard) class AgentTaskConfig(BaseModel): max_steps: int 5 max_token_budget: int 8000 timeout_per_tool_sec: float 3.0 class SafeAgentRunner: def __init__(self, llm_client, registered_tools: Dict[str, Any], config: AgentTaskConfig): self.llm_client llm_client self.tools registered_tools self.config config self.executed_tool_hashes set() # 用于幂等性防重 def _generate_idempotency_key(self, tool_name: str, args: Dict[str, Any]) - str: 根据工具名称和参数算出 Hash 值 raw_str f{tool_name}:{sorted(args.items())} return hashlib.md5(raw_str.encode(utf-8)).hexdigest() def _prune_history_context(self, history: List[Dict[str, Any]]) - List[Dict[str, Any]]: 强行滑窗剪枝保留 System Prompt 和最近 3 轮消息 if len(history) 6: return history system_msgs [m for m in history if m.get(role) system] recent_msgs history[-6:] return system_msgs recent_msgs async def execute_task_loop(self, user_input: str) - str: history [ {role: system, content: 你是一个严谨的 AI 助手请根据工具规范进行操作。}, {role: user, content: user_input} ] step_count 0 while step_count self.config.max_steps: step_count 1 logger.info(f执行 Step {step_count}/{self.config.max_steps}) # 1. 强行剪枝上下文 pruned_history self._prune_history_context(history) # 2. 调用模型 response await self.llm_client.async_generate(pruned_history) if not response.get(tool_call): return response.get(content, 任务完成) tool_name response[tool_call][name] tool_args response[tool_call][args] # 3. CR 防线工具存在性校验 if tool_name not in self.tools: history.append({role: system_error, content: f工具 {tool_name} 未注册}) continue tool_obj self.tools[tool_name] # 4. CR 防线Schema 参数强校验 try: validated_args tool_obj.schema(**tool_args) except ValidationError as ve: history.append({role: system_error, content: f参数格式校验失败: {ve.errors()}}) continue # 5. CR 防线幂等性去重 idempotency_key self._generate_idempotency_key(tool_name, tool_args) if idempotency_key in self.executed_tool_hashes: logger.warning(f检测到重复的工具调用: {tool_name}) history.append({role: system_error, content: 警告阻止了重复的工具调用。}) continue # 6. CR 防线带 Timeout 的工具隔离执行 try: tool_result await asyncio.wait_for( tool_obj.execute(validated_args), timeoutself.config.timeout_per_tool_sec ) self.executed_tool_hashes.add(idempotency_key) history.append({role: tool_output, content: str(tool_result)}) except asyncio.TimeoutError: logger.error(f工具 {tool_name} 执行超时 ({self.config.timeout_per_tool_sec}s)) history.append({role: system_error, content: f工具 {tool_name} 执行超时。}) return 【安全中断】超过最大允许执行步数Max Steps已终止任务。这段代码把所有在 CR 中容易漏掉的隐患全用固化的硬代码包揽了下来。5. 异步 Callback 与日志追踪如何防范被吞掉的异常在 Agent 架构中很多工具调用是通过异步事件驱动Async Event Listener / Callback完成的。如果异步 Callback 内部发生 Exception 却没有在最外层捕获Python 的asyncioTask 会静默结束导致 Agent 主线程一直处于await等待状态直到全局超时。在 CR 时必须严格审查所有异步 Callback必须包含try...except Exception兜底并把异常信息格式化为 Agent 主线程能识别的消息结构绝不允许让 Task 静默死亡。6. 团队 Agent CR 检查表合并 Pull Request 前的最终闸门最后建议每个开发 Agent 的团队在 Merge 代码前执行这张清单Max Steps是否有硬编码的循环次数限制默认 5Context Window历史消息是否有滑窗裁剪或 Summarize 机制Schema CheckTool Calling 参数是否经过了 Pydantic 等强类型校验Timeout Cover所有异步工具调用是否均包装了wait_for超时Idempotency敏感工具如修改数据是否有 Hash 去重或幂等控制把这几条细节盯牢了AI Agent 系统才能真正安全地放量给真实用户。