手写 ReAct Agent：用 100 行 Python 理解 Agent 核心循环

本章从零实现一个 ReAct Agent，不依赖任何框架。通过手写实现，可以清晰理解 Agent 循环的每一个环节，从而在使用 LangChain（一个用于构建 AI 应用的 Python 框架，封装了 LLM 调用、工具调用、链式执行等常用功能）等框架时，能准确判断问题出在哪里。

1.1 ReAct 的本质

ReAct 循环图
ReAct 核心循环 — Thought 思考、Action 行动、Observation 观察三步交替迭代

ReAct 来自 Shunyu Yao et al. 2022 年的论文《ReAct: Synergizing Reasoning and Acting in Language Models》。名字是 Reasoning + Acting 的合并，核心思路只有一句话：让 LLM 交替输出思考和行动，每次行动后把结果喂回去，继续下一轮思考。

循环是这样的：

这个循环的关键点有两个。第一，每次行动之前都有思考过程，模型不是盲目调用工具，而是先推理"现在应该做什么、为什么"。第二，每次观察结果都会被追加进对话历史，模型的下一步决策基于完整的上下文，而不是只看当前状态。

具体的输出格式长这样：

code

Thought: 用户问的是北京明天天气，我需要查询天气工具。
Action: get_weather
Action Input: {"city": "北京", "date": "明天"}
Observation: 北京明天：多云转晴，18-26°C，东风3级。

Thought: 已经拿到天气数据了，可以直接回答用户。
Finish: 北京明天多云转晴，气温18到26度，东风3级，不需要带伞。

这个格式是纯文本约定，不是什么特殊协议。LLM 按照 system prompt 里规定的格式输出，用正则表达式解析。这也是它最脆弱的地方，后面会讲。

1.2 从零实现

先看完整代码，再逐块解释：

python

import re
import json
import ast
import operator
from openai import OpenAI

client = OpenAI()

# ============================================================
# 1. 工具注册：字典存储工具名 -> {描述, 函数} 的映射
# ============================================================

def search(query: str) -> str:
    """模拟搜索工具，实际场景接入 Tavily / SerpAPI"""
    mock_data = {
        "北京天气": "北京今天晴，气温 22-31°C，南风 2 级",
        "上海天气": "上海今天多云，气温 24-29°C，东南风 3 级",
        "人口": "中国大陆人口约 14.1 亿（2023年数据）",
        "GDP": "中国 2023 年 GDP 约 126 万亿人民币",
    }
    for key, value in mock_data.items():
        if key in query:
            return value
    return f"未找到关于「{query}」的具体信息，请换个关键词试试。"


# ⚠️ 为什么不用 eval()？
# eval() 会执行任意 Python 代码，如果 LLM 生成了恶意表达式
# （如 "__import__('os').system('rm -rf /')"），会造成安全漏洞。
# 使用 AST 解析只允许数学运算符，安全可控。

def safe_eval_math(expression: str) -> float:
    """
    安全计算数学表达式，只允许基本运算符。
    使用 AST 解析而非 eval()，避免代码注入风险。
    """
    # 只允许的操作符
    allowed_operators = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.Mod: operator.mod,        # 支持取模运算，如 10 % 3
        ast.FloorDiv: operator.floordiv,  # 支持整除，如 10 // 3
        ast.USub: operator.neg,
    }

    def eval_node(node):
        if isinstance(node, ast.Constant):  # 数字
            return node.value
        elif isinstance(node, ast.BinOp):   # 二元运算
            op_type = type(node.op)
            if op_type not in allowed_operators:
                raise ValueError(f"不支持的运算符: {op_type.__name__}")
            left = eval_node(node.left)
            right = eval_node(node.right)
            return allowed_operators[op_type](left, right)
        elif isinstance(node, ast.UnaryOp): # 一元运算（如负号）
            op_type = type(node.op)
            if op_type not in allowed_operators:
                raise ValueError(f"不支持的运算符: {op_type.__name__}")
            return allowed_operators[op_type](eval_node(node.operand))
        else:
            raise ValueError(f"不支持的表达式类型: {type(node).__name__}")

    try:
        tree = ast.parse(expression, mode='eval')
        return eval_node(tree.body)
    except Exception as e:
        raise ValueError(f"无效的数学表达式 '{expression}': {e}")


def calculate(expression: str) -> str:
    """安全计算数学表达式"""
    try:
        result = safe_eval_math(expression)
        return str(result)
    except ValueError as e:
        return f"计算出错：{e}"


TOOLS = {
    "search": {
        "description": "搜索互联网获取信息。参数：query（字符串，搜索关键词）",
        "func": search,
    },
    "calculate": {
        "description": "执行数学计算。参数：expression（字符串，数学表达式，如 '3 * (12 + 5)'）",
        "func": calculate,
    },
}


# ============================================================
# 2. System Prompt：告诉 LLM 工具列表和输出格式
# ============================================================

