200行Python实现AI Agent核心架构与ReAct模式

1. 从零理解Agent架构：一个200行Python的实践之旅

最近在AI编程助手领域，像Claude Code、Cursor这样的工具越来越受欢迎。它们能帮我们完成复杂的编程任务，背后核心就是Agent技术。很多人好奇Agent到底是什么，其实最好的理解方式就是自己动手实现一个简化版。

今天我要分享的是用200行Python代码构建一个最小化的Agent系统。这个系统虽然简单，但包含了Agent最核心的三个要素：大语言模型（LLM）的推理能力、工具调用能力以及循环执行机制。通过这个实践，你会发现那些商业级Agent产品的设计思路其实和我们这个小项目一脉相承。

2. Agent核心架构解析

2.1 Agent的本质公式

Agent可以用一个简单的公式来定义：

code复制Agent = LLM + 工具 + 循环执行

• LLM：提供推理和决策能力，是Agent的"大脑"。它负责分析任务、制定计划并决定每一步该做什么。

• 工具：突破LLM的纯文本边界，让Agent能够与现实世界交互。在我们的实现中，工具包括读文件、写文件和执行终端命令。

• 循环执行：Agent不是一次性问答，而是通过多步骤迭代直到任务完成。比如Claude Code在帮你写一个完整项目时，可能会循环执行几十次：读文件→分析→写代码→运行测试→修Bug→再运行...

2.2 ReAct模式详解

ReAct（Reasoning + Acting）是目前最主流的Agent运行模式。它的工作流程如下：

1. 用户输入：接收任务描述
2. 思考阶段：模型先分析任务，决定下一步行动
3. 行动阶段：调用适当的工具执行具体操作
4. 观察结果：获取工具执行结果
5. 循环：根据结果继续思考并采取下一步行动，直到任务完成

这种模式模拟了人类解决问题的过程：先思考再行动，根据反馈调整策略。

3. 工具系统实现

3.1 基础工具定义

我们先实现三个最基础的工具，这些工具将扩展Agent的能力边界：

python复制# tools.py
import subprocess
from pathlib import Path

def read_file(path: str) -> str:
    """
    读取文件内容
    
    Args:
        path: 文件路径（相对于工作目录）
    
    Returns:
        文件内容字符串，如果文件不存在返回错误信息
    """
    try:
        file_path = Path(path)
        if not file_path.exists():
            return f"错误：文件 '{path}' 不存在"
        return file_path.read_text(encoding="utf-8")
    except Exception as e:
        return f"读取文件失败：{e}"

def write_to_file(path: str, content: str) -> str:
    """
    写入内容到文件（如果目录不存在会自动创建）
    
    Args:
        path: 文件路径
        content: 要写入的内容
    
    Returns:
        成功或失败的信息
    """
    try:
        file_path = Path(path)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(content, encoding="utf-8")
        return f"成功写入文件：{path}（{len(content)} 字符）"
    except Exception as e:
        return f"写入文件失败：{e}"

def run_terminal_command(command: str, working_dir: str = ".") -> str:
    """
    在终端执行命令
    
    Args:
        command: 要执行的 shell 命令
        working_dir: 命令执行的工作目录
    
    Returns:
        命令的标准输出和标准错误输出
    """
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=60,
            cwd=working_dir
        )
        output = ""
        if result.stdout:
            output += f"[stdout]\n{result.stdout}"
        if result.stderr:
            output += f"[stderr]\n{result.stderr}"
        if not output:
            output = f"命令执行完成，退出码：{result.returncode}"
        return output
    except subprocess.TimeoutExpired:
        return "命令执行超时（60秒）"
    except Exception as e:
        return f"命令执行失败：{e}"

# 工具注册表（名称 → 函数映射）
AVAILABLE_TOOLS = {
    "read_file": read_file,
    "write_to_file": write_to_file,
    "run_terminal_command": run_terminal_command,
}

3.2 工具设计要点

1. 错误处理：每个工具都包含完善的错误处理逻辑，确保任何异常情况都能被捕获并返回有意义的错误信息。

2. 路径处理：使用Python的pathlib模块处理文件路径，确保跨平台兼容性。

3. 安全考虑：对于执行终端命令的工具，我们设置了60秒的超时限制，防止长时间运行的命令阻塞系统。

4. 返回值标准化：所有工具都返回字符串格式的结果，方便LLM解析和理解。

4. ReAct Agent核心实现

4.1 系统提示设计

系统提示（System Prompt）是指导LLM行为的关键。我们的提示包含以下几个部分：

