从零构建AI自动化代理：基于ReAct模式的核心原理与工程实践-平芜编程栈

在实际项目中，将AI能力从简单的对话或生成工具，转变为能够自主执行任务、管理流程的“自动化代理”，是当前技术落地的一个关键跃迁。很多开发者学习了AI模型的基础调用，但面对“构建一个能独立工作的AI代理”这个目标时，却不知从何入手，陷入工具堆砌或设计复杂的困境。本文旨在为初学者提供一条清晰的路径，从核心概念理解到环境搭建，再到实现一个具备基础感知、决策和执行能力的AI代理原型，并最终探讨将其转化为可持续业务所需考虑的生产级要素。

本文适合已经了解Python基础、对主流大模型API（如OpenAI、智谱、DeepSeek等）有初步调用经验，并希望将这些能力系统化、自动化的开发者。我们将避开空泛的理论，聚焦于可运行的代码、可复现的配置以及从开发到部署的完整闭环。读完本文，你将能够搭建一个可以接收目标、分解任务、调用工具并持续运行的AI代理骨架，并理解其扩展为复杂业务系统的核心模式。

1. 理解AI自动化代理的核心：从工具到智能体

在开始写代码之前，必须厘清几个关键概念，这决定了我们构建的系统的设计方向和复杂度。

1.1 AI代理与简单API调用的本质区别

简单的AI API调用，例如让模型生成一段文本或分析一张图片，是一个“请求-响应”的瞬时过程。用户提供输入，模型返回输出，过程结束。而AI自动化代理则是一个具有状态持续性、目标导向性和自主行动能力的系统。

状态持续性：代理能记住之前的交互历史、执行结果和环境上下文，并基于此做出后续决策。这通常通过维护一个“记忆”或“会话历史”模块来实现。
目标导向性：代理接收一个高级目标（如“帮我分析本周销售数据并写一份报告”），而非具体的指令。代理需要自己规划如何达成这个目标。
自主行动能力：代理不仅会“思考”（生成文本），还会“行动”。它可以自主调用外部工具，如执行代码、查询数据库、调用第三方API、操作文件系统等。这是自动化得以实现的关键。

一个典型的AI代理工作流遵循“感知-规划-执行-反思”的循环：

感知：接收用户输入或环境信号。
规划：分析当前状态和目标，决定下一步做什么（思考），可能分解为子任务。
执行：调用合适的工具或能力来完成规划出的动作。
反思：评估执行结果，更新内部状态，决定是继续下一个子任务、调整规划还是向用户请求澄清。

1.2 关键组件与架构模式

要构建这样一个系统，我们需要设计几个核心组件：

大脑（LLM Core）：负责所有的推理、规划和决策。通常由一个大语言模型担任，它处理自然语言，理解目标，并生成下一步的行动计划（通常以结构化格式，如JSON）。
记忆（Memory）：存储代理的历史对话、工具执行结果、学到的知识等。可以是简单的列表、向量数据库（用于语义检索长期记忆），或更复杂的图结构。
工具（Tools）：代理可以调用的函数或API集合。这是代理与外部世界交互的“手”和“脚”。例如：search_web,execute_python_code,read_file,send_email。
执行器（Orchestrator）：控制整个循环流程的组件。它负责初始化代理、管理记忆、调用大脑进行规划、解析大脑的输出、调用对应的工具、处理工具返回结果，并决定下一步。

目前社区流行的架构模式主要有两种：

ReAct模式：强调将“推理”和“行动”明确分离。代理的输出格式为Thought: ... Action: ... Action Input: ...，执行器解析后执行对应动作，再将结果Observation: ...返回给代理进行下一轮思考。结构清晰，易于调试。
Plan-and-Execute模式：代理先制定一个完整的计划（任务列表），然后按顺序或并行地执行这些任务。适合目标明确、步骤可预先分解的场景。

对于初学者，从ReAct模式入手更容易理解代理的运作机制。

1.3 技术选型：框架与模型