def build_system_prompt(tools: dict) -> str:
    tool_descriptions = "\n".join(
        f"- {name}: {info['description']}"
        for name, info in tools.items()
    )
    return f"""你是一个智能助手，可以使用以下工具来回答问题：

{tool_descriptions}

每次回复必须严格遵守以下格式之一：

格式一（需要使用工具时）：
Thought: 你的分析和推理过程
Action: 工具名称（只写工具名，不加其他内容）
Action Input: {{"参数名": "参数值"}}

格式二（可以直接回答时）：
Thought: 你的分析过程
Finish: 最终答案

规则：
1. Action 那行只写工具名称，不要加任何解释
2. Action Input 必须是合法的 JSON 格式
3. 每次只输出一个 Thought 块，等待工具结果后再继续
4. 拿到足够信息后，用 Finish 输出最终答案"""


# ============================================================
# 3. 输出解析：用正则提取 Action 或 Finish
# ============================================================

def parse_llm_output(text: str) -> dict:
    """
    解析 LLM 输出，返回：
    - {"type": "action", "tool": "...", "input": {...}}
    - {"type": "finish", "answer": "..."}
    - {"type": "error", "raw": "..."}
    """
    # 尝试匹配 Finish
    finish_match = re.search(r'Finish:\s*(.+)', text, re.DOTALL)
    if finish_match:
        return {"type": "finish", "answer": finish_match.group(1).strip()}

    # 尝试匹配 Action
    action_match = re.search(r'Action:\s*(\w+)', text)
    input_match = re.search(r'Action Input:\s*(\{.+?\})', text, re.DOTALL)

    if action_match and input_match:
        tool_name = action_match.group(1).strip()
        try:
            tool_input = json.loads(input_match.group(1))
            return {"type": "action", "tool": tool_name, "input": tool_input}
        except json.JSONDecodeError:
            return {"type": "error", "raw": text}

    return {"type": "error", "raw": text}


# ============================================================
# 4. 核心循环：组装消息 -> 调用 LLM -> 解析 -> 执行工具 -> 追加结果
# ============================================================

def react_agent(user_question: str, max_steps: int = 8) -> str:
    messages = [
        {"role": "system", "content": build_system_prompt(TOOLS)},
        {"role": "user", "content": user_question},
    ]

    print(f"\n问题：{user_question}")
    print("=" * 50)

    for step in range(max_steps):
        # 调用 LLM
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0,  # 降低随机性，输出格式更稳定
        )
        llm_output = response.choices[0].message.content
        print(f"\n[Step {step + 1}] LLM 输出：\n{llm_output}")

        # 把 LLM 输出追加进历史
        messages.append({"role": "assistant", "content": llm_output})

        # 解析输出
        parsed = parse_llm_output(llm_output)

        if parsed["type"] == "finish":
            print(f"\n最终答案：{parsed['answer']}")
            return parsed["answer"]

        elif parsed["type"] == "action":
            tool_name = parsed["tool"]
            tool_input = parsed["input"]

            # 检查工具是否存在
            if tool_name not in TOOLS:
                observation = f"错误：工具 '{tool_name}' 不存在，可用工具：{list(TOOLS.keys())}"
            else:
                # 执行工具
                try:
                    func = TOOLS[tool_name]["func"]
                    observation = func(**tool_input)
                except Exception as e:
                    observation = f"工具执行出错：{e}"

            print(f"[工具执行] {tool_name}({tool_input}) -> {observation}")

            # 把观察结果追加进历史
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

        else:
            # 解析失败：把原始输出作为提示，要求重新输出
            print(f"[解析失败] 原始输出：{parsed['raw'][:100]}...")
            messages.append({
                "role": "user",
                "content": "格式错误，请严格按照规定格式重新输出（Thought + Action/Finish）"
            })

    return "达到最大步数限制，任务未完成"


# ============================================================
# 测试
# ============================================================

if __name__ == "__main__":
    react_agent("北京今天天气怎么样？气温加上100是多少度？")

整个实现大约 110 行（含注释和空行），核心逻辑不超过 80 行。

1.3 四个关键设计点

工具注册用字典，不是写死的 if-else。好处是加新工具只需要往 TOOLS 里加一条记录，核心循环不用改。这是最基础的可扩展性设计。

System Prompt 是灵魂。格式约定、工具描述、输出规则，全靠这里传达给 LLM。提示词写得模糊，输出就会乱。特别要注意的是，JSON 格式的要求要写得非常明确，否则 LLM 经常输出带注释的伪 JSON，正则解析直接挂。

temperature 设为 0。ReAct 的输出需要格式稳定，越确定越好。对话类任务可以提高 temperature 增加多样性，但在需要按格式输出的场景，temperature 越低越可靠。

最大步数必须设置。不设的话，LLM 陷入循环时任务永远不结束，一直消耗 token。8 步对大多数任务够用，复杂任务可以适当放宽到 15-20 步。

1.4 输出解析的脆弱性

这是手写 ReAct 最容易踩的坑。

正则表达式解析本质上是在赌 LLM 的输出格式。模型有时会：

把 Action: search 写成 Action: 使用 search 工具，action_match 失败
Action Input 里混入注释：{"query": "北京天气"} // 搜索天气信息，json.loads 报错
一次输出两个 Action，第一个被解析到，第二个直接丢失
直接回答了问题但没写 Finish: 标记，被误判为解析失败

