在这里插入图片描述

别让LLM再"瞎编"了!ReAct Agent实战:手把手教你打造会"思考-行动-观察"的智能搜索问答系统,让AI真正学会"先想后做"


ReAct Agent
搜索问答实战

核心概念

ReAct原理: 推理+行动

Agent架构设计

工具调用机制

环境搭建

LangChain核心组件

搜索工具集成

LLM配置与选型

实战开发

基础ReAct Agent实现

多工具协同调用

对话记忆管理

进阶优化

错误处理与重试

输出解析优化

性能监控与调试

场景应用

企业知识问答

实时信息检索

多轮对话交互

避坑指南

常见错误分析

调试技巧分享

生产环境建议

目录

  • 一、ReAct核心原理:为什么LLM需要"先想后做"
  • 二、环境搭建:从零配置LangChain Agent开发环境
  • 三、基础实战:手写第一个ReAct搜索问答Agent
  • 四、进阶玩法:多工具协同与对话记忆
  • 五、生产优化:错误处理、调试与性能监控
  • 六、场景落地:企业级搜索问答系统实战
  • 七、避坑指南:新手最常踩的8个坑

嗨,大家好呀,我是你的老朋友精通代码大仙。接下来我们一起学习 《LangChain核心技术与LLM项目实践》,震撼你的学习轨迹!


“脑子是个好东西,希望AI也有一个。”

这句调侃背后,藏着多少程序员被LLM" hallucination(幻觉)“折磨的血泪史。你问它"今天北京天气怎么样”,它一本正经给你编个温度;你让它查某个技术文档的最新版本,它把三年前的信息当成宝。

更扎心的是,当你兴冲冲把LLM接入产品,发现它就像个"自信满满的糊涂蛋"——不管知不知道,都敢给你瞎编答案。用户投诉来了,老板脸色变了,你的KPI悬了。

但好消息是,ReAct(Reasoning + Acting)框架的出现,让LLM终于学会了"先想后做"。今天这篇,我就手把手带你实战:用LangChain搭建一个会思考、会搜索、会验证的ReAct Agent。读完这篇,你能做出一个"不确定就查,查完再答"的靠谱AI。


一、ReAct核心原理:为什么LLM需要"先想后做"

点题:ReAct到底是什么

ReAct是2022年Google提出的一个框架,核心思想很简单:把"推理(Reasoning)"和"行动(Acting)"交错进行

传统LLM的问答是"一口气说完"——输入问题,直接输出答案。ReAct则把它拆成循环:思考→行动→观察→再思考→再行动……直到问题解决。

用户提问

Thought
我需要搜索什么

Action
调用搜索工具

Observation
获取搜索结果

是否足够?

Final Answer
给出最终答案

看明白没?这就像你解决一个技术问题:先想想可能是哪里的bug(Thought),然后去查日志/搜Stack Overflow(Action),看到报错信息后(Observation),再分析下一步怎么做。

痛点分析:新手最容易犯的错

错误1:以为ReAct就是"套个模板"

我见过太多人,复制一段ReAct的prompt模板,就以为自己在用ReAct了。结果呢?LLM要么不按照格式输出,要么陷入死循环,一直在"思考"从不"行动"。

# 错误示范:简单粗暴的prompt拼接
prompt = f"""
用户问题:{question}
请你按照ReAct格式回答:
Thought: ...
Action: ...
Observation: ...
"""

# 结果:LLM根本不理解要调用工具,或者格式混乱
response = llm.predict(prompt)  # 输出完全不可控

错误2:工具描述写得太抽象

新手写工具描述,经常这样写:“search工具,用于搜索信息”。LLM看了懵圈:搜什么信息?什么时候该用?返回什么格式?

# 错误的工具定义
@tool
def search(query: str):
    """搜索工具"""  # 太简略!
    return web_search(query)

错误3:忽略Observation的真实性

有些同学做demo时,为了省事,Observation直接让LLM自己编。这完全违背了ReAct的初衷——Observation必须是外部真实信息,不是LLM的想象。

