LangChain Agent 开发实战：Tool、AgentExecutor 与 ReAct 模式深度解析

导读： 如果说 Chain 是让 LLM 按固定流程"说"，那 Agent 就是让 LLM 自主决策"做"。本文将深入 LangChain Agent 的核心机制——从 Tool 的封装与设计，到 AgentExecutor 的执行原理，再到 ReAct 推理模式的工作流程。读完本文，你将能够构建出真正"智能"的 LLM 应用。

---

目录

1. Agent 是什么？为什么需要它？
2. Agent 的核心三要素
3. Tool：赋予 LLM 超能力
4. AgentExecutor：Agent 的发动机
5. ReAct Agent：推理与行动的交响曲
6. 实战一：构建智能数据分析助手
7. 实战二：构建多工具协作的 DevOps Agent
8. Agent 开发的进阶技巧与避坑指南
9. 总结与展望

---

一、Agent 是什么？为什么需要它？

1.1 Chain vs Agent

在上一篇 LangChain 快速入门中，我们学习了 Chain——它将 LLM 的调用流程固定下来：

输入 → Prompt → LLM → 输出

这种方式适用于流程确定的场景。但现实中，很多问题是开放式的：

• “帮我查一下上海明天的天气，如果下雨就提醒我带伞”——需要先调用天气 API，再判断
• “这个月的销售额比上个月增长了多少？”——需要查询数据库，然后计算
• “帮我把这份 PDF 翻译成英文并发送给张三”——需要读文件→翻译→查通讯录→发邮件

这些场景的共同特点是：你不知道 LLM 需要做什么、做几次、以什么顺序做——这就是 Agent 的用武之地。

1.2 Agent 的定义

Agent（智能体）= LLM（大脑）+ Tools（手脚）+ 推理循环（思考过程）

1.3 Agent 的工作流程

Agent 运行的核心是一个**"思考-行动-观察"循环**（Thought-Action-Observation Loop）：

Step 1 ─ Thought:     "我需要知道上海明天的天气"
         Action:      调用天气查询工具(城市="上海", 日期="明天")
         Observation: "上海明天：中雨，15-20℃"

Step 2 ─ Thought:     "明天下雨，用户需要带伞"
         Action:       不需要再调用工具了
         Final Answer: "上海明天有中雨，气温15-20℃，建议您带伞出门！"

这个循环会持续进行，直到 LLM 认为已经拿到了足够的信息来回答用户的问题。

---

二、Agent 的核心三要素

在深入代码之前，先理解 LangChain Agent 的三大核心概念：

2.1 Tool（工具）

Tool 是 Agent 的"手脚"——让 LLM 能够与外部世界交互。每个 Tool 包含：

• name：工具的唯一标识，LLM 通过名字来决定调用哪个工具
• description：工具的功能描述，LLM 根据描述来判断何时使用
• func / coroutine：工具的实际执行函数
• args_schema：工具的参数定义（JSON Schema），告诉 LLM 如何传参

2.2 AgentExecutor（执行器）

AgentExecutor 是 Agent 的"发动机"——它负责驱动那个核心循环：

1. 将用户输入 + 可用工具列表 + 历史记录发给 LLM
2. 解析 LLM 的响应（是调用工具？还是给出最终答案？）
3. 如果需要调用工具，执行工具并将结果反馈给 LLM
4. 重复上述过程，直到 LLM 给出最终答案或达到最大迭代次数

2.3 Agent 类型

LangChain 支持多种 Agent 类型，最常用的是：

Agent 类型	推理模式	适用场景
ReAct	Thought→Action→Observation 循环	通用场景，最常用
OpenAI Functions	基于 Function Calling API	使用 OpenAI 模型时
Structured Chat	支持多参数复杂工具调用	工具参数复杂时
Self-Ask with Search	自问自答 + 搜索	需要逐步推理的问答

本文重点讲解最经典的 ReAct Agent。

---

三、Tool：赋予 LLM 超能力

Tool 是 Agent 系统中最关键的组件。一个好的 Tool 定义，直接决定了 Agent 是否能正确地使用它。