几种缓解手段：

正则写宽松一点，比如用 re.IGNORECASE 忽略大小写，容忍多余的空格
JSON 解析出错时尝试修复，用 LLM 重新格式化一遍那段 JSON，再 parse
连续解析失败时提示 LLM 重试，把错误原因说清楚，而不是直接放弃
换用结构化输出，现在 OpenAI 支持 response_format={"type": "json_object"}，强制输出 JSON，避免格式混乱

根本解法是 Function Calling：让 API 层面强制结构化，彻底绕开文本解析。LangChain 的 ReAct Agent 现在底层就是基于 Function Calling 实现的，不再用文本正则。

1.5 和 LangChain 比，差距在哪

手写版和 LangChain 的 ReAct Agent 功能上差不多，但生产就绪程度差很远。

流式输出：LangChain 支持 streaming，用户能实时看到 LLM 在思考什么；手写版等 API 返回完整内容，用户盯着空屏等。

错误重试：LangChain 有内置的重试机制，包括解析失败重试、工具调用超时重试；手写版需要自己处理。

可观测性：LangChain 集成了 LangSmith，可以看到每一步的输入输出、token 消耗、耗时；手写版只有 print。

工具生态：LangChain 有几百个预制工具，搜索、数据库、文件操作都有现成的；手写版要自己写。

但这些"差距"恰恰是理解框架价值的路径。知道 LangChain 在 ReAct 循环的哪些环节做了封装，出了问题才知道从哪里插手。比如解析失败的报错，在 LangChain 里看 OutputParserException 就知道是提示词格式有问题，而不是框架 bug。

1.6 理解底层，才能用好框架

理解了底层循环之后，排查 Agent 问题的路径会变得清晰：

先看 prompt 模板对不对，再看工具描述是否够清晰，再看输出解析那步有没有异常，最后才考虑框架配置问题。

框架解决了大量工程问题，但框架封装不了对问题的理解。遇到奇怪的 bug，手写版的经验告诉你：ReAct 就是一个循环，一个提示词，一个解析器，三个工具函数。它能出什么问题，心里有数了。

1.7 工具超时：Agent的异常感知与应对

工具调用超时是生产环境最常见的问题之一，但很多教程不讲。

1.7.1 为什么工具超时特别棘手

普通代码超时：程序崩溃，用户看到错误。
Agent超时：Agent可能不知道工具超时了，继续"思考"——结果是无限等待或错误的后续决策。

1.7.2 三种超时场景和应对策略

场景	表现	应对策略
外部API超时（如天气API）	工具等待30秒无响应	设置超时限制，返回"工具暂时不可用"
LLM本身超时	生成响应时间过长	指数退避重试，最多3次
工具逻辑死循环	CPU占用100%，无响应	线程超时强制终止

1.7.3 带超时保护的工具执行

python

import asyncio
from typing import Any

async def execute_tool_with_timeout(
    tool_name: str,
    tool_args: dict,
    tools: dict,
    timeout_seconds: float = 10.0
) -> str:
    """
    带超时保护的工具执行。
    超时后返回友好的错误消息，而不是让Agent无限等待。
    """
    if tool_name not in tools:
        return f"错误：工具 '{tool_name}' 不存在，可用工具：{list(tools.keys())}"

    try:
        # asyncio.wait_for 会在超时后抛出 TimeoutError
        result = await asyncio.wait_for(
            asyncio.to_thread(tools[tool_name], **tool_args),
            timeout=timeout_seconds
        )
        return str(result)

    except asyncio.TimeoutError:
        # 超时后返回明确的错误信息，让Agent知道发生了什么
        return (
            f"工具 '{tool_name}' 执行超时（>{timeout_seconds}秒）。"
            f"请考虑：1) 用其他工具替代 2) 简化参数 3) 直接基于已有信息回答"
        )
    except Exception as e:
        return f"工具 '{tool_name}' 执行出错：{str(e)}"

1.7.4 超时后Agent的决策逻辑

关键：超时信息要清晰传给Agent，让它能做出合理决策。

python

# 在ReAct循环中，工具返回超时信息后
# Agent收到的Observation是：
# "工具 'search_web' 执行超时（>10秒）。请考虑：1) 用其他工具替代..."

# 好的System Prompt应该包含：
SYSTEM_PROMPT = """
...（其他内容）...

当工具执行超时时：
- 如果有替代工具，优先使用替代工具
- 如果没有替代工具，基于已有信息给出最佳答案，并说明"由于工具超时，答案可能不完整"
- 不要无限重试同一个超时的工具
"""

1.7.5 超时时间设置参考

工具类型	建议超时	原因
本地计算（数学、字符串处理）	5秒	本地操作应该很快
数据库查询	10秒	包含网络延迟
外部API调用	15秒	第三方服务不稳定
文件读写	30秒	大文件可能较慢
LLM子调用	60秒	LLM生成本身就慢

生产建议：超时时间不是越长越好。超时太长会让用户等待，超时太短会误杀正常请求。根据你的工具类型和用户容忍度调整。