手动从零搭建所有轮子是一个很好的学习过程，但对于快速构建原型，使用成熟的框架是更高效的选择。以下是一些常见选择及其特点：

框架/库	语言	特点	适用场景
LangChain	Python/JS	生态最丰富，组件齐全，文档庞大。概念较多，学习曲线稍陡。	快速构建复杂、生产级的AI应用，需要大量集成。
LlamaIndex	Python	专注于数据连接和检索（RAG），在构建知识库代理方面有优势。	代理需要深度处理私有文档、知识库的场景。
Semantic Kernel	C#/Python	微软出品，强于规划、插件（技能）管理和流程编排。	.NET生态或需要复杂工作流编排的项目。
AutoGen	Python	支持多代理对话和协作，擅长构建代理团队。	需要多个专业代理协同完成任务的场景。
简单自建	Python	完全可控，深度定制，依赖少。	学习原理、构建轻量级或特定场景的代理。

对于初学者指南，我们将采用“简单自建”结合LangChain的部分核心思想的方式，以确保你能理解每一行代码在做什么，同时又不至于陷入底层细节。模型方面，我们将使用OpenAI GPT-4/3.5-Turbo的API作为“大脑”，因为它提供了稳定且强大的推理能力。你也可以替换为其他兼容OpenAI API格式的模型服务。

2. 环境准备与项目初始化

我们将在Python环境中构建代理。请确保你的开发环境满足以下要求。

2.1 基础环境与依赖安装

首先，确保你已安装Python（推荐3.8以上版本）。然后创建一个新的项目目录并初始化虚拟环境，这能有效隔离依赖。

# 创建项目目录 mkdir ai_agent_project && cd ai_agent_project # 创建虚拟环境（以venv为例） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

接下来，安装核心依赖。我们将使用openai库与模型交互，langchain库主要借用其优秀的“工具”抽象和调用模板，但核心循环我们自己控制。

# 安装核心库 pip install openai langchain # 安装可能用到的工具库示例 pip install requests # 用于调用网页API pip install python-dotenv # 用于管理环境变量

2.2 配置模型API密钥

为了调用大模型，你需要一个API密钥。这里以OpenAI为例。永远不要将密钥硬编码在代码中。

在.env文件中填入你的密钥：

OPENAI_API_KEY=sk-your-actual-api-key-here

在代码中通过python-dotenv加载。

同时，创建一个config.py文件来集中管理配置：

# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Config: OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENAI_API_BASE = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1") # 支持自定义端点 MODEL_NAME = os.getenv("MODEL_NAME", "gpt-3.5-turbo") # 默认使用 3.5-turbo

2.3 项目结构设计

一个清晰的项目结构有助于后续扩展。建议如下：

ai_agent_project/ ├── .env # 环境变量（列入.gitignore） ├── config.py # 配置文件 ├── main.py # 主程序入口 ├── agent/ │ ├── __init__.py │ ├── core.py # 代理核心循环、大脑 │ ├── memory.py # 记忆模块 │ └── tools/ # 工具集目录 │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ ├── web_search.py # 网页搜索工具示例 │ └── calculator.py # 计算器工具示例 └── utils/ └── __init__.py

3. 构建核心组件：工具、记忆与大脑

现在，我们从下往上构建代理的各个部件。

3.1 定义工具（Tools）

工具是代理能力的扩展。我们定义一个简单的工具基类，所有具体工具都继承它。

# agent/tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): """工具基类""" name: str # 工具名称，代理通过这个名称来调用 description: str # 工具描述，用于帮助LLM理解何时使用此工具 parameters: Dict[str, Any] # 工具所需的参数定义 def __init__(self, name: str, description: str): self.name = name self.description = description self.parameters = {} @abstractmethod def execute(self, **kwargs) -> str: """执行工具的核心逻辑，返回字符串格式的结果""" pass def to_function_call_schema(self) -> Dict: """将工具转换为OpenAI Function Calling所需的格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": self.parameters, "required": list(self.parameters.keys()) } } }

