构建可控AI智能体循环:从ReAct到分阶段架构的设计与实践
2026/8/7 14:22:39 网站建设 项目流程

1. 项目概述:从“失控”到“可控”的Agent循环设计

最近和几个做AI应用开发的朋友聊天,大家不约而同地提到了同一个痛点:Agent(智能体)跑起来容易,但让它“听话”地停下来、或者按照我们预设的路径去执行,简直太难了。你肯定也遇到过类似场景:让一个基于Claude的Agent去分析一份市场报告,它要么在某个细节上无限循环追问,要么突然跳到一个完全不相关的任务上,最后输出的结果离题万里。这背后的核心问题,就是我们今天要深入探讨的——如何为Claude这类大模型驱动的Agent,设计一个真正“可控”的执行循环(Agent Loops)。

所谓Agent Loop,简单理解就是Agent感知环境、思考决策、执行动作、并观察结果的循环过程。一个失控的Loop,就像一辆没有刹车和方向盘的汽车,动力再强也只会横冲直撞。而一个可控的Loop,则是一位训练有素的司机,知道何时加速、何时转弯、何时抵达目的地后平稳停车。对于Claude Code、Claude Desktop或是任何基于Claude API构建的Agent项目,设计可控的Loop不仅是提升效率的关键,更是确保应用可靠、安全、可预测的基石。无论是开发一个自动化的代码助手,还是一个复杂的业务流程Agent,掌握Loop的控制权,就意味着掌握了项目的命脉。

2. 可控Agent Loop的核心设计哲学与架构拆解

2.1 理解“可控”的四个维度:边界、状态、流程与干预

在设计之前,我们必须先统一对“可控”的理解。在我看来,一个可控的Agent Loop必须满足以下四个维度的要求:

第一,边界可控。这是最基础的一层。Agent必须明确知道自己的任务边界是什么,什么该做,什么不该做。很多初级开发者只给Agent一个模糊的指令,比如“帮我优化代码”,结果Agent可能去修改了系统文件,或者尝试调用没有权限的API。边界控制需要通过清晰的系统提示词(System Prompt)和工具(Tools)权限管理来实现。例如,在给Claude Code设计Loop时,我会在System Prompt里明确写上:“你的工作空间仅限于当前项目目录./src,禁止访问或修改此目录外的任何文件。你可用的工具仅限于:文件读取、代码分析、代码重构(不改变外部依赖)。对于任何超出此范围的需求,你必须直接拒绝并说明原因。”

第二,状态可控。Agent在运行中会积累上下文、产生中间结果、并具备某种“记忆”。一个可控的Loop必须能清晰地追踪和管理这些状态。这包括:当前任务的目标状态(Goal State)、已完成的历史动作(Action History)、环境的最新观察(Latest Observation)、以及可能存在的子任务栈(Sub-task Stack)。状态管理不善,Agent就容易失忆、重复劳动或逻辑混乱。我通常会用一种结构化的状态对象(State Object)来封装这一切,确保每个循环迭代都能基于完整、准确的状态进行决策。

第三,流程可控。这是指Agent执行动作的逻辑流程是可预测、可引导的。我们不希望Agent像无头苍蝇一样随机选择动作。流程控制通常通过引入规划器(Planner)或工作流引擎来实现。例如,一个经典的“规划-执行-观察”循环(Plan-Execute-Observe),或者更复杂的基于目标的层次任务网络(HTN)。在Claude Agent中,我们可以让Claude自身扮演规划器的角色,在每个循环开始时,先输出一个清晰的下一步计划(Plan),然后由外部控制器(Orchestrator)来批准和执行这个计划,从而将“思考”和“行动”分离,实现流程的制动。

第四,干预可控。无论设计多么完美,总有意外。一个健壮的Loop必须为人类(或其他监控系统)预留干预的入口。这包括:在循环的关键节点设置检查点(Checkpoint)以供审核;设计优雅的中断(Interrupt)和继续(Resume)机制;以及当Agent陷入死循环或错误状态时,能通过外部信号(如超时机制、看门狗)强制将其拉回安全状态。干预是系统安全的最后一道防线。

2.2 主流Agent Loop模式剖析:ReAct与更优选择