3.1 最简单的 Tool：@tool 装饰器

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """将两个整数相加并返回结果。"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """将两个整数相乘并返回结果。"""
    return a * b

# 查看 Tool 自动生成的 schema
print(add.name)         # add
print(add.description)  # 将两个整数相加并返回结果。
print(add.args)         # {'a': {'title': 'A', 'type': 'integer'}, 'b': {...}}
print(add.args_schema.schema())  # JSON Schema

💡 关键提示： Tool 的 函数名 和 docstring 极其重要！LLM 就是根据这两个信息来决定何时调用哪个工具、如何传参的。docstring 要清晰、具体、包含参数说明。

3.2 结构化 Tool：使用 BaseTool

当 Tool 逻辑复杂时，可以继承 BaseTool 显式控制每个部分：

from langchain_core.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from pydantic import BaseModel, Field
from typing import Optional, Type


# 第一步：定义参数 Schema
class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，例如：北京、上海、深圳")
    date: Optional[str] = Field(
        default="今天",
        description="查询日期，格式为YYYY-MM-DD，默认为今天"
    )


# 第二步：实现 Tool 类
class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = """
    查询指定城市在指定日期的天气情况。
    返回信息包括：天气状况、温度范围、湿度、风力。
    使用场景：当用户询问某地天气时需要调用此工具。
    """
    args_schema: Type[BaseModel] = WeatherInput
    
    def _run(
        self,
        city: str,
        date: Optional[str] = "今天",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """
        实际的工具执行逻辑。
        生产环境中，这里会调用真实的天气 API。
        """
        # 模拟天气数据（实际项目中替换为 API 调用）
        import random
        conditions = ["晴", "多云", "阴", "小雨", "中雨", "雷阵雨"]
        temps = range(10, 36)
        
        condition = random.choice(conditions)
        temp_low = random.choice(temps)
        temp_high = temp_low + random.randint(2, 10)
        humidity = random.randint(30, 95)
        
        return (
            f"城市：{city}\n"
            f"日期：{date}\n"
            f"天气：{condition}\n"
            f"温度：{temp_low}℃ ~ {temp_high}℃\n"
            f"湿度：{humidity}%\n"
            f"风力：{random.choice(['1-2级', '2-3级', '3-4级', '4-5级'])}"
        )


# 测试 Tool
weather = WeatherTool()
result = weather.invoke({"city": "上海", "date": "2024-12-20"})
print(result)

3.3 带副作用的 Tool：文件操作示例

Tool 不仅限于查询，还可以执行有副作用的操作（写文件、发邮件、操作数据库等）：

from langchain_core.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import os

class FileWriteInput(BaseModel):
    filepath: str = Field(description="文件路径，例如：/tmp/output.txt")
    content: str = Field(description="要写入文件的内容")
    mode: Optional[str] = Field(
        default="w",
        description="写入模式：'w'覆盖写入，'a'追加写入"
    )

class FileWriteTool(BaseTool):
    name: str = "write_file"
    description: str = """
    将内容写入指定文件。
    如果文件不存在则创建，如果存在则根据 mode 参数决定覆盖或追加。
    使用场景：当需要保存生成的代码、报告或任何文本内容到文件时。
    """
    args_schema: Type[BaseModel] = FileWriteInput
    
    def _run(self, filepath: str, content: str, mode: str = "w") -> str:
        # 安全检查：确保路径合法
        allowed_dir = os.path.join(os.getcwd(), "agent_output")
        os.makedirs(allowed_dir, exist_ok=True)
        
        full_path = os.path.join(allowed_dir, os.path.basename(filepath))
        
        with open(full_path, mode, encoding="utf-8") as f:
            f.write(content)
        
        file_size = os.path.getsize(full_path)
        return f"成功写入文件：{full_path}（{file_size} 字节）"


class FileReadInput(BaseModel):
    filepath: str = Field(description="要读取的文件路径")

class FileReadTool(BaseTool):
    name: str = "read_file"
    description: str = """
    读取指定文件的内容并返回。
    使用场景：当需要查看之前保存的文件内容时。
    """
    args_schema: Type[BaseModel] = FileReadInput
    
    def _run(self, filepath: str) -> str:
        allowed_dir = os.path.join(os.getcwd(), "agent_output")
        full_path = os.path.join(allowed_dir, os.path.basename(filepath))
        
        if not os.path.exists(full_path):
            return f"错误：文件 {full_path} 不存在。"
        
        with open(full_path, "r", encoding="utf-8") as f:
            content = f.read()
        
        return f"文件 {filepath} 的内容（{len(content)} 字符）：\n\n{content}"

3.4 异步 Tool

对于网络请求、数据库查询等 IO 密集型操作，应使用异步 Tool 来避免阻塞：

from langchain_core.tools import tool
import asyncio

@tool
async def search_documents(query: str, top_k: int = 5) -> str:
    """在文档库中搜索与 query 最相关的内容。返回 top_k 条结果。"""
    # 模拟异步搜索（实际项目中可能是向量数据库查询）
    await asyncio.sleep(0.5)  # 模拟网络延迟
    return f"搜索「{query}」的结果（共 {top_k} 条）：\n" + "\n".join(
        f"{i+1}. 关于「{query}」的文档片段 {i+1}..."
        for i in range(top_k)
    )

# 异步调用（需要在 async 函数中运行）
async def main():
    result = await search_documents.ainvoke({"query": "LangChain Agent"})
    print(result)

asyncio.run(main())

3.5 Tool 设计的最佳实践

⚠️ Tool 设计的黄金法则：

1. 描述要精确：docstring 要说明"做什么"、“什么时候用”、“参数含义”。LLM 完全依赖这段文本来决策。
2. 参数要合理：使用 Pydantic Model 定义 args_schema，利用 Field(description=...) 详细说明每个参数
3. 返回要结构化：返回的字符串要清晰、信息密度高，便于 LLM 理解和后续处理
4. 职责要单一：一个 Tool 只做一件事，复杂的操作拆成多个 Tool
5. 错误要友好：捕获异常并返回有意义的错误信息，而不是让 Agent 看到 traceback
6. 要加安全边界：文件操作限制目录，网络请求限制域名，防止 Agent 越权

---

四、AgentExecutor：Agent 的发动机

4.1 AgentExecutor 的工作原理

AgentExecutor 是驱动 Agent 运行的核心组件。它实际上就是一个精心设计的 while 循环：

# AgentExecutor 的简化版核心逻辑
def agent_executor_simplified(agent, tools, user_input, max_iterations=15):
    """
    AgentExecutor 核心逻辑的精简教学版。
    实际源码更复杂，包含了错误处理、超时、回调等机制。
    """
    # 中间步骤记录
    intermediate_steps = []
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # 步骤1：调用 Agent（LLM），传入当前状态
        # LLM 返回两种可能：
        #   a) AgentAction：需要调用某个工具
        #   b) AgentFinish：已经可以给出最终答案
        output = agent.plan(
            intermediate_steps=intermediate_steps,
            input=user_input,
        )
        
        # 步骤2：判断输出类型
        if isinstance(output, AgentFinish):
            # LLM 认为已经可以回答了
            return {
                "output": output.return_values["output"],
                "intermediate_steps": intermediate_steps,
            }
        
        # 步骤3：是 AgentAction —— 执行工具
        # output.tool：要调用的工具名
        # output.tool_input：传入工具的参数
        tool = tools[output.tool]  # 根据名字找到工具
        observation = tool.run(output.tool_input)
        
        # 步骤4：记录这一步（工具调用 + 观察结果）
        intermediate_steps.append((output, observation))
        
        # 步骤5：循环继续，LLM 会看到之前的思考、行动和观察结果
    
    # 达到最大迭代次数仍未结束
    # AgentExecutor 会强制让 LLM 基于现有信息给出答案
    raise Exception(f"Agent 在 {max_iterations} 步后仍未完成任务")

4.2 理解 intermediate_steps

intermediate_steps 是 Agent 的"记忆链条"。它是一个列表，每个元素是一个 (AgentAction, observation) 元组：

intermediate_steps = [
    (
        AgentAction(tool="search", tool_input="Python 协程原理", log="Thought: 我需要先搜索..."),
        "Python 协程（Coroutine）是..."  # ← 搜索返回的结果
    ),
    (
        AgentAction(tool="calculator", tool_input="3 * 5", log="Thought: ..."),
        "15"  # ← 计算结果
    ),
]

这些中间步骤会在每次调用 LLM 时被格式化并添加到 Prompt 中，让 LLM "记住"自己做了什么、结果是什么——这正是 Agent 能够进行多步推理的关键。

4.3 完整的 AgentExecutor 配置

from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.chat_models import ChatTongyi
from langchain_core.tools import tool
import os

# ========== 1. 定义工具 ==========
@tool
def search(query: str) -> str:
    """在互联网上搜索信息。参数 query 为搜索关键词。"""
    # 模拟搜索
    return f"搜索「{query}」的结果：LangChain 是一个开源的 LLM 应用开发框架..."

@tool
def calculator(expression: str) -> str:
    """执行数学计算。参数 expression 为数学表达式，例如 '2+3*4'。"""
    try:
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算出错：{str(e)}"

tools = [search, calculator]

# ========== 2. 初始化 LLM ==========
llm = ChatTongyi(
    model="qwen-plus",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0.0,  # Agent 场景建议用低温，减少幻觉
)

# ========== 3. 创建 Agent ==========
agent = create_react_agent(llm, tools)

# ========== 4. 配置 AgentExecutor ==========
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,                # 打印详细执行过程
    max_iterations=10,           # 最多执行10轮
    max_execution_time=60,       # 最长执行60秒
    handle_parsing_errors=True,  # 自动处理格式解析错误
    early_stopping_method="generate",  # 超时后的处理方式
    return_intermediate_steps=False,   # 是否返回中间步骤
)

# ========== 5. 执行 ==========
result = agent_executor.invoke({
    "input": "请帮我计算 (156 * 23) / 12 的结果，然后搜索一下这个数字在计算机科学中有什么特殊含义。"
})

print(f"\n{'='*50}")
print(f"最终答案：{result['output']}")

4.4 AgentExecutor 的关键配置项

配置项	默认值	说明
max_iterations	15	Agent 最大推理步数，防止无限循环
max_execution_time	None	最大执行时间（秒），超时后触发 early_stopping_method
early_stopping_method	“force”	"force"=直接返回最后的输出；"generate"=让 LLM 基于已有信息生成最终答案
handle_parsing_errors	False	是否自动处理 LLM 输出的格式解析错误（建议开启）
verbose	False	是否打印完整的思考→行动→观察过程（调试时强烈建议开启）
return_intermediate_steps	False	是否在返回结果中包含中间步骤详情

---

五、ReAct Agent：推理与行动的交响曲

5.1 ReAct 是什么？

ReAct（Reasoning + Acting） 是 2022 年由 Google DeepMind 等机构提出的一种 prompting 模式。它的核心思想是：

让 LLM 交替进行"推理（Reasoning）“和"行动（Acting）”，两者相互增强。

传统 Chain-of-Thought（思维链）：
  Q: 小明有5个苹果，买了3个，吃了2个，还剩几个？
  A: 小明一开始有5个。买了3个变成8个。吃了2个变成6个。所以还剩6个苹果。

ReAct:
  Q: 2024年法国网球公开赛的冠军是谁？
  Thought: 我需要知道2024年法网的冠军。让我搜索一下。
  Action: search("2024 French Open winner")
  Observation: 2024年法国网球公开赛男子单打冠军是Carlos Alcaraz...
  Thought: 搜索结果告诉我冠军是Carlos Alcaraz。我可以回答用户了。
  Final Answer: 2024年法网男子单打冠军是卡洛斯·阿尔卡拉斯（Carlos Alcaraz）。

ReAct 的优势在于：

• 可追溯：每一步思考和行动都有记录
• 可纠错：如果工具调用出错，LLM 可以调整策略
• 可组合：可以组合多个工具完成复杂任务

5.2 LangChain 中 ReAct 的实现

LangChain 使用特定的 Prompt 模板来实现 ReAct。让我们看看它内部是怎么工作的：

# LangChain 的 ReAct Prompt 模板（简化版）
REACT_PROMPT_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}
"""

