云计算百科
云计算领域专业知识百科平台

从零手写一个 ReAct Agent:Python 实战(附完整代码)

从零手写一个 ReAct Agent:Python 实战(附完整代码)

前言

前三篇我们讲完了概念、架构和框架。这篇文章回到本质:不依赖任何 Agent 框架,用纯 Python 手写一个可运行的 ReAct Agent。

读完你会理解:

  • ReAct 循环在代码层面到底长什么样

  • Function Calling 的"声明-选择-执行-回填"如何落地

  • 如何对接任意 OpenAI 兼容 API(DeepSeek / 豆包 / 通义 / OpenAI 都行)

一、环境准备

只需要 Python 3.9+ 和 requests:

pip install requests

配置环境变量(以 DeepSeek 为例,可换成任意 OpenAI 兼容接口):

export LLM_API_KEY=sk-你的密钥
export LLM_BASE_URL=https://api.deepseek.com/v1
export LLM_MODEL=deepseek-chat

Windows PowerShell 下用 $env:LLM_API_KEY="sk-…" 设置。

二、整体设计

整个 Agent 就是一个"循环":

Action

Final Answer

用户问题

组装消息

调用 LLM

输出类型?

执行工具

Observation 回填

输出结果

三、Step 1:定义工具

Agent 的"手脚"就是普通 Python 函数,注册到 TOOLS 字典即可:

import re

def get_weather(city: str) -> str:
   """查询城市天气(演示用,真实场景替换为天气 API)"""
   table = {"深圳": "小雨 22~26℃", "北京": "晴 15~28℃", "上海": "多云 20~27℃"}
   return table.get(city, f"{city}:暂无天气数据")

def calculator(expr: str) -> str:
   """安全计算器:只允许数字和基本运算符"""
   expr = expr.replace(" ", "")
   if not re.fullmatch(r"[\\d+\\-*/().%]+", expr):
       return "Error: 表达式包含非法字符"
   try:
       return str(eval(expr, {"__builtins__": {}}, {}))
   except Exception as e:
       return f"Error: {e}"

TOOLS = {
   "get_weather": {"desc": "查询城市天气,参数 city(城市名)", "func": get_weather},
   "calculator": {"desc": "计算数学表达式,参数 expr(如 12*8+6)", "func": calculator},
}

注意 eval 里禁用了内置函数(__builtins__ 置空),避免代码注入——这是工具实现里必须守住的底线。

四、Step 2:构造 ReAct Prompt

通过提示词让模型学会"思考-行动-观察"的循环格式:

SYSTEM_PROMPT = """你是一个使用 ReAct 范式的 AI Agent。请严格按下面的格式循环输出:

Thought: 你当前的分析和思考
Action: 工具名
Action Input: {{"参数名": "参数值"}}

当信息足够时输出:

Thought: 我已经得到足够信息
Final Answer: 最终答案

可用工具:
{tools}

注意:需要查询或计算时必须调用工具,禁止凭空编造结果。"""

五、Step 3:解析模型输出

模型返回的是文本,我们要把它解析成结构化指令:

import json

def parse_response(text: str):
   """解析模型输出,返回 final 或 action 两种结构"""
   if "Final Answer:" in text:
       return {"type": "final", "value": text.split("Final Answer:", 1)[1].strip()}

   action = re.search(r"Action:\\s*(\\w+)", text)
   action_input = re.search(r"Action Input:\\s*(\\{.*?\\})", text, re.S)
   if action:
       params = {}
       if action_input:
           try:
               params = json.loads(action_input.group(1))
           except json.JSONDecodeError:
               params = {}
       return {"type": "action", "name": action.group(1), "params": params}

   return {"type": "final", "value": text}

六、Step 4:主循环(核心)

import os
import requests

API_KEY = os.environ.get("LLM_API_KEY", "sk-xxx")
BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com/v1")
MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")