解决方案:正确理解ReAct的精髓

正确做法1:用LangChain的Structured Agent,别自己造轮子

LangChain已经封装好了ReAct的完整流程,包括输出解析、工具调用、循环控制。你只需要关注业务逻辑:

from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate

# 正确的做法:使用官方封装的ReAct模板
react_template = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}"""

prompt = PromptTemplate.from_template(react_template)
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

正确做法2:工具描述要"说人话",带场景示例

from langchain.tools import Tool

# 好的工具描述:说明用途、场景、输入输出
search_tool = Tool(
    name="web_search",
    func=web_search,
    description="""用于搜索互联网上的实时信息。
    当用户询问当前事件、最新数据、或你不确定的事实时,必须使用此工具。
    输入:搜索关键词(字符串)
    输出:搜索结果摘要(字符串)
    示例:
    - 用户问"Python 3.12新特性",输入:"Python 3.12 new features release"
    - 用户问"今天北京天气",输入:"北京天气 今天"
    """
)

正确做法3:严格区分LLM生成内容和外部真实数据

# 正确的ReAct循环中,Observation必须来自真实工具调用
def run_react_step(agent, question, intermediate_steps):
    # Thought: LLM生成
    thought = llm.predict(f"Question: {question}\nWhat should I do?")
    
    # Action: LLM决定调用什么工具
    action = parse_action(thought)
    
    # Observation: 必须是真实工具返回,不能是LLM编造!
    observation = real_tool_call(action)  # 关键!真实调用
    
    return observation

小结

ReAct的核心不是prompt模板,而是**"推理-行动"的循环机制**。用LangChain的封装,写好工具描述,保证Observation的真实性,你就迈出了成功的第一步。


二、环境搭建:从零配置LangChain Agent开发环境

点题:需要准备什么

搭建ReAct Agent环境,核心是四件套:LLM接入、工具集成、记忆管理、Agent编排。我们一步步来。

ReAct Agent环境架构

LLM层
OpenAI/Claude/本地模型

工具层
搜索/数据库/计算

记忆层
对话历史/上下文

Agent核心
ReAct编排

用户输入

外部系统
搜索引擎/API

痛点分析:环境配置的坑

痛点1:版本地狱

LangChain迭代快,0.1.x和0.2.x的API差异巨大。你照着一篇半年前的教程写,发现initialize_agent已经废弃了,ZeroShotAgent找不到了。

# 旧版本(已废弃)
from langchain.agents import initialize_agent, ZeroShotAgent  # 报错!

agent = initialize_agent(tools, llm, agent="zero-shot-react-description")  # 跑不通

痛点2:API密钥管理混乱

新手常把密钥硬编码,或者到处复制.env文件。一不小心的提交,密钥泄露,账单爆炸。

# 危险做法
openai_api_key = "sk-xxxxxxxxxxxxxxxx"  # 千万别这么干!

痛点3:国内网络环境

OpenAI官方API访问困难,Claude注册麻烦,很多新手卡在这一步就放弃了。

解决方案:稳健的环境配置

步骤1:锁定版本,用新不用旧

# 创建独立环境
conda create -n react-agent python=3.10
conda activate react-agent

# 安装指定版本(以0.2.x为例)
pip install langchain==0.2.0 langchain-openai==0.1.0 langchain-community==0.2.0

# 核心依赖
pip install openai duckduckgo-search  # 免费搜索工具

步骤2:安全的密钥管理

# config.py - 集中管理配置
import os
from dataclasses import dataclass

@dataclass
class Config:
    openai_api_key: str = os.getenv("OPENAI_API_KEY", "")
    openai_base_url: str = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    model_name: str = os.getenv("MODEL_NAME", "gpt-3.5-turbo")
    
    def validate(self):
        if not self.openai_api_key:
            raise ValueError("OPENAI_API_KEY not set!")

# 使用
from config import Config
cfg = Config()
cfg.validate()

步骤3:灵活的LLM接入方案

from langchain_openai import ChatOpenAI

# 方案A:官方API(有访问条件时)
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    api_key=cfg.openai_api_key,
    base_url=cfg.openai_base_url
)

# 方案B:国内代理/中转服务
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    api_key="your-proxy-key",
    base_url="https://your-proxy-domain.com/v1"  # 国内中转
)

# 方案C:本地模型(Ollama/LM Studio)
from langchain_community.llms import Ollama
llm = Ollama(model="llama2-13b")  # 完全离线,免费

步骤4:集成搜索工具

from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool

# DuckDuckGo - 免费,无需API Key
search = DuckDuckGoSearchRun()

# 包装成标准Tool
search_tool = Tool(
    name="duckduckgo_search",
    func=search.run,
    description="Search the internet for current information. Input should be a search query."
)

tools = [search_tool]

小结

环境配置是地基,版本锁定、安全密钥、灵活LLM接入,这三件事做好了,后面开发事半功倍。别在起跑线上摔跤。


三、基础实战:手写第一个ReAct搜索问答Agent

点题:从0到1跑通ReAct

现在,我们把前面准备的组件组装起来,做一个真正能"搜索-回答"的Agent。

搜索工具 LLM ReAct Agent 用户 搜索工具 LLM ReAct Agent 用户 loop [ReAct循环] "Python 3.12有什么新特性?" 生成Thought + Action Thought: 需要搜索最新信息 Action: duckduckgo_search Action Input: Python 3.12 new features 调用搜索工具 Observation: [搜索结果...] 基于Observation继续推理 Thought: 信息已足够 Final Answer: ... 完整回答

痛点分析:第一次运行就报错

痛点1:输出格式解析失败

LLM没有严格按照ReAct格式输出,Agent解析不了,直接崩溃。

# LLM输出(格式错误)
我觉得应该搜索一下,用duckduckgo_search工具,输入Python 3.12 features

# 期望的格式
Thought: I need to search for Python 3.12 features
Action: duckduckgo_search
Action Input: Python 3.12 new features

痛点2:工具调用陷入死循环

LLM一直在"搜索",搜了10轮还不肯给答案,或者反复用同样的关键词搜索。

痛点3:Final Answer质量差

即使搜索到了信息,LLM的总结要么遗漏重点,要么加入自己的"脑补"。

解决方案:完整的可运行代码

import os
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool

# ========== 1. 配置 ==========
os.environ["OPENAI_API_KEY"] = "your-api-key"

# ========== 2. 初始化LLM ==========
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,  # 关键!ReAct需要确定性输出
    max_tokens=2000
)

# ========== 3. 定义工具 ==========
search = DuckDuckGoSearchRun()

search_tool = Tool(
    name="web_search",
    func=search.run,
    description="""Useful for searching current information on the internet.
    Use this when you need to answer questions about current events, latest data, or anything you're not sure about.
    Input should be a search query string."""
)

tools = [search_tool]

# ========== 4. 创建ReAct Agent ==========
# 使用官方推荐的create_react_agent(LangChain 0.2.x)
from langchain import hub

# 从LangChain Hub获取优化过的ReAct prompt
prompt = hub.pull("hwchase17/react")

# 或者自定义prompt(更可控)
custom_prompt = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: always think about what to do, step by step
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Important rules:
1. If you don't know something, you MUST use web_search tool
2. Don't make up information, always verify with search
3. Keep your answers concise and accurate

Begin!

Question: {input}
Thought:{agent_scratchpad}"""