关键解析：

• {tools}：工具列表的格式化描述（名称 + 功能说明）
• {tool_names}：工具名称列表，LLM 必须从中选择
• {input}：用户的问题
• {agent_scratchpad}：前面步骤的 Thought→Action→Observation 记录，让 LLM 看到历史

5.3 手动实现一个最小 ReAct Agent

为了让你真正理解 ReAct 的机制，下面我们不依赖 LangChain 的 create_react_agent，手动实现一个精简版的 ReAct Agent：

import re
import os
from langchain_community.chat_models import ChatTongyi
from langchain_core.tools import tool


# ==================== 第一步：定义工具 ====================
@tool
def web_search(query: str) -> str:
    """在互联网上搜索信息。查询关键词为 query。"""
    # 模拟搜索（实际应调用搜索 API）
    mock_db = {
        "python": "Python 是一种解释型、面向对象的高级编程语言，由 Guido van Rossum 于 1991 年发布。",
        "langchain": "LangChain 是一个用于构建 LLM 应用的框架，支持 Chain、Agent、RAG 等模式。",
        "agent": "在 AI 领域，Agent（智能体）指能够自主感知环境并采取行动以实现目标的系统。",
    }
    for key, value in mock_db.items():
        if key in query.lower():
            return value
    return f"未找到关于「{query}」的相关信息。"