python复制SYSTEM_PROMPT = """你是一个强大的 AI 编程助手，能够通过使用工具来帮助完成各种编程任务。

## 可用工具

### read_file
读取指定文件的内容
用法：
<tool_call>
<tool_name>read_file</tool_name>
<path>文件路径</path>
</tool_call>

### write_to_file
将内容写入文件
用法：
<tool_call>
<tool_name>write_to_file</tool_name>
<path>文件路径</path>
<content>文件内容</content>
</tool_call>

### run_terminal_command
在终端执行命令
用法：
<tool_call>
<tool_name>run_terminal_command</tool_name>
<command>要执行的命令</command>
</tool_call>

## 工作方式

1. 收到任务后，先在 <thinking> 标签中分析任务，制定计划
2. 按计划使用工具，每次只调用一个工具
3. 根据工具返回的结果继续推理
4. 任务完成后，输出 <final_answer> 标签

## 注意事项

- 写代码时先规划文件结构，再逐一写入
- 每次工具调用后等待结果，再决定下一步
- 遇到错误时分析原因，尝试修复
- 不要假设工具执行的结果，必须等实际返回
"""

这个系统提示有几个关键设计点：

1. 工具说明：清晰定义每个工具的用途和调用格式
2. 工作流程：明确ReAct循环的各个阶段
3. 注意事项：提供编程任务的最佳实践指导

4.2 Agent类实现

下面是Agent的核心实现代码：

python复制# agent.py
import re
import anthropic
from tools import AVAILABLE_TOOLS

MAX_ITERATIONS = 20  # 防止无限循环

class ReActAgent:
    def __init__(self, working_dir: str = "."):
        self.client = anthropic.Anthropic()
        self.working_dir = working_dir
        self.messages = []
    
    def run(self, task: str) -> str:
        """
        执行任务，返回最终结果
        """
        print(f"\n{'='*60}")
        print(f"任务：{task}")
        print('='*60)
        
        self.messages = [{"role": "user", "content": f"<task>\n{task}\n</task>"}]
        
        for iteration in range(MAX_ITERATIONS):
            print(f"\n--- 第 {iteration + 1} 轮推理 ---")
            
            # 调用 LLM
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=4096,
                system=SYSTEM_PROMPT,
                messages=self.messages
            )
            
            assistant_message = response.content[0].text
            print(f"\n[模型输出]\n{assistant_message[:500]}{'...' if len(assistant_message) > 500 else ''}")
            
            self.messages.append({
                "role": "assistant",
                "content": assistant_message
            })
            
            # 检查是否完成
            if "<final_answer>" in assistant_message:
                final_answer = self._extract_tag(assistant_message, "final_answer")
                print(f"\n{'='*60}")
                print(f"[任务完成]\n{final_answer}")
                return final_answer
            
            # 解析工具调用
            tool_call = self._parse_tool_call(assistant_message)
            
            if tool_call is None:
                # 没有工具调用也没有最终答案，让模型继续
                self.messages.append({
                    "role": "user",
                    "content": "请继续执行任务，或者输出 <final_answer> 表示完成。"
                })
                continue
            
            # 执行工具
            tool_result = self._execute_tool(tool_call)
            print(f"\n[工具执行结果]\n{tool_result[:300]}{'...' if len(tool_result) > 300 else ''}")
            
            # 把工具结果作为用户消息返回
            self.messages.append({
                "role": "user",
                "content": f"<tool_result>\n{tool_result}\n</tool_result>"
            })
        
        return "任务执行超过最大轮次限制，未能完成"
    
    def _parse_tool_call(self, text: str) -> dict | None:
        """解析 XML 格式的工具调用"""
        if "<tool_call>" not in text:
            return None
        
        tool_name = self._extract_tag(text, "tool_name")
        if not tool_name:
            return None
        
        tool_call = {"name": tool_name.strip()}
        
        # 根据工具名提取参数
        if tool_name == "read_file":
            tool_call["path"] = self._extract_tag(text, "path", "")
        elif tool_name == "write_to_file":
            tool_call["path"] = self._extract_tag(text, "path", "")
            tool_call["content"] = self._extract_tag(text, "content", "")
        elif tool_name == "run_terminal_command":
            tool_call["command"] = self._extract_tag(text, "command", "")
        
        return tool_call
    
    def _execute_tool(self, tool_call: dict) -> str:
        """执行工具调用"""
        tool_name = tool_call["name"]
        
        if tool_name not in AVAILABLE_TOOLS:
            return f"错误：未知工具 '{tool_name}'"
        
        print(f"\n[执行工具] {tool_name}")
        
        try:
            if tool_name == "read_file":
                return AVAILABLE_TOOLS[tool_name](tool_call.get("path", ""))
            elif tool_name == "write_to_file":
                return AVAILABLE_TOOLS[tool_name](
                    tool_call.get("path", ""),
                    tool_call.get("content", "")
                )
            elif tool_name == "run_terminal_command":
                return AVAILABLE_TOOLS[tool_name](
                    tool_call.get("command", ""),
                    self.working_dir
                )
        except Exception as e:
            return f"工具执行出错：{e}"
    
    def _extract_tag(self, text: str, tag: str, default: str = "") -> str:
        """从 XML 格式文本中提取标签内容"""
        pattern = f"<{tag}>(.*?)</{tag}>"
        match = re.search(pattern, text, re.DOTALL)
        return match.group(1).strip() if match else default

4.3 关键实现细节