prompt = PromptTemplate.from_template(custom_prompt)

# 创建agent
agent = create_react_agent(llm, tools, prompt)

# ========== 5. 创建执行器(关键!控制循环) ==========
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # 打印执行过程,调试用
    max_iterations=5,       # 防止死循环!最多5轮
    max_execution_time=60,  # 超时60秒
    handle_parsing_errors=True  # 自动处理解析错误
)

# ========== 6. 运行 ==========
def ask(question: str):
    """封装调用,方便测试"""
    result = agent_executor.invoke({"input": question})
    return result["output"]

# 测试
if __name__ == "__main__":
    # 测试1:需要搜索的问题
    print("=" * 50)
    print("测试1:实时信息查询")
    response = ask("2024年诺贝尔物理学奖颁给了谁?")
    print(f"回答:{response}")
    
    # 测试2:知识+搜索结合
    print("\n" + "=" * 50)
    print("测试2:需要验证的知识")
    response = ask("Python的GIL是什么?Python 3.13有改进吗?")
    print(f"回答:{response}")

关键配置解读:

参数 作用 建议值
temperature=0 降低随机性,保证格式稳定 必须0
max_iterations 防止无限循环 3-5足够
max_execution_time 超时保护 30-60秒
handle_parsing_errors 格式错误时自动重试 建议开启
verbose=True 打印思考过程 开发时必开

