LangChain ReAct 范式详解

ReAct 是 Reasoning（推理）和 Acting（行动）的组合词。它是 LangChain 中构建智能 Agent（智能体）最核心、最经典的范式。

简单来说，ReAct 范式让大模型不再只是“单纯的对话”，而是学会了**“先思考，再行动，根据结果再思考”**的循环过程，直到解决问题。

---

1. 核心理念：为什么需要 ReAct？

普通的 LLM（如直接对话）存在两个主要缺陷：

1. 幻觉（Hallucination）：不懂装懂，一本正经地胡说八道。
2. 数据滞后/封闭：无法获取当前最新的信息（如今天的股价、私有数据库）。

ReAct 范式通过让模型使用**工具（Tools）**来弥补这些缺陷。它要求模型在回答问题前，必须遵循一个严格的思考路径：

$$\text{Thought (思考)}\to \text{Action (行动)}\to \text{Observation (观察)}$$

---

2. ReAct 的执行流程（三步走循环）

在 LangChain 中，这个过程表现为一个**“思考-行动-观察”的循环**，直到模型认为获得了最终答案：

1. Question (用户输入)
   • 用户问：“特斯拉 Model 3 现在的价格是多少？”
2. Thought (推理/Reasoning)
   • 模型自言自语：“用户问的是价格，我不能瞎编，我应该查找数据库。”
3. Action (行动/Acting)
   • 模型决定调用工具：Action: 查询产品名称
   • 输入参数：Action Input: Model 3
4. Observation (观察/外部反馈)
   • 关键步骤：程序（AgentExecutor）截获模型的 Action，执行真正的 Python 函数或 API。
   • 结果：“Model 3 定价 23.19-33.19万”。
   • 系统将这个结果作为 Observation 喂回给模型。
5. Thought (再次推理)
   • 模型结合新信息思考：“我已经拿到价格了，可以回答用户了。”
6. Final Answer (最终回答)
   • 模型输出：“特斯拉 Model 3 的价格在 23.19万 到 33.19万之间。”

---

3. 直观对比案例

假设用户问：“马斯克现在的年龄乘以 2 是多少？”

❌ 没有 ReAct (普通 LLM)

LLM: 马斯克今年 50 岁（可能是模型训练时的旧数据），乘以 2 是 100 岁。
(结果可能是错的，且没有验证过程)

✅ 有 ReAct (LangChain Agent)

Question: 马斯克现在的年龄乘以 2 是多少？

Thought: 我需要先知道马斯克现在的真实年龄，这需要用搜索工具。
Action: Google Search
Action Input: “Elon Musk age”

Observation: (搜索引擎返回) 52 years old.

Thought: 我现在知道他 52 岁了。用户要求乘以 2。我应该计算 52 * 2。
Action: Calculator
Action Input: 52 * 2

Observation: (计算器返回) 104

Thought: 我已经得到最终结果了。
Final Answer: 马斯克现在的年龄乘以 2 等于 104。

---

4. 代码与概念的映射

结合实例代码，LangChain 的各个组件是如何实现 ReAct 的：

概念	对应代码组件	作用
剧本/规则	PromptTemplate	定义了 Thought/Action/Observation 的格式，告诉 LLM 必须按这个套路出牌。
大脑	LLM (Tongyi)	负责生成 Thought 和 Action 的文本。
手脚/技能	Tools (TeslaDataSource)	实际执行操作的函数（如 find_product_description），赋予模型查库的能力。
翻译官	OutputParser	负责利用正则表达式，从 LLM 的一大段输出中精准提取出要调用的工具名和参数。
控制中心	AgentExecutor	循环的驱动者。它负责： 1. 把 Prompt 发给 LLM。 2. 拿到 Action 去运行函数。 3. 把运行结果包装成 Observation 再次发给 LLM。 4. 直到出现 Final Answer 才停止。

---

5. 总结

ReAct 范式就是让大模型学会“使用工具”和“分步思考”的方法论。

