迷你版 ReAct Agent：让模型先“思考再行动”

在复杂任务中，直接让大模型生成答案常常会遇到两个问题：一是过程不可见，难以解释或审计；二是需要外部工具时容易失控，格式与调用都不稳定。ReAct（Reasoning + Acting）智能体的核心是把推理过程拆成“先思考（Thought）→再行动（Action）→随后观察（Observation）”的闭环，通过多轮迭代逐步逼近最终答案，其过程如图 1所示。这样的组织方式使轨迹可记录、可回放，错误也能被定位到具体某一步，便于修正与治理。

图 1 ReAct 智能体推理过程

本文围绕一个“最小可用”的ReAct智能体实现展开说明：输出采用严格的JSON结构，工具以注册表集中管理，循环有明确的退出条件，示例工具仅包含计算器与时区时间查询两项以突出方法本身。

本文示例ReAct智能体完整代码可通过文末提供的代码仓库地址获取，本文仅引用关键片段说明其实现逻辑。

1.基本思路与项目初始化

最小实现仅保留ReAct智能体的核心组成部分与逻辑。模型输出被强约束为仅包含thought、action、action_input的单个JSON；每个工具都有名称、描述、参数结构与执行函数；达到最大步数仍未完成时主动退出；所有异常都作为观察（Observation）回流，以便在下一轮交互中可以自我修复。

本示例ReAct智能体依赖项主要是OpenAI SDK及python-dotenv。OpenAI SDK用于访问DeepSeek或其他兼容OpenAI SDK的模型平台所提供的大模型服务，python-dotenv用于读取.env文件中保存的环境配置内容。

.env文件内容如下，其中DEEPSEEK_API_KEY需要替换为真实DeepSeek开放平台的API Key。如果选择其他兼容OpenAI SDK的模型平台，则需要对应修改文件中的所有三个环境变量配置。

DEEPSEEK_API_KEY=sk-xxxxxx
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
DEEPSEEK_MODEL=deepseek-chat

2.系统提示词与输出约束

智能体推理逻辑的实现由代码逻辑与系统提示词协同实现，系统提示词的职责是把模型牢靠地“框进”ReAct的推理轨道。
以下系统提示词定义了当前智能体的角色、推理思考过程说明与约束。指定Agent的角色有助于大模型从该角色的职责或专业能力出发思考问题；明确先思考再行动的推理过程是实现ReAct型智能体的核心逻辑；约束则要求模型必须只输出一个JSON，键名固定且不得附带任何解释或Markdown，这有助于代码在收到大模型的回复后进行解析与处理。

SYSTEM_PROMPT = """\
你是一个遵循 ReAct 模式的助理。必须先逐步思考（thought），再选择行动（action），并通过工具完成任务。
你必须只输出一个 JSON 对象，且不得包含任何额外文本或 Markdown。
JSON 结构如下：
{"thought": "...", "action": "<工具名或 final>", "action_input": {... 或 "<最终答案文本>"}}

规则：
- 若需要使用工具：将 "action" 设为该工具名，将 "action_input" 设为该工具的 JSON 参数对象。
- 若可以直接给出最终答案：将 "action" 设为 "final"，并把最终答案文本放入 "action_input"。
- 只允许这三个键：thought、action、action_input；严禁出现其它键。
- 严禁输出解释性文字、前后缀、或代码块标记（例如 ```）。
"""

上述系统提示词中的“只输出一个JSON对象”是智能体稳定性的关键所在。有了这样的约束，即便偶有偏差，后续的解析器也可以进行自我修复式重试，将返回拉回到可解析形态。

3.工具实现与工具注册表

工具使用注册表集中管理，结构清晰、便于扩展。工具实体包含名称、描述、参数模式（用于提示，不强校验）以及执行函数。工具注册表以字典形式实现，而每个工具实体则对应一个数据类Tool的实例。

from dataclasses import dataclass
from typing import Dict, Any, Callable

@dataclass
class Tool:
    name: str
    description: str
    schema: Dict[str, Any]
    func: Callable[[Dict[str, Any]], str]

TOOLS: Dict[str, Tool] = {}