@tool
def calculator(expression: str) -> str:
    """计算数学表达式。expression 为数学表达式字符串，如 '3*4+2'。"""
    try:
        # 安全地计算（仅允许数字和基本运算符）
        allowed = set("0123456789+-*/().% ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含不允许的字符"
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"


# ==================== 第二步：构建 ReAct Prompt ====================
def build_react_prompt(tools, question: str, scratchpad: str = "") -> str:
    """构建 ReAct 格式的 Prompt"""
    
    # 生成工具描述
    tool_descriptions = "\n".join(
        f"- {t.name}: {t.description}" for t in tools
    )
    tool_names = ", ".join(t.name for t in tools)
    
    prompt = f"""你是一个智能助手，可以使用以下工具来回答问题。

可用工具：
{tool_descriptions}

回答格式（严格遵守）：
Question: 用户的问题
Thought: 你的思考（应该先做什么？需要用什么工具？）
Action: 要使用的工具名称（可选值：{tool_names}）
Action Input: 传递给工具的参数
Observation: 工具返回的结果
...（Thought → Action → Action Input → Observation 可以重复）
Thought: 我现在知道最终答案了
Final Answer: 用中文回答用户的最终答案

现在开始！

Question: {question}
{scratchpad}"""
    
    return prompt


# ==================== 第三步：解析 LLM 输出 ====================
def parse_react_output(text: str) -> dict:
    """
    解析 LLM 的输出，判断是继续行动还是给出最终答案。
    
    Returns:
        {"type": "action", "tool": "...", "tool_input": "..."}
        {"type": "finish", "answer": "..."}
    """
    # 检查是否是最终答案
    final_match = re.search(r"Final Answer:\s*(.*?)$", text, re.DOTALL | re.IGNORECASE)
    if final_match:
        return {"type": "finish", "answer": final_match.group(1).strip()}
    
    # 检查是否是行动指令
    action_match = re.search(r"Action:\s*(.*?)\n", text, re.IGNORECASE)
    action_input_match = re.search(r"Action Input:\s*(.*?)(?:\n|$)", text, re.IGNORECASE)
    
    if action_match and action_input_match:
        return {
            "type": "action",
            "tool": action_match.group(1).strip(),
            "tool_input": action_input_match.group(1).strip(),
        }
    
    # 无法解析
    return {"type": "unknown", "raw": text}


# ==================== 第四步：ReAct 主循环 ====================
class SimpleReActAgent:
    """手动实现的 ReAct Agent，展示核心机制"""
    
    def __init__(self, llm, tools, max_steps=10, verbose=True):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps
        self.verbose = verbose
    
    def run(self, question: str) -> str:
        scratchpad = ""  # 累积的 Thought→Action→Observation 历史
        steps_taken = []
        
        for step in range(self.max_steps):
            if self.verbose:
                print(f"\n{'='*50}")
                print(f"🔄 第 {step + 1} 轮推理")
                print(f"{'='*50}")
            
            # 1. 构建 Prompt 并调用 LLM
            prompt = build_react_prompt(
                list(self.tools.values()), question, scratchpad
            )
            response = self.llm.invoke(prompt)
            
            # ChatModel 返回 AIMessage，取 content
            if hasattr(response, 'content'):
                response_text = response.content
            else:
                response_text = str(response)
            
            if self.verbose:
                print(f"💭 LLM 输出：\n{response_text}")
            
            # 2. 解析输出
            parsed = parse_react_output(response_text)
            
            if parsed["type"] == "finish":
                if self.verbose:
                    print(f"\n✅ Agent 完成！共 {step + 1} 步")
                return parsed["answer"]
            
            elif parsed["type"] == "action":
                tool_name = parsed["tool"]
                tool_input = parsed["tool_input"]
                
                # 3. 执行工具
                if tool_name not in self.tools:
                    observation = f"错误：没有名为 '{tool_name}' 的工具。可用工具：{list(self.tools.keys())}"
                else:
                    try:
                        tool = self.tools[tool_name]
                        observation = tool.invoke(tool_input)
                    except Exception as e:
                        observation = f"工具执行出错：{str(e)}"
                
                if self.verbose:
                    print(f"🔧 Action: {tool_name}({tool_input})")
                    print(f"👁️ Observation: {observation}")
                
                # 4. 将本轮的结果追加到 scratchpad
                scratchpad += f"\nThought: {response_text}\nObservation: {observation}\n"
                steps_taken.append({
                    "step": step + 1,
                    "tool": tool_name,
                    "tool_input": tool_input,
                    "observation": observation,
                })
            
            else:
                # 解析失败，将原始输出作为 "Thought" 继续
                scratchpad += f"\nThought: {response_text}\n"
                if self.verbose:
                    print("⚠️ 解析 LLM 输出失败，继续尝试...")
        
        # 超过最大步数，强制让 LLM 总结
        if self.verbose:
            print(f"\n⚠️ 达到最大步数 {self.max_steps}，强制生成最终答案...")
        
        final_prompt = build_react_prompt(
            list(self.tools.values()), question, scratchpad
        ) + "\nThought: 我已经收集了足够的信息。\nFinal Answer:"
        
        final_response = self.llm.invoke(final_prompt)
        if hasattr(final_response, 'content'):
            return final_response.content
        return str(final_response)


# ==================== 第五步：运行测试 ====================
def main():
    llm = ChatTongyi(
        model="qwen-plus",
        dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
        temperature=0.0,
    )
    
    tools = [web_search, calculator]
    agent = SimpleReActAgent(llm=llm, tools=tools, max_steps=10, verbose=True)
    
    # 测试问题
    questions = [
        "LangChain 是什么？简要介绍。",
        "帮我算一下 (1234 + 5678) * 2 等于多少？",
        "搜索 Python 的信息，然后告诉我它是什么时候发布的。",
    ]
    
    for q in questions:
        print(f"\n\n{'#'*60}")
        print(f"❓ 用户问题：{q}")
        print(f"{'#'*60}")
        
        answer = agent.run(q)
        
        print(f"\n{'─'*60}")
        print(f"🎯 最终答案：{answer}")
        print(f"{'─'*60}")


if __name__ == "__main__":
    main()

5.4 ReAct 为什么有效？

         ReAct 的认知科学基础                  
                                                      
   人类解决问题时，也是"想一步、做一步"：                
                                                      
   问题: "去北京出差需要带什么衣服？"                     
                                                       
   💭 思考：我需要知道北京这几天的天气                    
   🔧 行动：打开天气APP查北京                            
   👁️ 观察：北京明天-5℃~3℃，有雪                        
                                                       
   💭 思考：零下且有雪，需要保暖衣物和防滑鞋              
   💭 思考：不需要再查什么了，可以回答                    
   📝 答案：建议带羽绒服、保暖内衣、防滑鞋、手套和帽子    
                                                       
   ReAct 就是把这种人类自然的推理模式，                   
   用 Prompt Engineering 的方式教给了 LLM。              

---

六、实战一：构建智能数据分析助手

理论知识已经足够了。现在让我们构建一个真正有用的 Agent——智能数据分析助手。它能够：

1. 读取 CSV 文件
2. 执行 Python 代码进行数据分析
3. 生成可视化图表

6.1 完整代码

import os
import sys
import io
import traceback
import pandas as pd
import matplotlib
matplotlib.use('Agg')  # 非交互式后端
import matplotlib.pyplot as plt
from typing import Optional, Type
from pydantic import BaseModel, Field
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.chat_models import ChatTongyi
from langchain_core.tools import BaseTool


# ==================== Tool 1: 读取 CSV ====================
class ReadCSVInput(BaseModel):
    filepath: str = Field(description="CSV 文件的完整路径")
    max_rows: Optional[int] = Field(
        default=20,
        description="最多显示的行数（用于预览），默认20行"
    )

class ReadCSVTool(BaseTool):
    name: str = "read_csv"
    description: str = """
    读取 CSV 文件并返回前 N 行预览，包括列名和数据类型。
    使用场景：在分析数据之前，先用此工具了解数据结构。
    """
    args_schema: Type[BaseModel] = ReadCSVInput
    
    def _run(self, filepath: str, max_rows: int = 20) -> str:
        try:
            df = pd.read_csv(filepath)
            buffer = io.StringIO()
            df.info(buf=buffer)
            
            return (
                f"✅ 成功读取 {filepath}\n"
                f"━━━━━━━━━━━━━━━━━━━━━━\n"
                f"📊 数据概览：{df.shape[0]} 行 × {df.shape[1]} 列\n"
                f"📋 列名：{list(df.columns)}\n\n"
                f"📝 数据类型：\n{buffer.getvalue()}\n"
                f"━━━━━━━━━━━━━━━━━━━━━━\n"
                f"📄 前 {min(max_rows, len(df))} 行预览：\n"
                f"{df.head(max_rows).to_string()}"
            )
        except Exception as e:
            return f"❌ 读取失败：{str(e)}"


# ==================== Tool 2: 执行 Python 分析 ====================
class PythonAnalysisInput(BaseModel):
    code: str = Field(description="""
        要执行的 Python 代码。
        可以使用的变量：
        - df: 已加载的 DataFrame（需要先用 read_csv 读取数据）
        - pd: pandas 模块
        - np: numpy 模块
        - plt: matplotlib.pyplot 模块
        
        代码的最后一行如果是表达式，其值会被自动捕获并返回。
        推荐使用 print() 来输出分析结果。
    """)

class PythonAnalysisTool(BaseTool):
    name: str = "python_analysis"
    description: str = """
    执行 Python 代码进行数据分析。
    前置条件：需要先用 read_csv 将数据加载到 DataFrame 中。
    支持的操作：数据过滤、分组聚合、统计计算、数据可视化等。
    注意：代码中不要包含危险操作（文件删除、系统命令等）。
    """
    args_schema: Type[BaseModel] = PythonAnalysisInput
    
    # 存储已加载的 DataFrame（简易状态管理）
    _dataframes: dict = {}
    
    def _run(self, code: str) -> str:
        # 安全检查
        dangerous_keywords = ['os.system', 'subprocess', '__import__', 'eval(',
                              'exec(', 'open(', 'shutil', 'importlib', 'compile(']
        for keyword in dangerous_keywords:
            if keyword in code:
                return f"❌ 安全限制：代码中包含禁止的关键词 '{keyword}'。"
        
        try:
            # 准备执行环境
            import numpy as np
            
            local_vars = {
                'pd': pd,
                'np': np,
                'plt': plt,
                **self._dataframes,  # 注入已加载的 DataFrame
            }
            
            # 捕获打印输出（必须在 exec 之前重定向）
            old_stdout = sys.stdout
            sys.stdout = captured = io.StringIO()
            
            # 执行代码
            exec(code, {'__builtins__': __builtins__}, local_vars)
            
            # 恢复 stdout
            sys.stdout = old_stdout
            printed = captured.getvalue()
            
            # 保存可能被修改的 DataFrame
            for key, value in local_vars.items():
                if isinstance(value, pd.DataFrame) and not key.startswith('_'):
                    self._dataframes[key] = value
            
            # 如果代码中使用了 plt，保存图表
            plot_path = None
            if 'plt.' in code:
                plot_path = os.path.join(os.getcwd(), "agent_output", "plot.png")
                os.makedirs(os.path.dirname(plot_path), exist_ok=True)
                plt.tight_layout()
                plt.savefig(plot_path, dpi=150, bbox_inches='tight')
                plt.close()
            
            result_parts = []
            if printed.strip():
                result_parts.append(f"📊 分析输出：\n{printed}")
            if plot_path:
                result_parts.append(f"📈 图表已保存至：{plot_path}")
            if not result_parts:
                result_parts.append("✅ 代码执行完成，无打印输出。")
            
            return "\n\n".join(result_parts)
            
        except Exception as e:
            return f"❌ 代码执行出错：\n{traceback.format_exc()}"


# ==================== Tool 3: 生成报告 ====================
class GenerateReportInput(BaseModel):
    title: str = Field(description="报告标题")
    findings: str = Field(description="分析发现的要点，用换行分隔")
    save_path: Optional[str] = Field(
        default="analysis_report.md",
        description="报告保存路径，默认为 analysis_report.md"
    )

class GenerateReportTool(BaseTool):
    name: str = "generate_report"
    description: str = """
    将分析发现整理为一份结构化的 Markdown 报告并保存。
    使用场景：在数据分析完成后，总结分析结果。
    """
    args_schema: Type[BaseModel] = GenerateReportInput
    
    def _run(self, title: str, findings: str, save_path: str = "analysis_report.md") -> str:
        report = f"""# {title}

> 📅 生成时间：{pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}

---

## 📊 分析发现

{findings}

---

## 🔍 分析方法

本次分析使用了 Python 数据分析栈（pandas + numpy + matplotlib），
通过 LangChain Agent 驱动自动化分析流程。

---

*本报告由 AI 数据分析助手自动生成*
"""
        output_dir = os.path.join(os.getcwd(), "agent_output")
        os.makedirs(output_dir, exist_ok=True)
        full_path = os.path.join(output_dir, save_path)
        
        with open(full_path, "w", encoding="utf-8") as f:
            f.write(report)
        
        return f"✅ 报告已保存至：{full_path}\n\n{report}"


# ==================== 组装 Agent ====================
def create_data_analyst_agent(api_key: str):
    """创建数据分析 Agent"""
    
    llm = ChatTongyi(
        model="qwen-plus",
        dashscope_api_key=api_key,
        temperature=0.0,
    )
    
    tools = [
        ReadCSVTool(),
        PythonAnalysisTool(),
        GenerateReportTool(),
    ]
    
    agent = create_react_agent(llm, tools)
    
    return AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=15,
        max_execution_time=120,
        handle_parsing_errors=True,
        early_stopping_method="generate",
    )