运行效果示例

> 输入:2024年诺贝尔物理学奖颁给了谁?

[开始ReAct循环]

Thought: I need to search for the 2024 Nobel Prize in Physics winner.
Action: web_search
Action Input: 2024 Nobel Prize Physics winner

Observation: [搜索结果... "The Nobel Prize in Physics 2024 was awarded to John J. Hopfield and Geoffrey E. Hinton..."]

Thought: I have found the answer. The 2024 Nobel Prize in Physics was awarded to John J. Hopfield and Geoffrey E. Hinton for their work on machine learning with artificial neural networks.
Final Answer: 2024年诺贝尔物理学奖授予了约翰·霍普菲尔德(John J. Hopfield)和杰弗里·辛顿(Geoffrey E. Hinton),以表彰他们在使用人工神经网络进行机器学习方面的基础性发现和发明。

[结束,共2轮]

小结

一个能用的ReAct Agent = 确定性LLM(temperature=0)+ 清晰工具描述 + 循环控制(max_iterations)。先跑通这个基础版本,再谈优化。


四、进阶玩法:多工具协同与对话记忆

点题:让Agent更"聪明"

基础版本只能搜索,实际场景中我们需要:多个工具配合、记住对话历史、处理复杂任务。

多工具Agent

计算

搜索

代码

数据库

用户问题

工具选择

Calculator
数学计算

WebSearch
信息检索

PythonREPL
代码执行

SQLQuery
数据查询

结果整合

ConversationMemory
对话记忆

最终回答

痛点分析:工具多了反而更乱

痛点1:工具选择困难症

给Agent 5个工具,它不知道该用哪个,或者把简单问题搞复杂(明明能直接答,非要调用工具)。

痛点2:多轮对话"失忆"

用户问"刚才说的那个框架",Agent一脸懵:刚才?什么框架?

痛点3:工具结果冲突

搜索说A版本最新,数据库查出来是B版本,Agent不知道怎么处理矛盾信息。

解决方案:系统化设计

方案1:多工具配置与路由优化

from langchain.tools import Tool, BaseTool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_experimental.tools import PythonREPLTool
import math

# 工具1:搜索(已有)
search = DuckDuckGoSearchRun()
search_tool = Tool(
    name="web_search",
    func=search.run,
    description="Search for current information on the internet. Use for news, latest versions, current events."
)

# 工具2:计算器(精确数学计算)
class CalculatorTool(BaseTool):
    name: str = "calculator"
    description: str = "Useful for mathematical calculations. Input should be a valid Python math expression."
    
    def _run(self, expression: str) -> str:
        try:
            # 安全计算:只允许math模块
            safe_dict = {
                'sqrt': math.sqrt, 'pow': math.pow,
                'sin': math.sin, 'cos': math.cos,
                'pi': math.pi, 'e': math.e
            }
            result = eval(expression, {"__builtins__": {}}, safe_dict)
            return str(result)
        except Exception as e:
            return f"Error: {str(e)}"

calculator_tool = CalculatorTool()

# 工具3:Python代码执行(数据分析)
python_tool = Tool(
    name="python_executor",
    func=PythonREPLTool().run,
    description="Execute Python code for data analysis, file processing, or complex calculations. Use when calculator is not enough."
)

