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很多人第一个想到的是ReActReasoning and Acting模式。它让模型以“Thought: ... Action: ... Observation: ...”的格式循环将推理和行动交织在一起。对于快速原型验证ReAct非常有效。但在追求“可控性”的生产环境中ReAct的固有缺陷就暴露出来了思考与行动耦合过紧模型输出的“Thought”和“Action”在一个响应里外部系统很难在模型“思考”之后、“行动”之前插入校验或修改。状态管理隐式且脆弱整个对话历史就是它的状态容易受到上下文长度限制且历史中的任何干扰都可能带偏后续循环。缺乏显式的流程阶段所有步骤看起来都一样难以实施差异化的控制策略比如在“决策”阶段加强审核在“执行”阶段放宽限制。因此对于可控性要求高的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_history和conversation_contextaction_history记录所有对环境的操作如调用了哪个工具输入输出是什么用于回溯和审计。conversation_context则只保留与模型推理相关的关键对话防止上下文被无关历史淹没。通常我只保留最近3-5轮的关键思考。sub_task_stack是复杂任务的生命线当主任务被拆解后子任务压入栈中。Agent永远只处理栈顶任务完成后再弹出。这天然实现了任务的分解与串行执行避免了思维混乱。对于可以并行的任务可以设计多个工作线程或协程每个线程管理自己的栈但这属于高级话题。pause_points实现软控制你可以在状态初始化时就指定在哪些环节后暂停。例如pause_points [“after_plan”]那么当current_stage从PLANNING切换到DECIDING之前路由器会检测到这个暂停点并将状态置为PAUSED_FOR_REVIEW等待外部指令。max_iterations是硬保险无论如何必须设置一个循环上限。这是防止无限循环的最后手段。达到上限后将状态置为FAILED并终止。3.2 阶段路由器与流程引擎Loop的节拍器阶段路由器是驱动Loop运转的节拍器。它的逻辑并不复杂但却需要严谨。其核心是一个基于当前状态主要是current_stage和latest_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, modelclaude-3-5-sonnet-latest): self.client AsyncAnthropic(api_keyapi_key) self.model model retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) 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( modelself.model, max_tokens4096, # 根据阶段调整 messagesmessages, system你是一个严谨的AI助手必须严格按照用户指定的格式输出。, # 可覆盖 temperature0.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_REVIEW或FAILED状态并通知人工处理。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(default100, 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( nameread_file, description读取指定文本文件的内容, args_schemaReadFileArgs, funcread_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)工具设计黄金法则无副作用验证工具函数内部必须进行严格的输入验证如路径合法性、参数范围这是安全的第一道防线。权限隔离根据Agent的职责注册不同的工具集。一个代码分析Agent可能只有read_file,analyze_code工具而一个部署Agent则拥有run_shell,deploy_service等工具。结果标准化所有工具返回字符串结果方便记录到action_history和作为latest_observation。对于复杂结果可以返回JSON字符串。异常捕获工具内部必须捕获所有异常并返回格式化的错误信息而不是抛出异常导致整个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_seconds30): 带超时限制的阶段执行 try: # 对于异步函数使用asyncio.wait_for return await asyncio.wait_for(stage_func(state), timeouttimeout_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 state3. 状态重复检测Agent可能在不同迭代中进入相似或相同的状态原地打转。可以在状态更新器中加入检测逻辑。def detect_stagnation(state: AgentState, history_window5): 检测状态是否停滞最近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_REVIEW4. 看门狗Watchdog进程对于极其重要的Agent进程可以启动一个独立的看门狗进程来监控主Agent进程的心跳。如果主进程卡死看门狗可以重启它或上报警报。这属于系统级的设计在此不展开。将这些策略组合使用就能构建一个既有强大行动力又不会“发疯”或“卡死”的可靠Agent Loop。5. 实战构建一个可控的代码分析Agent让我们将上述所有设计付诸实践构建一个简单的、可控的“代码库分析Agent”。它的目标是给定一个代码目录自动分析其技术栈、主要模块和潜在问题。5.1 定义目标与状态初始化目标“分析项目目录./my_project中的主要技术栈、核心模块结构并找出可能存在的代码问题如安全漏洞、性能瓶颈。”初始化状态initial_state AgentState( task_idcode_analysis_001, main_goal分析项目目录 ./my_project 中的主要技术栈、核心模块结构并找出可能存在的代码问题如安全漏洞、性能瓶颈。, current_stageAgentStage.INITIALIZING, max_iterations30, 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改为DECIDINGLoop继续。后续自动化循环决策阶段Claude看到栈顶任务是task_1列出文件并从工具描述中知道有list_files工具可用。它输出决策使用list_files工具输入目录路径./my_project。执行阶段引擎调用list_files_tool.run(directory./my_project)得到文件列表字符串。评估阶段Claude评估文件列表结果判断task_1完成并总结出关键文件有哪些。状态更新器将task_1移入completed_tasks并将文件列表作为latest_observation。路由器发现sub_task_stack不为空于是状态回到DECIDING开始处理task_2分析依赖文件... 如此循环直到所有子任务完成或达到迭代上限。5.4 最终输出与总结当所有子任务完成且评估阶段判断目标已达成或达到迭代上限时循环终止。最终状态中的action_history和completed_tasks就构成了完整的分析报告。外部控制器可以将这些信息整理成一份格式化的文档输出给用户。通过这个实战例子你可以看到一个原本可能杂乱无章、四处碰壁的代码分析过程被我们设计的可控Loop分解成了清晰、有序、可监控、可干预的步骤。Agent的每一步都在预设的轨道内运行既发挥了Claude强大的理解和推理能力又确保了整个过程的安全与可靠。设计可控的Agent Loop本质是在赋予AI自主性的同时为它建立一套可靠的行为规范和管理体系。这不仅仅是技术实现更是一种工程哲学任何自动化系统其可控性必须优先于其自主性。希望这套从理论到实践的设计方案能帮助你打造出真正强大且可靠的Claude Agent。
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很多人第一个想到的是ReActReasoning and Acting模式。它让模型以“Thought: ... Action: ... Observation: ...”的格式循环将推理和行动交织在一起。对于快速原型验证ReAct非常有效。但在追求“可控性”的生产环境中ReAct的固有缺陷就暴露出来了思考与行动耦合过紧模型输出的“Thought”和“Action”在一个响应里外部系统很难在模型“思考”之后、“行动”之前插入校验或修改。状态管理隐式且脆弱整个对话历史就是它的状态容易受到上下文长度限制且历史中的任何干扰都可能带偏后续循环。缺乏显式的流程阶段所有步骤看起来都一样难以实施差异化的控制策略比如在“决策”阶段加强审核在“执行”阶段放宽限制。因此对于可控性要求高的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_history和conversation_contextaction_history记录所有对环境的操作如调用了哪个工具输入输出是什么用于回溯和审计。conversation_context则只保留与模型推理相关的关键对话防止上下文被无关历史淹没。通常我只保留最近3-5轮的关键思考。sub_task_stack是复杂任务的生命线当主任务被拆解后子任务压入栈中。Agent永远只处理栈顶任务完成后再弹出。这天然实现了任务的分解与串行执行避免了思维混乱。对于可以并行的任务可以设计多个工作线程或协程每个线程管理自己的栈但这属于高级话题。pause_points实现软控制你可以在状态初始化时就指定在哪些环节后暂停。例如pause_points [“after_plan”]那么当current_stage从PLANNING切换到DECIDING之前路由器会检测到这个暂停点并将状态置为PAUSED_FOR_REVIEW等待外部指令。max_iterations是硬保险无论如何必须设置一个循环上限。这是防止无限循环的最后手段。达到上限后将状态置为FAILED并终止。3.2 阶段路由器与流程引擎Loop的节拍器阶段路由器是驱动Loop运转的节拍器。它的逻辑并不复杂但却需要严谨。其核心是一个基于当前状态主要是current_stage和latest_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, modelclaude-3-5-sonnet-latest): self.client AsyncAnthropic(api_keyapi_key) self.model model retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) 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( modelself.model, max_tokens4096, # 根据阶段调整 messagesmessages, system你是一个严谨的AI助手必须严格按照用户指定的格式输出。, # 可覆盖 temperature0.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_REVIEW或FAILED状态并通知人工处理。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(default100, 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( nameread_file, description读取指定文本文件的内容, args_schemaReadFileArgs, funcread_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)工具设计黄金法则无副作用验证工具函数内部必须进行严格的输入验证如路径合法性、参数范围这是安全的第一道防线。权限隔离根据Agent的职责注册不同的工具集。一个代码分析Agent可能只有read_file,analyze_code工具而一个部署Agent则拥有run_shell,deploy_service等工具。结果标准化所有工具返回字符串结果方便记录到action_history和作为latest_observation。对于复杂结果可以返回JSON字符串。异常捕获工具内部必须捕获所有异常并返回格式化的错误信息而不是抛出异常导致整个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_seconds30): 带超时限制的阶段执行 try: # 对于异步函数使用asyncio.wait_for return await asyncio.wait_for(stage_func(state), timeouttimeout_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 state3. 状态重复检测Agent可能在不同迭代中进入相似或相同的状态原地打转。可以在状态更新器中加入检测逻辑。def detect_stagnation(state: AgentState, history_window5): 检测状态是否停滞最近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_REVIEW4. 看门狗Watchdog进程对于极其重要的Agent进程可以启动一个独立的看门狗进程来监控主Agent进程的心跳。如果主进程卡死看门狗可以重启它或上报警报。这属于系统级的设计在此不展开。将这些策略组合使用就能构建一个既有强大行动力又不会“发疯”或“卡死”的可靠Agent Loop。5. 实战构建一个可控的代码分析Agent让我们将上述所有设计付诸实践构建一个简单的、可控的“代码库分析Agent”。它的目标是给定一个代码目录自动分析其技术栈、主要模块和潜在问题。5.1 定义目标与状态初始化目标“分析项目目录./my_project中的主要技术栈、核心模块结构并找出可能存在的代码问题如安全漏洞、性能瓶颈。”初始化状态initial_state AgentState( task_idcode_analysis_001, main_goal分析项目目录 ./my_project 中的主要技术栈、核心模块结构并找出可能存在的代码问题如安全漏洞、性能瓶颈。, current_stageAgentStage.INITIALIZING, max_iterations30, 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改为DECIDINGLoop继续。后续自动化循环决策阶段Claude看到栈顶任务是task_1列出文件并从工具描述中知道有list_files工具可用。它输出决策使用list_files工具输入目录路径./my_project。执行阶段引擎调用list_files_tool.run(directory./my_project)得到文件列表字符串。评估阶段Claude评估文件列表结果判断task_1完成并总结出关键文件有哪些。状态更新器将task_1移入completed_tasks并将文件列表作为latest_observation。路由器发现sub_task_stack不为空于是状态回到DECIDING开始处理task_2分析依赖文件... 如此循环直到所有子任务完成或达到迭代上限。5.4 最终输出与总结当所有子任务完成且评估阶段判断目标已达成或达到迭代上限时循环终止。最终状态中的action_history和completed_tasks就构成了完整的分析报告。外部控制器可以将这些信息整理成一份格式化的文档输出给用户。通过这个实战例子你可以看到一个原本可能杂乱无章、四处碰壁的代码分析过程被我们设计的可控Loop分解成了清晰、有序、可监控、可干预的步骤。Agent的每一步都在预设的轨道内运行既发挥了Claude强大的理解和推理能力又确保了整个过程的安全与可靠。设计可控的Agent Loop本质是在赋予AI自主性的同时为它建立一套可靠的行为规范和管理体系。这不仅仅是技术实现更是一种工程哲学任何自动化系统其可控性必须优先于其自主性。希望这套从理论到实践的设计方案能帮助你打造出真正强大且可靠的Claude Agent。
