Agent的概念、原理与构建模式 - 代码深度解析

项目概述

本项目是一个完整的ReAct (Reasoning and Acting) Agent实现，展示了现代AI Agent的核心架构和工作原理。通过具体的代码实现，演示了如何构建一个能够进行推理、决策和行动的智能代理系统。
项目地址：Agent的概念、原理与构建模式

核心架构

ReActAgent类 (agent.py)

1. 类初始化

class ReActAgent:
    def __init__(self, tools: List[Callable], model: str, project_directory: str):
        self.tools = { func.__name__: func for func in tools }
        self.model = model
        self.project_directory = project_directory
        self.client = OpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key=ReActAgent.get_api_key(),
        )

Python语法科普：字典推导式

代码中的 self.tools = { func.__name__: func for func in tools } 使用了字典推导式语法：

语法结构：

{ key_expression: value_expression for item in iterable }

具体解析：

• func.__name__ - 键：函数的名称字符串
• func - 值：函数对象本身
• for func in tools - 遍历工具函数列表

等价的传统写法：

self.tools = {}
for func in tools:
    self.tools[func.__name__] = func

实际效果示例：

# 假设 tools = [read_file, run_command]
# 结果：self.tools = {
#     "read_file": read_file,
#     "run_command": run_command
# }

这种设计实现了函数名到函数对象的映射，支持Agent根据LLM返回的工具名称动态调用相应函数。

设计要点:

• 工具管理: 将函数列表转换为字典，便于按名称快速查找
• 模型集成: 使用OpenRouter作为统一的LLM API网关
• 环境感知: 记录项目目录，为Agent提供上下文信息

2. 核心执行循环

def run(self, user_input: str):
    messages = [
        {"role": "system", "content": self.render_system_prompt(react_system_prompt_template)},
        {"role": "user", "content": f"<question>{user_input}</question>"}
    ]

    while True:
        # 请求模型
        content = self.call_model(messages)
        
        # 解析思考过程
        thought_match = re.search(r"<thought>(.*?)</thought>", content, re.DOTALL)
        if thought_match:
            thought = thought_match.group(1)
            print(f"\n\n💭 Thought: {thought}")
        
        # 检查是否完成任务
        if "<final_answer>" in content:
            final_answer = re.search(r"<final_answer>(.*?)</final_answer>", content, re.DOTALL)
            return final_answer.group(1)
        
        # 解析并执行动作
        action_match = re.search(r"<action>(.*?)</action>", content, re.DOTALL)
        if not action_match:
            raise RuntimeError("模型未输出 <action>")
        
        action = action_match.group(1)
        tool_name, args = self.parse_action(action)
        
        # 执行工具并获取观察结果
        try:
            observation = self.tools[tool_name](*args)
        except Exception as e:
            observation = f"工具执行错误：{str(e)}"
        
        # 将观察结果反馈给模型
        obs_msg = f"<observation>{observation}</observation>"
        messages.append({"role": "user", "content": obs_msg})

ReAct循环的关键步骤:

1. Reasoning (推理): 模型分析当前状态，输出思考过程
2. Acting (行动): 选择合适的工具并执行
3. Observation (观察): 获取工具执行结果
4. Iteration (迭代): 基于观察结果继续推理，直到完成任务

3. 动作解析器

def parse_action(self, code_str: str) -> Tuple[str, List[str]]:
    match = re.match(r'(\w+)\((.*?)\)', code_str, re.DOTALL)
    if not match:
        raise ValueError("Invalid function call syntax")
    
    func_name = match.group(1)
    args_str = match.group(2).strip()
    
    # 复杂的参数解析逻辑
    args = []
    current_arg = ""
    in_string = False
    string_char = None
    paren_depth = 0
    
    # ... 详细的字符串解析逻辑

技术亮点:

• 健壮的解析: 处理多行字符串、嵌套括号、转义字符
• 类型推断: 使用ast.literal_eval安全解析Python字面量
• 错误处理: 优雅处理解析失败的情况

提示词工程 (prompt_template.py)

ReAct提示词模板设计

react_system_prompt_template = """
你需要解决一个问题。为此，你需要将问题分解为多个步骤。对于每个步骤，首先使用 <thought> 思考要做什么，然后使用可用工具之一决定一个 <action>。接着，你将根据你的行动从环境/工具中收到一个 <observation>。持续这个思考和行动的过程，直到你有足够的信息来提供 <final_answer>。

所有步骤请严格使用以下 XML 标签格式输出：
- <question> 用户问题
- <thought> 思考
- <action> 采取的工具操作
- <observation> 工具或环境返回的结果
- <final_answer> 最终答案
"""

