Python 与 AI 开发教程

ReAct 和先规划后执行

边想边做,还是想好再做?这是 Agent 最基本的两种组织方式。这一章把它们都实现一遍,再看什么时候根本不需要 Agent。

  • Agent 开发 · 第 2 章
  • 11 分钟
  • 2 组对比
  • 基于 OpenAI Python SDK 3.42 · Pydantic 2.14

ReAct

ReAct 来自 2022 年的论文,名字是 Reasoning 和 Acting 的组合。它让模型交替进行三件事:

  • Thought:想一想现在知道什么,下一步该做什么
  • Action:调用一个工具
  • Observation:读取工具返回的结果

上一章的循环其实就是 ReAct。论文发表时还没有 Function Calling,模型只能输出文字,所以程序要自己解析。这段代码能看清它的原理。

Python
import re

from openai import OpenAI

client = OpenAI()

PROMPT = """按这个格式回答:
Thought: 你的思考
Action: 工具名[参数]
工具返回后你会收到 Observation。得出结论时输出 Answer: 最终回答。
可用工具:get_weather[城市]"""

TOOLS = {"get_weather": lambda city: f"{city}:晴"}

def react(task: str) -> str:
    messages = [
        {"role": "system", "content": PROMPT},
        {"role": "user", "content": task},
    ]
    for _ in range(5):
        response = client.chat.completions.create(model=MODEL, messages=messages)
        text = response.choices[0].message.content
        messages.append({"role": "assistant", "content": text})

        if match := re.search(r"Answer: (.*)", text):
            return match.group(1)

        name, argument = re.search(r"Action: (\w+)\[(.*)\]", text).groups()
        observation = TOOLS[name](argument)
        messages.append({"role": "user", "content": f"Observation: {observation}"})
    raise RuntimeError("超过最大步数")

print(react("北京今天天气如何?"))  # => 北京今天晴

现在不用这样解析文字了,tool_calls 比正则可靠得多,而且新模型自带思考能力。这种写法只在模型不支持工具调用时才有用。ReAct 留下来的是思路:每次只决定下一步,根据结果再决定再下一步。

ReAct 的优缺点

ReAct 适合下一步取决于上一步结果的任务,比如调试程序、查资料。它能随时调整,遇到意外情况也能应对。

它的缺点也来自「走一步看一步」:

  • 容易陷入局部,做很多无关的步骤
  • 任务复杂时会忘记整体目标
  • 每一步都要调用模型,累积的历史让费用增长很快

先规划后执行

另一种做法是把思考和行动分开。先让模型列出完整的计划,再逐步执行,每一步可以用更便宜的模型,只拿到这一步需要的上下文。

计划用结构化输出,程序才能读取每一步。

计划的数据结构

Python
from openai import OpenAI
from pydantic import BaseModel, Field

class Step(BaseModel):
    id: int
    task: str = Field(description="这一步要完成的事")
    depends_on: list[int] = Field(description="依赖的步骤编号")

class Plan(BaseModel):
    goal: str
    steps: list[Step]

client = OpenAI()

def make_plan(task: str) -> Plan:
    response = client.chat.completions.parse(
        model=MODEL,
        messages=[
            {"role": "system", "content": "把任务拆成三到七个步骤"},
            {"role": "user", "content": task},
        ],
        response_format=Plan,
    )
    return response.choices[0].message.parsed

plan = make_plan("写一份销售报告")
[step.task for step in plan.steps]  # => ['收集数据', '写总结']
record Step(int id, String task, List<Integer> dependsOn) {}

record Plan(String goal, List<Step> steps) {}

interface Planner {
    @SystemMessage("把任务拆成三到七个步骤")
    Plan makePlan(String task);
}

Plan run(ChatModel model) {
    var planner = AiServices.create(Planner.class, model);
    return planner.makePlan("写一份销售报告");
}
对比项PythonJava
结构Pydantic 模型record
调用parse接口方法返回 Plan
校验PydanticJackson 反序列化

depends_on 把计划变成依赖图,没有依赖关系的步骤可以并行执行。

执行时按顺序处理,每一步只拿到前面步骤的结果,不带上全部历史。

Python
def execute_plan(plan: Plan) -> dict[int, str]:
    results: dict[int, str] = {}
    for step in plan.steps:
        context = "\n".join(results[i] for i in step.depends_on)
        response = client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": f"总目标:{plan.goal}。只完成当前步骤。"},
                {"role": "user", "content": f"已有结果:\n{context}\n\n当前步骤:{step.task}"},
            ],
        )
        results[step.id] = response.choices[0].message.content
    return results

execute_plan(plan)[2]  # => '总结:销量增长良好'

计划可能赶不上变化。实际项目里常加一步重新规划:每执行完一步,检查结果和计划是否还相符,不符合就调整剩下的步骤。这部分用 LangGraph 写起来很方便,后面会讲。

两种方式怎么选

两者经常组合使用:外层规划,内层每个步骤是一个小的 ReAct 循环。

可能不需要 Agent

很多任务用固定的流程就能完成,更便宜,也更容易测试。Anthropic 总结了五种常见的工作流模式:

路由是最常用的,用结构化输出做分类,后面就是普通的 if。

路由

Python
from typing import Literal

from openai import OpenAI
from pydantic import BaseModel

class Route(BaseModel):
    category: Literal["refund", "shipping", "other"]

client = OpenAI()

def classify(text: str) -> str:
    response = client.chat.completions.parse(
        model=MODEL,
        messages=[{"role": "user", "content": text}],
        response_format=Route,
    )
    return response.choices[0].message.parsed.category

def handle(text: str) -> str:
    match classify(text):
        case "refund":
            return "转到退款流程"
        case "shipping":
            return "转到物流查询"
        case _:
            return "转人工"

print(handle("我想退货"))  # => 转到退款流程
enum Category { REFUND, SHIPPING, OTHER }

interface Router {
    @UserMessage("判断这个请求的类别:{{it}}")
    Category classify(String text);
}

String handle(Router router, String text) {
    return switch (router.classify(text)) {
        case REFUND -> "转到退款流程";
        case SHIPPING -> "转到物流查询";
        case OTHER -> "转人工";
    };
}
对比项PythonJava
类别Literalenum
分支matchswitch
完整性检查类型检查工具才会提示编译器检查

路由之后的分支都是确定的代码,能用普通测试覆盖。模型只负责分类这一步。

本章要点

  • ReAct 交替思考、行动和观察,上一章的循环就是它的现代写法
  • 走一步看一步很灵活,但费用高,也容易偏离目标
  • 先规划后执行:结构化输出计划,每步只带需要的上下文,必要时重新规划
  • 探索性任务用 ReAct,目标明确的任务用规划,两者可以组合
  • 优先考虑提示链、路由、并行这些工作流,确实需要时才用 Agent

参考资料:ReAct 论文构建有效的 AgentLangGraph 工作流和 Agent