接着，实现两个简单的工具：一个计算器和一个模拟的网页搜索。

# agent/tools/calculator.py from .base_tool import BaseTool import math class CalculatorTool(BaseTool): def __init__(self): super().__init__( name="calculator", description="执行数学计算。支持加(+)、减(-)、乘(*)、除(/)、乘方(**)、平方根(sqrt)等。" ) self.parameters = { "expression": {"type": "string", "description": "数学表达式，例如：'3 + 5 * 2' 或 'sqrt(16)'"} } def execute(self, expression: str) -> str: try: # 警告：在生产环境中，直接eval用户输入是极度危险的！ # 这里仅为演示，实际应使用安全的表达式解析库（如 ast.literal_eval 限制操作）。 # 此处我们做一个极简的安全过滤和计算。 allowed_chars = set("0123456789+-*/.() sqrt") if not all(c in allowed_chars for c in expression): return "错误：表达式中包含不安全字符。" # 替换sqrt为math.sqrt expression = expression.replace('sqrt', 'math.sqrt') result = eval(expression, {"__builtins__": {}}, {"math": math}) return f"计算结果：{result}" except Exception as e: return f"计算出错：{e}" # agent/tools/web_search.py from .base_tool import BaseTool import requests import json class WebSearchTool(BaseTool): def __init__(self, api_key: str = None, search_engine: str = "google"): # 注意：这里需要接入真实的搜索API，如SerpAPI、Google Custom Search等。 # 此处仅为模拟。 super().__init__( name="web_search", description="在互联网上搜索信息。当需要获取最新、未知或事实性信息时使用。" ) self.api_key = api_key self.search_engine = search_engine self.parameters = { "query": {"type": "string", "description": "搜索关键词"} } def execute(self, query: str) -> str: # 模拟搜索返回 # 真实情况下，这里应调用第三方API mock_results = [ f"关于'{query}'的搜索结果1：...", f"关于'{query}'的搜索结果2：...", f"关于'{query}'的搜索结果3：..." ] return f"搜索完成。摘要：{'; '.join(mock_results[:2])}"

在agent/tools/__init__.py中导出工具，方便管理。

# agent/tools/__init__.py from .calculator import CalculatorTool from .web_search import WebSearchTool __all__ = ["CalculatorTool", "WebSearchTool"]

3.2 实现记忆（Memory）

我们实现一个简单的对话历史记忆，只保存最近的几轮交互。

# agent/memory.py from typing import List, Dict, Any class SimpleConversationMemory: """简单的对话记忆，保存固定轮数的历史""" def __init__(self, max_turns: int = 10): self.max_turns = max_turns self.history: List[Dict[str, str]] = [] # 每个元素是 {"role": "user"/"assistant"/"tool", "content": "..."} def add_user_message(self, message: str): self.history.append({"role": "user", "content": message}) self._trim() def add_assistant_message(self, message: str): self.history.append({"role": "assistant", "content": message}) self._trim() def add_tool_message(self, tool_name: str, result: str): # 工具执行结果通常以“Observation”的形式加入历史 self.history.append({"role": "tool", "content": f"[{tool_name}] 返回: {result}"}) self._trim() def get_conversation_history(self) -> List[Dict[str, str]]: """获取用于LLM提示的对话历史格式""" return self.history.copy() def _trim(self): """保持历史记录不超过最大轮数""" if len(self.history) > self.max_turns: # 简单策略：移除最早的一对用户/助手消息（如果可能） # 更复杂的策略可以基于token数裁剪 self.history = self.history[-self.max_turns:] def clear(self): self.history = []

3.3 构建大脑与执行器（Core）

这是最核心的部分，我们将实现一个遵循ReAct模式的代理循环。