# ==================== 演示 ====================
def demo():
    """演示数据分析 Agent"""
    
    API_KEY = os.getenv("DASHSCOPE_API_KEY", "sk-xxx")
    agent = create_data_analyst_agent(API_KEY)
    
    # 先创建一个示例 CSV
    sample_data = pd.DataFrame({
        "月份": ["1月", "2月", "3月", "4月", "5月", "6月",
                 "7月", "8月", "9月", "10月", "11月", "12月"],
        "销售额": [12000, 13500, 14800, 14200, 16800, 18900,
                  19500, 21000, 22500, 20800, 23500, 28000],
        "成本": [8000, 8500, 9000, 8800, 9500, 10200,
                 10800, 11500, 12000, 11200, 12500, 14500],
        "客户数": [45, 52, 58, 55, 62, 70, 75, 82, 88, 80, 90, 105],
    })
    
    data_dir = os.path.join(os.getcwd(), "agent_output")
    os.makedirs(data_dir, exist_ok=True)
    csv_path = os.path.join(data_dir, "sales_data.csv")
    sample_data.to_csv(csv_path, index=False)
    
    # Agent 任务
    task = f"""
    请对 {csv_path} 中的销售数据进行全面分析。请按以下步骤操作：

    1. 先用 read_csv 读取数据，了解数据结构
    2. 用 python_analysis 执行以下分析：
       - 计算销售额和成本的基本统计（平均值、最大值、最小值）
       - 计算每个月的利润率（(销售额-成本)/销售额*100%）
       - 找出销售额最高和最低的月份
       - 分析销售额的增长趋势
       - 生成一张折线图展示销售额和利润的月度变化
    3. 用 generate_report 总结你的分析发现

    请用中文给出分析结果。
    """
    
    result = agent.invoke({"input": task})
    print(f"\n{'='*60}")
    print("🎯 最终分析结果：")
    print(result['output'])