1. 消息循环管理：Agent维护一个消息列表，包含用户输入、模型响应和工具执行结果，形成完整的对话上下文。

2. 工具调用解析：使用正则表达式解析模型输出的XML格式工具调用指令，确保参数提取准确。

3. 迭代限制：设置MAX_ITERATIONS防止无限循环，确保任务能在合理时间内完成或失败。

4. 执行跟踪：添加详细的打印语句，方便调试和理解Agent的工作过程。

5. 实际应用示例

5.1 运行Agent

我们可以这样运行Agent来创建一个贪吃蛇游戏：

bash复制python agent.py ./snake "写一个贪吃蛇游戏，用 HTML、CSS 和 JavaScript 实现，代码分别放在 index.html、style.css、game.js 三个文件中"

Agent会执行以下步骤：

1. 分析任务，制定文件结构
2. 写index.html（骨架）
3. 写style.css（样式）
4. 写game.js（游戏逻辑）
5. 用run_terminal_command检查文件是否创建成功
6. 输出完成报告

5.2 执行过程解析

让我们看看Agent执行这个任务时的典型思考过程：

code复制[模型输出]
<thinking>
首先需要创建一个完整的贪吃蛇游戏，这需要三个文件：
1. index.html - 主HTML文件，包含游戏容器和引入其他文件
2. style.css - 游戏样式表
3. game.js - 游戏逻辑

我应该先创建HTML骨架，然后添加样式，最后实现游戏逻辑。
</thinking>

<tool_call>
<tool_name>write_to_file</tool_name>
<path>index.html</path>
<content>
<!DOCTYPE html>
<html>
<head>
    <title>贪吃蛇游戏</title>
    <link rel="stylesheet" href="style.css">
</head>
<body>
    <div id="game-container">
        <canvas id="game-canvas"></canvas>
    </div>
    <script src="game.js"></script>
</body>
</html>
</content>
</tool_call>

然后根据写入结果，Agent会继续创建其他文件，直到任务完成。

6. ReAct与Plan-and-Execute模式对比

除了ReAct，还有一种常见的Agent模式叫做Plan-and-Execute：

我们的实现采用了ReAct模式，因为它更简单直观，适合教学目的。商业级Agent通常会结合两种模式的优势，比如先做整体规划，再在每个步骤中使用ReAct进行细化执行。

7. 扩展与优化方向

虽然我们的200行实现已经展示了Agent的核心原理，但还有很多可以改进的地方：

7.1 错误恢复机制

当前实现中，如果工具执行失败，Agent只能依靠LLM的分析能力来尝试修复。可以添加专门的错误处理逻辑，比如：

• 自动重试机制
• 错误分类与标准修复流程
• 用户确认步骤（对于高风险操作）

7.2 上下文管理

长任务可能会超出LLM的上下文窗口限制。可以引入：

• 上下文摘要与压缩
• 长期记忆存储
• 关键信息提取与保留

7.3 工具扩展

基础三工具虽然有用，但可以添加更多实用工具：

• Git操作（提交、拉取、推送）
• 网络请求（API调用）
• 数据库查询
• 代码静态分析

7.4 用户交互

添加交互功能可以提升用户体验：

• 进度反馈
• 操作确认
• 多选项决策
• 实时状态显示

8. 实现中的经验与教训

在实际开发这个简化版Agent的过程中，我积累了一些有价值的经验：

1. 工具调用格式：最初尝试使用JSON格式定义工具调用，但发现LLM在生成结构化内容时更容易遵循XML格式。这可能是因为XML有明显的开始和结束标签，减少了语法错误。

2. 迭代限制：必须设置最大迭代次数，早期版本因为没有这个限制，当任务无法完成时会导致无限循环。20次对于简单任务已经足够，复杂任务可以适当增加。

3. 错误处理：工具执行中的错误信息应该尽可能详细但结构化，这样LLM才能准确理解问题所在。过于简略的错误信息会让LLM难以诊断问题。

4. 系统提示设计：系统提示需要明确定义工具的使用格式和工作流程。最初版本提示不够详细时，LLM经常尝试一次性调用多个工具或跳过思考步骤。

5. 执行可视化：添加详细的打印输出对调试和理解Agent的思考过程非常有帮助。在实际产品中，这些信息可以转化为用户可见的执行日志。

9. 从简化版到商业级Agent

理解了这个最小实现后，再看Claude Code等商业产品，你会发现它们的设计思路其实和我们这个小项目一脉相承，只是在以下方面做了增强：

1. 更完善的工具集：支持代码补全、调试、测试等专业开发者需要的全套工具。

2. 上下文管理：能够处理大型代码库和长会话，有效管理上下文窗口。

3. 用户交互：提供友好的界面和交互方式，比如代码差异展示、修改建议确认等。

4. 性能优化：减少不必要的LLM调用，提高响应速度。

5. 安全控制：限制危险操作，增加用户确认步骤。

这个200行的实现虽然简单，但已经包含了Agent最核心的思想。通过这个项目，我希望你能更深入地理解Agent技术，并在此基础上继续探索和扩展。