# agent/core.py import json import openai from typing import List, Dict, Any, Optional from .memory import SimpleConversationMemory from .tools.base_tool import BaseTool class SimpleAIAgent: def __init__(self, model_name: str = "gpt-3.5-turbo", api_key: str = None, base_url: str = None): self.client = openai.OpenAI(api_key=api_key, base_url=base_url) self.model_name = model_name self.memory = SimpleConversationMemory() self.tools: Dict[str, BaseTool] = {} # 工具名 -> 工具实例的映射 self.max_iterations = 10 # 防止无限循环 def register_tool(self, tool: BaseTool): """注册一个工具到代理""" self.tools[tool.name] = tool def _build_messages(self, user_input: str) -> List[Dict[str, str]]: """构建发送给LLM的消息列表""" messages = [ { "role": "system", "content": """你是一个有帮助的AI助手，可以调用工具来解决问题。请遵循以下格式： 思考：你需要先思考当前情况和目标。 行动：调用工具时，请严格按照以下JSON格式输出： {"action": "工具名称", "action_input": {"参数名": "参数值"}} 如果你已经得到最终答案，或者无需使用工具，请直接输出最终答案。 """ } ] # 加入历史记忆 messages.extend(self.memory.get_conversation_history()) # 加入当前用户输入 messages.append({"role": "user", "content": user_input}) return messages def _parse_llm_response(self, response: str) -> Dict[str, Any]: """解析LLM的回复，判断是最终答案还是工具调用""" response = response.strip() # 尝试解析JSON格式的工具调用 if response.startswith("{") and response.endswith("}"): try: data = json.loads(response) if "action" in data and "action_input" in data: return {"type": "action", "data": data} except json.JSONDecodeError: pass # 否则视为最终答案 return {"type": "final_answer", "data": response} def _execute_action(self, action_name: str, action_input: Dict) -> str: """执行工具调用""" if action_name not in self.tools: return f"错误：未知工具 '{action_name}'。" tool = self.tools[action_name] try: # 这里简化处理，实际需要根据工具定义的参数进行校验和适配 result = tool.execute(**action_input) return result except Exception as e: return f"工具执行出错：{e}" def run(self, user_input: str) -> str: """运行代理的主要循环""" self.memory.add_user_message(user_input) print(f"用户: {user_input}") for i in range(self.max_iterations): # 1. 构建消息并调用LLM messages = self._build_messages(user_input) try: response = self.client.chat.completions.create( model=self.model_name, messages=messages, temperature=0.1, # 低温度保证输出稳定，易于解析 stream=False, ) llm_output = response.choices[0].message.content except Exception as e: return f"调用模型API失败：{e}" print(f"助手思考: {llm_output}") # 2. 解析LLM输出 parsed = self._parse_llm_response(llm_output) if parsed["type"] == "final_answer": final_answer = parsed["data"] self.memory.add_assistant_message(final_answer) print(f"最终答案: {final_answer}") return final_answer elif parsed["type"] == "action": action_data = parsed["data"] action_name = action_data["action"] action_input = action_data["action_input"] print(f"执行动作: {action_name}, 输入: {action_input}") # 3. 执行工具 observation = self._execute_action(action_name, action_input) print(f"观察结果: {observation}") # 4. 将观察结果加入记忆，开始下一轮循环 self.memory.add_assistant_message(llm_output) # 记录助手的“思考/行动” self.memory.add_tool_message(action_name, observation) # 记录工具结果 # 下一轮循环的“用户输入”实际上是上一步的观察结果，但我们的消息构建方式已包含历史。 # 这里只需更新user_input为观察结果，以便在下一轮构建消息时，历史是最新的。 user_input = observation # 简化处理，实际ReAct中，观察是作为下一轮LLM输入的一部分。 else: return "无法解析LLM的输出。" return f"达到最大迭代次数({self.max_iterations})，未完成目标。"

4. 组装与运行：你的第一个AI代理

现在，我们将所有组件组装起来，并运行一个完整的示例。

4.1 主程序入口

创建main.py作为启动文件。

