跳转至

第4章 复杂 Chain 模式

学习目标

  • 理解 Chain-of-Thought(CoT)推理原理及其适用场景
  • 掌握 ReAct 模式(Reasoning + Acting)的思考-行动-观察循环
  • 了解 Tree of Thoughts(ToT)树形搜索思想
  • 掌握工具调用链的设计与多步推理实现
  • 能够从零实现一个 ReAct Agent

问题导入

某团队开发了一个"数据分析助手",用户问:"今年三季度华东区销售额比二季度增长多少?"这个问题需要多步处理:①查询三季度华东区销售额;②查询二季度华东区销售额;③计算增长率;④组织回答。

直接把问题抛给 LLM,它会"一本正经地胡说八道"——因为它既没有数据,也不会调用数据库。即便接入了查询工具,模型也常常"一次调用就想给答案",不会分步思考。

这类需要多步推理 + 工具调用的任务,必须用 Chain 模式来解决。本章讲解四种主流 Chain 模式,让你能应对复杂业务推理。

核心概念

1. Chain-of-Thought(CoT 思维链)

CoT 让模型在给出答案前,先展示"思考过程",逐步推理得出结论。

两种触发方式: - Zero-shot CoT:在 Prompt 末尾加"让我们一步一步思考" - Few-shot CoT:在示例中展示推理过程

原理:LLM 是自回归模型,每个 token 的生成都以之前所有 token 为条件。展示中间推理步骤,相当于给模型更多"思考 token"来逼近正确答案,尤其对数学、逻辑题有效。

经典示例:

问题:小明有 5 个苹果,给了小红 2 个,又买了 3 个,现在有几个?
CoT 推理:小明原有 5 个,给出 2 个后剩 5-2=3 个,又买 3 个后 3+3=6 个。
答案:6 个

2. ReAct 模式(Reasoning + Acting)

ReAct = Reasoning(推理)+ Acting(行动),是 Agent 最经典的模式。它让模型在"思考"和"行动"之间交替进行:

  • Thought:思考下一步该做什么
  • Action:选择并调用工具
  • Observation:观察工具返回结果
  • 循环直到得出最终答案

ReAct 的核心价值:把 LLM 的推理能力与外部工具的执行能力结合,模型不再只是"说",还能"做"。

3. Tree of Thoughts(ToT 思维树)

ToT 把推理过程组织为树形结构:

  • 在每个推理节点生成多个候选想法(分支)
  • 评估每个想法的可行性(打分)
  • 选择最优分支继续探索(搜索)
  • 支持广度优先(BFS)或深度优先(DFS)

适用于解空间大、需要回溯的复杂问题(如数学证明、游戏策略)。缺点是 Token 消耗大、实现复杂。

4. 工具调用链

现代 LLM(如 GPT-4)原生支持 Function Calling,工具调用链包括:

  1. 模型决定调用哪个工具 + 参数
  2. 系统执行工具,返回结果
  3. 模型基于结果继续推理或调用下一个工具
  4. 循环直到任务完成

5. 多步推理的工程要点

  • 终止条件:最大轮数限制,防止死循环
  • 错误处理:工具调用失败时反馈给模型重试
  • 上下文累积:每轮的 Thought/Action/Observation 都加入历史
  • 成本控制:监控 Token 消耗,复杂任务可能多轮调用

原理图

flowchart TB
    Start([用户问题]) --> Thought1[Thought 1<br/>分析问题,决定第一步]

    subgraph ReAct["ReAct 循环"]
        Thought1 --> Action1[Action 1<br/>调用工具 A]
        Action1 --> Obs1[Observation 1<br/>工具 A 返回结果]
        Obs1 --> Thought2[Thought 2<br/>基于结果分析]
        Thought2 --> Action2[Action 2<br/>调用工具 B]
        Action2 --> Obs2[Observation 2<br/>工具 B 返回结果]
        Obs2 --> Thought3[Thought 3<br/>信息充分?]
    end

    Thought3 -->|否| Thought2
    Thought3 -->|是| Final[Final Answer<br/>组织最终回答]
    Final --> End([输出答案])

    style Thought1 fill:#fff9c4
    style Thought2 fill:#fff9c4
    style Thought3 fill:#fff9c4
    style Action1 fill:#bbdefb
    style Action2 fill:#bbdefb
    style Obs1 fill:#c8e6c9
    style Obs2 fill:#c8e6c9
    style Final fill:#c8e6c9

ReAct 模式通过 Thought-Action-Observation 循环,让模型在推理与行动间交替,直到信息充分后给出最终答案。

示例代码

1. CoT 思维链示例

"""
第4章示例:Chain-of-Thought 思维链
"""
from openai import OpenAI

client = OpenAI()


