ReAct框架解析：从原理到实战的AI推理行动范式

ReAct是谷歌研究人员在2022年提出的一种结合 推理(Reasoning) 与 行动(Acting) 的AI提示词/框架范式。

在这之前,零样本、少样本、甚至是CoT提示词模式,都是单纯的机遇提示词惊喜直接回答或者机遇提示词提供的样本进行推理,但是这些方式存在一些局限,比如,不能实时的获取外部真实的信息,在这基础上进行再思考、推理和下结论。如果你还不了解上述提到的基础提示词内容,推荐阅读 # 提示工程入门：零样本与少样本提示的全面解析

ReAct 提示 LLMs 为任务生成口头推理轨迹和操作。这使得系统执行动态推理来创建、维护和调整操作计划，同时还支持与外部环境(例如，Wikipedia)的交互，以将额外信息合并到推理中。

工作流程

ReAct每一步循环包含三个环节:

Thought（思考） → Action（行动） → Observation（观察）

问题：今天北京天气如何？

Thought: 我需要查询实时天气，我应该调用天气工具。
Action: search("北京今天天气")
Observation: 结果显示北京气温 22°C，晴天。

Thought: 我已经得到了答案，可以回复用户了。
Answer: 北京今天晴天，气温约 22°C。

实战演练

1. 模型客户端初始化

"""
真实 ReAct Agent
- LLM：DeepSeek-V4-Pro（thinking enabled）
- 搜索工具：Tavily（真实联网搜索）
- 计算工具：Python eval（本地执行）
流程：Think → Act → Observe → Think → ... → 最终答案
"""
import requests
import json
import os

from openai import OpenAI
from tavily import TavilyClient

# ─────────────────────────────────────────
# 1. 初始化客户端
#    建议 Key 从环境变量读取，避免硬编码泄露
#    终端执行：
#      export DEEPSEEK_API_KEY=sk-xxx
#      export TAVILY_API_KEY=tvly-xxx
# ─────────────────────────────────────────
client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
)
tavily_client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))

MODEL = "deepseek-v4-pro"

如果你觉得使用环境变量麻烦,也可以暂时将密钥信息直接硬编码在代码中。

2. 工具实现

这里主要实现agent执行过程中需要调用到的工具函数,比如联网搜索、数学计算。

# ─────────────────────────────────────────
# 2. 工具实现
# ─────────────────────────────────────────
def search(query: str) -> str:
    """
    Tavily 真实联网搜索
    返回前 3 条结果的标题 + 摘要 + 来源
    """
    response = tavily_client.search(
        query=query,
        max_results=3,
        search_depth="basic",  # basic 省 token；advanced 更深入
    )
    results = []
    for r in response.get("results", []):
        results.append(f"【{r['title']}】\n{r['content']}\n来源：{r['url']}")
    return "\n\n".join(results) if results else "未找到相关结果"