# main.py from config import Config from agent.core import SimpleAIAgent from agent.tools import CalculatorTool, WebSearchTool def main(): # 1. 初始化配置 config = Config() # 2. 创建代理实例 agent = SimpleAIAgent( model_name=config.MODEL_NAME, api_key=config.OPENAI_API_KEY, base_url=config.OPENAI_API_BASE ) # 3. 注册工具 agent.register_tool(CalculatorTool()) # 如果需要真实的搜索，在这里传入API Key # agent.register_tool(WebSearchTool(api_key="your_serpapi_key")) agent.register_tool(WebSearchTool()) # 使用模拟搜索 # 4. 运行代理 print("=== AI 代理已启动，输入 'quit' 或 'exit' 退出 ===") while True: try: user_input = input("\n请输入你的问题或指令: ").strip() if user_input.lower() in ['quit', 'exit', 'q']: print("再见！") break if not user_input: continue final_answer = agent.run(user_input) # run方法内部已打印过程，这里可以额外处理最终答案 # 例如，保存到文件或触发其他操作 except KeyboardInterrupt: print("\n程序被中断。") break except Exception as e: print(f"运行过程中发生错误：{e}") if __name__ == "__main__": main()

4.2 运行与验证

在终端中，确保虚拟环境已激活，并且.env文件中的OPENAI_API_KEY已正确设置。

python main.py

程序启动后，你可以尝试输入一些指令来测试代理的能力：

纯推理问题：“法国的首都是哪里？”代理可能直接回答，无需调用工具。
需要计算的问题：“计算一下 15 的平方加上 28 除以 4 等于多少？”代理应该识别出需要计算，调用calculator工具，并返回结果。
需要搜索的问题：“今天北京的天气怎么样？”代理会调用web_search工具（模拟），并基于“搜索结果”给出回答。
混合任务：“先搜索一下爱因斯坦的生日，然后计算他到今天（假设是2024年1月1日）活了多少天？”代理需要先搜索获取生日信息，再进行日期计算（这里我们的计算器工具不支持日期，可以观察代理如何处理）。

观察控制台输出，你会看到类似以下的流程：

用户: 计算一下 15 的平方加上 28 除以 4 等于多少？ 助手思考: 思考：用户需要计算一个数学表达式。我需要使用计算器工具。 行动：{"action": "calculator", "action_input": {"expression": "15**2 + 28/4"}} 执行动作: calculator, 输入: {'expression': '15**2 + 28/4'} 观察结果: 计算结果：232.0 助手思考: 思考：我已经得到了计算结果。 最终答案: 15的平方是225，28除以4是7，两者相加等于232。

这个流程清晰地展示了代理的“思考-行动-观察”循环。

5. 从原型到业务：关键考量与进阶方向

一个能跑通的原型距离一个可靠、可用的AI自动化代理业务还有很长的路。以下是几个必须深入考虑的方向。

5.1 稳定性与可靠性提升

错误处理与重试：当前代码对工具执行和API调用错误处理很基础。生产环境需要更健壮的机制，例如网络超时重试、模型API的退避重试、工具执行异常的回退方案等。
循环检测与中断：代理可能陷入死循环。除了设置max_iterations，还可以检测重复的动作、无进展的对话，并引入人工中断或策略调整。
输出格式校验：我们简单使用JSON解析，很脆弱。应使用更严格的校验（如Pydantic模型），并为LLM提供更清晰的输出格式指导，甚至采用支持“函数调用”功能的模型，让模型直接返回结构化调用对象。

5.2 记忆与上下文管理优化

短期与长期记忆：SimpleConversationMemory是短期记忆。对于需要记住大量历史或知识的场景，需要引入向量数据库（如Chroma, Pinecone）作为长期记忆，实现基于语义的检索。
上下文窗口限制：大模型有token限制。需要实现智能的上下文窗口管理，例如总结冗长对话、选择性保留关键信息、使用更高效的token压缩技术。
记忆的结构化：将记忆分类存储，如用户偏好、任务历史、学到的知识等，便于代理更精准地利用。

5.3 工具生态与安全性