本示例智能体可调用的工具有两个：一个简单的计算器及一个可获取时间的工具。工具实现不是本文关注的重点，故采用相对简单的设计实现。
计算器工具的实现对安全性作了初步保证。它实现采用ast白名单方式，仅放行与数值四则及幂运算相关的语法节点，并限制常量只能是int/float，禁止变量与函数调用。这一设计可以在几乎不牺牲表达能力的前提下，显著降低注入风险。具体实现参见如下safe_eval_expr()函数。

import ast

def safe_eval_expr(expr: str) -> float:
    """
    安全算式求值：仅允许 + - * / ** 与括号、数值常量。
    禁止变量名、函数调用，以及非数值常量（如字符串、布尔、None）。
    """
    allowed_nodes = {
        ast.Expression, ast.BinOp, ast.UnaryOp, ast.Load,
        ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.USub, ast.UAdd,
        ast.Constant,  # 仅允许数值型常量，见下面的类型检查
    }

    tree = ast.parse(expr, mode='eval')

    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            raise ValueError("不允许函数调用")
        if isinstance(node, ast.Name):
            raise ValueError("不允许变量名")
        if type(node) not in allowed_nodes:
            raise ValueError(f"不允许的语法: {type(node).__name__}")
        # 关键：仅放行数值常量（int/float）
        if isinstance(node, ast.Constant) and not isinstance(node.value, (int, float)):
            raise ValueError("只允许数值常量（int/float）")

return eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}}, {})

在上述安全约束下，计算器工具的执行函数保持简洁，统一以字符串形式返回观察（Observation），从而简化轨迹记录与调试。计算器工具执行过程由函数tool_calculator()实现。其代码如下。

def tool_calculator(params: Dict[str, Any]) -> str:
    expr = str(params.get("expression", "")).strip()
    if not expr:
        return "错误：缺少 expression"
    try:
        val = safe_eval_expr(expr)
        return f"结果={val}"
    except Exception as e:
        return f"错误：无法计算（{e}）"

另一个可获取时间的工具由tool_time_now()函数实现，它支持以IANA时区标识（如Asia/Shanghai、Asia/Singapore）作为可选参数zone，以便指定目标时区，若未提供该参数则返回本地系统时区时间。它输出ISO 8601格式的时间字符串作为观察（Observation）写回轨迹，便于在后续步骤中直接解析小时、日期等信息。若传入非法时区，将返回形如“错误：无效时区（…）”的文本。与计算器工具一样，获取时间的工具无论成功或失败均保持统一的字符串返回约定，以便记录与调试。tool_time_now()函数的实现代码如下。

from zoneinfo import ZoneInfo
import datetime

def tool_time_now(params: Dict[str, Any]) -> str:
    zone = params.get("zone")
    try:
        if zone:
            now = datetime.datetime.now(tz=ZoneInfo(zone))
        else:
            now = datetime.datetime.now()  # 本地时区
        return now.isoformat()
    except Exception as e:
        return f"错误：无效时区（{e}）"

已完成实现的两项工具需要添加到注册表中，注册表一方面用于将工具暴露给模型，另一方面用于在执行工具时将工具名称映射到工具函数名称。本示例中注册表通过字典实现，其代码如下。

TOOLS: Dict[str, Tool] = {
    "calculator": Tool(
        name="calculator",
        description="安全计算算式，支持 + - * / ** 与括号；参数：expression (string)。",
        schema={"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"]},
        func=tool_calculator
    ),
    "time_now": Tool(
        name="time_now",
        description="返回当前时间（ISO 8601 字符串）；可选参数：zone (IANA 时区，如 'Asia/Shanghai')。",
        schema={"type": "object", "properties": {"zone": {"type": "string"}}},
        func=tool_time_now
    ),
}

至此，ReAct智能体所需的工具准备完成。

4.用户提示构造与“思维草稿”拼接