if __name__ == "__main__":
    demo()

6.2 Agent 执行过程解析

当运行上述代码时，Agent 的执行过程大致如下：

Step 1 ─ 💭 Thought: 我需要先读取 CSV 文件来了解数据结构
         🔧 Action: read_csv("agent_output/sales_data.csv")
         👁️ Observation: 数据有12行4列，列名：月份、销售额、成本、客户数...

Step 2 ─ 💭 Thought: 现在我了解了数据结构。接下来需要进行统计分析。
         🔧 Action: python_analysis("df.describe()")
         👁️ Observation: 销售额平均值...最大值...最小值...

Step 3 ─ 💭 Thought: 我需要计算利润率和增长趋势，并生成图表
         🔧 Action: python_analysis("df['利润率'] = (df['销售额'] - df['成本']) / df['销售额'] * 100...")
         👁️ Observation: 图表已保存至 agent_output/plot.png

Step 4 ─ 💭 Thought: 分析已经完成，让我生成最终报告
         🔧 Action: generate_report("销售数据分析报告", "1. 销售额呈上升趋势...")
         👁️ Observation: 报告已保存

Step 5 ─ 💭 Thought: 所有任务已完成
         📝 Final Answer: 分析完成！以下是主要发现：...

---

七、实战二：构建多工具协作的 DevOps Agent

再来看一个更复杂的 Agent——DevOps 运维助手，它集成多个外部工具来完成运维任务。

import os
from typing import Optional, Type
from pydantic import BaseModel, Field
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.chat_models import ChatTongyi
from langchain_core.tools import BaseTool, tool
from datetime import datetime


# ==================== Tool 1: 系统信息 ====================
class SystemInfoTool(BaseTool):
    name: str = "system_info"
    description: str = """
    获取当前系统的基本信息，包括操作系统、CPU使用率、内存使用情况、磁盘空间。
    使用场景：当用户询问服务器状态、系统资源时调用。
    """
    
    def _run(self, tool_input: str = "") -> str:
        import platform
        import psutil
        
        info = {
            "操作系统": platform.system(),
            "主机名": platform.node(),
            "CPU核心数": psutil.cpu_count(),
            "CPU使用率": f"{psutil.cpu_percent(interval=1)}%",
            "内存总量": f"{psutil.virtual_memory().total / (1024**3):.1f} GB",
            "内存使用率": f"{psutil.virtual_memory().percent}%",
            "磁盘总量": f"{psutil.disk_usage('/').total / (1024**3):.1f} GB",
            "磁盘使用率": f"{psutil.disk_usage('/').percent}%",
        }
        
        return "\n".join(f"{k}: {v}" for k, v in info.items())


# ==================== Tool 2: 日志查询（模拟） ====================
class LogSearchInput(BaseModel):
    keyword: str = Field(description="搜索关键词，例如 'ERROR', 'timeout', 'Exception'")
    lines: Optional[int] = Field(default=50, description="返回最近 N 行匹配的日志")
    service: Optional[str] = Field(
        default="all",
        description="服务名称，例如 'nginx', 'api-server', 'database'"
    )

