从零构建基于Claude的代码生成智能体：深入ReAct原理与工程实践

1. 项目缘起为什么我要“重造轮子”最近几个月大语言模型LLM驱动的智能体Agent开发框架火得一塌糊涂。无论是学术界的ReAct还是工业界的LangChain、LlamaIndex都试图为开发者提供一套标准化的“脚手架”让我们能快速构建出能思考、能执行、能使用工具的AI应用。作为一名长期泡在AI工程化一线的开发者我自然也第一时间上手了这些框架。LangChain的链式调用很优雅ReAct的思想很深刻用它们快速搭建一个Demo原型体验确实很棒。但当我试图将一个基于Claude的代码生成智能体投入实际生产环境去处理我们团队内部复杂的、定制化的代码库时问题开始接踵而至。框架的抽象层带来了便利但也带来了“黑盒”。当智能体行为不符合预期或者需要深度定制其决策逻辑、工具调用流程时我发现自己像是在一个精心设计但结构复杂的迷宫里调试而不是在亲手搭建的系统里解决问题。框架的通用性牺牲了极致的性能和可控性一些我们业务特有的需求比如对代码变更进行极其精细的审计、与内部CI/CD系统的深度集成实现起来异常别扭甚至需要大量“打补丁”式的Hack。于是一个念头冒了出来既然我对ReAct的思想和LangChain的架构如此熟悉也对Claude API了如指掌为什么不抛开这些框架从零开始亲手实现一个专属于我们团队的“Claude Code”智能体呢这个想法并非为了否定现有框架的价值而是一次深入的“技术考古”和“肌肉记忆训练”。我想彻底弄明白一个能写代码的智能体它的“大脑”推理、“手”执行和“记忆”上下文究竟是如何协同工作的。这个过程远比调用AgentExecutor.run()要复杂和有趣得多。最终我完成了一个简洁、高效、完全可控的代码生成助手。它没有LangChain那么庞大的生态但每一个组件都是我亲手打磨知其然更知其所以然。接下来我就把这趟“造轮子”之旅的完整过程、核心设计以及踩过的坑毫无保留地分享给你。无论你是想深入理解Agent原理还是面临类似定制化需求相信都能从中获得启发。2. 核心蓝图拆解一个代码生成智能体的三大支柱在动手写第一行代码之前我们必须先想清楚目标。我们要构建的是一个能接收自然语言需求如“在用户服务类中添加一个根据邮箱查找用户的方法”并能在指定代码库的上下文中正确生成、修改甚至执行代码的Claude智能体。它不能天马行空必须脚踏实地基于现有代码库的实际情况来操作。通过对ReAct论文的研读和对LangChain等框架的逆向工程我提炼出了这样一个智能体不可或缺的三大核心支柱支柱一结构化推理与规划Reasoning Planning这是智能体的“大脑”。它决定了智能体如何理解任务、分解步骤、并决定下一步该做什么。ReAct范式Reasoning Acting的精髓就在这里模型在输出最终答案或执行动作前必须先进行一系列连贯的“思考”Reasoning这些思考会被记录在上下文中形成一条清晰的思维链Chain-of-Thought。对于代码生成任务推理可能包括“用户想要添加一个查询方法。首先我需要定位到用户服务类文件。然后分析现有类的结构和导入。接着设计符合项目规范的方法签名。最后编写方法实现并考虑异常处理。”支柱二工具与执行能力Tools Actuators这是智能体的“手”和“感官”。智能体不能只“想”必须能“做”。在代码上下文中核心工具包括文件读取工具读取指定路径的源代码文件获取当前内容。代码搜索工具在代码库中全局搜索特定的类、方法或模式。代码写入/修改工具这是最核心且最危险的工具。它负责将模型生成的代码变更应用到实际文件中。必须包含严格的验证和备份机制。代码执行/测试工具可选但推荐运行单元测试或简单的语法检查验证生成代码的有效性。支柱三上下文与状态管理Context State Management这是智能体的“记忆”和“工作台”。它需要维护一个贯穿整个交互过程的会话状态State这个状态至少包含对话历史用户的所有请求和模型的回复。当前任务正在处理的任务描述及其分解后的子目标。已执行动作历史智能体调用过哪些工具输入输出分别是什么。这对于防止循环操作和后续调试至关重要。代码库的当前快照信息例如刚刚读取了哪个文件的内容搜索到了哪些相关类。这三大支柱相互耦合共同运作。推理模块根据当前状态和任务决定调用哪个工具工具执行后产生的结果新的代码内容、搜索结果等会更新到状态中更新后的状态又为下一轮推理提供了新的依据。整个循环持续进行直到任务被标记为完成或失败。3. 从零搭建手把手实现核心引擎有了清晰的蓝图我们就可以开始编码了。我选择Python作为实现语言并使用anthropic官方SDK来调用Claude模型。整个项目的核心是一个CodeAgent类。3.1 定义智能体的状态与工具首先我们定义智能体的状态。这里我使用一个简单的dataclass来封装确保状态的可序列化和清晰性。from dataclasses import dataclass, field from typing import List, Dict, Any, Optional dataclass class AgentState: 智能体的核心状态容器 conversation_history: List[Dict[str, str]] field(default_factorylist) 格式[{role: user, content: ...}, {role: assistant, content: ...}] current_task: Optional[str] None action_history: List[Dict[str, Any]] field(default_factorylist) 记录每次工具调用的详情用于调试和防止循环 code_context: Dict[str, Any] field(default_factorydict) 临时存储如当前查看的文件路径、内容等 # 工具基类定义 class Tool: def __init__(self, name: str, description: str): self.name name self.description description def run(self, **kwargs) - str: 执行工具返回结果字符串。必须被具体工具重写。 raise NotImplementedError class ReadFileTool(Tool): def __init__(self): super().__init__(read_file, 读取指定路径的源代码文件内容。) def run(self, file_path: str) - str: try: with open(file_path, r, encodingutf-8) as f: content f.read() return f文件 {file_path} 的内容如下\n\n{content}\n except FileNotFoundError: return f错误找不到文件 {file_path}。 except Exception as e: return f读取文件 {file_path} 时发生错误{str(e)} # 类似地可以定义 SearchCodeTool, WriteFileTool 等。 # WriteFileTool 必须极其谨慎实现前备份、差异对比等功能。3.2 实现ReAct推理循环的核心逻辑这是整个智能体的“发动机”。我们需要设计一个提示词Prompt引导Claude按照“思考 - 行动 - 观察”的循环进行。import anthropic import re class CodeAgent: def __init__(self, api_key: str, model: str claude-3-sonnet-20240229): self.client anthropic.Anthropic(api_keyapi_key) self.model model self.state AgentState() self.tools { read_file: ReadFileTool(), # ... 注册其他工具 } # 核心系统提示词 self.system_prompt 你是一个专业的软件开发助手专门负责在给定的代码库中执行代码生成和修改任务。 你必须严格按照以下格式进行响应 思考在这里详细分析当前任务、已有信息并规划下一步。 行动要执行的动作必须是以下工具之一{tool_names} 行动输入以JSON格式提供该工具所需的参数例如 {{file_path: src/service/user.py}} 观察工具执行后的结果文本 规则 1. 你必须先“思考”再“行动”。 2. “行动”只能从可用工具中选择。 3. “观察”部分将由系统在你执行行动后自动填充你不需要在第一次响应中填写。 4. 如果任务已经完成或无法继续请在“思考”中得出结论并输出“最终答案你的总结或最终代码”。 5. 始终基于“观察”到的事实进行下一步推理不要臆测。 可用工具 {tool_descriptions} 当前代码库根目录/project/src 现在开始处理任务。 self._update_system_prompt() def _update_system_prompt(self): 动态更新提示词中的工具列表 tool_names list(self.tools.keys()) tool_descs \n.join([f- {name}: {tool.description} for name, tool in self.tools.items()]) self.system_prompt_formatted self.system_prompt.format( tool_names, .join(tool_names), tool_descriptionstool_descs ) def run(self, task: str) - str: 执行一个任务 self.state.current_task task self.state.conversation_history.append({role: user, content: task}) max_steps 10 # 防止无限循环 for step in range(max_steps): # 1. 构建给Claude的完整消息历史 messages self._build_messages() # 2. 调用Claude API response self.client.messages.create( modelself.model, max_tokens2048, systemself.system_prompt_formatted, messagesmessages ) assistant_message response.content[0].text # 3. 解析Claude的响应思考、行动、行动输入 thought, action, action_input self._parse_response(assistant_message) # 记录思考 self.state.conversation_history.append({role: assistant, content: f思考{thought}}) # 4. 检查是否是最终答案 if action.lower() final_answer: final_result self._extract_final_answer(assistant_message) return final_result # 5. 执行工具调用 if action not in self.tools: error_msg f错误未知行动 {action}。可用行动{list(self.tools.keys())} self.state.conversation_history.append({role: system, content: f观察{error_msg}}) continue try: # 安全解析action_input这里假设是JSON字符串 import json params json.loads(action_input) if action_input else {} tool_result self.tools[action].run(**params) observation f观察{tool_result} except Exception as e: observation f观察执行工具 {action} 时出错{str(e)} # 6. 记录行动和观察结果到历史 self.state.action_history.append({ step: step, thought: thought, action: action, input: action_input, observation: observation }) self.state.conversation_history.append({role: system, content: observation}) return 错误达到最大步数限制任务可能未完成。 def _build_messages(self): 将状态中的对话历史转换为API所需的格式 # 这里需要将我们内部格式的历史转换为Claude API的messages格式 # 简化处理直接将conversation_history作为messages # 注意需要将之前步骤的“观察”也作为user或assistant消息的一部分融入历史 # 具体实现略取决于对话历史的组织方式 pass def _parse_response(self, text: str): 解析Claude的响应提取思考、行动和行动输入 thought_pattern r思考(.?)(?\n行动|\n最终答案|$) action_pattern r行动(.?)(?\n行动输入|$) input_pattern r行动输入(.?)(?\n观察|$) thought re.search(thought_pattern, text, re.DOTALL) action re.search(action_pattern, text, re.DOTALL) action_input re.search(input_pattern, text, re.DOTALL) thought thought.group(1).strip() if thought else action action.group(1).strip() if action else action_input action_input.group(1).strip() if action_input else return thought, action, action_input提示上述代码是一个高度简化的骨架。在实际实现中_build_messages方法需要精心设计确保将完整的“思考-行动-观察”链有效地组织到对话上下文中。一个常见的技巧是将整个链包括系统的“观察”都作为assistant消息的一部分或者交替使用user放观察和assistant放思考和行动消息以符合模型训练时的多轮对话格式。3.3 构建最关键的“写文件”工具对于代码生成智能体WriteFileTool是威力最大也最危险的工具。绝不能让它直接覆盖文件。我的实现包含了多层安全防护import os import shutil from datetime import datetime class WriteFileTool(Tool): def __init__(self, project_root: str, backup_dir: str ./agent_backups): super().__init__(write_file, 将内容写入指定文件。如果文件存在会生成差异对比并创建备份。) self.project_root project_root self.backup_dir backup_dir os.makedirs(backup_dir, exist_okTrue) def run(self, file_path: str, content: str) - str: # 1. 路径安全校验 abs_path os.path.join(self.project_root, file_path) if not os.path.abspath(abs_path).startswith(os.path.abspath(self.project_root)): return 错误尝试写入项目根目录之外的文件路径操作被拒绝。 # 2. 创建备份 backup_info if os.path.exists(abs_path): timestamp datetime.now().strftime(%Y%m%d_%H%M%S) backup_file os.path.join(self.backup_dir, f{os.path.basename(file_path)}.{timestamp}.bak) shutil.copy2(abs_path, backup_file) backup_info f已创建备份{backup_file}\n # 3. 生成差异对比 (使用difflib) import difflib with open(abs_path, r, encodingutf-8) as f: old_lines f.readlines() new_lines content.splitlines(keependsTrue) diff difflib.unified_diff(old_lines, new_lines, lineterm, fromfilea/file_path, tofileb/file_path) diff_text .join(diff) diff_info f变更差异如下\ndiff\n{diff_text}\n\n if diff_text else 文件内容无变化。\n else: diff_info 新文件\n # 4. 模拟写入请求确认在实际自动化中这里可以接入审批流或人工确认 # 本例中我们直接写入但在生产环境强烈建议加入确认环节。 try: os.makedirs(os.path.dirname(abs_path), exist_okTrue) with open(abs_path, w, encodingutf-8) as f: f.write(content) return f{backup_info}{diff_info}文件 {file_path} 已成功写入。 except Exception as e: return f错误写入文件 {file_path} 失败{str(e)}这个工具实现了四个关键安全层路径约束、自动备份、差异预览和异常捕获。在生产系统中你甚至可以在第4步前插入一个暂停将差异发送到Slack或邮件等待人工批准从而实现“人机协同”的代码变更。4. 实战演练让智能体完成一个真实任务理论说得再多不如跑一个实例。假设我们的项目结构如下/project/src ├── service │ └── user_service.py └── main.pyuser_service.py内容如下class UserService: def get_user_by_id(self, user_id: int): # 模拟数据库查询 return {id: user_id, name: Test User}现在我们给智能体下达任务“在UserService类中添加一个根据邮箱查找用户的方法get_user_by_email。”启动智能体并运行后我们可以通过日志观察它的思考过程以下为模拟输出用户在 UserService 类中添加一个根据邮箱查找用户的方法 get_user_by_email。 助理思考用户想要在UserService类中添加新方法。首先我需要找到这个类所在的文件。根据代码库根目录很可能在src/service目录下。常见的命名是user_service.py或user.py。我应该先读取这个文件确认类的结构和现有方法然后再编写新方法。 行动read_file 行动输入{file_path: src/service/user_service.py} 系统观察文件 src/service/user_service.py 的内容如下class UserService: def get_user_by_id(self, user_id: int): # 模拟数据库查询 return {id: user_id, name: Test User}助理思考很好文件找到了类结构很简单。现在我需要设计get_user_by_email方法。它应该接收一个email字符串参数返回用户信息或None。为了保持一致性返回格式可以类似get_user_by_id。现在开始编写新内容。我将生成整个文件的新版本。 行动write_file 行动输入{file_path: src/service/user_service.py, content: class UserService:\n def get_user_by_id(self, user_id: int):\n # 模拟数据库查询\n return {\id\: user_id, \name\: \Test User\}\n\n def get_user_by_email(self, email: str):\n # 根据邮箱查询用户逻辑\n # 此处为示例实际应连接数据库\n if email \knownexample.com\:\n return {\id\: 2, \name\: \Known User\, \email\: email}\n return None}系统观察已创建备份./agent_backups/user_service.py.20231027_143022.bak 变更差异如下--- a/src/service/user_service.py b/src/service/user_service.py -1,4 1,11 class UserService: def get_user_by_id(self, user_id: int): # 模拟数据库查询 return {id: user_id, name: Test User} def get_user_by_email(self, email: str): # 根据邮箱查询用户逻辑 # 此处为示例实际应连接数据库 if email \knownexample.com\: return {\id\: 2, \name\: \Known User\, \email\: email} return None文件src/service/user_service.py已成功写入。助理思考文件已成功更新添加了get_user_by_email方法。任务完成。 最终答案已在src/service/user_service.py文件的UserService类中成功添加get_user_by_email(self, email: str)方法。方法实现了基本的邮箱查询逻辑示例并保持了与现有方法一致的返回格式。变更前已自动备份原文件。通过这个简单的例子你可以清晰地看到智能体“思考-行动-观察”的完整循环。它先通过read_file工具获取上下文然后规划行动最后使用write_file工具实施变更并给出了清晰的最终答案。 ## 5. 深度优化超越基础实现的进阶技巧 一个能跑起来的原型只是第一步。要让这个“手搓”的智能体真正可靠、强大需要在以下几个方向进行深度优化 ### 5.1 提示词工程让推理更稳定、更可控 最初的系统提示词虽然能工作但智能体有时会“叛逆”不按格式输出或者做出匪夷所思的推理。通过大量测试我总结出几个优化点 - **强化格式指令**在提示词开头和结尾重复强调输出格式并使用“你必须”、“严格遵循”等强约束性词语。甚至可以提供更详细的格式示例。 - **分阶段任务分解**对于复杂任务可以在用户请求中或系统提示词里引导智能体进行更细粒度的分解。例如“处理此任务时请按顺序考虑1. 理解需求2. 定位相关文件3. 分析现有代码4. 设计变更方案5. 执行变更6. 验证变更。” - **提供领域知识**在提示词中嵌入项目特定的编码规范、常用库的导入风格、目录结构说明等。这能极大提升生成代码的契合度。 ### 5.2 状态管理的艺术平衡上下文长度与信息完整性 Claude等模型有上下文窗口限制。随着对话轮次和工具调用增加历史记录会迅速膨胀。我们需要一个智能的状态管理策略 - **选择性记忆**不是所有“观察”都需要完整保留。对于read_file读出的长篇代码可以只存储文件路径和一个内容摘要如前N行后N行当智能体需要再次引用时可以动态重新读取或从缓存中提取关键片段。 - **动作摘要**将action_history中的详细观察结果在几轮之后总结成一句话例如“已读取user_service.py文件并确认类结构。” 这样可以释放大量token。 - **分层上下文**将上下文分为“会话记忆”高层次任务进展和“工作记忆”最近几步的详细操作。每次请求主要提供“工作记忆”和高层次的“会话记忆”摘要。 ### 5.3 工具设计的边界与安全 工具是智能体能力的延伸也是主要的风险点。 - **工具权限粒度化**不要只有一个万能的write_file。可以拆分为create_file仅创建新文件、edit_file_section通过指定行号范围进行编辑、replace_code_pattern搜索替换等。更细的粒度意味着更可控的操作。 - **操作前模拟与验证**对于写操作可以设计一个dry_run模式。在此模式下WriteFileTool只生成差异报告而不实际写入供人工审核。 - **集成代码质量工具**在WriteFileTool的run方法内部写入文件后可以自动调用项目的格式化工具如black、prettier和linter如pylint、flake8并将检查结果作为“观察”的一部分反馈给智能体让它有机会自行修正问题。 ### 5.4 处理复杂任务与错误恢复 智能体不是万能的它会犯错也会遇到无法理解的任务。 - **设置步数限制与超时**正如基础代码中的max_steps这是防止无限循环的保险丝。 - **定义明确的终止状态**除了“最终答案”还应定义“任务失败”状态并让智能体学会在遇到不可逾越的障碍如关键文件缺失、工具连续错误时清晰地报告失败原因而不是陷入死循环。 - **实现“撤销”或“回滚”工具**当智能体执行了一系列错误操作后一个连接到备份系统的revert_change工具可以快速将代码库恢复到指定时间点这比手动修复要快得多。 - **人类干预点**在关键决策点如是否要删除一个文件、是否要修改一个被多处引用的函数签名设计流程让智能体暂停并生成一份清晰的摘要报告给人类开发者做决策。 ## 6. 与LangChain的对比我们获得了什么又失去了什么 从头实现一遍之后再回头看LangChain视角完全不同了。这不是一个“孰优孰劣”的问题而是一个“选择与权衡”的问题。 **我们获得的东西** 1. **极致的透明度和可控性**每一行代码你都了如指掌。当智能体行为异常时你可以像调试普通程序一样在推理循环、状态管理或工具调用的任何一环设置断点精准定位问题。没有“魔法”。 2. **轻量与高性能**移除了框架的所有抽象层和通用适配器你的智能体引擎只包含必需的功能。这意味着更少的依赖、更快的启动速度和更低的内存开销。在需要高并发或资源敏感的环境中这一点至关重要。 3. **深度定制能力**你可以轻松地将智能体与公司内部的任何系统项目管理、代码审核、监控告警无缝集成。工具的定义完全自由不受框架生态的限制。 4. **深刻的理解**这个过程强迫你深入思考ReAct的每一个细节理解智能体与环境的交互本质。这种理解是使用任何高级框架都无法替代的底层知识。 **我们失去的东西** 1. **开箱即用的丰富生态**LangChain提供了上百种现成的工具搜索引擎、计算器、各种API连接器、几十种文档加载器、以及多种记忆策略。要自己实现这些需要巨大的工作量。 2. **社区与最佳实践**使用主流框架意味着你站在巨人的肩膀上。遇到的大多数问题都能在社区找到答案框架本身也集成了很多经过验证的最佳实践如错误的处理、提示词的优化。 3. **开发速度**对于大多数标准场景如基于文档的QA、简单的工具调用使用LangChain可能在几分钟内就能搭出可用的原型而从零开始则需要数天甚至数周。 4. **长期维护成本**自己实现的框架需要自己维护、升级、修复安全漏洞。而LangChain有专业的团队和活跃的社区在持续做这件事。 **所以我的结论是****“从零实现”是一次绝佳的学习和深度定制路径而“使用成熟框架”是大多数生产应用的正确起点。** 对于核心的、差异化的、对性能和控制力有极端要求的智能体应用在充分理解原理后进行自研是合理的。而对于需要快速验证想法、利用丰富生态、或功能相对标准的场景LangChain无疑是更高效的选择。经过这次实践我再使用LangChain时能更清楚地知道它在背后帮我做了什么也更能判断何时应该绕过它的抽象直接操作底层组件。这种“知其所以然”的自信是这次“造轮子”之旅带给我的最大财富。