从零构建基于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）这是智能体的“手”和“感官”。智能体不能只“想”，必须能“做”。在代码上下文中，核心工具包括：

1. 文件读取工具：读取指定路径的源代码文件，获取当前内容。
2. 代码搜索工具：在代码库中全局搜索特定的类、方法或模式。
3. 代码写入/修改工具：这是最核心且最危险的工具。它负责将模型生成的代码变更应用到实际文件中。必须包含严格的验证和备份机制。
4. 代码执行/测试工具（可选但推荐）：运行单元测试或简单的语法检查，验证生成代码的有效性。

支柱三：上下文与状态管理（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_factory=list) """格式：[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]""" current_task: Optional[str] = None action_history: List[Dict[str, Any]] = field(default_factory=list) """记录每次工具调用的详情，用于调试和防止循环""" code_context: Dict[str, Any] = field(default_factory=dict) """临时存储如当前查看的文件路径、内容等""" # 工具基类定义 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', encoding='utf-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_key=api_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_descriptions=tool_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( model=self.model, max_tokens=2048, system=self.system_prompt_formatted, messages=messages ) 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_ok=True) 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', encoding='utf-8') as f: old_lines = f.readlines() new_lines = content.splitlines(keepends=True) diff = difflib.unified_diff(old_lines, new_lines, lineterm='', fromfile='a/'+file_path, tofile='b/'+file_path) diff_text = ''.join(diff) diff_info = f"变更差异如下：\n```diff\n{diff_text}\n```\n" if diff_text else "文件内容无变化。\n" else: diff_info = "（新文件）\n" # 4. 模拟写入，请求确认（在实际自动化中，这里可以接入审批流或人工确认） # 本例中，我们直接写入，但在生产环境强烈建议加入确认环节。 try: os.makedirs(os.path.dirname(abs_path), exist_ok=True) with open(abs_path, 'w', encoding='utf-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.py

user_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 == \"known@example.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 == \"known@example.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时，能更清楚地知道它在背后帮我做了什么，也更能判断何时应该绕过它的抽象，直接操作底层组件。这种“知其所以然”的自信，是这次“造轮子”之旅带给我的最大财富。