class LogSearchTool(BaseTool):
    name: str = "search_logs"
    description: str = """
    搜索系统日志。支持按关键词、服务名过滤。
    使用场景：排查线上问题时，搜索错误日志或异常堆栈。
    """
    args_schema: Type[BaseModel] = LogSearchInput
    
    def _run(self, keyword: str, lines: int = 50, service: str = "all") -> str:
        # 模拟日志数据（实际项目中替换为 Elasticsearch / Loki 查询）
        mock_logs = [
            f"[2024-12-20 10:23:45] [{service}] INFO  Request processed in 234ms",
            f"[2024-12-20 10:24:12] [{service}] {keyword} Database connection timeout after 30s",
            f"[2024-12-20 10:24:30] [{service}] {keyword} Failed to connect to redis:6379",
            f"[2024-12-20 10:25:01] [{service}] WARN  High memory usage detected: 87%",
            f"[2024-12-20 10:25:45] [{service}] {keyword} Exception in worker-3: OutOfMemoryError",
        ]
        
        matched = [log for log in mock_logs if keyword.lower() in log.lower()]
        
        if not matched:
            return f"未找到包含「{keyword}」的日志。"
        
        return f"找到 {len(matched)} 条匹配的日志（服务：{service}）：\n" + "\n".join(matched)


# ==================== Tool 3: HTTP 健康检查 ====================
class HealthCheckInput(BaseModel):
    url: str = Field(description="要检查的URL，例如 'https://api.example.com/health'")
    timeout: Optional[int] = Field(default=10, description="超时时间（秒）")

class HealthCheckTool(BaseTool):
    name: str = "health_check"
    description: str = """
    对指定的 HTTP 端点进行健康检查，返回状态码和响应时间。
    使用场景：检查服务是否正常运行，排查接口可用性问题。
    """
    args_schema: Type[BaseModel] = HealthCheckInput
    
    def _run(self, url: str, timeout: int = 10) -> str:
        import time
        try:
            import urllib.request
            start = time.time()
            req = urllib.request.Request(url, method='HEAD')
            resp = urllib.request.urlopen(req, timeout=timeout)
            elapsed = (time.time() - start) * 1000
            
            return (
                f"URL: {url}\n"
                f"状态码: {resp.status}\n"
                f"响应时间: {elapsed:.1f}ms\n"
                f"✅ 服务可用"
            )
        except Exception as e:
            return f"URL: {url}\n❌ 健康检查失败: {str(e)}"


# ==================== Tool 4: 时间工具（用于定时任务等） ====================
@tool
def get_current_time(format: str = "%Y-%m-%d %H:%M:%S") -> str:
    """获取当前系统时间。format 为时间格式字符串。"""
    return datetime.now().strftime(format)


# ==================== 组装 DevOps Agent ====================
def create_devops_agent(api_key: str):
    """创建 DevOps 运维 Agent"""
    
    llm = ChatTongyi(
        model="qwen-plus",
        dashscope_api_key=api_key,
        temperature=0.0,  # DevOps 场景对准确性要求高，必须低温
    )
    
    tools = [
        SystemInfoTool(),
        LogSearchTool(),
        HealthCheckTool(),
        get_current_time,
    ]
    
    # 使用增强的 System Prompt
    agent = create_react_agent(llm, tools)
    
    return AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=12,
        max_execution_time=90,
        handle_parsing_errors=True,
        early_stopping_method="generate",
    )


# ==================== 演示 ====================
def demo_devops():
    API_KEY = os.getenv("DASHSCOPE_API_KEY", "sk-xxx")
    agent = create_devops_agent(API_KEY)
    
    # 场景1：系统健康检查
    print("\n" + "="*60)
    print("场景1：系统综合健康检查")
    print("="*60)
    
    result1 = agent.invoke({
        "input": "请全面检查一下服务器的状态：先查看系统资源（CPU、内存、磁盘），然后检查 API 服务的健康状态（https://api.example.com/health），最后搜一下最近的错误日志。"
    })

    print(f"\n📊 检查结果：\n{result1['output']}")
    
    # 场景2：故障排查
    print("\n" + "="*60)
    print("场景2：数据库连接故障排查")
    print("="*60)
    
    result2 = agent.invoke({
        "input": "用户反馈系统响应很慢。请帮排查：先看系统资源占用是否正常，然后搜一下包含 'timeout' 或 'ERROR' 的日志，判断问题原因。"
    })
    
    print(f"\n🔍 排查结果：\n{result2['output']}")


if __name__ == "__main__":
    demo_devops()

---

八、Agent 开发的进阶技巧与避坑指南

8.1 工具设计的三层架构

在实际项目中，推荐将工具按职责分为三层：

8.2 错误处理模式

from langchain.agents import AgentExecutor
from langchain_core.tools import tool
from typing import Optional
import functools


# ========== 模式1：工具级重试 ==========
def retry_on_failure(max_retries: int = 3, backoff_factor: float = 2.0):
    """装饰器：为工具添加重试逻辑"""
    import time
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        wait = backoff_factor ** attempt
                        time.sleep(wait)
            
            return f"❌ 工具执行失败（已重试{max_retries}次）：{last_error}"
        return wrapper
    return decorator


# ========== 模式2：降级处理 ==========
def primary_search(query: str) -> str:
    """主搜索引擎（示例）"""
    # 实际项目中调用 Google/Bing API
    raise RuntimeError("主搜索引擎不可用")

def fallback_search(query: str) -> str:
    """备用搜索引擎（示例）"""
    return f"使用备用搜索找到关于「{query}」的结果..."

@tool
def search_with_fallback(query: str) -> str:
    """
    搜索信息，主搜索失败时自动降级。
    当主搜索引擎不可用时，使用备用搜索引擎。
    """
    try:
        # 尝试主搜索
        return primary_search(query)
    except Exception:
        return f"⚠️ 主搜索不可用，使用备用搜索。\n结果：{fallback_search(query)}"