# 工具集合
tools = [search_tool, calculator_tool, python_tool]

方案2:添加对话记忆

from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory
from langchain.agents import AgentExecutor

# 方案A:完整记忆(适合短对话)
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="output"
)

# 方案B:滑动窗口(适合长对话,防止token爆炸)
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=5,  # 只保留最近5轮
    return_messages=True,
    output_key="output"
)

# 修改prompt,加入记忆占位符
prompt_with_memory = """Answer the following questions as best you can...

Previous conversation:
{chat_history}

Current question: {input}
Thought:{agent_scratchpad}"""

# 创建Agent时传入memory
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,  # 关键!
    verbose=True,
    max_iterations=5
)

方案3:完整的多工具+记忆示例

from langchain_core.prompts import MessagesPlaceholder

# 更优雅的prompt结构
prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful AI assistant with access to tools.
Follow the ReAct format: Thought -> Action -> Observation -> Final Answer.
Always verify uncertain information with web_search."""),
    MessagesPlaceholder(variable_name="chat_history"),  # 历史消息
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 中间步骤
])

# 创建agent
agent = create_openai_tools_agent(llm, tools, prompt)  # 使用OpenAI函数调用格式,更稳定

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 测试多轮对话
def chat():
    print("开始对话(输入'quit'退出):\n")
    while True:
        user_input = input("你:")
        if user_input.lower() == 'quit':
            break
        
        result = agent_executor.invoke({"input": user_input})
        print(f"AI:{result['output']}\n")

# 对话示例:
# 你:Python 3.12的性能提升了多少?
# AI:[搜索并回答...]
# 你:那和3.11相比呢?  ← 能记住上下文!
# AI:[基于记忆,直接比较或补充搜索]

方案4:工具结果冲突处理

# 在prompt中加入冲突解决指导
conflict_resolution_prompt = """
When you receive conflicting information from different sources:
1. Prefer more recent information (check dates)
2. Prefer authoritative sources (official docs > forums)
3. If still uncertain, use web_search to verify
4. In your Final Answer, briefly mention the discrepancy and your reasoning
"""

小结

多工具协同的关键是清晰的工具描述让LLM选对工具,对话记忆让交互更自然,冲突处理策略让答案更可信。这三板斧下去,Agent的实用度提升一个档次。


五、生产优化:错误处理、调试与性能监控

点题:从Demo到生产

实验室里跑通的代码,放到生产环境可能各种问题:超时、格式错误、API限流、成本失控。

生产优化体系

错误处理层

重试机制

降级策略

熔断保护

调试观测层

结构化日志

调用链追踪

沙箱测试

成本控制层

Token监控

结果缓存

模型路由
复杂任务GPT-4
简单任务GPT-3.5

痛点分析:生产环境的"惊喜"

痛点1:LLM输出不稳定

同样的输入,有时格式对,有时不对;有时3轮出答案,有时5轮还搞不定。

痛点2:API调用成本高

用户问个"你好",Agent也要走完整ReAct流程,调好几次LLM,账单蹭蹭涨。

痛点3:问题难定位

用户投诉"答案错了",你翻日志只看到最终输出,不知道中间哪步出了问题。

解决方案:工程化实践

方案1:多层错误处理

from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.exceptions import OutputParserException

class RobustAgentExecutor:
    def __init__(self, agent_executor):
        self.agent_executor = agent_executor
        self.fallback_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        retry=lambda e: isinstance(e, (TimeoutError, OutputParserException))
    )
    def invoke_with_retry(self, inputs):
        """带重试的调用"""
        return self.agent_executor.invoke(inputs)
    
    def invoke_with_fallback(self, inputs):
        """主流程失败时的降级策略"""
        try:
            return self.invoke_with_retry(inputs)
        except Exception as e:
            # 记录错误
            logger.error(f"Agent failed: {e}")
            
            # 降级:直接用LLM回答,不走ReAct
            simple_prompt = f"Answer this question directly: {inputs['input']}"
            return {
                "output": self.fallback_llm.predict(simple_prompt),
                "fallback_used": True
            }
    
    def invoke_with_circuit_breaker(self, inputs):
        """熔断保护:连续失败时直接降级"""
        if self.failure_count > 5:
            return self.invoke_with_fallback(inputs)
        
        try:
            result = self.invoke_with_retry(inputs)
            self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            raise

方案2:结构化日志与追踪

import json
import time
from datetime import datetime

class AgentTracer:
    def __init__(self):
        self.traces = []
    
    def trace_step(self, step_num, thought, action, observation, latency):
        """记录每一步的详细信息"""
        trace = {
            "timestamp": datetime.now().isoformat(),
            "step": step_num,
            "thought": thought,
            "action": action,
            "observation_preview": str(observation)[:200],  # 截断避免过大
            "latency_ms": latency * 1000
        }
        self.traces.append(trace)
        return trace
    
    def finalize(self, final_answer, total_tokens, success):
        """完成一次调用的记录"""
        return {
            "trace_id": generate_trace_id(),
            "start_time": self.traces[0]["timestamp"] if self.traces else None,
            "steps": len(self.traces),
            "total_tokens": total_tokens,
            "success": success,
            "final_answer_preview": str(final_answer)[:100],
            "step_details": self.traces
        }

# 使用
tracer = AgentTracer()
start = time.time()

# ... 执行agent步骤 ...
tracer.trace_step(1, thought, action, observation, time.time() - start)

# 最终记录
log_entry = tracer.finalize(answer, token_usage, success=True)
logger.info(json.dumps(log_entry, ensure_ascii=False))

方案3:成本优化策略

class CostOptimizedAgent:
    def __init__(self):
        self.cache = {}  # 简单缓存,生产用Redis
        self.smart_router = {
            "simple": ChatOpenAI(model="gpt-3.5-turbo", temperature=0),
            "complex": ChatOpenAI(model="gpt-4", temperature=0)
        }
    
    def should_use_agent(self, question: str) -> bool:
        """判断是否需要走ReAct流程"""
        # 简单问题直接回答
        simple_patterns = ["你好", "谢谢", "再见", "什么是", "介绍一下"]
        return not any(p in question for p in simple_patterns)
    
    def route_by_complexity(self, question: str) -> str:
        """根据问题复杂度选择模型"""
        # 简单启发式:长度、关键词
        if len(question) < 20 and "?" not in question:
            return "simple"
        if any(kw in question for kw in ["计算", "分析", "比较", "为什么"]):
            return "complex"
        return "simple"
    
    def invoke(self, question: str):
        # 1. 检查缓存
        cache_key = hash(question)
        if cache_key in self.cache:
            return {"output": self.cache[cache_key], "cached": True}
        
        # 2. 判断是否需要Agent
        if not self.should_use_agent(question):
            llm = self.smart_router["simple"]
            answer = llm.predict(f"简短回答:{question}")
            return {"output": answer, "direct": True}
        
        # 3. 选择模型并执行
        model_tier = self.route_by_complexity(question)
        # ... 执行agent ...
        
        # 4. 更新缓存
        self.cache[cache_key] = result["output"]
        return result

小结

生产环境的关键词是容错、可观测、成本控制。重试+降级保证可用性,结构化日志保证可调试,智能路由+缓存控制成本。这三件事做好了,才能放心上线。


六、场景落地:企业级搜索问答系统实战

点题:解决真实业务问题

把ReAct Agent应用到企业场景,常见的是:内部知识库问答、产品文档助手、客服智能回复。

企业搜索问答系统

员工/客户

API网关

权限验证

请求路由

ReAct Agent

工具集群

企业搜索
ES/OpenSearch

业务数据库
MySQL/PostgreSQL

文档系统
Confluence/Notion

内部API
HR/财务/CRM

Redis缓存

日志审计

格式化回答

痛点分析:企业场景的特有问题

痛点1:权限与数据隔离

不同员工能查的信息不同,Agent不能"越权"回答。

痛点2:回答需要可溯源

企业场景不能说"我觉得",要说"根据XX文档第3章"。

痛点3:与现有系统集成

不能推倒重来,要对接现有的搜索、数据库、权限系统。

解决方案:企业级架构设计

方案1:带权限的工具调用

class SecureToolWrapper:
    """带权限检查的工具包装器"""
    
    def __init__(self, tool, required_roles):
        self.tool = tool
        self.required_roles = required_roles
    
    def run(self, query: str, user_context: dict):
        # 检查用户权限
        user_roles = user_context.get("roles", [])
        if not any(r in self.required_roles for r in user_roles):
            return "Error: Insufficient permissions to access this resource."
        
        # 记录审计日志
        audit_log.info(f"User {user_context['user_id']} accessed {self.tool.name}")
        
        # 执行工具
        return self.tool.run(query)

# 使用
hr_search = SecureToolWrapper(
    tool=hr_database_tool,
    required_roles=["HR", "MANAGER"]
)

# Agent执行时传入用户上下文
result = agent_executor.invoke({
    "input": question,
    "user_context": {"user_id": "u123", "roles": ["ENGINEER"]}  # 无HR权限
})
# 返回:Error: Insufficient permissions...

方案2:溯源与引用

class CitationTool:
    """带引用信息的搜索工具"""
    
    def search_with_citation(self, query: str):
        results = self.internal_search(query)
        
        # 格式化带来源的结果
        formatted = []
        for r in results:
            formatted.append({
                "content": r["content"],
                "source": r["document_name"],
                "page": r.get("page", "N/A"),
                "url": r.get("url", ""),
                "last_updated": r["timestamp"]
            })
        
        return {
            "answer_snippets": formatted,
            "citation_format": "Source: {source}, Page {page}"
        }

# 在Final Answer中强制要求引用
citation_prompt = """
When providing your Final Answer:
1. Base your answer strictly on the Observation
2. Include citations like [Source: Employee Handbook, Page 12]
3. If information is incomplete, say so explicitly
"""

方案3:完整的企业级Agent服务

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class QuestionRequest(BaseModel):
    question: str
    session_id: str
    user_id: str

class AgentService:
    def __init__(self):
        self.agent_executor = self._init_agent()
        self.session_manager = SessionManager()  # 管理多轮对话
    
    async def answer(self, request: QuestionRequest):
        # 1. 获取用户上下文
        user_ctx = await get_user_context(request.user_id)
        
        # 2. 获取/创建会话记忆
        memory = self.session_manager.get_memory(request.session_id)
        
        # 3. 执行Agent(带超时和熔断)
        try:
            result = await asyncio.wait_for(
                self._execute_agent(request.question, memory, user_ctx),
                timeout=30.0
            )
        except asyncio.TimeoutError:
            return {"answer": "请求超时,请简化问题重试", "status": "timeout"}
        
        # 4. 保存会话状态
        self.session_manager.save_memory(request.session_id, memory)
        
        # 5. 返回结构化结果
        return {
            "answer": result["output"],
            "citations": result.get("citations", []),
            "tools_used": result.get("intermediate_steps", []),
            "confidence": self._calculate_confidence(result)
        }

@app.post("/api/ask")
async def ask_question(request: QuestionRequest):
    service = AgentService()
    return await service.answer(request)

小结

企业级应用的核心是安全、可控、可集成。权限隔离保安全,溯源引用保可信,标准接口保集成。这三点做到,ReAct Agent才能真正落地产生价值。


七、避坑指南:新手最常踩的8个坑

点题:前人的教训,你的财富

我整理了8个最常见的坑,都是血泪换来的。

25% 20% 15% 15% 10% 10% 3% 2% ReAct Agent常见错误分布 Prompt格式问题 工具描述不清 温度参数错误 循环控制缺失 Observation伪造 内存/Token爆炸 错误处理缺失 版本兼容问题

8个坑与填坑方案

坑号 症状 原因 解决方案
1 OutputParserException频繁报错 LLM输出格式不对 temperature=0 + handle_parsing_errors=True
2 工具永远不被调用 工具描述太抽象 加使用场景、输入示例、返回值说明
3 陷入死循环,不停搜索 没有终止条件 设置max_iterationsmax_execution_time
4 答案还是瞎编 Observation让LLM自己生成 必须真实调用工具,禁止伪造
5 长对话后报错/变慢 历史消息太多,token超限 ConversationBufferWindowMemory限制轮数
6 简单问题也走完整流程 没有判断机制 should_use_agent判断,简单问题直接答
7 升级LangChain后代码全挂 API变动大 锁定版本,关注官方迁移指南
8 生产环境偶发卡死 缺少超时和熔断 asyncio.wait_for和断路器模式

快速检查清单

# 上线前逐项检查
CHECKLIST = {
    "temperature": 0,  # 必须是0
    "max_iterations": 3-5,  # 不能无限
    "handle_parsing_errors": True,  # 必须开启
    "tool_descriptions": "有场景示例",  # 不能少于50字
    "memory_window": 5,  # 长对话必设
    "timeout_seconds": 30,  # 必须有
    "fallback_strategy": "已定义",  # 降级方案
    "logging_level": "INFO",  # 生产需要
}

写在最后

走到这里,你已经掌握了ReAct Agent从原理到生产的完整路径。

回想一下,我们解决了什么问题?让LLM不再"瞎编",学会"先想后做";从单工具到多工具协同,从单轮到多轮记忆;从本地Demo到企业级部署。这每一步,都是把"玩具"变成"工具"的关键跳跃。

我知道,学习Agent开发不容易。你要理解LLM的脾性,要调试各种奇怪的输出,要在成本和效果之间找平衡。但请相信,这些投入都是值得的——能自主决策、调用工具、完成任务的AI,正是未来应用的主流形态

编程之路不易,但每一步成长都算数。今天你啃下的ReAct原理,写下的每一行Agent代码,都在为明天的竞争力筑基。保持好奇,持续迭代,你也能做出让人眼前一亮的智能应用。

最后送大家一句话:好的Agent不是一次性设计出来的,是不断观察它的"思考过程",逐步调教出来的。多打verbose=True,多看它的Thought,你会越来越懂怎么和它配合。

我们下篇见!


关注私信备注:“资料代找获取”,全网计算机学习资料代找:例如:
《课程:2026 年多模态大模型实战训练营》
《课程:AI 大模型工程师系统课程 (22 章完整版 持续更新)》
《课程:AI 大模型系统实战课第四期 (2026 年开课 持续更新)》
《课程:2026 年 AGI 大模型系统课 23 期》
《课程:2026 年 AGI 大模型系统课 21 期》
《课程:AI 大模型实战课 8 期 (2026 年 2 月最新完结版)》
《课程:AI 大模型系统实战课三期》
《课程:AI 大模型系统课程 (2026 年 2 月开课 持续更新)》
《课程:AI 大模型全阶课程 (2025 年 12 月开课 2026 年 6 月结课)》
《课程:AI 大模型工程师全阶课程 (2025 年 10 月开课 2026 年 4 月结课)》
《课程:2026 年最新大模型 Agent 开发系统课 (持续更新)》
《课程:LLM 多模态视觉大模型系统课》
《课程:大模型 AI 应用开发企业级项目实战课 (2026 年 1 月开课)》
《课程:大模型智能体线上速成班 V2.0》
《课程:Java+AI 大模型智能应用开发全阶课》
《课程:Python+AI 大模型实战视频教程》
《书籍:软件工程 3.0: 大模型驱动的研发新范式.pdf》
《课程:人工智能大模型系统课 (2026 年 1 月底完结版)》
《课程:AI 大模型零基础到商业实战全栈课第五期》
《课程:Vue3.5+Electron + 大模型跨平台 AI 桌面聊天应用实战 (2025)》
《课程:AI 大模型实战训练营 从入门到实战轻松上手》
《课程:2026 年 AI 大模型 RAG 与 Agent 智能体项目实战开发课》
《课程:大模型训练营配套补充资料》

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