ReAct智能体的推理过程是一个与大模型多轮交互的过程，而多轮交互能够持续推进的前提是交互的上下文中包含此前交互得到的中间推理结果（必要时，也可以包含推理过程）。这个上下文类似一张可递增的“思维草稿”，这张思维草稿在每一轮交互时都会把之前的Thought/Action/Observation作为参考信息拼回输入提供上下文，使模型具备连续的推理语境。当然，工具清单及工具的schema也会同步注入这张“思维草稿”，以便辅助模型做出正确的工具选择并填好工具调用所需的参数。
build_user_prompt()函数用于生成本轮发送给模型的“思维草稿”。函数将当前任务描述、由build_tools_block()函数拼接好的工具描述，以及已有的思维轨迹（scratchpad）按固定模板整合为一段文本，并在其中明确“只输出 JSON（不要解释，不要 markdown），字段为 thought/action/action_input”的格式约束。当历史轨迹为空时，会以占位语“（无）”填充，以保持结构稳定。

import textwrap

def build_user_prompt(task: str, scratchpad: str) -> str:
    return textwrap.dedent(f"""\
    任务：{task}

    {build_tools_block(TOOLS)}

    现在请只输出 JSON（不要解释，不要 markdown），字段为 thought/action/action_input。
    历史轨迹（供你参考）：
    {scratchpad if scratchpad.strip() else "（无）"}
    """).strip()

上述代码中被调用的build_tools_block()函数从注册表中读取工具的名称、说明与参数结构，拼接成一段“给模型看的工具说明”作为返回值，并被插入到“思维草稿”之中。其实现代码如下。

import json, textwrap

def build_tools_block(tools: Dict[str, Tool]) -> str:
    lines = ["可用工具列表："]
    for t in tools.values():
        lines.append(f"- {t.name}: {t.description}")
        schema = json.dumps(t.schema, ensure_ascii=False)
        lines.append(f"  schema={schema}")
    return "\n".join(lines)

通过将可用工具与过往步骤同时呈现，模型能够在完整语境中做出本轮决策，既减少无效调用，也便于ReAct多轮推理持续收敛。

5.调用与解析：一次“自我修复”保证格式回正

智能体的“思维大脑”是大模型，因而智能体的每一轮推理决策都依赖与大模型的交互。本示例中与大模型的交互以及对大模型返回信息的形式校验由llm_json_decision()函数实现。
llm_json_decision()函数负责向模型发起一次“结构化决策”请求，并将自由文本回复收敛为可解析、可执行的JSON对象。它将外部模型调用与解析细节统一封装，向上层仅暴露稳定的返回契约，即包含thought、action、action_input三个键的字典。通过这一收口，ReAct主循环无需关心模型输出的多样性与偶发噪声，只需据此三元信息驱动下一步行为。
llm_json_decision()函数的代码如下。

from openai import OpenAI

client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

def llm_json_decision(messages) -> Dict[str, Any]:
    """
    调用 DeepSeek（OpenAI 兼容），解析出 JSON 对象；
    若第一次未返回合规 JSON，进行一次自我修复重试。
    """
    resp = client.chat.completions.create(
        model=MODEL,
        messages=messages,
        temperature=0.2,
        top_p=0.9,
    )
    content = resp.choices[0].message.content.strip()

    def try_parse(txt: str) -> Optional[Dict[str, Any]]:
        try:
            left = txt.find("{")
            right = txt.rfind("}")
            if left != -1 and right != -1 and right > left:
                txt = txt[left:right+1]
            return json.loads(txt)
        except Exception:
            return None

    data = try_parse(content)
    if data is None:
        repair_messages = messages + [
            {"role": "assistant", "content": content},
            {"role": "user", "content": "你没有按要求输出可解析 JSON。请严格只输出 JSON（thought/action/action_input）。"}
        ]
        resp2 = client.chat.completions.create(
            model=MODEL,
            messages=repair_messages,
            temperature=0.0
        )
        content2 = resp2.choices[0].message.content.strip()
        data = try_parse(content2)

    if not isinstance(data, dict) or not all(k in data for k in ("thought", "action", "action_input")):
        preview = content[:200].replace("\n", " ")
        raise ValueError(f"模型未返回正确 JSON：{preview}...")

    return data