• Reasoning：避免盲目行动，先思考（生成 Thought）。
• Acting：获取外部信息或进行计算，调用工具（生成 Action）。
• Loop：通过 Observation 修正认知，不断循环直到解决问题。

6.本地客服代码实现

import warnings

from langchain_classic.agents import AgentOutputParser, LLMSingleActionAgent, AgentExecutor
from langchain_classic.chains.llm import LLMChain

warnings.filterwarnings('ignore', category=DeprecationWarning)

import os
import re
import textwrap
import time
from typing import List, Union

from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.tools import Tool
from langchain_community.llms import Tongyi
from langchain_core.prompts import PromptTemplate, StringPromptTemplate
from langchain_core.language_models import BaseLLM
from langchain_core.messages import HumanMessage  # 确保导入这个

# --- Prompt 定义保持不变 ---
CONTEXT_QA_TMPL = """
根据以下提供的信息，回答用户的问题
信息：{context}

问题：{query}
"""
CONTEXT_QA_PROMPT = PromptTemplate(
    input_variables=["query", "context"],
    template=CONTEXT_QA_TMPL,
)


def output_response(response: str) -> None:
    if isinstance(response, dict):
        output = response.get("output", "")
    else:
        output = str(response)

    if not output:
        return

    output = output.rstrip("#")
    for line in textwrap.wrap(output, width=60):
        for word in line.split():
            for char in word:
                print(char, end="", flush=True)
                time.sleep(0.02)  # 稍微调快一点演示速度
            print(" ", end="", flush=True)
        print()
    print("-" * 60)


class TeslaDataSource:
    def __init__(self, llm: BaseLLM):
        self.llm = llm

    def find_product_description(self, product_name: str) -> str:
        product_info = {
            "model 3": "具有简洁、动感的外观设计，流线型车身和现代化前脸。定价23.19-33.19万",
            "model y": "在外观上与Model 3相似，但采用了更高的车身和更大的后备箱空间。定价26.39-36.39万",  # 统一小写方便匹配
            "model x": "拥有独特的翅子门设计和更加大胆的外观风格。定价89.89-105.89万",
        }
        # 简单做个小写处理，提高命中率
        key = product_name.lower().strip()
        if key in product_info:
            return product_info[key]
        else:
            available_products = ", ".join(product_info.keys())
            return f"抱歉，我只能提供以下产品的信息：{available_products}。您询问的 '{product_name}' 不在我们的产品列表中。"

    def find_company_info(self, query: str) -> str:
        context = """
        特斯拉最知名的产品是电动汽车，其中包括Model S、Model 3、Model X和Model Y等多款车型。
        特斯拉以其技术创新、高性能和领先的自动驾驶技术而闻名。
        """
        prompt = CONTEXT_QA_PROMPT.format(query=query, context=context)
        messages = [HumanMessage(content=prompt)]

        # 【修复点1】确保返回的是字符串，而不是对象
        response = self.llm.invoke(messages)
        # 兼容不同版本的 LangChain 返回值
        if hasattr(response, 'content'):
            return response.content
        return str(response)


AGENT_TMPL = """你是一个严格遵守规则的AI助手。你的职责是只能使用提供的工具来回答问题。

你可以使用下面这些工具：
{tools}

回答时请遵循以下格式（注意：Thought/Action/Observation 可以重复多次，直到得到最终答案）：
Question: 用户提出的问题
Thought: 这里的思考过程...
Action: {tool_names} 中的一个工具名
Action Input: 工具需要的具体输入参数
Observation: 工具返回的结果
... (重复 Thought/Action/Observation)
Thought: 我现在知道最终答案了
Final Answer: 原始输入问题的最终答案

重要规则：
1. 必须从 Observation 中获取信息，禁止编造。
2. 得到 Observation 后，请立即分析并给出 Final Answer。

Question: {input}
{agent_scratchpad}
"""