def solve_with_cot(question: str) -> str:
    """Few-shot CoT:通过示例引导模型展示推理过程"""
    prompt = """请按以下示例格式回答数学问题,先展示推理步骤。

示例1:
问题:小明有 5 个苹果,给了小红 2 个,又买了 3 个,现在有几个?
推理:原有 5 个,给出 2 个剩 3 个,又买 3 个共 6 个。
答案:6 个

示例2:
问题:一件衣服原价 200 元,先打 8 折,再用 30 元优惠券,最终多少钱?
推理:200 × 0.8 = 160 元,160 - 30 = 130 元。
答案:130 元

现在请回答:
问题:{q}
推理:""".format(q=question)

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    print(solve_with_cot("商店有 120 件商品,上午卖出 35 件,下午又进货 50 件,现在有多少件?"))

2. ReAct Agent 完整实现

"""
ReAct Agent 实现:推理 + 行动循环
"""
from openai import OpenAI
import json
import re

client = OpenAI()


# ========== 工具定义 ==========
def search_product(query: str) -> str:
    """模拟商品搜索"""
    db = {
        "iPhone 15": {"price": 5999, "stock": 100},
        "MacBook Pro": {"price": 14999, "stock": 30},
        "AirPods": {"price": 1299, "stock": 200},
    }
    for name, info in db.items():
        if query.lower() in name.lower():
            return f"商品:{name},价格:{info['price']}元,库存:{info['stock']}件"
    return f"未找到商品:{query}"


def calculate(expression: str) -> str:
    """模拟计算器"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果:{expression} = {result}"
    except Exception as e:
        return f"计算失败:{e}"


TOOLS = {
    "search_product": {
        "function": search_product,
        "description": "搜索商品信息(价格、库存)。参数:query(商品名称)",
    },
    "calculate": {
        "function": calculate,
        "description": "数学计算。参数:expression(计算表达式,如 '100*0.8-30')",
    },
}


# ========== ReAct Prompt 模板 ==========
REACT_PROMPT = """你是一个能使用工具的助手。请按以下格式回答问题:

Thought: 思考下一步该做什么
Action: 工具名称
Action Input: 工具参数(JSON 格式)
Observation: 工具返回结果
...(重复 Thought/Action/Observation 直到得出答案)
Thought: 我现在知道最终答案
Final Answer: 最终回答

可用工具:
{tools}

问题:{question}