该函数首先以给定的消息序列调用大模型（调用时指定温度与top_p以控制生成稳定性），然后在取得大模型返回的文本后通过定位首尾花括号截取JSON片段并解码为字典的方式尝试解析。若解析失败，函数会追加一轮“自我修复”对话——把上一轮的输出原样作为assistant消息回显，再附上一条明确要求只输出JSON且仅含指定键名的用户指令，随后重新请求并再次解析。最终结果会经过最小结构校验（必须同时存在thought、action、action_input），若仍不满足契约，则抛出异常，将错误显式上浮给调用方。
从工程视角看，llm_json_decision()函数的作用在于把模型的不确定输出转化为上层需要的确定输入。它在I/O边界处完成容错与纠偏，一方面尽量容纳模型偶发的冗余文字或格式偏差，另一方面坚持输出契约不被破坏。由此，ReAct智能体流程的稳健性与可维护性显著提升，错误也能被快速定位在“决策—解析”这一层次，而不会扩散到工具执行或轨迹管理的后续环节。

6.ReAct智能体主循环：把“思—做—看”落成代码

ReAct闭环的调度核心通过react_loop()函数实现，它负责把“决策—执行—记录—迭代”的全过程组织起来。函数接收自然语言任务task、最大步数max_steps与可选的调试开关verbose三个参数，返回最终答案与完整轨迹文本。轨迹由若干步的Thought、Action、Observation串联而成，可直接用于回放与排障。
进入循环前，会先准备消息上下文，将系统提示词SYSTEM_PROMPT作为system消息固定在最前，随后在每一轮动态构造user消息。user消息通过build_user_prompt()函数构建，如前所述它由当前任务描述、build_tools_block()生成的工具自描述以及既往的“思维草稿”（scratchpad）三部分拼接而成。
每一轮迭代以一次llm_json_decision()调用开始。该函数对模型回复进行严格解析，期望得到只包含thought、action、action_input三个键的JSON。若解析失败，会在llm_json_decision()函数内部追加一次自我修复请求，尽最大可能把输出拉回合规形态。react_loop()在拿到合规决策后，会先将相关记录在调试输出中（当verbose=True时）。
随后进入分支处理，当action == “final”时，认为模型已具备直接给出结论的依据，action_input即为最终答案。此时将本轮Thought与最终结论写入轨迹，函数立即返回。若action指向某个工具，则从注册表查找对应实现，并对action_input做最小结构检查。未找到工具、参数类型不匹配或工具执行抛出异常时，都会生成带“错误：……”字样的Observation文本；成功时则将工具返回值作为Observation。无论成功或失败，Observation都会被追加到轨迹中，成为下一轮推理的上下文依据。
循环的退出条件有两类。一类是上述的final分支，属于正常完成；另一类是达到max_steps仍未收敛，此时返回一段提示性结论，指引缩小任务粒度或提高步数上限。通过显式的步数上限限制，能够避免模型在不明确目标时陷入长时间的无效尝试，从而控制时延与成本。
在可观测性方面，react_loop()函数将每步的Thought、Action、Action Input与Observation以统一文本格式写入scratchpad_steps，既能在控制台直观看到推理过程，也便于后续持久化到日志系统或事件总线进行回放分析。该设计配合工具层的统一字符串返回约定，使轨迹具备一致的语义与结构，方便在不同任务与环境中进行比对与调试。
react_loop()函数的实现代码如下。

def react_loop(task: str, max_steps: int = 10, verbose: bool = True) -> Tuple[str, str]:
    """
    执行 ReAct 闭环，返回 (final_answer, trajectory_text)
    """
    scratchpad_steps = []  # 展示轨迹：Thought/Action/Observation
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
    ]

    for step in range(1, max_steps + 1):
        scratchpad_text = "\n".join(scratchpad_steps)
        user_prompt = build_user_prompt(task, scratchpad_text)
        messages_step = messages + [{"role": "user", "content": user_prompt}]

        decision = llm_json_decision(messages_step)
        thought = str(decision.get("thought", "")).strip()
        action = str(decision.get("action", "")).strip()
        action_input = decision.get("action_input")

        if verbose:
            print(f"\n[Step {step}] Thought: {thought}")
            print(f"[Step {step}] Action:  {action}")
            print(f"[Step {step}] Action Input: {action_input}")

        if action == "final":
            final_answer = str(action_input)
            scratchpad_steps.append(f"Thought: {thought}\nFinal: {final_answer}")
            return final_answer, "\n".join(scratchpad_steps)

        tool = TOOLS.get(action)
        if not tool:
            obs = f"错误：未知工具 {action}"
        else:
            if not isinstance(action_input, dict):
                obs = "错误：action_input 不是 JSON 对象"
            else:
                try:
                    obs = tool.func(action_input)
                except Exception as e:
                    obs = f"错误：工具执行异常（{e}）"

        scratchpad_steps.append(
            f"Thought: {thought}\nAction: {action}\nAction Input: {json.dumps(action_input, ensure_ascii=False)}\nObservation: {obs}"
        )

    final = "达到最大步数仍未完成，请缩小问题或增加步数。"
    return final, "\n".join(scratchpad_steps)