class CustomPromptTemplate(StringPromptTemplate):
    template: str
    tools: List[Tool]

    def format(self, **kwargs) -> str:
        intermediate_steps = kwargs.pop("intermediate_steps")
        thoughts = ""
        for action, observation in intermediate_steps:
            thoughts += action.log
            # 确保 observation 是字符串，避免格式错误
            thoughts += f"\nObservation: {str(observation)}\nThought: "

        kwargs["agent_scratchpad"] = thoughts
        kwargs["tools"] = "\n".join([f"{tool.name}: {tool.description}" for tool in self.tools])
        kwargs["tool_names"] = ", ".join([tool.name for tool in self.tools])

        return self.template.format(**kwargs)


class CustomOutputParser(AgentOutputParser):
    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        # 1. 检查是否包含 Final Answer
        if "Final Answer:" in llm_output:
            return AgentFinish(
                return_values={"output": llm_output.split("Final Answer:")[-1].strip()},
                log=llm_output,
            )

        # 2. 尝试解析 Action
        regex = r"Action\s*\d*\s*:(.*?)\nAction\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"
        match = re.search(regex, llm_output, re.DOTALL)

        if not match:
            # 【修复点2】容错处理：
            # 如果 LLM 没按套路出牌（既没有 Action 也没有 Final Answer），
            # 但它输出了内容，我们假设这可能就是答案（或者是胡言乱语），避免直接 Crash。
            # 通常在 Observation 之后，LLM 可能会直接说 "答案是xxx" 而忘了加 Final Answer 前缀。

            # 简单的策略：如果内容看起来不像是在调用工具，就把它当做最终回复
            return AgentFinish(
                return_values={"output": llm_output.strip()},
                log=llm_output,
            )

        action = match.group(1).strip()
        action_input = match.group(2)
        action_input = action_input.strip().strip('"').split('\n')[0].strip()

        return AgentAction(
            tool=action, tool_input=action_input, log=llm_output
        )


# 设置通义千问API密钥
DASHSCOPE_API_KEY = os.environ.get('DASHSCOPE_API_KEY')

if __name__ == "__main__":
    if not DASHSCOPE_API_KEY:
        print("请设置环境变量 DASHSCOPE_API_KEY")
        exit(1)

    llm = Tongyi(model_name="qwen-plus", dashscope_api_key=DASHSCOPE_API_KEY)
    tesla_data_source = TeslaDataSource(llm)

    tools = [
        Tool(
            name="查询产品名称",
            func=tesla_data_source.find_product_description,
            description="通过产品名称找到产品描述时用的工具，输入的是产品名称，例如：Model 3",
        ),
        Tool(
            name="公司相关信息",
            func=tesla_data_source.find_company_info,
            description="当用户询问公司相关的问题（如公司介绍、历史、总体情况）时使用",
        ),
    ]

    agent_prompt = CustomPromptTemplate(
        template=AGENT_TMPL,
        tools=tools,
        input_variables=["input", "intermediate_steps"],
    )

    output_parser = CustomOutputParser()
    llm_chain = LLMChain(llm=llm, prompt=agent_prompt)

    tool_names = [tool.name for tool in tools]

    agent = LLMSingleActionAgent(
        llm_chain=llm_chain,
        output_parser=output_parser,
        # 这里非常关键：告诉 LLM 遇到 Observation 就停下来，把控制权交给代码
        stop=["\nObservation:"],
        allowed_tools=tool_names,
    )

    agent_executor = AgentExecutor.from_agent_and_tools(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=3,
        handle_parsing_errors=True  # 开启内置的解析错误处理，防止程序崩溃
    )

    while True:
        try:
            print("\n" + "=" * 30)
            user_input = input("请输入您的问题(输入 q 退出)：")
            if user_input.lower() in ['q', 'quit', 'exit']:
                break
            response = agent_executor.invoke({"input": user_input})
            output_response(response)
        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"发生错误: {e}")

7.实现效果