设计原则:

• 结构化输出: 使用XML标签确保输出格式的一致性
• 示例驱动: 提供具体的使用示例指导模型行为
• 动态注入: 通过模板变量注入工具列表和环境信息

工具系统

内置工具函数

1. 文件操作工具

def read_file(file_path):
    """用于读取文件内容"""
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()

def write_to_file(file_path, content):
    """将指定内容写入指定文件"""
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(content.replace("\\n", "\n"))
    return "写入成功"

2. 系统命令工具

def run_terminal_command(command):
    """用于执行终端命令"""
    import subprocess
    run_result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return "执行成功" if run_result.returncode == 0 else run_result.stderr

安全特性:

• 用户确认: 终端命令执行前需要用户确认
• 错误捕获: 完整的异常处理和错误信息返回
• 编码处理: 统一使用UTF-8编码避免乱码问题

技术栈与依赖

技术栈

• Python 3.12+: 现代Python版本，支持最新语言特性
• OpenAI SDK: 用于与LLM API交互
• OpenRouter API: 多模型聚合服务，支持多种LLM选择
• Click: 命令行界面框架
• dotenv: 环境变量管理

OpenRouter 服务商背景

OpenRouter 是由 OpenRouter Inc. 公司开发和运营的AI聚合服务。该公司是一家专注于AI基础设施的美国科技公司，主要特点包括：

• 公司定位: AI聚合服务商，专门提供多模型AI API聚合服务
• 商业模式: 通过统一接口访问多种AI模型，按使用付费
• 核心价值: 解决开发者需要集成多个不同AI服务商API的复杂性问题
• 服务特色: 模型中立、实时路由、成本优化、统一标准（OpenAI兼容格式）

与传统的单一模型提供商（如OpenAI、Anthropic）不同，OpenRouter作为聚合服务商，在AI模型提供商和最终用户之间搭建桥梁，让开发者通过一个API就能访问市面上主流的AI模型。

核心依赖 (pyproject.toml)

[project]
name = "agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "click>=8.2.1",      # 命令行界面框架
    "openai>=1.91.0",    # OpenAI API客户端
    "python-dotenv>=1.1.1", # 环境变量管理
]

技术选型说明

• Click: 提供优雅的命令行接口，支持参数验证和帮助文档
• OpenAI SDK: 统一的LLM API接口，兼容多种模型提供商
• python-dotenv: 安全的API密钥管理，避免硬编码敏感信息

运行机制

启动流程

@click.command()
@click.argument('project_directory',
                type=click.Path(exists=True, file_okay=False, dir_okay=True))
def main(project_directory):
    project_dir = os.path.abspath(project_directory)
    
    tools = [read_file, write_to_file, run_terminal_command]
    agent = ReActAgent(tools=tools, model="openai/gpt-4o", project_directory=project_dir)
    
    task = input("请输入任务：")
    final_answer = agent.run(task)
    
    print(f"\n\n✅ Final Answer：{final_answer}")

使用示例

# 基本用法
uv run agent.py /path/to/project

# 具体示例
uv run agent.py snake  # 在snake目录下运行Agent

核心特性

1. 自适应推理

• 动态规划: Agent根据任务复杂度自动分解步骤
• 上下文感知: 利用项目目录信息提供环境上下文
• 错误恢复: 工具执行失败时能够重新规划

2. 可扩展工具系统

• 插件化设计: 新工具只需实现标准函数接口
• 自动发现: 通过反射机制自动生成工具文档
• 类型安全: 参数解析支持多种Python数据类型

3. 安全机制

• 用户控制: 危险操作需要用户确认
• 沙箱执行: 工具执行在受控环境中进行
• 错误隔离: 单个工具失败不会影响整个Agent

实际应用场景

1. 代码生成与修改: 自动编写、调试和优化代码
2. 项目管理: 文件操作、依赖管理、构建部署
3. 数据处理: 文件解析、格式转换、批量操作
4. 系统运维: 日志分析、配置管理、监控报警

Agent.py 能力演示 🤖

🔧 核心架构能力

ReActAgent 是一个基于 ReAct（Reasoning + Acting）模式的智能代理系统，具备以下核心能力：

1. 智能推理与行动循环

• 思考阶段：使用 <thought> 标签分析问题和规划步骤
• 行动阶段：通过 <action> 标签调用具体工具
• 观察阶段：接收 <observation> 反馈并继续推理
• 结论阶段：输出 <final_answer> 完成任务

2. 内置工具集

• read_file(file_path)：读取文件内容
• write_to_file(file_path, content)：写入文件内容
• run_terminal_command(command)：执行系统命令（需用户确认）