工具的动态注册与发现：在大型系统中，工具可能来自不同模块。需要设计机制让代理能动态发现和调用新工具。
工具权限与安全：这是重中之重。必须为工具调用设置严格的权限控制。例如，execute_python_code工具必须在安全的沙箱环境中运行；文件操作工具必须有路径白名单；数据库操作工具必须有查询限制。永远不要在生产环境直接eval用户输入。
工具描述的质量：工具的描述（description）和参数定义直接影响LLM调用的准确性。需要精心编写，并可以通过少量示例（few-shot）来提升模型的理解。

5.4 监控、评估与成本控制

全链路日志：记录每一轮的用户输入、LLM思考、工具调用、工具结果、最终输出。这是排查问题、分析代理行为和改进提示词的基石。
关键指标监控：
- 延迟：每轮循环的耗时。
- 成本：API调用token消耗。
- 工具调用分布：哪些工具被频繁使用。
- 成功率：任务完成的比例。
- 人工接管率：需要人工干预的次数。
评估体系：如何判断代理做得好不好？需要设计评估任务（Benchmark），可以是基于规则（输出是否包含关键词）、基于模型（用另一个LLM评估回答质量）或基于人工的评估。

5.5 架构模式扩展

多代理协作：对于复杂任务，可以引入多个各司其职的代理（如规划者、执行者、审查者），让它们通过对话协同工作。AutoGen框架专门擅长于此。
分层规划：对于宏大目标，代理可以先制定高层计划，再为每个子目标创建子代理或进行更细粒度的规划。
与外部系统集成：代理需要融入现有的业务系统。这意味着它需要能监听消息队列（如Kafka）、调用内部RPC/HTTP服务、写入业务数据库等。此时，代理更像一个智能的“流程自动化中心”。

6. 常见问题排查清单

在开发和使用AI代理过程中，你会遇到各种问题。以下是一个快速排查清单。

问题现象	可能原因	检查步骤	解决方案
代理不调用工具，直接回答	1. 系统提示词不清晰。 2. 工具描述不够准确。 3. 模型温度（temperature）过高，输出随机。	1. 检查`_build_messages`中的system prompt。 2. 检查工具的`name`和`description`是否清晰。 3. 检查模型调用参数，`temperature`建议设为0.1-0.3。	1. 优化提示词，明确要求使用指定JSON格式。 2. 参考OpenAI Function Calling文档，优化工具描述。 3. 降低temperature，使用支持function calling的模型（如gpt-3.5-turbo-1106以上版本）。
JSON解析失败	1. LLM输出格式不符合预期。 2. 输出中包含多余标记或思考过程。	打印出完整的`llm_output`，检查其格式。	1. 在提示词中强制要求输出纯JSON，并给出更严格的示例。 2. 使用模型的“函数调用”功能，而非文本生成。 3. 在解析前，用正则表达式提取JSON部分。
工具执行出错	1. 参数类型或数量不匹配。 2. 工具内部代码有bug。 3. 网络或依赖问题。	1. 检查`_execute_action`中传递给工具的参数字典。 2. 在工具`execute`方法内添加详细日志和异常捕获。 3. 单独测试工具函数。	1. 在工具调用前增加参数验证和转换逻辑。 2. 修复工具内部代码。 3. 确保工具所需的环境和依赖已正确安装。
达到最大迭代次数	1. 任务过于复杂。 2. 代理陷入循环或无效动作。 3. 工具无法满足任务需求。	查看完整的历史日志，观察代理在循环做什么。	1. 增加`max_iterations`或设计任务分解机制。 2. 引入循环检测，当动作重复时强制改变策略或终止。 3. 检查是否需要增加新的工具能力。
API调用超时或失败	1. 网络问题。 2. API密钥无效或额度不足。 3. 服务端限流。	1. 检查网络连接。 2. 检查API密钥和账单。 3. 查看错误信息中是否包含`rate_limit`。	1. 实现重试机制（如指数退避）。 2. 更换有效的API密钥。 3. 降低请求频率，或升级API套餐。