def calculator(expression: str) -> str:
    """本地计算器，执行 Python 数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算出错：{e}"

def get_current_time() -> str:
    """
    返回当前日期和时间
    """
    from datetime import datetime

    return f"当前时间:{datetime.now().strftime('%Y年%m月%d日 %H:%M:%S')}"


def get_exchange_rate(base: str, target: str) -> str:
    """实时汇率查询"""
    url = f"https://open.er-api.com/v6/latest/{base.upper()}"
    response = requests.get(url)
    data = response.json()

    rate = data["rates"].get(target.upper())

    if rate:
        return f"1 {base.upper()} = {rate} {target.upper()}"

    return f"未找到 {target} 的汇率"

# 工具注册表：名字 → 函数
TOOLS = {
    "search": search,
    "calculator": calculator,
    "get_current_time": get_current_time,
    "get_exchange_rate": get_exchange_rate,
}

3. Tool Schema 定义

定义Tool Schema是为了让模型知道有哪些可以调用的工具。

# ─────────────────────────────────────────
# 3. Tool Schema（告诉模型有哪些工具可用）
# ─────────────────────────────────────────
TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "联网搜索实时信息，适合查新闻、天气、人物、事件、最新数据等",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词，尽量具体，如「2025年最新 AI 模型排行」",
                    }
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "执行数学计算，输入合法的 Python 数学表达式",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '3 * (12 + 5)' 或 '2 ** 10'",
                    }
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前日期和时间,返回年月日、时分秒格式",
            "parameters": {
                "type": "object",
                "properties": {},  # 无参数，留空
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_exchange_rate",
            "description": "查询两种货币之间的实时汇率，如美元兑人民币",
            "parameters": {
                "type": "object",
                "properties": {
                    "base": {
                        "type": "string",
                        "description": "基础货币代码，如 USD、CNY、EUR",
                    },
                    "target": {
                        "type": "string",
                        "description": "目标货币代码，如 CNY、USD、JPY",
                    },
                },
                "required": ["base", "target"],
            },
        },
    },
]

4. 模型调用封装

• 统一加上 thinking 参数，stream=False

• ReAct 循环内必须关流式，否则 tool_calls 无法正常解析

# ─────────────────────────────────────────
# 4. 封装 DeepSeek 调用
# ─────────────────────────────────────────
def call_model(messages: list) -> object:
    return client.chat.completions.create(
        model=MODEL,
        messages=messages,
        tools=TOOL_SCHEMAS,
        tool_choice="auto",
        reasoning_effort="high",
        extra_body={"thinking": {"type": "enabled"}},
        stream=False,
    )

5. ReAct 主循环

# ─────────────────────────────────────────
# 5. ReAct 主循环
# ─────────────────────────────────────────
def run_agent(user_question: str, max_steps: int = 6):
    print(f"\n{'=' * 55}")
    print(f"问题：{user_question}")
    print(f"{'=' * 55}")

    # 对话历史，每轮追加 — 这就是 Agent 的短期记忆
    messages = [
        {
            "role": "system",
            "content": (
                "回答前先判断是否需要工具获取信息。"
                "需要时调用工具，拿到结果后再综合给出最终答案。"
                "如果问题可以直接回答则无需调用工具。"
                "请用中文回答。"
            ),
        },
        {"role": "user", "content": user_question},
    ]

    for step in range(max_steps):
        print(f"\n── Step {step + 1} ──────────────────────────────")

        # ① THINK：调用模型，让它决定下一步
        response = call_model(messages)
        msg = response.choices[0].message

        # 如果开启了 thinking，打印模型的推理过程（可选，调试用）
        if hasattr(msg, "reasoning_content") and msg.reasoning_content:
            preview = msg.reasoning_content[:150].replace("\n", " ")
            print(f"[THINK]   {preview}...")

        if msg.tool_calls:
            # ② ACT：模型决定调用工具
            messages.append(msg)  # 把模型决策存入历史

            for tool_call in msg.tool_calls:
                fn_name = tool_call.function.name
                fn_args = json.loads(tool_call.function.arguments)

                print(f"[ACT]     工具：{fn_name}")
                print(f"          参数：{fn_args}")

                # ③ OBSERVE：执行工具，拿到真实结果
                if fn_name in TOOLS:
                    result = TOOLS[fn_name](**fn_args)
                else:
                    result = f"工具 {fn_name} 不存在"

                # 只打印前 200 字，避免刷屏
                preview = result[:200] + "..." if len(result) > 200 else result
                print(f"[OBSERVE] {preview}")

                # 把工具结果追加到历史，下一轮模型会看到
                messages.append(
                    {"role": "tool", "tool_call_id": tool_call.id, "content": result}
                )

        else:
            # 模型认为信息足够，直接给出最终答案
            print(f"\n[最终答案]\n{msg.content}")
            return msg.content

    print("\n[已达最大步数，强制结束]")
    return "达到最大推理步数，未能完成任务。"

6. 测试运行

# ─────────────────────────────────────────
# 6. 测试运行
# ─────────────────────────────────────────
if __name__ == "__main__":
    # 测试1：需要真实搜索
    # run_agent("今天贵州省毕节市织金县的天气如何？")

    # 测试2：搜索 + 计算组合
    # run_agent("特斯拉今天的股价是多少？如果我买 100 股需要多少美元？")

    # 测试3：直接回答，不需要工具
    # run_agent("用一句话解释什么是 ReAct Agent")

    # 测试4:当前时间
    # run_agent("当前北京时间")

    # 测试5:汇率测试
    run_agent("现在 1000 美元能换多少人民币？")

运行结果展示(部分)