本文是AI Agent实战系列的第一篇。我们将从零开始,用Python实现一个基于ReAct框架的智能体,它能自主思考、调用工具、完成任务。全文含完整可运行代码,约3500字。—## 目录- 一、什么是AI Agent- 二、ReAct框架:思考-行动-观察循环- 三、核心架构设计- 四、从零实现:完整代码- 五、运行效果演示- 六、局限性讨论- 总结—## 一、什么是AI Agent普通的AI对话是这样的:你问一个问题,AI回答一个答案。一问一答,AI不会主动做任何事情。AI Agent的区别在于:它能自主规划和执行多步骤任务。举个例子:普通AI对话: 用户:今天北京天气怎么样? AI:我无法获取实时天气数据,建议你查看天气应用。AI Agent: 用户:今天北京天气怎么样? Agent思考:我需要调用天气API获取北京的天气数据 Agent行动:调用get_weather("北京") Agent观察:{"temp": 28, "condition": "晴", "humidity": 45} Agent回答:北京今天晴天,气温28°C,湿度45%,适合出行。关键区别:Agent有工具使用能力自主决策能力。> 需要说明的是,Agent不是万能的。它受限于工具的能力和大模型的推理水平。本文实现的是一个基础但完整的Agent框架,后续文章会逐步增强能力。—## 二、ReAct框架:思考-行动-观察循环ReAct(Reasoning + Acting)是目前最主流的Agent框架,核心思路很简单:循环 { Thought(思考):分析当前状态,决定下一步 Action(行动):调用一个工具 Observation(观察):获取工具返回结果}直到任务完成为什么这个框架有效?因为它模拟了人类解决问题的过程:1. 你先想想要做什么(Thought)2. 你去做某个动作(Action)3. 你看结果怎么样(Observation)4. 基于结果继续思考下一步### 与纯推理的区别| 方式 | 特点 | 适用场景 ||------|------|---------|| 纯推理(CoT) | 只思考不行动 | 数学题、逻辑分析 || 纯行动 | 只执行不思考 | 简单脚本 || ReAct | 思考+行动结合 | 需要外部信息的复杂任务 |—## 三、核心架构设计我们的Agent由4个核心组件构成:┌─────────────────────────────────┐│ Agent Core ││ ┌───────────┐ ┌─────────────┐ ││ │ Planner │ │ Executor │ ││ │ (思考规划) │ │ (执行动作) │ ││ └───────────┘ └─────────────┘ ││ ┌───────────┐ ┌─────────────┐ ││ │ Memory │ │ Tools │ ││ │ (记忆存储) │ │ (工具集) │ ││ └───────────┘ └─────────────┘ │└─────────────────────────────────┘- Planner:调用大模型进行推理,生成Thought和Action- Executor:执行工具调用,返回Observation- Memory:保存对话历史和中间结果- Tools:可供调用的工具集合—## 四、从零实现:完整代码### 4.1 定义工具基类python# tools/base.pyfrom abc import ABC, abstractmethodfrom typing import Anyclass Tool(ABC): """工具基类,所有工具必须实现name、description和run方法""" @property @abstractmethod def name(self) -> str: """工具名称,Agent通过名称调用""" ... @property @abstractmethod def description(self) -> str: """工具描述,告诉Agent什么时候该用这个工具""" ... @abstractmethod def run(self, **kwargs) -> str: """执行工具,返回字符串结果""" ... def get_prompt(self) -> str: """生成工具描述供大模型理解""" return f"- {self.name}: {self.description}"### 4.2 实现具体工具python# tools/calculator.pyfrom tools.base import Toolimport mathclass CalculatorTool(Tool): """计算器工具:执行数学运算""" @property def name(self) -> str: return "calculator" @property def description(self) -> str: return "执行数学计算。输入一个数学表达式(如 '2+3*4'),返回计算结果。" def run(self, expression: str = "", **kwargs) -> str: try: # 安全计算:只允许数学表达式 allowed = set("0123456789+-*/.() ") if not all(c in allowed for c in expression): return "错误:表达式包含不允许的字符" # ⚠️ 安全提示:eval()有代码注入风险,此处已通过白名单限制可用字符, # 并禁用了__builtins__。生产环境建议使用 ast.literal_eval 或专用解析库(如 numexpr)。 result = eval(expression, {"__builtins__": {}}, {"math": math}) return f"{expression} = {result}" except Exception as e: return f"计算错误:{e}"class WeatherTool(Tool): """天气工具:模拟天气查询(实际项目中替换为真实API)""" @property def name(self) -> str: return "get_weather" @property def description(self) -> str: return "查询指定城市的天气。输入城市名称,返回天气信息。" def run(self, city: str = "", **kwargs) -> str: # 模拟数据,实际项目中调用天气API mock_data = { "北京": "晴天,28°C,湿度45%", "上海": "多云,25°C,湿度60%", "深圳": "阵雨,30°C,湿度75%", } return mock_data.get(city, f"未找到{city}的天气数据(模拟数据仅支持:北京/上海/深圳)")class SearchTool(Tool): """搜索工具:模拟搜索(实际项目中替换为真实搜索API)""" @property def name(self) -> str: return "search" @property def description(self) -> str: return "搜索互联网信息。输入搜索关键词,返回相关结果摘要。" def run(self, query: str = "", **kwargs) -> str: # 模拟搜索结果 return f"[模拟搜索结果] 关于'{query}'的信息:这是一条模拟的搜索结果。实际项目中应接入Perplexity API或Tavily等搜索服务。"### 4.3 Agent核心实现python# agent.pyfrom openai import OpenAIfrom dotenv import load_dotenvfrom typing import Optionalimport jsonimport reload_dotenv()class ReActAgent: """基于ReAct框架的AI Agent""" def __init__(self, tools: list, model: str = "deepseek-chat", max_steps: int = 5): self.tools = {tool.name: tool for tool in tools} self.model = model self.max_steps = max_steps self.client = OpenAI() # 从.env加载配置 # 构建系统提示词 tools_desc = "\n".join(tool.get_prompt() for tool in tools) self.system_prompt = f"""你是一个AI智能体,能够通过思考和调用工具来完成任务。可用工具:{tools_desc}请严格按照以下格式回复:Thought: 分析当前情况,决定下一步行动Action: tool_name(query)Observation: (由系统填充,你不需要写这个)当你认为已经获得足够信息来回答用户问题时,使用以下格式:Thought: 我已经获得了足够的信息Answer: 你的最终回答注意:- 每次只调用一个工具- 仔细分析观察结果后再决定下一步- 不要编造工具返回的数据""" self.conversation = [] def _parse_action(self, response: str) -> Optional[tuple]: """从模型回复中解析工具调用""" # 匹配 Action: tool_name(argument) 模式 match = re.search(r'Action:\s*(\w+)\(([^)]*)\)', response) if match: tool_name = match.group(1) argument = match.group(2).strip().strip('"').strip("'") return tool_name, argument return None def _is_final_answer(self, response: str) -> Optional[str]: """检查是否包含最终答案""" match = re.search(r'Answer:\s*(.+)', response, re.DOTALL) if match: return match.group(1).strip() return None def run(self, task: str) -> str: """执行任务""" print(f"\n{'='*60}") print(f"任务: {task}") print(f"{'='*60}\n") self.conversation = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": task}, ] for step in range(self.max_steps): print(f"--- 步骤 {step + 1} ---") # 1. 调用大模型思考 response = self.client.chat.completions.create( model=self.model, messages=self.conversation, temperature=0.3, max_tokens=1000, ) reply = response.choices[0].message.content print(f"模型回复:\n{reply}\n") # 2. 检查是否有最终答案 final_answer = self._is_final_answer(reply) if final_answer: print(f"最终答案: {final_answer}") return final_answer # 3. 解析工具调用 action = self._parse_action(reply) if not action: # 模型没有调用工具也没有给出答案,继续引导 self.conversation.append({"role": "assistant", "content": reply}) self.conversation.append({ "role": "user", "content": "请使用Thought和Action格式继续,或直接给出Answer。" }) continue tool_name, argument = action # 4. 执行工具 if tool_name not in self.tools: observation = f"错误:工具'{tool_name}'不存在。可用工具:{list(self.tools.keys())}" else: tool = self.tools[tool_name] observation = tool.run(query=argument) if hasattr(tool.run, '__code__') and 'query' in tool.run.__code__.co_varnames else tool.run(expression=argument) print(f"工具调用: {tool_name}({argument})") print(f"观察结果: {observation}\n") # 5. 将结果添加到对话历史 self.conversation.append({"role": "assistant", "content": reply}) self.conversation.append({ "role": "user", "content": f"Observation: {observation}" }) return "达到最大步数限制,任务未完成。可以尝试增加max_steps参数。"# 运行示例if __name__ == "__main__": from tools.calculator import CalculatorTool, WeatherTool, SearchTool # 创建Agent agent = ReActAgent( tools=[CalculatorTool(), WeatherTool(), SearchTool()], max_steps=5, ) # 测试任务 result = agent.run("北京今天天气怎么样?如果出去玩需要带伞吗?") print(f"\n最终结果: {result}") print("\n" + "="*60) # 测试计算任务 result2 = agent.run("帮我算一下 (128 + 256) * 3 的结果") print(f"\n最终结果: {result2}")—## 五、运行效果演示执行上面的代码,你会看到类似这样的输出:============================================================任务: 北京今天天气怎么样?如果出去玩需要带伞吗?============================================================--- 步骤 1 ---模型回复:Thought: 用户想知道北京的天气情况,我需要调用天气查询工具Action: get_weather(北京)工具调用: get_weather(北京)观察结果: 晴天,28°C,湿度45%--- 步骤 2 ---模型回复:Thought: 已经获取到北京天气数据,晴天28°C,不需要带伞Answer: 北京今天是晴天,气温28°C,湿度45%。天气很好,不需要带伞,可以放心出门!建议做好防晒。最终答案: 北京今天是晴天,气温28°C,湿度45%。天气很好,不需要带伞,可以放心出门!建议做好防晒。### 关键观察1. 第一步:Agent正确选择了天气工具2. 第二步:基于观察结果(晴天),给出了合理的回答3. 自主决策:没有人工指定先查天气再分析,Agent自己规划了步骤—## 六、局限性讨论老实说,这个基础Agent有不少限制:1. 工具调用解析依赖正则:如果模型输出格式不规范就会失败。生产环境建议用Function Calling API2. 没有长期记忆:每次对话都是独立的,不记得上次交互3. 错误处理简单:工具失败时缺乏重试和恢复机制4. 单工具调用:每步只能调用一个工具,复杂任务效率不高5. 成本不可控:没有token计数和预算限制这些问题会在本系列后续文章中逐步解决。—## 总结本文实现了一个基于ReAct框架的AI Agent,核心要素:1. Thought-Action-Observation循环:Agent的核心运行模式2. 工具抽象:统一的工具接口,方便扩展3. 对话历史管理:通过messages维护上下文代码量不到200行,但包含了Agent的核心要素。理解这个基础框架比直接用LangChain等框架更有价值——因为你知道底层在做什么。> 下一篇《AI Agent实战(二):给智能体加上工具使用和记忆能力》将实现Function Calling集成和持久化记忆系统。—*本文由AI辅助整理,经作者亲自验证和编辑。代码在Python 3.12 + DeepSeek API环境下测试通过。如有问题欢迎评论区交流。*最后更新:2026年5月

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