谈到Agent Loop,很多人第一个想到的是ReAct(Reasoning and Acting)模式。它让模型以“Thought: ... Action: ... Observation: ...”的格式循环,将推理和行动交织在一起。对于快速原型验证,ReAct非常有效。但在追求“可控性”的生产环境中,ReAct的固有缺陷就暴露出来了:

  1. 思考与行动耦合过紧:模型输出的“Thought”和“Action”在一个响应里,外部系统很难在模型“思考”之后、“行动”之前插入校验或修改。
  2. 状态管理隐式且脆弱:整个对话历史就是它的状态,容易受到上下文长度限制,且历史中的任何干扰都可能带偏后续循环。
  3. 缺乏显式的流程阶段:所有步骤看起来都一样,难以实施差异化的控制策略(比如,在“决策”阶段加强审核,在“执行”阶段放宽限制)。

因此,对于可控性要求高的Claude Agent,我强烈建议采用“规划与执行分离”的架构。下面是一个我经过多个项目锤炼后的基础架构设计:

外部控制器 (Orchestrator) | | (1. 分发任务 & 初始状态) v +-------------------------------+ | Agent 循环引擎 | +-------------------------------+ | 当前状态 (State) | | - 目标 (Goal) | | - 历史 (History) | | - 观察 (Observation) | | - 子任务栈 (Sub-task Stack) | +-------------------------------+ | | (2. 基于状态,决定下一步阶段) v +-------------------------------+ | 阶段路由器 (Stage Router) | +-------------------------------+ | | (3. 路由到特定处理阶段) v +-------+-------+-------+ | | | | v v v v 规划阶段 决策阶段 执行阶段 评估阶段 (Plan) (Decide) (Execute)(Evaluate) | | | | | | | | (4. 调用对应模块) v v v v +-----------------------------------+ | Claude 大模型引擎 | | (或特定功能模块,如代码执行器) | +-----------------------------------+ | | (5. 返回结果,更新状态) v +-------------------------------+ | 状态更新器 (State Updater) | +-------------------------------+ | | (6. 检查循环终止条件) v [任务完成?] --> 是 --> 退出循环,返回结果 | 否 | v (回到步骤2,继续循环)

在这个架构中,外部控制器是总指挥,负责启动任务和注入初始指令。Agent循环引擎是核心容器,持有并管理着状态对象阶段路由器是大脑,它根据当前状态判断现在应该进入哪个阶段。每个阶段(规划、决策、执行、评估)都是独立的模块,有明确的输入输出规范。Claude大模型引擎在这里更像一个“全能员工”,在不同阶段被调用去做不同的事:在规划阶段做战略思考,在决策阶段做选择判断,等等。状态更新器则负责将每个阶段的结果规整地写入状态,为下一轮循环做好准备。

这种架构的优势在于,控制点(Checkpoint)可以非常方便地加在各个阶段之间。例如,你可以在“规划阶段”完成后,让路由器暂停,将生成的计划提交给人工审核,审核通过后再进入“决策阶段”。这才是真正的“可控”。

3. 构建可控Loop的三大核心组件详解

3.1 状态管理:设计一个健壮的状态机

状态是Loop的“记忆”和“情境”,设计的好坏直接决定Agent是否清醒。我反对将整个对话历史直接作为状态。一个精炼的、结构化的状态对象才是王道。

一个我常用的状态对象结构如下(以Python Pydantic模型为例):

from enum import Enum from typing import List, Optional, Any from pydantic import BaseModel class AgentStage(Enum): INITIALIZING = "initializing" PLANNING = "planning" DECIDING = "deciding" EXECUTING = "executing" EVALUATING = "evaluating" PAUSED_FOR_REVIEW = "paused_for_review" FINISHED = "finished" FAILED = "failed" class SubTask(BaseModel): id: str description: str status: str # 'pending', 'in_progress', 'completed', 'failed' result: Optional[Any] = None class AgentState(BaseModel): # 核心标识 task_id: str main_goal: str current_stage: AgentStage = AgentStage.INITIALIZING # 执行历史与上下文 action_history: List[dict] = [] # 记录每一步动作及结果 conversation_context: List[dict] = [] # 与模型交互的关键上下文,非全部历史 latest_observation: Optional[str] = None # 上一次行动后的环境反馈 # 任务分解与管理 sub_task_stack: List[SubTask] = [] # 待处理的子任务栈 completed_tasks: List[SubTask] = [] # 已完成的任务 # 控制与元信息 iteration_count: int = 0 max_iterations: int = 50 # 防死循环硬限制 pause_points: List[str] = [] # 预设的暂停点,如 [“after_plan”, “before_file_write”] metadata: dict = {} # 存放任意自定义信息

