Python实现AI智能体:200行代码理解ReAct模式

📅 2026/7/22 15:50:20
Python实现AI智能体:200行代码理解ReAct模式
1. 项目概述用Python实现一个极简版Claude Code200行Python代码能做什么在这个AI技术爆炸的时代我们可以用它来构建一个具备基础推理和执行能力的智能体Agent。这个简化版的Claude Code项目将带你深入理解现代AI Agent的核心机制——ReActReasoningActing模式。这个项目特别适合想了解AI Agent底层原理的开发者希望掌握Python实现智能体基础架构的编程爱好者对Claude等大模型应用感兴趣的技术探索者通过这个项目你将掌握ReAct模式的核心工作流程如何用Python构建一个具备工具调用能力的智能体大模型与外部工具协同工作的基本原理2. 核心架构设计2.1 ReAct模式解析ReActReasoningActing是当前AI Agent的主流架构范式其核心在于交替进行推理Reasoning模型分析当前问题状态执行Acting调用适当工具获取信息观察Observation处理工具返回结果迭代直到问题解决或达到最大尝试次数典型的工作流示例如下用户提问篮球队员人数乘以曲棍球队员人数是多少 → 思考需要先查询两种运动的队员人数 → 行动调用维基百科查询篮球队员人数 ← 观察篮球每队5人 → 思考现在需要查询曲棍球人数 → 行动调用维基百科查询曲棍球人数 ← 观察曲棍球每队10人 → 行动调用计算器计算5*10 ← 观察结果为50 → 最终答案502.2 系统架构设计我们的极简实现包含以下核心组件class ChatBot: def __init__(self, system_prompt): self.system system_prompt self.messages [] # 保存对话历史 def __call__(self, user_input): self.messages.append({role: user, content: user_input}) response self._execute() self.messages.append({role: assistant, content: response}) return response def _execute(self): # 调用大模型获取响应 response client.messages.create( modelclaude-3-5-sonnet, messagesself.messages, systemself.system ) return response.content3. 关键实现细节3.1 系统提示词设计系统提示词是Agent的大脑定义了其行为模式。我们的基础提示词包含prompt 你运行在Thought(思考)、Action(行动)、Observation(观察)、Answer(回答)的循环中。 最终你需要输出Answer。 可用工具 calculate: 执行数学计算如calculate: 3 * 4 wikipedia: 查询维基百科如wikipedia: Python 示例流程 问题法国的首都是 思考我应该查询法国相关信息 行动wikipedia: France 观察巴黎是法国首都... 回答法国的首都是巴黎 提示实际项目中应该将工具描述动态化避免硬编码。这里为简化演示采用固定格式。3.2 工具系统实现工具是Agent能力的延伸我们实现两个基础工具def wikipedia(query): 调用维基百科API获取摘要 response httpx.get(https://en.wikipedia.org/w/api.php, params{ action: query, list: search, srsearch: query, format: json }) return response.json()[query][search][0][snippet] def calculate(expression): 执行数学计算注意安全风险 return eval(expression) # 生产环境应替换为安全计算方式 tools { wikipedia: wikipedia, calculate: calculate }警告实际项目中应避免使用eval()这里仅为演示用途。生产环境应使用ast.literal_eval或专用数学库。3.3 主循环逻辑核心执行循环处理ReAct流程import re action_pattern re.compile(r^Action: (\w): (.*)$) def run_agent(question, max_iter5): bot ChatBot(prompt) current_state question for _ in range(max_iter): response bot(current_state) print(fAgent响应: {response}) # 解析行动指令 action_match action_pattern.search(response) if not action_match: return response # 没有行动指令返回最终答案 tool_name, tool_input action_match.groups() if tool_name not in tools: raise ValueError(f未知工具: {tool_name}) print(f执行工具: {tool_name}({tool_input})) result tools[tool_name](tool_input) print(f工具返回: {result}) current_state fObservation: {result}4. 完整实现与测试4.1 完整代码整合import anthropic import httpx import re # 初始化Claude客户端 client anthropic.Anthropic(api_keyyour_api_key) # 工具实现 def wikipedia(query): response httpx.get(https://en.wikipedia.org/w/api.php, params{ action: query, list: search, srsearch: query, format: json }) return response.json()[query][search][0][snippet] def calculate(expr): return eval(expr) tools {wikipedia: wikipedia, calculate: calculate} # 系统提示词 prompt (...如前所述...) # ChatBot类 class ChatBot: (...) # 如前所述 # 主执行函数 def run_agent(question, max_iter5): (...) # 如前所述 # 测试案例 if __name__ __main__: print(run_agent(篮球队员人数乘以曲棍球队员人数是多少)) print(run_agent(20的平方加上30的立方是多少))4.2 典型执行流程输入问题篮球队员人数乘以曲棍球队员人数是多少执行日志Agent响应: 思考需要先查询篮球和曲棍球的队员人数... 行动wikipedia: 篮球 执行工具: wikipedia(篮球) 工具返回: 篮球是团队运动每队5名球员... Agent响应: 思考现在查询曲棍球人数... 行动wikipedia: 曲棍球 执行工具: wikipedia(曲棍球) 工具返回: 曲棍球每队10名球员... Agent响应: 行动calculate: 5 * 10 执行工具: calculate(5 * 10) 工具返回: 50 Agent响应: 回答结果是505. 进阶优化方向5.1 生产环境注意事项安全加固替换eval()为安全计算方式添加输入验证和过滤实现API调用速率限制工具系统增强class Tool: def __init__(self, name, description, func): self.name name self.desc description self.func func tools { calculate: Tool( namecalculate, description执行数学计算如: calculate: 3 4, funcsafe_calculate ), # 其他工具... }对话历史管理添加对话历史截断策略实现短期记忆缓存添加对话状态跟踪5.2 性能优化技巧并行工具调用import asyncio async def call_tool(tool_name, input): tool tools[tool_name] if inspect.iscoroutinefunction(tool.func): return await tool.func(input) return tool.func(input)结果缓存from functools import lru_cache lru_cache(maxsize100) def cached_wikipedia(query): return wikipedia(query)流式响应def stream_response(prompt): with client.messages.stream(...) as stream: for chunk in stream: yield chunk.text6. 常见问题排查6.1 典型错误与解决方案问题现象可能原因解决方案工具未触发提示词格式不匹配检查Action正则表达式API超时网络问题/配额不足添加重试机制无效响应模型理解偏差优化提示词工程安全警告eval()使用风险替换为安全计算库6.2 调试技巧详细日志记录def log_interaction(role, content): with open(agent.log, a) as f: f.write(f[{role}] {content}\n)交互式调试def debug_agent(): while True: question input(用户 ) if question.lower() quit: break print(Agent, run_agent(question))单元测试覆盖import unittest class TestAgent(unittest.TestCase): def test_calculation(self): self.assertEqual(run_agent(22是多少), 4)这个200行的Python实现虽然简单但完整展现了现代AI Agent的核心工作机制。通过这个项目你应该已经理解了ReAct模式的基本原理以及如何将大语言模型与外部工具结合来构建智能系统。