1. 从一次“卡死”的对话说起为什么需要理解主循环那天下午我正在调试一个通过 Claude API 构建的代码生成助手。需求很简单用户输入一段自然语言描述比如“写一个Python函数计算斐波那契数列的第n项”助手需要生成代码然后我本地执行并返回结果。一开始很顺利但当我尝试一个更复杂的请求——“帮我写一个完整的Flask Web应用包含用户注册、登录和JWT鉴权”时问题出现了。助手基于Claude的模型开始“滔滔不绝”地输出代码一行接一行仿佛没有尽头。它生成了app.py、models.py、requirements.txt甚至还有config.py和templates目录下的HTML文件。然而整个过程是“一次性”的模型在生成一个长达数百行的响应后就陷入了沉默。我无法在它生成app.py的中途打断它问一句“等等你打算用SQLAlchemy还是直接用SQLite3数据库连接池的配置考虑了吗” 我只能被动地等待整个“巨无霸”响应结束然后在一大坨代码里费力地寻找可能的问题点。这次经历让我意识到许多开发者包括当时的我对像Claude这类大型语言模型的运作机制存在一个根本性的误解我们常常把它当作一个“黑盒函数”输入问题输出答案一次交互就结束。但在构建复杂、交互式的编码助手或智能体Agent时这种“单次请求-响应”模式是远远不够的。我们需要模型能够“思考”能够“暂停”能够根据中间结果或我的反馈来“调整”后续的生成方向。这就引出了“主循环”Main Loop这个概念。在Claude Code或更广义的基于LLM的代码生成/智能体系统的上下文中主循环远不止是编程里那个while True的循环。它是一个协调推理、执行、观察和决策的核心控制流架构。理解它意味着你从“调用API的用户”变成了“设计智能工作流的架构师”。简单来说Claude Code的主循环机制就是为了解决上述“一次性输出”的痛点让模型具备状态保持、多步推理、工具调用和环境交互的能力。它让AI从“静态应答机”进化成了“动态协作者”。2. 主循环的核心组件拆解不只是While True当我们谈论Claude Code的主循环时不能把它想象成一个孤立的魔法循环。它是一个由多个精密组件协同工作的系统。我们可以将其类比为一个经验丰富的程序员在解决问题时的思维和工作流程。2.1 状态管理State Management循环的“记忆体”这是主循环的基石。没有状态每次循环迭代都是一次失忆后的重启无法进行连贯的多步任务。会话历史Conversation History这是最基础的状态。它不仅仅保存了用户和AI的对话记录更重要的是保存了上下文Context。在主循环中每一次迭代产生的思考、执行结果、用户反馈都会被追加到历史中作为下一次迭代的输入。这解决了传统单次调用中上下文长度有限和无法累积的问题。任务目标与中间结果Task Intermediate Results循环需要一个明确的目标例如“修复这个报错的函数”并且在循环过程中会不断产生中间产物比如“第一步生成的代码片段”、“第二步执行后的错误日志”、“第三步根据错误调整后的新代码”。主循环必须能妥善地保存、引用和更新这些中间状态。工具调用状态Tool Call Status如果Claude Code可以调用外部工具如执行终端命令、读取文件、调用API那么哪些工具被调用过、调用的参数和返回结果是什么这些状态必须被维护。这避免了重复调用或基于过时结果进行决策。一个常见的误区是认为状态管理只是把所有的messages数组不断变长。在实际的高效实现中你需要考虑状态压缩与摘要。例如当历史会话非常长时你可以让模型自己生成一个对之前步骤的简短摘要替代冗长的原始记录以节省宝贵的上下文窗口并聚焦于核心信息。2.2 推理与规划模块Reasoning Planning Module循环的“大脑”这是主循环的智能核心。在每次循环迭代开始时模型需要基于当前状态决定“接下来做什么”。任务分解Task Decomposition面对复杂需求如“搭建一个博客系统”模型需要能将其分解为一系列可执行的子任务例如1. 设计数据库模型2. 创建后端API3. 实现前端页面4. 配置部署脚本。主循环会逐个攻克这些子任务。下一步行动预测Next Action Prediction在每一个子步骤中模型需要决定具体的行动。是直接生成代码还是需要先查询文档或者是执行一段已有的代码来验证假设这个决策过程就是模型根据当前代码状态、错误信息、任务目标进行推理的结果。思维链Chain-of-Thought的集成主循环天然是思维链的实践场。模型在输出最终行动前可以在内部先进行一番“思考”并将思考过程以特定格式如“让我们先分析一下这个错误它说的是导入失败...”输出到状态中。这使得模型的决策过程更加透明、可控也更容易纠偏。在我的Flask应用例子中一个具备良好规划能力的Claude Code主循环可能会先输出“目标创建带JWT的Flask应用。我将分步进行1. 设置项目结构和依赖2. 编写用户模型和数据库配置3. 实现注册/登录API4. 添加JWT生成与验证中间件。” 然后逐步执行并在每一步后等待我的确认或提供反馈。2.3 执行与工具调用层Execution Tool Call Layer循环的“手脚”规划好了就需要执行。这是主循环与外界代码执行环境、文件系统、网络等交互的接口。代码执行器Code Executor这是Claude Code最核心的工具之一。主循环可以生成一段代码Python、Shell等然后调用执行器来运行它。执行结果标准输出、标准错误、返回值被捕获并反馈给状态管理模块成为下一轮推理的依据。例如模型生成代码后执行如果遇到ModuleNotFoundError这个错误信息会被送入下一轮循环驱动模型去生成pip install命令或检查导入语句。文件系统操作File System Operations读取文件内容、写入新文件、列出目录结构。这使得模型能理解项目现状并对其进行增删改查。比如在添加新功能前先读取现有的models.py以了解已有的数据模型。外部API调用查询文档、获取天气数据、调用第三方服务等。扩展了模型的能力边界。关键点在于工具调用的标准化。通常这会通过一个“工具定义”的列表来实现每个工具都有名称、描述、参数schema。模型在推理后会输出一个结构化的请求如{action: execute_code, code: print(hello), language: python}主循环解析这个请求调用对应的工具函数并将结果格式化后返回给模型。2.4 观察与评估步骤Observation Evaluation Step循环的“感官”与“反思”执行之后必须观察结果并评估当前进展。观察执行结果捕获执行输出、错误、文件变化等所有“副作用”。这些观察需要被清晰、结构化地呈现给模型的推理模块。评估任务进度模型或一个外部评估器需要判断当前子任务是否完成整体目标完成了百分之多少是否遇到了无法自动解决的阻塞需要人工干预评估结果决定了循环的走向是继续下一个子任务还是重试当前任务或是请求帮助。错误处理与恢复这是体现主循环鲁棒性的关键。当执行出错时循环不应崩溃而应进入“调试子循环”。模型需要分析错误信息提出假设并尝试修复。例如执行Python代码报语法错误模型应能定位错误行尝试修正然后重新执行。将以上四个组件串联起来就形成了一个完整的主循环流程保持状态 - 推理规划 - 执行动作 - 观察评估 - 更新状态 - 继续推理...直到任务完成或达到终止条件如最大迭代次数。3. 实现一个简易的Claude Code主循环代码实战理论说得再多不如动手实现一个简化版。下面我们用Python和OpenAI/Anthropic API概念相通来模拟一个Claude Code的主循环完成一个实际任务“在当前目录创建一个‘calculator.py’文件实现加减乘除函数并写一个简单的测试。”我们将使用openai库假设Claude API调用方式类似和subprocess作为代码执行器。3.1 定义系统角色与工具首先定义系统的“角色”提示词让模型进入状态。import openai import subprocess import json import os # 系统提示词定义了AI的角色、能力和循环规则 SYSTEM_PROMPT 你是一个强大的编程助手可以编写和执行代码。 你将在一个主循环中工作循环的每一步你需要根据当前对话历史和最新结果决定下一步行动。 你可以使用以下工具 1. execute_python: 执行一段Python代码。参数{code: 要执行的python代码字符串} 2. write_file: 创建或覆盖一个文件。参数{path: 文件路径, content: 文件内容} 3. read_file: 读取一个文件的内容。参数{path: 文件路径} 4. task_complete: 当你认为任务已经完成时调用此工具结束循环。参数{summary: 任务总结} 你思考的格式如下 思考你基于当前情况的分析和推理 行动要调用的工具名称必须是上面之一 参数一个严格的JSON对象符合工具参数要求 用户会给出任务。从第一步开始请开始你的思考和行动。 3.2 构建主循环引擎接下来是主循环的核心逻辑。我们将维护一个消息历史并在每次循环中让模型决定行动。class SimpleCodingAgent: def __init__(self, api_key, modelgpt-4): # 此处仅为示例实际使用Claude需调整 self.client openai.OpenAI(api_keyapi_key) self.model model self.messages [{role: system, content: SYSTEM_PROMPT}] self.is_task_complete False def execute_python(self, code): 执行Python代码并返回结果 try: # 使用subprocess在安全隔离环境中执行这里为简化直接exec。 # 警告在生产环境中直接exec用户生成的代码极其危险必须使用沙箱如Docker容器、安全执行环境。 local_vars {} exec(code, {}, local_vars) output f代码执行成功。局部变量{local_vars} except Exception as e: output f代码执行出错{type(e).__name__}: {e} return output def write_file(self, path, content): 写入文件 try: with open(path, w, encodingutf-8) as f: f.write(content) return f文件 {path} 写入成功。 except Exception as e: return f写入文件失败{e} def read_file(self, path): 读取文件 try: with open(path, r, encodingutf-8) as f: content f.read() return f文件 {path} 的内容\n\n{content}\n except FileNotFoundError: return f文件 {path} 不存在。 except Exception as e: return f读取文件失败{e} def process_action(self, action_name, action_args): 根据模型指令调用对应工具 if action_name execute_python: return self.execute_python(action_args.get(code, )) elif action_name write_file: return self.write_file(action_args.get(path, ), action_args.get(content, )) elif action_name read_file: return self.read_file(action_args.get(path, )) elif action_name task_complete: self.is_task_complete True return f任务结束。总结{action_args.get(summary, )} else: return f错误未知行动 {action_name}。 def run_step(self): 运行主循环的一步 # 调用模型获取下一步的思考和行动 response self.client.chat.completions.create( modelself.model, messagesself.messages, temperature0.1, # 低温度保证决策稳定 max_tokens500 ) assistant_message response.choices[0].message.content self.messages.append({role: assistant, content: assistant_message}) print(f\n AI 响应 \n{assistant_message}) # 解析响应提取“思考”、“行动”、“参数” lines assistant_message.strip().split(\n) thought, action, args , , {} for line in lines: if line.startswith(思考): thought line[3:].strip() elif line.startswith(行动): action line[3:].strip() elif line.startswith(参数): args_str line[3:].strip() # 尝试解析JSON参数 try: # 这里假设参数是JSON字符串。更健壮的做法是用正则表达式提取。 args json.loads(args_str) except json.JSONDecodeError: args {raw_args: args_str} # 执行行动 if action: print(f\n 执行行动{action} ) print(f参数{args}) result self.process_action(action, args) print(f结果{result}) # 将结果作为“系统”或“用户”消息加入历史供下一轮参考 self.messages.append({role: user, content: f行动结果{result}}) else: print(未解析到明确行动。) def run(self, user_task): 启动主循环执行用户任务 print(f开始任务{user_task}) self.messages.append({role: user, content: user_task}) step_count 0 max_steps 10 # 防止无限循环 while not self.is_task_complete and step_count max_steps: step_count 1 print(f\n{*30} 步骤 {step_count} {*30}) self.run_step() if self.is_task_complete: print(\n✅ 任务成功完成) else: print(f\n⚠️ 达到最大步数{max_steps}任务未完成。)3.3 运行与结果分析现在我们初始化智能体并运行任务。# 假设已经设置了环境变量 OPENAI_API_KEY agent SimpleCodingAgent(api_keyos.getenv(OPENAI_API_KEY)) # 启动任务 agent.run(在当前目录创建一个calculator.py文件实现加、减、乘、除四个函数并写一个简单的测试。)可能的循环过程模拟步骤1AI思考“用户要求创建calculator.py并实现函数。我应该先创建文件写入函数定义。”AI行动write_file参数{path: calculator.py, content: def add(a, b): return a b\ndef subtract(a, b): return a - b\ndef multiply(a, b): return a * b\ndef divide(a, b): return a / b if b ! 0 else Error: Division by zero}步骤2AI观察上一步文件写入成功。AI思考“文件已创建。现在需要写一个测试。我应该读取刚创建的文件确认内容然后执行一个测试脚本。”AI行动read_file参数{path: calculator.py}步骤3AI观察成功读取到calculator.py的内容。AI思考“内容正确。现在编写并执行一个测试。”AI行动execute_python参数{code: import calculator\nprint(Test add:, calculator.add(5, 3))\nprint(Test subtract:, calculator.subtract(5, 3))\nprint(Test multiply:, calculator.multiply(5, 3))\ntry:\n print(Test divide:, calculator.divide(5, 3))\n print(Test divide by zero:, calculator.divide(5, 0))\nexcept Exception as e:\n print(Error:, e)}步骤4AI观察测试代码执行成功打印出预期结果。AI思考“所有函数测试通过任务完成。”AI行动task_complete参数{summary: 已成功创建calculator.py包含add, subtract, multiply, divide函数并执行了基础测试验证功能正常。}通过这个简化的例子你可以清晰地看到主循环如何一步步推进任务规划写文件- 执行 - 观察读文件确认- 再规划写测试- 再执行 - 最终评估完成。每个步骤的决策都依赖于之前步骤积累的状态历史消息。4. 生产级主循环的关键考量与避坑指南上面的简易版揭示了核心原理但要投入生产环境还有无数个“坑”需要填平。以下是基于实战经验的深度剖析。4.1 安全性第一生命线在允许AI执行代码甚至文件操作的主循环中安全是重中之重。代码执行沙箱化绝对不能在宿主机器上直接使用exec()或subprocess.run()执行AI生成的代码。必须使用隔离环境。方案一Docker容器。每个会话或每个任务在一个崭新的、资源受限的Docker容器中运行。任务完成后销毁容器。这是最彻底的隔离方案。方案二专用安全执行服务。使用像piston一个开源的多语言代码执行引擎或E2B、Replit的容器化API。它们提供了安全的沙箱和资源限制。关键配置限制CPU、内存、运行时间、网络访问通常应禁止外网访问、文件系统写入限制在/tmp等临时目录。工具调用的权限最小化仔细定义每个工具的能力。write_file工具应该限制可写入的目录路径防止覆盖系统关键文件。execute_command工具如果提供应限制可执行的命令白名单。输入过滤与审查对模型生成的行动参数进行严格的验证和过滤。例如检查文件路径是否包含..路径遍历攻击检查执行的命令是否在黑名单内。血的教训我曾在一个早期原型中让AI拥有执行任意Shell命令的能力。结果在一次调试中AI为了“清理临时文件”生成了rm -rf /tmp/*但由于路径解析的一个小bug命令变成了rm -rf /tmp /*注意空格差点酿成大祸。从此以后我彻底放弃了直接Shell命令调用所有操作都通过严格的API进行。4.2 循环控制与终止条件避免“鬼打墙”主循环必须能自己停下来否则会消耗大量资源并可能陷入死循环。最大迭代次数最基本的保障如最多循环20步。超过则强制终止并标记任务失败。超时控制整个任务或单步执行应有时间限制。目标达成判定除了依赖模型自己调用task_complete系统最好有一个外部评估器。这个评估器可以是一个简单的规则如“检测到文件X已创建且内容包含关键词Y”也可以是另一个轻量级AI模型来判断当前输出是否已满足用户初始请求。检测无效循环如果连续多步的“观察”结果没有实质性变化例如反复修改同一行代码但错误依旧或者行动模式重复如“写文件-读文件-写文件”循环应触发中断并可能请求人工干预。4.3 状态管理的优化与上下文窗口的博弈LLM的上下文长度是有限的如Claude 3的200K Token。在主循环中历史消息会快速增长很快触及上限。选择性记忆不是所有中间步骤都需要完整保留。可以只保留关键的决策点、错误信息和最终结果而省略冗长的、成功的代码执行输出除非出错。自动摘要这是一个高级技术。在历史达到一定长度后可以调用模型本身对之前的对话历史生成一个简洁、准确的摘要。然后用这个摘要替换掉大部分旧历史作为新的上下文起点。这要求摘要能保留任务目标、当前进展和关键决策依据。向量数据库外挂记忆将长篇的代码文件、文档内容、历史对话片段编码成向量存入向量数据库如Chroma、Pinecone。当模型需要参考时通过查询检索最相关的片段注入上下文。这相当于为模型扩展了一个外部“记忆体”。4.4 错误处理与鲁棒性让循环更“坚韧”错误不是例外而是常态。主循环必须优雅地处理错误。工具调用失败网络超时、文件不存在、权限不足等。循环应捕获这些异常并将清晰的错误信息如“写入文件失败权限被拒绝”反馈给模型让模型有机会调整策略例如尝试另一个路径或提示用户。模型输出解析失败模型可能不严格按照指定的“思考/行动/参数”格式输出。你的解析逻辑需要有容错能力比如使用正则表达式进行模糊匹配或者在解析失败时给模型一个友好的错误提示让它重新格式化输出。计划失败与重试如果模型制定的多步计划在某一步卡住例如一个无法修复的编译错误循环应该有能力“回退”或“重新规划”。可以设计一个机制当连续失败N次后清空最近几步的“错误尝试”历史让模型以更干净的状态重新思考整个任务或者提升推理的“温度”temperature来激发不同思路。5. 从主循环到智能体Agent架构的演进当你熟练掌握了主循环的构建你会发现你其实已经在构建一个智能体Agent。现代AI智能体框架如LangChain、AutoGen、CrewAI的核心就是一个高度优化和模块化的主循环。LangChain它的AgentExecutor本质上就是一个主循环运行器。你定义Tools工具、LLM大脑和AgentType推理逻辑如ReActAgentExecutor就负责维护状态、调用LLM决策、执行工具、处理观察的循环过程。专业代码智能体像OpenAI Codex、GitHub Copilot背后的系统以及Claude Code本身其内部的主循环更为复杂。它们可能集成了代码理解Code Understanding、测试生成Test Generation、静态分析Static Analysis等专用工具并且在规划时能调用复杂的代码知识图谱。理解主循环机制是理解和定制这些高级框架的基础。它让你不再满足于简单的单次问答而是能够设计出能够自主探索、试错并最终完成复杂目标的AI伙伴。无论是自动化脚本编写、代码库重构辅助还是交互式数据分析一个健壮的主循环都是实现这些愿景的引擎。
1. 从一次“卡死”的对话说起为什么需要理解主循环那天下午我正在调试一个通过 Claude API 构建的代码生成助手。需求很简单用户输入一段自然语言描述比如“写一个Python函数计算斐波那契数列的第n项”助手需要生成代码然后我本地执行并返回结果。一开始很顺利但当我尝试一个更复杂的请求——“帮我写一个完整的Flask Web应用包含用户注册、登录和JWT鉴权”时问题出现了。助手基于Claude的模型开始“滔滔不绝”地输出代码一行接一行仿佛没有尽头。它生成了app.py、models.py、requirements.txt甚至还有config.py和templates目录下的HTML文件。然而整个过程是“一次性”的模型在生成一个长达数百行的响应后就陷入了沉默。我无法在它生成app.py的中途打断它问一句“等等你打算用SQLAlchemy还是直接用SQLite3数据库连接池的配置考虑了吗” 我只能被动地等待整个“巨无霸”响应结束然后在一大坨代码里费力地寻找可能的问题点。这次经历让我意识到许多开发者包括当时的我对像Claude这类大型语言模型的运作机制存在一个根本性的误解我们常常把它当作一个“黑盒函数”输入问题输出答案一次交互就结束。但在构建复杂、交互式的编码助手或智能体Agent时这种“单次请求-响应”模式是远远不够的。我们需要模型能够“思考”能够“暂停”能够根据中间结果或我的反馈来“调整”后续的生成方向。这就引出了“主循环”Main Loop这个概念。在Claude Code或更广义的基于LLM的代码生成/智能体系统的上下文中主循环远不止是编程里那个while True的循环。它是一个协调推理、执行、观察和决策的核心控制流架构。理解它意味着你从“调用API的用户”变成了“设计智能工作流的架构师”。简单来说Claude Code的主循环机制就是为了解决上述“一次性输出”的痛点让模型具备状态保持、多步推理、工具调用和环境交互的能力。它让AI从“静态应答机”进化成了“动态协作者”。2. 主循环的核心组件拆解不只是While True当我们谈论Claude Code的主循环时不能把它想象成一个孤立的魔法循环。它是一个由多个精密组件协同工作的系统。我们可以将其类比为一个经验丰富的程序员在解决问题时的思维和工作流程。2.1 状态管理State Management循环的“记忆体”这是主循环的基石。没有状态每次循环迭代都是一次失忆后的重启无法进行连贯的多步任务。会话历史Conversation History这是最基础的状态。它不仅仅保存了用户和AI的对话记录更重要的是保存了上下文Context。在主循环中每一次迭代产生的思考、执行结果、用户反馈都会被追加到历史中作为下一次迭代的输入。这解决了传统单次调用中上下文长度有限和无法累积的问题。任务目标与中间结果Task Intermediate Results循环需要一个明确的目标例如“修复这个报错的函数”并且在循环过程中会不断产生中间产物比如“第一步生成的代码片段”、“第二步执行后的错误日志”、“第三步根据错误调整后的新代码”。主循环必须能妥善地保存、引用和更新这些中间状态。工具调用状态Tool Call Status如果Claude Code可以调用外部工具如执行终端命令、读取文件、调用API那么哪些工具被调用过、调用的参数和返回结果是什么这些状态必须被维护。这避免了重复调用或基于过时结果进行决策。一个常见的误区是认为状态管理只是把所有的messages数组不断变长。在实际的高效实现中你需要考虑状态压缩与摘要。例如当历史会话非常长时你可以让模型自己生成一个对之前步骤的简短摘要替代冗长的原始记录以节省宝贵的上下文窗口并聚焦于核心信息。2.2 推理与规划模块Reasoning Planning Module循环的“大脑”这是主循环的智能核心。在每次循环迭代开始时模型需要基于当前状态决定“接下来做什么”。任务分解Task Decomposition面对复杂需求如“搭建一个博客系统”模型需要能将其分解为一系列可执行的子任务例如1. 设计数据库模型2. 创建后端API3. 实现前端页面4. 配置部署脚本。主循环会逐个攻克这些子任务。下一步行动预测Next Action Prediction在每一个子步骤中模型需要决定具体的行动。是直接生成代码还是需要先查询文档或者是执行一段已有的代码来验证假设这个决策过程就是模型根据当前代码状态、错误信息、任务目标进行推理的结果。思维链Chain-of-Thought的集成主循环天然是思维链的实践场。模型在输出最终行动前可以在内部先进行一番“思考”并将思考过程以特定格式如“让我们先分析一下这个错误它说的是导入失败...”输出到状态中。这使得模型的决策过程更加透明、可控也更容易纠偏。在我的Flask应用例子中一个具备良好规划能力的Claude Code主循环可能会先输出“目标创建带JWT的Flask应用。我将分步进行1. 设置项目结构和依赖2. 编写用户模型和数据库配置3. 实现注册/登录API4. 添加JWT生成与验证中间件。” 然后逐步执行并在每一步后等待我的确认或提供反馈。2.3 执行与工具调用层Execution Tool Call Layer循环的“手脚”规划好了就需要执行。这是主循环与外界代码执行环境、文件系统、网络等交互的接口。代码执行器Code Executor这是Claude Code最核心的工具之一。主循环可以生成一段代码Python、Shell等然后调用执行器来运行它。执行结果标准输出、标准错误、返回值被捕获并反馈给状态管理模块成为下一轮推理的依据。例如模型生成代码后执行如果遇到ModuleNotFoundError这个错误信息会被送入下一轮循环驱动模型去生成pip install命令或检查导入语句。文件系统操作File System Operations读取文件内容、写入新文件、列出目录结构。这使得模型能理解项目现状并对其进行增删改查。比如在添加新功能前先读取现有的models.py以了解已有的数据模型。外部API调用查询文档、获取天气数据、调用第三方服务等。扩展了模型的能力边界。关键点在于工具调用的标准化。通常这会通过一个“工具定义”的列表来实现每个工具都有名称、描述、参数schema。模型在推理后会输出一个结构化的请求如{action: execute_code, code: print(hello), language: python}主循环解析这个请求调用对应的工具函数并将结果格式化后返回给模型。2.4 观察与评估步骤Observation Evaluation Step循环的“感官”与“反思”执行之后必须观察结果并评估当前进展。观察执行结果捕获执行输出、错误、文件变化等所有“副作用”。这些观察需要被清晰、结构化地呈现给模型的推理模块。评估任务进度模型或一个外部评估器需要判断当前子任务是否完成整体目标完成了百分之多少是否遇到了无法自动解决的阻塞需要人工干预评估结果决定了循环的走向是继续下一个子任务还是重试当前任务或是请求帮助。错误处理与恢复这是体现主循环鲁棒性的关键。当执行出错时循环不应崩溃而应进入“调试子循环”。模型需要分析错误信息提出假设并尝试修复。例如执行Python代码报语法错误模型应能定位错误行尝试修正然后重新执行。将以上四个组件串联起来就形成了一个完整的主循环流程保持状态 - 推理规划 - 执行动作 - 观察评估 - 更新状态 - 继续推理...直到任务完成或达到终止条件如最大迭代次数。3. 实现一个简易的Claude Code主循环代码实战理论说得再多不如动手实现一个简化版。下面我们用Python和OpenAI/Anthropic API概念相通来模拟一个Claude Code的主循环完成一个实际任务“在当前目录创建一个‘calculator.py’文件实现加减乘除函数并写一个简单的测试。”我们将使用openai库假设Claude API调用方式类似和subprocess作为代码执行器。3.1 定义系统角色与工具首先定义系统的“角色”提示词让模型进入状态。import openai import subprocess import json import os # 系统提示词定义了AI的角色、能力和循环规则 SYSTEM_PROMPT 你是一个强大的编程助手可以编写和执行代码。 你将在一个主循环中工作循环的每一步你需要根据当前对话历史和最新结果决定下一步行动。 你可以使用以下工具 1. execute_python: 执行一段Python代码。参数{code: 要执行的python代码字符串} 2. write_file: 创建或覆盖一个文件。参数{path: 文件路径, content: 文件内容} 3. read_file: 读取一个文件的内容。参数{path: 文件路径} 4. task_complete: 当你认为任务已经完成时调用此工具结束循环。参数{summary: 任务总结} 你思考的格式如下 思考你基于当前情况的分析和推理 行动要调用的工具名称必须是上面之一 参数一个严格的JSON对象符合工具参数要求 用户会给出任务。从第一步开始请开始你的思考和行动。 3.2 构建主循环引擎接下来是主循环的核心逻辑。我们将维护一个消息历史并在每次循环中让模型决定行动。class SimpleCodingAgent: def __init__(self, api_key, modelgpt-4): # 此处仅为示例实际使用Claude需调整 self.client openai.OpenAI(api_keyapi_key) self.model model self.messages [{role: system, content: SYSTEM_PROMPT}] self.is_task_complete False def execute_python(self, code): 执行Python代码并返回结果 try: # 使用subprocess在安全隔离环境中执行这里为简化直接exec。 # 警告在生产环境中直接exec用户生成的代码极其危险必须使用沙箱如Docker容器、安全执行环境。 local_vars {} exec(code, {}, local_vars) output f代码执行成功。局部变量{local_vars} except Exception as e: output f代码执行出错{type(e).__name__}: {e} return output def write_file(self, path, content): 写入文件 try: with open(path, w, encodingutf-8) as f: f.write(content) return f文件 {path} 写入成功。 except Exception as e: return f写入文件失败{e} def read_file(self, path): 读取文件 try: with open(path, r, encodingutf-8) as f: content f.read() return f文件 {path} 的内容\n\n{content}\n except FileNotFoundError: return f文件 {path} 不存在。 except Exception as e: return f读取文件失败{e} def process_action(self, action_name, action_args): 根据模型指令调用对应工具 if action_name execute_python: return self.execute_python(action_args.get(code, )) elif action_name write_file: return self.write_file(action_args.get(path, ), action_args.get(content, )) elif action_name read_file: return self.read_file(action_args.get(path, )) elif action_name task_complete: self.is_task_complete True return f任务结束。总结{action_args.get(summary, )} else: return f错误未知行动 {action_name}。 def run_step(self): 运行主循环的一步 # 调用模型获取下一步的思考和行动 response self.client.chat.completions.create( modelself.model, messagesself.messages, temperature0.1, # 低温度保证决策稳定 max_tokens500 ) assistant_message response.choices[0].message.content self.messages.append({role: assistant, content: assistant_message}) print(f\n AI 响应 \n{assistant_message}) # 解析响应提取“思考”、“行动”、“参数” lines assistant_message.strip().split(\n) thought, action, args , , {} for line in lines: if line.startswith(思考): thought line[3:].strip() elif line.startswith(行动): action line[3:].strip() elif line.startswith(参数): args_str line[3:].strip() # 尝试解析JSON参数 try: # 这里假设参数是JSON字符串。更健壮的做法是用正则表达式提取。 args json.loads(args_str) except json.JSONDecodeError: args {raw_args: args_str} # 执行行动 if action: print(f\n 执行行动{action} ) print(f参数{args}) result self.process_action(action, args) print(f结果{result}) # 将结果作为“系统”或“用户”消息加入历史供下一轮参考 self.messages.append({role: user, content: f行动结果{result}}) else: print(未解析到明确行动。) def run(self, user_task): 启动主循环执行用户任务 print(f开始任务{user_task}) self.messages.append({role: user, content: user_task}) step_count 0 max_steps 10 # 防止无限循环 while not self.is_task_complete and step_count max_steps: step_count 1 print(f\n{*30} 步骤 {step_count} {*30}) self.run_step() if self.is_task_complete: print(\n✅ 任务成功完成) else: print(f\n⚠️ 达到最大步数{max_steps}任务未完成。)3.3 运行与结果分析现在我们初始化智能体并运行任务。# 假设已经设置了环境变量 OPENAI_API_KEY agent SimpleCodingAgent(api_keyos.getenv(OPENAI_API_KEY)) # 启动任务 agent.run(在当前目录创建一个calculator.py文件实现加、减、乘、除四个函数并写一个简单的测试。)可能的循环过程模拟步骤1AI思考“用户要求创建calculator.py并实现函数。我应该先创建文件写入函数定义。”AI行动write_file参数{path: calculator.py, content: def add(a, b): return a b\ndef subtract(a, b): return a - b\ndef multiply(a, b): return a * b\ndef divide(a, b): return a / b if b ! 0 else Error: Division by zero}步骤2AI观察上一步文件写入成功。AI思考“文件已创建。现在需要写一个测试。我应该读取刚创建的文件确认内容然后执行一个测试脚本。”AI行动read_file参数{path: calculator.py}步骤3AI观察成功读取到calculator.py的内容。AI思考“内容正确。现在编写并执行一个测试。”AI行动execute_python参数{code: import calculator\nprint(Test add:, calculator.add(5, 3))\nprint(Test subtract:, calculator.subtract(5, 3))\nprint(Test multiply:, calculator.multiply(5, 3))\ntry:\n print(Test divide:, calculator.divide(5, 3))\n print(Test divide by zero:, calculator.divide(5, 0))\nexcept Exception as e:\n print(Error:, e)}步骤4AI观察测试代码执行成功打印出预期结果。AI思考“所有函数测试通过任务完成。”AI行动task_complete参数{summary: 已成功创建calculator.py包含add, subtract, multiply, divide函数并执行了基础测试验证功能正常。}通过这个简化的例子你可以清晰地看到主循环如何一步步推进任务规划写文件- 执行 - 观察读文件确认- 再规划写测试- 再执行 - 最终评估完成。每个步骤的决策都依赖于之前步骤积累的状态历史消息。4. 生产级主循环的关键考量与避坑指南上面的简易版揭示了核心原理但要投入生产环境还有无数个“坑”需要填平。以下是基于实战经验的深度剖析。4.1 安全性第一生命线在允许AI执行代码甚至文件操作的主循环中安全是重中之重。代码执行沙箱化绝对不能在宿主机器上直接使用exec()或subprocess.run()执行AI生成的代码。必须使用隔离环境。方案一Docker容器。每个会话或每个任务在一个崭新的、资源受限的Docker容器中运行。任务完成后销毁容器。这是最彻底的隔离方案。方案二专用安全执行服务。使用像piston一个开源的多语言代码执行引擎或E2B、Replit的容器化API。它们提供了安全的沙箱和资源限制。关键配置限制CPU、内存、运行时间、网络访问通常应禁止外网访问、文件系统写入限制在/tmp等临时目录。工具调用的权限最小化仔细定义每个工具的能力。write_file工具应该限制可写入的目录路径防止覆盖系统关键文件。execute_command工具如果提供应限制可执行的命令白名单。输入过滤与审查对模型生成的行动参数进行严格的验证和过滤。例如检查文件路径是否包含..路径遍历攻击检查执行的命令是否在黑名单内。血的教训我曾在一个早期原型中让AI拥有执行任意Shell命令的能力。结果在一次调试中AI为了“清理临时文件”生成了rm -rf /tmp/*但由于路径解析的一个小bug命令变成了rm -rf /tmp /*注意空格差点酿成大祸。从此以后我彻底放弃了直接Shell命令调用所有操作都通过严格的API进行。4.2 循环控制与终止条件避免“鬼打墙”主循环必须能自己停下来否则会消耗大量资源并可能陷入死循环。最大迭代次数最基本的保障如最多循环20步。超过则强制终止并标记任务失败。超时控制整个任务或单步执行应有时间限制。目标达成判定除了依赖模型自己调用task_complete系统最好有一个外部评估器。这个评估器可以是一个简单的规则如“检测到文件X已创建且内容包含关键词Y”也可以是另一个轻量级AI模型来判断当前输出是否已满足用户初始请求。检测无效循环如果连续多步的“观察”结果没有实质性变化例如反复修改同一行代码但错误依旧或者行动模式重复如“写文件-读文件-写文件”循环应触发中断并可能请求人工干预。4.3 状态管理的优化与上下文窗口的博弈LLM的上下文长度是有限的如Claude 3的200K Token。在主循环中历史消息会快速增长很快触及上限。选择性记忆不是所有中间步骤都需要完整保留。可以只保留关键的决策点、错误信息和最终结果而省略冗长的、成功的代码执行输出除非出错。自动摘要这是一个高级技术。在历史达到一定长度后可以调用模型本身对之前的对话历史生成一个简洁、准确的摘要。然后用这个摘要替换掉大部分旧历史作为新的上下文起点。这要求摘要能保留任务目标、当前进展和关键决策依据。向量数据库外挂记忆将长篇的代码文件、文档内容、历史对话片段编码成向量存入向量数据库如Chroma、Pinecone。当模型需要参考时通过查询检索最相关的片段注入上下文。这相当于为模型扩展了一个外部“记忆体”。4.4 错误处理与鲁棒性让循环更“坚韧”错误不是例外而是常态。主循环必须优雅地处理错误。工具调用失败网络超时、文件不存在、权限不足等。循环应捕获这些异常并将清晰的错误信息如“写入文件失败权限被拒绝”反馈给模型让模型有机会调整策略例如尝试另一个路径或提示用户。模型输出解析失败模型可能不严格按照指定的“思考/行动/参数”格式输出。你的解析逻辑需要有容错能力比如使用正则表达式进行模糊匹配或者在解析失败时给模型一个友好的错误提示让它重新格式化输出。计划失败与重试如果模型制定的多步计划在某一步卡住例如一个无法修复的编译错误循环应该有能力“回退”或“重新规划”。可以设计一个机制当连续失败N次后清空最近几步的“错误尝试”历史让模型以更干净的状态重新思考整个任务或者提升推理的“温度”temperature来激发不同思路。5. 从主循环到智能体Agent架构的演进当你熟练掌握了主循环的构建你会发现你其实已经在构建一个智能体Agent。现代AI智能体框架如LangChain、AutoGen、CrewAI的核心就是一个高度优化和模块化的主循环。LangChain它的AgentExecutor本质上就是一个主循环运行器。你定义Tools工具、LLM大脑和AgentType推理逻辑如ReActAgentExecutor就负责维护状态、调用LLM决策、执行工具、处理观察的循环过程。专业代码智能体像OpenAI Codex、GitHub Copilot背后的系统以及Claude Code本身其内部的主循环更为复杂。它们可能集成了代码理解Code Understanding、测试生成Test Generation、静态分析Static Analysis等专用工具并且在规划时能调用复杂的代码知识图谱。理解主循环机制是理解和定制这些高级框架的基础。它让你不再满足于简单的单次问答而是能够设计出能够自主探索、试错并最终完成复杂目标的AI伙伴。无论是自动化脚本编写、代码库重构辅助还是交互式数据分析一个健壮的主循环都是实现这些愿景的引擎。