从工程实践出发，react_loop()将不确定性的边界尽量前置与收口：决策解析的不确定性交由llm_json_decision()兜底，工具执行的不确定性用Observation文本承载并回流给模型，循环控制用步数上限与早终止策略把握成本。由此，整个ReAct流程能够在最小实现的复杂度下，获得可解释、可治理且可扩展的运行特性。

7.运行与观测：从轨迹理解行为

上述解析说明已经对ReAct智能体的主体内容进行了说明。随后可以尝试运行智能体并对其行为进行观测，以进一步理解ReAct智能体的推理过程。
示例任务可以设为“先计算 3.5 的平方，再加上我所在时区（Asia/Shanghai）的当前小时数，然后给出最终结果。”。测试代码如下。

if __name__ == "__main__":
    demo_task = "先计算 3.5 的平方，再加上我所在时区（Asia/Shanghai）的当前小时数，然后给出最终结果。"
    answer, trace = react_loop(demo_task, max_steps=5, verbose=True)

    print("\n====== 最终答案 ======")
    print(answer)

典型情况下，示例智能体第一轮会先计算3.5的平方，第二轮取得对应时区的小时数，第三轮将前两轮的结果相加，第四轮输出最终值。控制台会打印每一步的Thought、Action以及Observation，轨迹清晰、便于分析。
示例输出可能类似如图 2所示（具体文案与步数可能略有差异）。

图 2 运行过程与结果

8.常见问题与改进方向

若模型未按JSON输出，需要检查系统提示词是否足够明确，并可将温度调低到0.0以降低模型的“发挥欲”。示例已提供一次自我修复式重试，通常可以把输出拉回到合规格式。时区错误往往来自于非IANA标识，使用“Asia/Shanghai”或“Asia/Singapore”等标准名称即可。
在智能体工具使用能力扩展层面，可以把数据库查询、HTTP请求、文件读写等能力封装为工具，并继续沿用“白名单 + 参数校验 + 统一Observation”策略维护可控边界；将检索能力以工具形式接入，即可把RAG结果结构化地反馈给模型；引入MCP（Model Context Protocol）后，工具将具备自描述的发现能力，跨项目复用成本会大幅降低。对于生产场景，可观测性与治理尤为关键：轨迹落库、回放面板、速率限制、费用统计与审批机制，均能在保持灵活性的同时，保证安全与合规。

9.小结

本文所呈现的最小ReAct智能体内核的核心价值在于把不确定的生成过程压缩进确定的交互契约，即用固定键的JSON承接决策，用受限工具面向执行，用统一“观察”文本回流上下文。由此形成的“决策—执行—回流”闭环，使轨迹天然可解释，错误可被精准定位，工程边界清晰可控。
治理能力落在三处关节点：其一，决策接口只接受结构化JSON，从源头减少格式噪声；其二，工具以白名单与参数校验收紧权限，把风险隔离在最小面；其三，循环以步数上限与早终止策略控制成本，并以观察文本承载异常，保持语义上可自洽的回合推进。三者组合，既保障可用性，也为后续扩展留下稳固的支点。
落地实践可从“可观测”做起：将Thought、Action、Observation落库建立最小回放视图，需要时还可加入耗时、token用量等信息。评估与优化可围绕任务达成率、工具选择准确率、平均步数、单题时延与单位答案成本等指标展开，配合小步试错的Prompt与工具迭代，逐步收敛到业务所需的稳定区间。常见失效模式包括动作幻觉、参数漂移与轨迹过长，可通过更严格的输出约束、参数模式提示与轨迹截断/摘要化策略加以缓解。
本文示例完整代码参见：
https://gitcode.com/gtyan/AgentHandBook/tree/main/03