请从 Thought 开始回答:"""


def build_tools_desc() -> str:
    return "\n".join(
        f"- {name}: {info['description']}" for name, info in TOOLS.items()
    )


def parse_action(text: str):
    """从模型输出中解析 Action 和 Action Input"""
    action_match = re.search(r"Action:\s*(\w+)", text)
    input_match = re.search(r"Action Input:\s*(.+?)(?:\n|$)", text)
    if action_match and input_match:
        return action_match.group(1), input_match.group(1).strip()
    return None, None


def react_agent(question: str, max_steps: int = 5) -> str:
    """ReAct Agent 主循环"""
    messages = [
        {
            "role": "user",
            "content": REACT_PROMPT.format(
                tools=build_tools_desc(), question=question
            ),
        }
    ]

    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0,
        )
        output = resp.choices[0].message.content
        messages.append({"role": "assistant", "content": output})

        print(f"\n--- 第 {step + 1} 步 ---")
        print(output)

        # 检查是否得到最终答案
        if "Final Answer:" in output:
            return output.split("Final Answer:")[1].strip()

        # 解析并执行工具
        action, action_input = parse_action(output)
        if not action:
            messages.append({
                "role": "user",
                "content": "请按格式输出 Action 和 Action Input,或给出 Final Answer。",
            })
            continue

        if action not in TOOLS:
            observation = f"错误:未知工具 {action}"
        else:
            try:
                observation = TOOLS[action]["function"](action_input)
            except Exception as e:
                observation = f"工具执行错误:{e}"

        messages.append({
            "role": "user",
            "content": f"Observation: {observation}",
        })

    return "已达最大步数,未能得出答案。"


if __name__ == "__main__":
    question = "买 2 个 iPhone 15 和 1 个 AirPods,总共多少钱?"
    answer = react_agent(question)
    print(f"\n最终答案:{answer}")

3. 基于 Function Calling 的工具调用链

"""
使用 OpenAI Function Calling 实现工具调用链(推荐方式)
"""
from openai import OpenAI
import json

client = OpenAI()

TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "search_product",
            "description": "搜索商品价格和库存",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "商品名称"}
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "计算表达式"}
                },
                "required": ["expression"],
            },
        },
    },
]


def function_call_agent(question: str, max_steps: int = 5) -> str:
    """基于 Function Calling 的多步工具调用"""
    messages = [{"role": "user", "content": question}]

    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS_SCHEMA,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            result = TOOLS[name]["function"](**args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result),
            })

    return "已达最大调用次数。"

企业案例

复杂推理任务

某跨境电商公司的"选品分析 Agent"需要回答这类问题:"对比过去 30 天,A 产品和 B 产品哪个销量增长更快?建议补货哪个?"

这个任务需要多步推理 + 多次工具调用:

  1. Thought:需要先查 A 和 B 过去 30 天销量数据
  2. Action:调用 get_sales(product="A", days=30)
  3. Observation:A 销量 1000 件,环比增长 15%
  4. Action:调用 get_sales(product="B", days=30)
  5. Observation:B 销量 800 件,环比增长 25%
  6. Thought:B 增长率更高(25% > 15%),但 A 销量更大
  7. Action:调用 get_inventory(product="B")
  8. Observation:B 库存仅剩 50 件
  9. Thought:B 增长快且库存低,建议补货 B
  10. Final Answer:B 产品销量增长更快(25% vs 15%),且库存仅剩 50 件,建议优先补货 B。

团队选择 ReAct 模式而非 CoT 的原因:

  • 需要外部数据:销量、库存必须实时查询,模型无法"想出来"
  • 多步骤决策:不是一次查询就能回答,需要根据中间结果决定下一步
  • 可解释性:每步 Thought 记录推理过程,便于审计

关键经验:

  1. 工具设计要原子化:一个工具做一件事,组合调用而非做大而全的工具
  2. 终止条件必须设:max_steps 防止死循环,生产环境设 5-10 步
  3. Observation 要精简:工具返回内容过长会撑爆上下文,需摘要
  4. 成本监控:ReAct 多轮调用,Token 消耗是单次调用的 3-5 倍

常见问题

Q1:CoT 什么时候有效,什么时候无效?

A:CoT 对数学计算、逻辑推理、多步规划类任务有效;对简单事实问答、分类任务效果不明显甚至有反作用。判断标准:任务是否需要"中间推理步骤"。

Q2:ReAct 和 Function Calling 是什么关系?

A:ReAct 是一种"思考-行动-观察"的推理范式;Function Calling 是 OpenAI 提供的工具调用接口。Function Calling 是实现 ReAct 的推荐方式——模型自动决定调用哪个工具,比正则解析 Action 更可靠。

Q3:ToT 什么时候用?

A:ToT 适用于解空间大、需要探索与回溯的问题(如 24 点游戏、创意写作、数学证明)。日常业务任务用 ReAct 足够,ToT 的 Token 成本是 ReAct 的 5-10 倍。

Q4:ReAct 循环停不下来怎么办?

A:①设置 max_steps(建议 5-10);②在 Prompt 中明确"信息充分后立即给出 Final Answer";③监控每步的 Thought,若连续 3 步无进展则强制终止;④用单独的"判断器"判断是否该停止。

Q5:多步推理如何控制成本?

A:①精简 Observation(工具结果摘要);②合并相关工具调用(并行 Function Calling);③简单子任务用小模型,复杂推理用大模型(路由策略);④缓存中间结果,避免重复查询。

面试题

  1. 简答题:CoT 的原理是什么?为什么"让我们一步一步思考"能提升推理效果?

  2. 对比题:ReAct 模式与纯 CoT 的区别是什么?ReAct 解决了 CoT 的什么局限?

参考要点:CoT 只能基于模型内部知识推理;ReAct 引入外部工具,能获取实时数据、执行计算。

  1. 设计题:设计一个 ReAct Agent 的工具集,用于回答"明天北京到上海的航班有哪些?最便宜的是哪个?"

  2. 实现题:用 Function Calling 实现 ReAct 时,如何处理"模型连续调用同一工具但参数错误"的情况?

参考要点:记录失败参数,在工具返回中提示正确参数格式,或限制同一工具最大调用次数。

  1. 原理题:Tree of Thoughts 相比 Chain of Thoughts 的优势是什么?为什么实际项目中用得少?

练习题

  1. CoT 实验:用 Zero-shot CoT 和 Few-shot CoT 分别解答 10 道数学题,对比准确率。尝试去掉"一步一步思考"提示,观察效果差异。

  2. ReAct 实现:基于本章 ReAct 模板,实现一个"天气查询 Agent",工具包括 get_weather(city) 和 get_activity_suggestion(weather),能回答"北京今天适合户外运动吗?"

  3. Function Calling 改造:将本章示例 2 的正则解析版 ReAct 改造为使用 Function Calling(参考示例 3),对比两种实现的稳定性与代码复杂度。

  4. 终止策略:为实现 ReAct Agent 增加一个"进度判断器"——每 2 步调用一次 LLM 判断"当前信息是否足够回答问题",若足够则强制输出 Final Answer。

  5. ToT 简化版:实现一个简化版 ToT——对一个问题生成 3 个候选推理路径,用 LLM 打分选最优路径继续,最多 3 层。

小结

复杂 Chain 模式让 LLM 从"单轮回答"进化为"多步推理 + 工具调用"的智能体。本章核心要点:

  • CoT 是基础:通过展示推理过程提升复杂问题准确率,适合数学、逻辑任务。
  • ReAct 是 Agent 核心:Thought-Action-Observation 循环,结合推理与工具调用,是生产环境首选。
  • Function Calling 是实现利器:比正则解析更可靠,OpenAI 原生支持多工具并行调用。
  • ToT 适合探索性问题:树形搜索支持回溯,但成本高,按需选用。
  • 工程要点:设终止条件、精简 Observation、错误重试、成本监控,缺一不可。

下一章将学习 Prompt 调试与 Guardrail,为 Agent 加上安全护栏,确保输出可控、合规。