3. 智能解析能力

• XML标签解析：精确提取AI的思考过程和行动指令
• 函数调用解析：支持复杂参数解析，包括多行字符串和转义字符
• 错误处理：工具执行失败时提供详细错误信息

4. 上下文感知

• 项目目录感知：自动获取当前目录文件列表
• 操作系统适配：根据系统类型（macOS/Windows/Linux）调整行为
• 对话历史管理：维护完整的对话上下文

🚀 技术特性

ReAct 提示词模板

• 结构化的XML标签系统
• 详细的示例和使用规范
• 动态变量替换（工具列表、文件列表、操作系统信息）

安全机制

• 终端命令执行需要用户确认
• API密钥通过环境变量安全管理
• 错误隔离和异常处理

模型集成

• 支持 OpenRouter API
• 使用 GPT-4o 模型
• 完整的对话历史传递

💡 使用场景示例

1. 文件操作：批量处理、内容分析、格式转换
2. 代码生成：自动编写和修改代码文件
3. 系统管理：执行系统命令、环境配置
4. 数据处理：文本分析、格式化、清理
5. 项目管理：文件组织、依赖管理

⚙️ 运行要求

环境配置：

• Python 3.12+
• UV 包管理器
• OpenRouter API Key（需在 .env 文件中配置）

启动命令：

cd Agent的概念、原理与构建模式
uv run agent.py <项目目录>

🎯 配置说明

由于需要配置 OPENROUTER_API_KEY 环境变量，项目中已提供 .env.example 配置示例文件：

# OpenRouter API Key 配置
# 请将此文件复制为 .env 并填入真实的API Key
OPENROUTER_API_KEY=your_openrouter_api_key_here

# 获取API Key的步骤：
# 1. 访问 https://openrouter.ai/
# 2. 注册账号并登录
# 3. 在API Keys页面生成新的API Key
# 4. 将API Key填入上面的配置中

完整的Agent工作流程：

1. 用户输入任务描述
2. Agent分析任务并制定计划
3. 逐步执行工具调用
4. 根据反馈调整策略
5. 输出最终结果

这个ReAct Agent展现了现代AI代理系统的核心特征：结构化推理、工具集成、上下文感知和安全执行。

提示词模板设计解析

XML标签设计：

• <question></question> 是结构化标记，用于明确标识内容类型
• 帮助LLM理解这是"用户问题"而不是其他内容
• 提供语义上下文，增强模型的理解能力

Python变量语法：

• {user_input} 是Python的f-string格式化语法
• 运行时会将 user_input 变量的值插入到字符串中
• 例如：user_input = "今天天气怎么样？" → "<question>今天天气怎么样？</question>"

ReAct模式核心：

• 结构化思维：强制AI按照 思考→行动→观察→思考 的循环进行
• 可控性：通过XML标签确保输出格式的一致性
• 可追踪性：每个步骤都有明确的标记，便于调试和理解

大模型记忆机制解析

核心原理：大模型本身是无状态的，每次API调用都是独立的，不会"记住"上一次对话。ReAct Agent通过messages数组来模拟"记忆"。

工作机制：

# 每次调用都发送完整的对话历史
messages = [
    {"role": "system", "content": "系统提示"},
    {"role": "user", "content": "第1轮用户输入"},
    {"role": "assistant", "content": "第1轮AI回复"},
    {"role": "user", "content": "第2轮用户输入"},
    {"role": "assistant", "content": "第2轮AI回复"},
    # ... 所有历史都在这里
]

实际调用流程：

• 第1次调用: [系统提示, 用户问题] → AI回复A
• 第2次调用: [系统提示, 用户问题, AI回复A, 观察结果] → AI回复B
• 第3次调用: [系统提示, 用户问题, AI回复A, 观察结果, AI回复B, 新观察] → AI回复C

类比理解：就像每次都给健忘症患者看完整的聊天记录，患者本身没记忆（大模型），但通过完整记录重建上下文。

优缺点：

• ✅ 简单可靠的上下文保持
• ✅ 完全可控的对话历史
• ❌ Token消耗随对话增长
• ❌ 有最大长度限制

扩展建议

工具扩展

def search_web(query: str) -> str:
    """网络搜索工具"""
    # 实现网络搜索逻辑
    pass

def send_email(to: str, subject: str, body: str) -> str:
    """邮件发送工具"""
    # 实现邮件发送逻辑
    pass

模型优化

• 本地模型: 集成Ollama等本地LLM
• 多模态: 支持图像、音频输入
• 流式输出: 实时显示推理过程