def chat(messages, temperature=0.0):
   resp = requests.post(
       f"{BASE_URL}/chat/completions",
       headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
       json={"model": MODEL, "messages": messages, "temperature": temperature},
       timeout=60,
  )
   resp.raise_for_status()
   return resp.json()["choices"][0]["message"]["content"]

def run_agent(question: str, max_steps: int = 8) -> str:
   tool_desc = "\\n".join(f"- {name}: {info['desc']}" for name, info in TOOLS.items())
   messages = [
      {"role": "system", "content": SYSTEM_PROMPT.format(tools=tool_desc)},
      {"role": "user", "content": question},
  ]

   for step in range(1, max_steps + 1):
       print(f"\\n[Step {step}] 调用 LLM …")
       reply = chat(messages)
       print(reply)

       parsed = parse_response(reply)

       # 1. 模型给出最终答案,循环结束
       if parsed["type"] == "final":
           return parsed["value"]

       # 2. 模型要调用工具
       tool = TOOLS.get(parsed["name"])
       if tool is None:
           messages.append({"role": "assistant", "content": reply})
           messages.append({"role": "user",
                            "content": f"Observation: 工具 {parsed['name']} 不存在,请改用可用工具"})
           continue

       result = tool["func"](**parsed["params"])
       print(f"   工具返回: {result}")

       # 3. 关键一步:把工具结果作为 Observation 回填给 LLM
       messages.append({"role": "assistant", "content": reply})
       messages.append({"role": "user", "content": f"Observation: {result}"})

   return "已达到最大步数,任务未完成。"

if __name__ == "__main__":
   print("结果:", run_agent("深圳今天会下雨吗?气温多少度?"))
   print("结果:", run_agent("帮我计算 (23*7+15)/4 等于多少"))

七、运行效果

配好 API Key 后直接运行:

[Step 1] 调用 LLM …
Thought: 用户想知道深圳的天气,我需要调用 get_weather 工具
Action: get_weather
Action Input: {"city": "深圳"}
  工具返回: 小雨 22~26℃
[Step 2] 调用 LLM …
Thought: 深圳今天有雨,我应该提醒用户带伞
Final Answer: 深圳今天小雨,气温 22~26℃,记得带伞哦。
结果: 深圳今天小雨,气温 22~26℃,记得带伞哦。

[Step 1] 调用 LLM …
Thought: 这是一个数学计算题,我需要用 calculator 工具
Action: calculator
Action Input: {"expr": "(23*7+15)/4"}
  工具返回: 44.0
[Step 2] 调用 LLM …
Thought: 计算完成
Final Answer: (23*7+15)/4 = 44.0
结果: (23*7+15)/4 = 44.0

完整的 ReAct 循环——思考、行动、观察、再思考——就这样在代码里跑通了。

八、可优化方向

方向说明
记忆 接入向量库(如 Chroma / Milvus)+ RAG,支持长期记忆
多工具与重试 工具调用失败时自动重试、纠错
流式输出 用 SSE 流式返回,提升体验
结构化解析 用 JSON Mode 或 Function Calling 原生接口替代正则解析
安全护栏 工具白名单、敏感操作人工确认(Human-in-the-Loop)

九、总结

  • ReAct Agent 的核心就是循环:Thought → Action → Observation → 再思考

  • Function Calling 的落地 = 声明工具 + 模型选择 + 程序执行 + 结果回填

  • 本教程代码约 100 行,却已经具备真实 Agent 的骨架,后续所有框架都只是在这套骨架上的封装和增强

至此,【AI Agent 系列】四篇完结:概念 → 架构 → 框架 → 实战。建议按顺序学习,动手跑一遍第四篇的代码,你对 Agent 的理解会上一个台阶。

如果本文对你有帮助,欢迎点赞、收藏、关注!有问题评论区见。

赞(0)
未经允许不得转载:网硕互联帮助中心 » 从零手写一个 ReAct Agent:Python 实战(附完整代码)
分享到: 更多 (0)

评论 抢沙发

评论前必须登录!