设计要点与避坑指南:

  • current_stage是关键:它是一个枚举值,明确告知系统和Agent自身“现在处在哪个阶段”。这为阶段路由器提供了决策依据。
  • 区分action_historyconversation_contextaction_history记录所有对环境的操作(如调用了哪个工具,输入输出是什么),用于回溯和审计。conversation_context则只保留与模型推理相关的关键对话,防止上下文被无关历史淹没。通常,我只保留最近3-5轮的关键思考。
  • sub_task_stack是复杂任务的生命线:当主任务被拆解后,子任务压入栈中。Agent永远只处理栈顶任务,完成后再弹出。这天然实现了任务的分解与串行执行,避免了思维混乱。对于可以并行的任务,可以设计多个工作线程或协程,每个线程管理自己的栈,但这属于高级话题。
  • pause_points实现软控制:你可以在状态初始化时,就指定在哪些环节后暂停。例如,pause_points = [“after_plan”],那么当current_stagePLANNING切换到DECIDING之前,路由器会检测到这个暂停点,并将状态置为PAUSED_FOR_REVIEW,等待外部指令。
  • max_iterations是硬保险:无论如何,必须设置一个循环上限。这是防止无限循环的最后手段。达到上限后,将状态置为FAILED并终止。

3.2 阶段路由器与流程引擎:Loop的节拍器

阶段路由器是驱动Loop运转的节拍器。它的逻辑并不复杂,但却需要严谨。其核心是一个基于当前状态(主要是current_stagelatest_observation)的状态转移函数。

一个简化的路由器逻辑如下:

def determine_next_stage(current_state: AgentState) -> AgentStage: """根据当前状态,决定下一个阶段""" # 检查硬性终止条件 if current_state.iteration_count >= current_state.max_iterations: return AgentStage.FAILED if current_state.main_goal is not None and goal_is_achieved(current_state): # 自定义的目标达成检查函数 return AgentStage.FINISHED # 检查是否处于暂停点 if current_state.current_stage == AgentStage.PAUSED_FOR_REVIEW: # 等待外部指令,这里返回自身表示保持暂停 return AgentStage.PAUSED_FOR_REVIEW # 状态转移逻辑 if current_state.current_stage == AgentStage.INITIALIZING: # 初始化完成后,进入规划阶段 return AgentStage.PLANNING elif current_state.current_stage == AgentStage.PLANNING: # 规划完成后,检查是否有预设的“规划后暂停点” if "after_plan" in current_state.pause_points: return AgentStage.PAUSED_FOR_REVIEW # 否则,进入决策阶段,选择第一个要执行的子任务或动作 return AgentStage.DECIDING elif current_state.current_stage == AgentStage.DECIDING: # 决策完成后,进入执行阶段 return AgentStage.EXECUTING elif current_state.current_stage == AgentStage.EXECUTING: # 执行完成后,必然进入评估阶段,分析执行结果 return AgentStage.EVALUATING elif current_state.current_stage == AgentStage.EVALUATING: # 评估完成后,判断下一步 if current_state.sub_task_stack: # 还有子任务 # 下一个子任务,回到决策阶段 return AgentStage.DECIDING else: # 所有子任务完成,返回规划阶段看是否需要生成新任务,或直接结束 # 这里可以加入更复杂的逻辑,比如评估整体目标是否达成 if goal_is_achieved(current_state): return AgentStage.FINISHED else: # 可能需要重新规划 return AgentStage.PLANNING # ... 其他阶段处理 # 默认情况,返回当前阶段(相当于暂停) return current_state.current_stage

流程引擎则是包裹着路由器和各阶段模块的循环体。它的伪代码如下:

def controlled_agent_loop(initial_goal: str, controller): """可控的Agent主循环""" state = initialize_state(initial_goal) while state.current_stage not in [AgentStage.FINISHED, AgentStage.FAILED]: # 1. 决定阶段 next_stage = determine_next_stage(state) if next_stage != state.current_stage: logger.info(f"状态转移: {state.current_stage} -> {next_stage}") state.current_stage = next_stage # 2. 如果进入暂停阶段,则跳出循环,等待外部唤醒 if state.current_stage == AgentStage.PAUSED_FOR_REVIEW: controller.notify_paused(state) break # 或 yield state, 取决于异步实现 # 3. 执行当前阶段的核心工作 if state.current_stage == AgentStage.PLANNING: state = planning_stage(state, claude_client) elif state.current_stage == AgentStage.DECIDING: state = deciding_stage(state, claude_client) elif state.current_stage == AgentStage.EXECUTING: state = executing_stage(state, tool_registry) # 工具执行可能不经过Claude elif state.current_stage == AgentStage.EVALUATING: state = evaluating_stage(state, claude_client) # ... 其他阶段 # 4. 更新迭代计数 state.iteration_count += 1 # 5. (可选) 每个循环后都检查一次外部中断信号 if controller.should_interrupt(): state.current_stage = AgentStage.PAUSED_FOR_REVIEW controller.notify_paused(state) break # 循环结束,返回最终状态 controller.notify_finished(state) return state

这个引擎清晰地将“状态判断”、“阶段执行”和“循环控制”分离开,使得整个Loop的流程一目了然,并且极易插入监控和干预点。

3.3 提示词工程:为每个阶段定制Claude的“角色卡”

很多开发者用一个通用的提示词(Prompt)让Claude干所有事,这是导致Agent行为不可控的重要原因之一。在我们的分阶段架构中,每个阶段都应该有专属的、高度定制的提示词,这就像给Claude在不同环节戴上不同的“角色面具”。

规划阶段提示词示例:

你是一个资深的项目规划专家。你的任务是将一个宏观目标分解为具体、可执行、有序的子任务。 当前总体目标:{state.main_goal} 已完成的任务:{state.completed_tasks} 当前环境反馈:{state.latest_observation} 请基于以上信息,规划接下来的步骤。 你的输出必须是严格的JSON格式: { "reasoning": "你的思考过程,分析当前状况和下一步方向", "new_sub_tasks": [ {"id": "task_1", "description": "第一个子任务的清晰描述"}, {"id": "task_2", "description": "第二个子任务的清晰描述"} ], "is_goal_achieved": false // 根据当前信息,判断总体目标是否已完全达成 } 注意:子任务描述必须具体、可操作,且一个任务应能在几步内完成。避免创建模糊或庞大的任务。
  • 设计意图:引导Claude进行战略分解,并强制其输出结构化数据,方便程序解析并压入sub_task_stack

决策阶段提示词示例:

你是一个决策者。当前需要从待办任务中选择下一个要执行的具体动作。 待处理子任务栈(栈顶在最前): {state.sub_task_stack} 可用工具列表: {tool_descriptions} 历史动作记录(最近3条): {state.action_history[-3:]} 请决定下一步做什么。你只能选择以下两种行动之一: 1. 从“可用工具列表”中选择一个工具来执行栈顶的子任务。 2. 如果认为栈顶任务无法用现有工具完成,或需要更多信息,可以请求“人工协助”。 你的输出必须是严格的JSON格式: { "reasoning": "选择该行动的理由", "action": "tool_name" 或 "human_help", "action_input": { // 如果action是工具,这里是对应的输入参数 "param1": "value1", ... }, "query_to_human": "" // 如果action是human_help,这里是你想问人的具体问题 }
  • 设计意图:严格限制Claude的行动选项(只能选工具或求助),防止其天马行空。结构化输出确保动作能被准确执行。

评估阶段提示词示例:

你是一个质量评估员。请评估刚刚执行的动作的结果,并判断相关子任务是否完成。 执行的子任务:{current_sub_task.description} 执行的动作:{last_action} 动作结果:{state.latest_observation} 请分析: 1. 该动作结果是否成功解决了子任务的目标? 2. 结果中是否包含了需要关注的新信息或错误? 3. 基于当前结果,总体目标`{state.main_goal}`的完成度是否有变化? 你的输出必须是严格的JSON格式: { "reasoning": "详细的评估分析", "sub_task_status": "completed" 或 "failed" 或 "needs_more_work", "summary": "对本次执行结果的简要总结", "suggested_next_step": "根据评估,对后续步骤的建议(例如:继续本任务、标记完成、重新规划等)" }
  • 设计意图:让Claude从“执行者”切换到“评审者”角色,客观评估工作质量,并为状态更新器提供明确的指令(如将子任务标记为完成)。

通过这种分阶段的提示词设计,Claude在每个环节的行为都被高度约束和引导,输出的结果也是结构化的,极大提升了Loop的可预测性和可解析性。

4. 实现细节、工具集成与防死循环策略

4.1 与Claude API的集成模式

在实际编码中,如何调用Claude API也有讲究。我推荐使用异步非阻塞的方式,并做好错误重试和降级处理

import asyncio from tenacity import retry, stop_after_attempt, wait_exponential from anthropic import AsyncAnthropic class ClaudeClient: def __init__(self, api_key, model="claude-3-5-sonnet-latest"): self.client = AsyncAnthropic(api_key=api_key) self.model = model @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def call_with_prompt(self, stage_prompt: str, state_context: dict) -> dict: """调用Claude,并尝试解析JSON返回""" try: # 1. 构建完整的消息上下文 messages = self._build_messages(stage_prompt, state_context) # 2. 发起API调用 response = await self.client.messages.create( model=self.model, max_tokens=4096, # 根据阶段调整 messages=messages, system="你是一个严谨的AI助手,必须严格按照用户指定的格式输出。", # 可覆盖 temperature=0.2 if "planning" in stage_prompt else 0.1, # 规划时创造性稍高,决策时更低 ) # 3. 解析响应内容 content = response.content[0].text # 尝试提取JSON部分(Claude有时会在JSON外加说明) parsed_response = self._extract_and_parse_json(content) return parsed_response except Exception as e: logger.error(f"调用Claude API失败: {e}") # 降级策略:返回一个预定义的错误结构,让状态机可以处理 return { "error": str(e), "reasoning": "API调用失败,无法进行本阶段推理。", "suggested_next_step": "retry_or_fail" # 由状态更新器决定 } def _build_messages(self, prompt, context): # 这里可以插入历史上下文管理逻辑 # 例如,只保留最近N轮与本阶段相关的对话,防止token超限 messages = [] # ... 构建消息列表的逻辑 return messages

关键点:

  • 使用tenacity等库实现重试机制:网络波动、API限流是常事,自动重试能提升鲁棒性。
  • 分阶段设置temperature:规划阶段可以稍高(如0.2)以鼓励创造性;决策、评估阶段应更低(如0.1甚至0),以确保输出稳定、可预测。
  • 实现JSON解析的健壮性:Claude即使被要求输出JSON,有时也会加上前言后语。编写一个_extract_and_parse_json函数,使用正则表达式或字符串查找来定位并提取第一个完整的JSON对象。
  • 必须有降级策略:当API彻底失败时,不能直接崩溃。应返回一个错误标识,让状态更新器能将Agent置为PAUSED_FOR_REVIEWFAILED状态,并通知人工处理。

4.2 工具(Tools)的设计与管理:Agent的“手脚”

工具是Agent与环境交互的桥梁。一个设计良好的工具系统,是边界控制的关键。我建议使用类似LangChain Tools或自定义Pydantic模型的方式来定义工具。

from pydantic import BaseModel, Field from typing import Type, Optional import inspect class Tool(BaseModel): """工具基类""" name: str description: str args_schema: Type[BaseModel] # 用Pydantic模型定义参数 func: callable class Config: arbitrary_types_allowed = True async def run(self, **kwargs): """执行工具,并返回字符串结果""" try: # 1. 参数验证 validated_args = self.args_schema(**kwargs) # 2. 执行实际函数 result = await self.func(**validated_args.dict()) return str(result) except Exception as e: return f"工具执行错误: {e}" # 工具参数模型示例 class ReadFileArgs(BaseModel): file_path: str = Field(description="要读取的文件的路径,必须是相对当前工作目录的路径") max_lines: Optional[int] = Field(default=100, description="最大读取行数,防止读取过大文件") # 工具函数 async def read_file_func(file_path: str, max_lines: int = 100) -> str: # 实现安全的文件读取逻辑,包含路径校验、权限检查等 if not os.path.exists(file_path): return f"错误:文件 '{file_path}' 不存在。" if not file_path.startswith('./src'): # 边界控制! return f"错误:无权访问 '{file_path}',工作空间限制为'./src'目录。" # ... 读取文件内容 return content # 注册工具 read_file_tool = Tool( name="read_file", description="读取指定文本文件的内容", args_schema=ReadFileArgs, func=read_file_func ) class ToolRegistry: """工具注册中心""" def __init__(self): self._tools: Dict[str, Tool] = {} def register(self, tool: Tool): if tool.name in self._tools: raise ValueError(f"工具 '{tool.name}' 已注册。") self._tools[tool.name] = tool def get_tool(self, name: str) -> Optional[Tool]: return self._tools.get(name) def get_tool_descriptions(self) -> str: """生成给Claude看的工具描述字符串""" descriptions = [] for name, tool in self._tools.items(): # 利用Pydantic模型的schema自动生成参数描述 schema = tool.args_schema.schema() args_desc = ", ".join([f"{k}: {v.get('description', '')}" for k, v in schema['properties'].items()]) descriptions.append(f"- {name}: {tool.description} 参数: ({args_desc})") return "\n".join(descriptions)

工具设计黄金法则:

  1. 无副作用验证:工具函数内部必须进行严格的输入验证(如路径合法性、参数范围),这是安全的第一道防线。
  2. 权限隔离:根据Agent的职责,注册不同的工具集。一个代码分析Agent可能只有read_file,analyze_code工具,而一个部署Agent则拥有run_shell,deploy_service等工具。
  3. 结果标准化:所有工具返回字符串结果,方便记录到action_history和作为latest_observation。对于复杂结果,可以返回JSON字符串。
  4. 异常捕获:工具内部必须捕获所有异常,并返回格式化的错误信息,而不是抛出异常导致整个Agent崩溃。

4.3 防死循环与超时控制:为Loop装上保险丝

即使有完美的架构,Agent也可能陷入逻辑怪圈或等待一个永远不会发生的外部事件。必须设置多层保险。

1. 迭代次数限制:如前所述,在AgentState中设置max_iterations(如50或100)。这是最直接、最有效的硬性停止条件。

2. 超时控制(Timeout):为每个循环迭代或每个阶段执行设置时间限制。

import asyncio import signal class TimeoutException(Exception): pass def timeout_handler(signum, frame): raise TimeoutException("操作超时") async def run_stage_with_timeout(stage_func, state, timeout_seconds=30): """带超时限制的阶段执行""" try: # 对于异步函数,使用asyncio.wait_for return await asyncio.wait_for(stage_func(state), timeout=timeout_seconds) except asyncio.TimeoutError: logger.warning(f"阶段 {stage_func.__name__} 执行超时(>{timeout_seconds}秒)") # 更新状态,记录超时,并可能进入暂停或失败状态 state.latest_observation = f"阶段执行超时,限制为{timeout_seconds}秒。" state.current_stage = AgentStage.PAUSED_FOR_REVIEW return state

3. 状态重复检测:Agent可能在不同迭代中进入相似或相同的状态,原地打转。可以在状态更新器中加入检测逻辑。

def detect_stagnation(state: AgentState, history_window=5): """检测状态是否停滞(最近N次迭代的核心状态未变)""" if len(state.action_history) < history_window: return False recent_actions = state.action_history[-history_window:] # 检查最近N个动作是否高度相似(例如,调用的工具和输入参数都相同) first_action = recent_actions[0] for action in recent_actions[1:]: if action.get('tool_name') != first_action.get('tool_name'): return False if action.get('input') != first_action.get('input'): return False # 如果最近N个动作都一样,说明停滞了 logger.warning(f"检测到状态停滞,最近{history_window}次动作重复。") return True # 在状态更新器中调用 if detect_stagnation(state): state.latest_observation = "检测到可能陷入循环,暂停以等待审查。" state.current_stage = AgentStage.PAUSED_FOR_REVIEW

4. 看门狗(Watchdog)进程:对于极其重要的Agent进程,可以启动一个独立的看门狗进程来监控主Agent进程的心跳。如果主进程卡死,看门狗可以重启它或上报警报。这属于系统级的设计,在此不展开。

将这些策略组合使用,就能构建一个既有强大行动力,又不会“发疯”或“卡死”的可靠Agent Loop。

5. 实战:构建一个可控的代码分析Agent

让我们将上述所有设计付诸实践,构建一个简单的、可控的“代码库分析Agent”。它的目标是:给定一个代码目录,自动分析其技术栈、主要模块和潜在问题。

5.1 定义目标与状态初始化

目标:“分析项目目录./my_project中的主要技术栈、核心模块结构,并找出可能存在的代码问题(如安全漏洞、性能瓶颈)。”

初始化状态:

initial_state = AgentState( task_id="code_analysis_001", main_goal="分析项目目录 ./my_project 中的主要技术栈、核心模块结构,并找出可能存在的代码问题(如安全漏洞、性能瓶颈)。", current_stage=AgentStage.INITIALIZING, max_iterations=30, pause_points=["after_plan"] # 我们希望在生成计划后,先让人看一眼 )

5.2 配置工具集

我们只为这个Agent注册必要的、安全的工具:

tool_registry = ToolRegistry() tool_registry.register(read_file_tool) # 前面定义的读文件工具 tool_registry.register(list_files_tool) # 新增:列出目录文件的工具 tool_registry.register(analyze_code_with_ast_tool) # 新增:使用AST进行简单代码分析的工具 # 注意:没有写文件、执行shell命令的工具,确保安全边界。

5.3 分阶段提示词与执行跟踪

循环开始:状态从INITIALIZING进入PLANNING阶段。规划提示词会引导Claude输出类似这样的计划:

{ "reasoning": "这是一个代码分析任务。我需要先探索项目结构,识别主要文件类型,然后针对关键代码文件进行深入分析以判断技术栈和发现问题。", "new_sub_tasks": [ {"id": "task_1", "description": "使用list_files工具,递归列出./my_project目录下的所有文件,并过滤出.py, .js, .json, package.json, requirements.txt等关键文件。"}, {"id": "task_2", "description": "读取package.json或requirements.txt等依赖管理文件,确定项目的主要技术栈和版本。"}, {"id": "task_3", "description": "选取项目中的核心入口文件(如main.py, app.js)进行AST分析,理解模块结构和主要函数。"}, {"id": "task_4", "description": "根据已了解的技术栈,对关键代码文件进行模式匹配,寻找常见的安全漏洞(如SQL注入、硬编码密码)或性能问题(如循环内的重复计算)。"} ], "is_goal_achieved": false }

状态更新器会将new_sub_tasks压入sub_task_stack,并将状态置为PAUSED_FOR_REVIEW(因为我们在pause_points中设置了after_plan)。

人工干预点:此时,外部控制器(可能是Web界面或命令行)将状态展示给用户。用户可以审核这个计划,批准、修改或添加子任务。批准后,控制器将状态current_stage改为DECIDING,Loop继续。

后续自动化循环:

  1. 决策阶段:Claude看到栈顶任务是task_1(列出文件),并从工具描述中知道有list_files工具可用。它输出决策:使用list_files工具,输入目录路径./my_project
  2. 执行阶段:引擎调用list_files_tool.run(directory="./my_project"),得到文件列表字符串。
  3. 评估阶段:Claude评估文件列表结果,判断task_1完成,并总结出关键文件有哪些。状态更新器将task_1移入completed_tasks,并将文件列表作为latest_observation
  4. 路由器发现sub_task_stack不为空,于是状态回到DECIDING,开始处理task_2(分析依赖文件)... 如此循环,直到所有子任务完成,或达到迭代上限。

5.4 最终输出与总结

当所有子任务完成,且评估阶段判断目标已达成(或达到迭代上限)时,循环终止。最终状态中的action_historycompleted_tasks就构成了完整的分析报告。外部控制器可以将这些信息整理成一份格式化的文档输出给用户。

通过这个实战例子,你可以看到,一个原本可能杂乱无章、四处碰壁的代码分析过程,被我们设计的可控Loop分解成了清晰、有序、可监控、可干预的步骤。Agent的每一步都在预设的轨道内运行,既发挥了Claude强大的理解和推理能力,又确保了整个过程的安全与可靠。

设计可控的Agent Loop,本质是在赋予AI自主性的同时,为它建立一套可靠的行为规范和管理体系。这不仅仅是技术实现,更是一种工程哲学:任何自动化系统,其可控性必须优先于其自主性。希望这套从理论到实践的设计方案,能帮助你打造出真正强大且可靠的Claude Agent。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询