# ========== 模式3：Agent 级别的错误处理 ==========
class RobustAgentExecutor(AgentExecutor):
    """
    增强版 AgentExecutor，添加了更完善的错误处理。
    """
    
    def _take_next_step(self, *args, **kwargs):
        try:
            return super()._take_next_step(*args, **kwargs)
        except Exception as e:
            # 将异常转换为 Observation，让 Agent 自己处理
            error_msg = f"Observation: 执行出错 - {str(e)}\nThought: 我需要尝试其他方法..."
            return self.agent.plan(
                intermediate_steps=self.intermediate_steps,
                input=self.input,
            )

8.3 Agent 输出控制

如何确保 Agent 的输出格式符合你的预期？

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel
from typing import List, Optional


class TaskResult(BaseModel):
    """定义 Agent 的结构化输出格式"""
    summary: str           # 任务摘要
    steps_taken: int       # 完成的步骤数
    findings: List[str]    # 发现的问题或结果
    recommendations: List[str]  # 建议的后续操作
    confidence: float      # 置信度 0-1


# 在 System Prompt 中嵌入输出格式要求
FORMATTED_AGENT_PROMPT = """
... (原有 ReAct prompt) ...

在给出 Final Answer 时，必须使用以下 JSON 格式：
{{
    "summary": "任务的简要总结",
    "steps_taken": 完成的步骤数,
    "findings": ["发现1", "发现2"],
    "recommendations": ["建议1", "建议2"],
    "confidence": 0.0到1.0之间的数值
}}

现在开始！
"""

8.4 调试技巧

# ========== 技巧1：开启 verbose 模式 ==========
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # ← 关键开关，打印完整推理过程
)

# ========== 技巧2：获取中间步骤 ==========
result = agent_executor.invoke({
    "input": "你的问题"
})

# 如果你需要查看中间步骤：
agent_executor.return_intermediate_steps = True
result_with_steps = agent_executor.invoke({"input": "你的问题"})
for i, (action, observation) in enumerate(result_with_steps["intermediate_steps"]):
    print(f"Step {i+1}: {action.tool}({action.tool_input}) → {observation[:100]}...")

# ========== 技巧3：使用 LangSmith 跟踪 ==========
# 设置环境变量即可自动记录每次 Agent 运行
# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"
# os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

# ========== 技巧4：回调函数 ==========
from langchain_core.callbacks import BaseCallbackHandler

class AgentCallback(BaseCallbackHandler):
    """自定义回调，记录 Agent 的每一步"""
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"🔧 [Tool Start] {serialized.get('name', 'unknown')}: {input_str}")
    
    def on_tool_end(self, output, **kwargs):
        print(f"👁️ [Tool End] {str(output)[:200]}...")
    
    def on_agent_action(self, action, **kwargs):
        print(f"💭 [Agent Action] {action.tool}: {action.tool_input}")

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[AgentCallback()],
)

8.5 常见问题与解决方案

问题	原因	解决方案
Agent 陷入死循环	LLM 反复调用同一个 Tool 得不到想要的结果	增加 max_iterations 限制；优化 Tool 的描述和返回值
Agent 使用错误的 Tool	Tool 描述不够清晰，LLM 判断失误	改进 docstring，加入"使用场景"说明
Tool 参数传递错误	LLM 没有正确理解参数格式	使用 Pydantic Model 定义 args_schema，添加 Field description
输出格式不稳定	LLM 温度太高	Agent 场景建议 temperature=0.0
工具执行超时	外部 API 响应慢	为工具添加 timeout 和重试逻辑
解析错误（Parsing Error）	LLM 输出不符合 ReAct 格式	开启 handle_parsing_errors=True

---

九、总结与展望

9.1 本文回顾

知识点	核心要点
Agent 概念	LLM（大脑）+ Tools（手脚）+ 推理循环（Thought→Action→Observation）
Tool 设计	函数名 + docstring 是关键，用 Pydantic 定义参数 Schema，职责单一
AgentExecutor	驱动推理循环的引擎，配置 max_iterations、max_execution_time 等参数控制行为
ReAct 模式	交替进行"推理"和"行动"，每一步都可追溯、可纠错
实战案例	数据分析 Agent（读取CSV→Python分析→生成报告）、DevOps Agent（系统检查→日志搜索→健康检查）

9.2 Agent vs Chain：什么时候用哪个？

                       需要自主决策？
                       /           \
                     否            是
                    │               │
                    ▼               ▼
              用 Chain         需要调用外部工具？
              固定流程         /             \
                             否              是
                             │               │
                             ▼               ▼
                        优化 Prompt      用 Agent
                        （Few-shot等）   （ReAct 等）

9.3 进阶方向

LangChain Agent 基础（本文）
        │
        ├──▶ 多 Agent 协作（AutoGen / CrewAI）
        │    多个 Agent 分工协作，像一个团队一样工作
        │
        ├──▶ 自定义 Agent 类型
        │    继承 BaseAgent 实现特定推理模式
        │
        ├──▶ LangGraph 状态图 Agent
        │    用有向图定义复杂多分支 Agent 行为
        │
        ├──▶ Agent 记忆系统
        │    长期记忆 + 短期记忆 + 语义检索
        │
        └──▶ 生产化部署
        │    LangServe + FastAPI + Docker + 监控
        │
        └──▶ MCP (Model Context Protocol)
             标准化的工具接口协议，让 Agent 可以接入任何 MCP Server

9.4 核心心法

💡 构建好的 Agent，80% 的工作在于设计好的 Tool。

Tool 是 Agent 与世界的接口——Tool 的设计直接决定了 Agent 的能力边界和可靠性。花时间打磨你的 Tool 描述、参数设计和错误处理，比调整 Agent 的配置参数有效得多。

💡 Agent 不是魔法——它是精心设计的 Prompt Engineering + 工具编排。

ReAct 的 “智能” 来自于两点：(1) 清晰的结构化 Prompt 模板；(2) 恰到好处的中间结果反馈。理解了这两点，你就理解了 Agent 的本质。

---

本文代码基于 langchain ≥ 0.2.0 编写，使用通义千问（Qwen）作为示例模型。所有 Agent 模式（ReAct、Tool、AgentExecutor）均为框架通用设计，切换模型只需替换 LLM 类。