Agent 智能代理与工具调用

HSXISAWD2026/5/14大约 27 分钟

Agent 智能代理与工具调用

什么是 Agent

在 LangChain 的体系中，Agent（智能代理） 是一种能够自主决策、选择工具并执行操作的智能系统。与传统的 Chain（链）不同，Agent 不需要预先定义固定的执行流程，而是根据用户的输入动态决定下一步该做什么。

Agent 与普通 Chain 的对比

特性	Chain（链）	Agent（代理）
执行流程	固定、预定义	动态、自主决策
工具使用	固定步骤	按需选择工具
灵活性	低，确定性输出	高，根据输入自适应
适用场景	结构化任务	开放性、多步骤任务
调试难度	简单	较复杂
成本控制	容易预估	需要额外策略
典型应用	RAG 问答、翻译流水线	研究助手、代码生成、数据分析

[!tip] 何时选择 Agent？
当你的任务需要以下能力时，应该考虑使用 Agent：
需要根据不同输入采取不同行动
需要调用外部工具（搜索、数据库、API）
任务步骤不确定，需要 LLM 自主规划
需要多轮推理和决策

Agent 工作原理

Agent 的核心是 ReAct（Reasoning + Acting） 模式。它遵循"观察 - 思考 - 行动"的循环，直到得到最终答案。

ReAct 循环

ReAct 的四个阶段

用户输入
  |
  v
[1] Observation（观察）
    接收用户输入或上一轮工具返回的结果
  |
  v
[2] Thought（思考）
    LLM 分析当前信息，判断是否需要更多信息
  |
  v
[3] Action（行动）
    如果信息不足 → 选择并调用工具
    如果信息充足 → 生成最终答案
  |
  v
[4] 回到 Observation（循环）

下面是一个 Agent 执行过程的实际输出示例：

问题: 2024年法国GDP是多少？折合人民币是多少？

> Entering new AgentExecutor chain...

Thought: 我需要查找2024年法国的GDP数据，然后将欧元或美元转换为人民币。
Action: tavily_search
Action Input: "2024年法国GDP"
Observation: 根据世界银行数据，2024年法国GDP约为3.16万亿美元。

Thought: 我得到了法国GDP数据，现在需要查询美元兑人民币的汇率。
Action: tavily_search
Action Input: "美元兑人民币汇率 2024"
Observation: 2024年美元兑人民币平均汇率约为7.15。

Thought: 现在我可以计算了。
Action: calculator
Action Input: 3.16 * 7.15
Observation: 22.594

Thought: 我已经得到了所有需要的信息，可以给出最终答案了。
Final Answer: 2024年法国GDP约为3.16万亿美元，按7.15的汇率折算，约合22.59万亿人民币。

> Finished chain.

工具定义

工具（Tool）是 Agent 与外部世界交互的桥梁。LangChain 提供了三种定义工具的方式。

方式一：@tool 装饰器（推荐）

最简洁的方式，使用 @tool 装饰器将普通 Python 函数转化为工具：

from langchain_core.tools import tool


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

    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a * b


@tool
def get_word_length(word: str) -> int:
    """返回一个单词的字符长度。"""
    return len(word)


# 使用工具
result = multiply.invoke({"a": 3, "b": 4})
print(result)  # 12

# 查看工具信息
print(multiply.name)         # "multiply"
print(multiply.description)  # 自动从 docstring 提取
print(multiply.args_schema)  # 自动从类型注解生成

[!warning] docstring 非常重要
@tool 装饰器会自动从函数的 docstring 中提取工具描述。LLM 通过描述来判断何时调用该工具，因此 docstring 必须清晰、准确地描述工具的功能和参数。模糊的描述会导致 Agent 错误地选择工具。

方式二：StructuredTool

当需要更精细的控制时，可以使用 StructuredTool：

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field


# 定义参数 Schema
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最大返回结果数")


# 定义工具函数
def search_database(query: str, max_results: int = 5) -> str:
    """在数据库中搜索相关信息。"""
    # 模拟数据库搜索
    results = [
        f"结果{i}: 关于'{query}'的信息"
        for i in range(1, max_results + 1)
    ]
    return "\n".join(results)


# 创建 StructuredTool
search_tool = StructuredTool.from_function(
    func=search_database,
    name="database_search",
    description="在企业数据库中搜索信息，支持指定返回结果数量",
    args_schema=SearchInput,
    return_direct=False,  # 如果为 True，工具结果直接返回给用户，不再经过 LLM
)


# 使用
result = search_tool.invoke({"query": "LangChain", "max_results": 3})
print(result)

方式三：继承 BaseTool

最灵活的方式，适合复杂的工具逻辑：

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


class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位: celsius 或 fahrenheit")


class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = "获取指定城市的当前天气信息"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, unit: str = "celsius") -> str:
        """同步执行天气查询。"""
        # 模拟 API 调用
        weather_data = {
            "北京": {"temp": 22, "condition": "晴天"},
            "上海": {"temp": 25, "condition": "多云"},
            "深圳": {"temp": 30, "condition": "雷阵雨"},
        }

        data = weather_data.get(city, {"temp": 20, "condition": "未知"})
        temp = data["temp"]

        if unit == "fahrenheit":
            temp = temp * 9 / 5 + 32
            return f"{city}: {data['condition']}，温度 {temp}°F"

        return f"{city}: {data['condition']}，温度 {temp}°C"

    async def _arun(self, city: str, unit: str = "celsius") -> str:
        """异步执行天气查询。"""
        # 异步版本的实现
        return self._run(city, unit)


# 使用
weather = WeatherTool()
result = weather.invoke({"city": "北京"})
print(result)  # 北京: 晴天，温度 22°C

三种工具定义方式对比

方式	适用场景	优点	缺点
`@tool` 装饰器	简单函数	简洁、代码量少	灵活性有限
`StructuredTool`	需要自定义参数验证	支持自定义 Schema	需要额外定义 Pydantic 模型
`BaseTool`	复杂逻辑、需要状态管理	最大灵活性、支持异步	代码量较多

[!tip] 选择建议
优先使用 @tool 装饰器。只有当需要自定义参数验证 Schema 或复杂的状态管理时，才考虑使用 StructuredTool 或 BaseTool。

内置工具

LangChain 提供了一系列开箱即用的内置工具，可以快速集成到 Agent 中。

Tavily Search

Tavily 是专为 AI Agent 设计的搜索引擎，返回结构化的搜索结果。

# 安装依赖
# pip install langchain-community tavily-python

import os
os.environ["TAVILY_API_KEY"] = "your-tavily-api-key"

from langchain_community.tools.tavily_search import TavilySearchResults

# 创建搜索工具
search = TavilySearchResults(
    max_results=3,           # 最多返回3条结果
    search_depth="advanced",  # 搜索深度: basic 或 advanced
    include_raw_content=True, # 是否包含原始网页内容
)

# 使用
results = search.invoke("LangChain Agent 最新教程")
print(results)

Wikipedia

查询维基百科获取百科知识：

# pip install wikipedia

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# 创建 Wikipedia 工具
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(
    top_k_results=2,       # 返回前2条结果
    doc_content_chars_max=2000,  # 内容最大字符数
))

# 使用
result = wikipedia.invoke("人工智能")
print(result)

Python REPL

让 Agent 能够执行 Python 代码（慎用）：

from langchain_community.tools import PythonREPLTool

python_repl = PythonREPLTool()

# Agent 可以用它执行代码
result = python_repl.invoke("print([x**2 for x in range(10)])")
print(result)

[!warning] Python REPL 安全风险
Python REPL 工具允许 Agent 执行任意 Python 代码，这在生产环境中存在严重的安全风险。建议：
仅在沙箱环境中使用
使用 Docker 容器隔离执行环境
限制文件系统和网络访问权限
生产环境中考虑使用代码审查机制

Requests 工具

用于发送 HTTP 请求：

from langchain_community.tools import RequestsGetTool
from langchain_community.utilities import TextRequestsWrapper

# 创建 HTTP GET 工具
requests_tool = RequestsGetTool(
    requests_wrapper=TextRequestsWrapper(),
    allow_dangerous_requests=True,  # 必须显式允许
)

# 使用
result = requests_tool.invoke("https://api.github.com/users/langchain-ai")
print(result)

内置工具一览

工具名	功能	依赖包	安全等级
TavilySearchResults	AI 搜索引擎	`tavily-python`	安全
WikipediaQueryRun	维基百科查询	`wikipedia`	安全
PythonREPLTool	执行 Python 代码	内置	高危
RequestsGetTool	HTTP GET 请求	`requests`	中危
RequestsPostTool	HTTP POST 请求	`requests`	中危
DuckDuckGoSearchRun	DuckDuckGo 搜索	`duckduckgo-search`	安全
ArxivQueryRun	论文搜索	`arxiv`	安全
PubmedQueryRun	医学文献搜索	`xmltodict`	安全
RedditSearchRun	Reddit 搜索	`praw`	安全

自定义工具开发

在实际项目中，通常需要开发自定义工具来满足特定业务需求。以下是几个常见的自定义工具示例。

计算器工具

import math
from langchain_core.tools import tool


@tool
def calculator(expression: str) -> str:
    """计算数学表达式并返回结果。

    支持基本的四则运算、幂运算、三角函数等。

    Args:
        expression: 数学表达式字符串，例如 "2 + 3 * 4"、"sin(3.14)"
    """
    # 安全的数学计算 - 仅允许数学函数和基本运算
    allowed_names = {
        "abs": abs, "round": round, "min": min, "max": max,
        "sqrt": math.sqrt, "pow": math.pow,
        "sin": math.sin, "cos": math.cos, "tan": math.tan,
        "log": math.log, "log10": math.log10,
        "pi": math.pi, "e": math.e,
    }

    try:
        # 编译表达式，限制可用函数
        code = compile(expression, "<string>", "eval")
        for name in code.co_names:
            if name not in allowed_names:
                return f"错误: 不支持的操作 '{name}'"

        result = eval(code, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"


@tool
def unit_converter(value: float, from_unit: str, to_unit: str) -> str:
    """在不同度量单位之间进行转换。

    Args:
        value: 要转换的数值
        from_unit: 原始单位（如 km, mile, kg, lb, celsius, fahrenheit）
        to_unit: 目标单位
    """
    conversions = {
        ("km", "mile"): lambda x: x * 0.621371,
        ("mile", "km"): lambda x: x * 1.60934,
        ("kg", "lb"): lambda x: x * 2.20462,
        ("lb", "kg"): lambda x: x * 0.453592,
        ("celsius", "fahrenheit"): lambda x: x * 9 / 5 + 32,
        ("fahrenheit", "celsius"): lambda x: (x - 32) * 5 / 9,
    }

    key = (from_unit, to_unit)
    if key in conversions:
        result = conversions[key](value)
        return f"{value} {from_unit} = {result:.4f} {to_unit}"
    return f"不支持的转换: {from_unit} -> {to_unit}"

数据库查询工具

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Optional


class QueryInput(BaseModel):
    sql: str = Field(description="SQL 查询语句，仅支持 SELECT 查询")
    limit: int = Field(default=10, description="返回结果的最大行数")


@tool(args_schema=QueryInput)
def query_database(sql: str, limit: int = 10) -> str:
    """查询企业数据库，执行 SQL SELECT 语句并返回结果。

    注意：此工具仅允许 SELECT 查询，禁止 INSERT/UPDATE/DELETE 操作。
    """
    # 安全检查：只允许 SELECT
    normalized_sql = sql.strip().upper()
    if not normalized_sql.startswith("SELECT"):
        return "错误: 仅允许 SELECT 查询"

    # 防止危险的 SQL 操作
    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
    for keyword in dangerous_keywords:
        if keyword in normalized_sql:
            return f"错误: 禁止使用 {keyword} 操作"

    # 添加 LIMIT（如果用户没有指定）
    if "LIMIT" not in normalized_sql:
        sql = f"{sql.rstrip(';')} LIMIT {limit}"

    try:
        # 模拟数据库查询（实际项目中替换为真实的数据库连接）
        # import sqlite3
        # conn = sqlite3.connect("your_database.db")
        # cursor = conn.cursor()
        # cursor.execute(sql)
        # columns = [desc[0] for desc in cursor.description]
        # rows = cursor.fetchall()
        # conn.close()

        # 模拟返回结果
        mock_results = [
            {"id": 1, "name": "产品A", "price": 99.9, "stock": 100},
            {"id": 2, "name": "产品B", "price": 199.0, "stock": 50},
            {"id": 3, "name": "产品C", "price": 49.9, "stock": 200},
        ]

        # 格式化为表格字符串
        header = " | ".join(mock_results[0].keys())
        separator = "-" * len(header)
        rows = [" | ".join(str(v) for v in row.values()) for row in mock_results]
        return f"{header}\n{separator}\n" + "\n".join(rows)

    except Exception as e:
        return f"查询错误: {str(e)}"

文件读取工具

import os
from langchain_core.tools import tool
from pydantic import BaseModel, Field


class FileReadInput(BaseModel):
    file_path: str = Field(description="要读取的文件路径")
    encoding: str = Field(default="utf-8", description="文件编码")
    max_lines: int = Field(default=100, description="最大读取行数")


@tool(args_schema=FileReadInput)
def read_file(file_path: str, encoding: str = "utf-8", max_lines: int = 100) -> str:
    """读取指定路径的文本文件内容。

    支持读取 .txt, .md, .py, .json, .csv 等文本文件。
    """
    # 安全检查：限制可读取的目录范围
    allowed_directories = ["/data", "/docs", "/uploads"]
    abs_path = os.path.abspath(file_path)

    if not any(abs_path.startswith(d) for d in allowed_directories):
        return f"错误: 只允许读取以下目录的文件: {allowed_directories}"

    if not os.path.exists(abs_path):
        return f"错误: 文件不存在 '{file_path}'"

    if os.path.getsize(abs_path) > 1024 * 1024:  # 1MB 限制
        return "错误: 文件过大，超过 1MB 限制"

    try:
        with open(abs_path, "r", encoding=encoding) as f:
            lines = f.readlines()[:max_lines]
            content = "".join(lines)
            total_lines = len(content.split("\n"))
            return f"文件: {file_path} (显示前 {total_lines} 行)\n---\n{content}"
    except UnicodeDecodeError:
        return "错误: 文件编码不正确，请尝试其他编码"
    except Exception as e:
        return f"读取错误: {str(e)}"

[!warning] 工具安全原则
开发自定义工具时，务必遵循以下安全原则：
输入验证 — 对所有输入进行严格的类型和范围检查
权限最小化 — 只开放必要的最小权限
路径限制 — 文件操作限制在指定目录内
SQL 注入防护 — 数据库查询使用参数化查询
超时控制 — 设置执行超时，防止工具卡死

create_react_agent

create_react_agent 是 LangChain 提供的高层 API，用于快速创建基于 ReAct 模式的 Agent。

基本用法

# pip install langchain langchain-openai langchain-community tavily-python

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent


# 1. 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
)


# 2. 定义工具
@tool
def get_current_time(city: str) -> str:
    """获取指定城市的当前时间。"""
    from datetime import datetime
    import pytz

    timezone_map = {
        "北京": "Asia/Shanghai",
        "东京": "Asia/Tokyo",
        "纽约": "America/New_York",
        "伦敦": "Europe/London",
    }

    tz_name = timezone_map.get(city, "UTC")
    tz = pytz.timezone(tz_name)
    now = datetime.now(tz)
    return f"{city}当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')}"


search = TavilySearchResults(max_results=3)

# 3. 创建 Agent
agent = create_react_agent(
    model=llm,
    tools=[get_current_time, search],
    prompt="你是一个有用的助手，可以搜索信息并查询时间。请用中文回答。",
)


# 4. 执行 Agent
result = agent.invoke(
    {"messages": [{"role": "user", "content": "北京现在几点了？最近有什么AI新闻？"}]}
)

# 5. 查看结果
for message in result["messages"]:
    print(f"[{message.type}] {message.content[:200]}")

LangGraph 方式（推荐）

从 LangChain v0.3 开始，推荐使用 LangGraph 的 create_react_agent：

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults


# 定义工具列表
tools = [
    TavilySearchResults(max_results=3),
]


# 创建 Agent
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=tools,
)


# 流式输出
from langgraph.prebuilt import create_react_agent

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "LangChain和LangGraph有什么区别？"}]}
):
    print(chunk)
    print("---")

Agent 决策流程图

AgentExecutor

AgentExecutor 是 Agent 的运行引擎，负责管理 Agent 的执行循环、错误处理和资源控制。

基本配置

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults


# 1. 准备工具
tools = [TavilySearchResults(max_results=3)]

# 2. 创建 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能研究助手，请用中文回答问题。你可以使用以下工具来帮助回答。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),  # Agent 的中间思考过程
])

# 3. 创建 Agent
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_tool_calling_agent(llm, tools, prompt)

# 4. 创建 AgentExecutor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,

    # 执行控制
    max_iterations=10,           # 最大迭代次数
    max_execution_time=60,       # 最大执行时间（秒）
    verbose=True,                # 打印详细执行过程

    # 错误处理
    handle_parsing_errors=True,  # 自动处理解析错误
    max_consecutive_failures=3,  # 连续失败最大次数

    # 返回控制
    return_intermediate_steps=True,  # 返回中间步骤
    early_stopping_method="generate",  # 超时时的处理方式
)

# 5. 执行
result = agent_executor.invoke({"input": "今天的科技新闻有哪些？"})
print("最终答案:", result["output"])
print("中间步骤:", result["intermediate_steps"])

错误处理策略

# 方式一：自动处理解析错误
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,  # 发生解析错误时，将错误信息返回给 LLM 重试
)

# 方式二：自定义错误处理函数
def handle_error(error):
    return f"工具调用出错: {str(error)}，请尝试其他方式回答问题。"

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=handle_error,
)

# 方式三：字符串消息
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors="工具调用失败，请检查输入格式后重试。",
)

流式输出

# 流式输出 Agent 的执行过程
for chunk in agent_executor.stream({"input": "解释量子计算的原理"}):
    # 不同类型的输出
    if "actions" in chunk:
        for action in chunk["actions"]:
            print(f"[工具调用] {action.tool} -> {action.tool_input}")
    elif "steps" in chunk:
        for step in chunk["steps"]:
            print(f"[观察结果] {step.observation[:100]}...")
    elif "output" in chunk:
        print(f"[最终答案] {chunk['output']}")

AgentExecutor 配置参数一览

参数	类型	默认值	说明
`max_iterations`	int	15	最大迭代次数
`max_execution_time`	float	None	最大执行时间（秒）
`verbose`	bool	False	是否打印详细过程
`handle_parsing_errors`	bool/str/callable	False	解析错误处理策略
`return_intermediate_steps`	bool	False	是否返回中间步骤
`early_stopping_method`	str	"force"	提前停止方式: "force" 或 "generate"
`max_consecutive_failures`	int	None	连续失败最大次数
`tags`	list	None	运行标签
`metadata`	dict	None	运行元数据

LangGraph Agent

LangGraph 是 LangChain 团队推出的 Agent 框架，专门用于构建复杂的、有状态的 Agent 工作流。相比传统的 AgentExecutor，LangGraph 提供了更精细的控制。

为什么需要 LangGraph

使用 LangGraph 创建 Agent

# pip install langgraph langchain-openai

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition


# 1. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]


# 2. 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网上的信息。"""
    # 模拟搜索结果
    return f"搜索 '{query}' 的结果: 这里是最新的相关信息..."


@tool
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"


tools = [search_web, calculate]

# 3. 初始化 LLM 并绑定工具
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)


# 4. 定义节点函数
def chatbot(state: State):
    """Agent 主节点：调用 LLM 决策"""
    return {"messages": [llm_with_tools.invoke(state["messages"])]}


# 5. 构建图
graph_builder = StateGraph(State)

# 添加节点
graph_builder.add_node("chatbot", chatbot)

# 添加工具节点（ToolNode 自动处理工具调用）
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)

# 添加边
graph_builder.add_edge(START, "chatbot")

# 条件边：如果 LLM 返回工具调用，则跳转到 tools 节点；否则结束
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")  # 工具执行完毕后回到 chatbot

# 编译图
graph = graph_builder.compile()


# 6. 执行
result = graph.invoke({
    "messages": [{"role": "user", "content": "搜索 LangGraph 的最新版本信息，并计算 2024 - 2020 的值"}]
})

for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content[:150]}")

LangGraph 可视化

# 生成 Agent 工作流的 Mermaid 图
from IPython.display import Image, display

try:
    img = Image(graph.get_graph().draw_mermaid_png())
    display(img)
except Exception:
    # 需要 pip install grandalf
    pass

添加记忆（Memory）

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点
memory = MemorySaver()

# 编译时传入 checkpointer
graph_with_memory = graph_builder.compile(checkpointer=memory)

# 第一轮对话
config = {"configurable": {"thread_id": "conversation-1"}}
result1 = graph_with_memory.invoke(
    {"messages": [{"role": "user", "content": "我叫小明"}]},
    config=config,
)
print(result1["messages"][-1].content)

# 第二轮对话 — Agent 会记住你的名字
result2 = graph_with_memory.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config=config,
)
print(result2["messages"][-1].content)  # "你叫小明"

[!tip] LangGraph vs AgentExecutor
简单任务：使用 create_react_agent 或 AgentExecutor 即可满足需求
复杂任务：需要多步决策、条件分支、人类审批、持久化状态时，选择 LangGraph
新项目：推荐直接使用 LangGraph，它是 LangChain 团队主推的 Agent 框架

多Agent协作

多 Agent 系统是指多个 Agent 各自承担不同角色，协同完成复杂任务的架构。

多 Agent 架构

使用 LangGraph 构建多 Agent 系统

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


# 1. 定义共享状态
class MultiAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str


# 2. 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)


# 3. 定义 Supervisor Agent
def supervisor_node(state: MultiAgentState) -> dict:
    """协调者：决定将任务分配给哪个 Agent。"""
    system_prompt = """你是一个任务协调者。根据用户的请求，决定应该由哪个 Agent 处理：
    - "researcher": 负责信息搜索、数据收集、事实核查
    - "coder": 负责代码编写、技术问题解答、程序调试
    - "writer": 负责内容创作、文章撰写、文本润色
    - "FINISH": 任务已完成，不需要进一步处理

    只需返回一个 Agent 名称或 FINISH。"""

    messages = [HumanMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)

    agent_name = response.content.strip().lower()
    if agent_name not in ["researcher", "coder", "writer", "finish"]:
        agent_name = "finish"

    return {"next_agent": agent_name, "messages": [response]}


# 4. 定义各个 Agent 节点
def researcher_node(state: MultiAgentState) -> dict:
    """研究员 Agent：负责信息搜索和分析。"""
    system_prompt = "你是一个专业的研究员。请根据用户的问题进行深入分析，提供全面、准确的信息。"
    messages = [HumanMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}


def coder_node(state: MultiAgentState) -> dict:
    """编程 Agent：负责代码相关的任务。"""
    system_prompt = "你是一个高级程序员。请根据用户的需求编写高质量代码，并提供详细解释。"
    messages = [HumanMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}


def writer_node(state: MultiAgentState) -> dict:
    """写作 Agent：负责内容创作。"""
    system_prompt = "你是一个专业的内容创作者。请根据提供的信息撰写清晰、有吸引力的内容。"
    messages = [HumanMessage(content=system_prompt)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}


# 5. 路由函数
def route_to_agent(state: MultiAgentState) -> str:
    next_agent = state["next_agent"]
    if next_agent == "finish":
        return END
    return next_agent


# 6. 构建多 Agent 图
builder = StateGraph(MultiAgentState)

# 添加节点
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("coder", coder_node)
builder.add_node("writer", writer_node)

# 添加边
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_agent)

# 各 Agent 执行完毕后回到 Supervisor
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
builder.add_edge("writer", "supervisor")

# 编译
multi_agent = builder.compile()


# 7. 执行
result = multi_agent.invoke({
    "messages": [HumanMessage(content="帮我写一篇关于 Python 异步编程的技术文章")],
    "next_agent": "",
})

多 Agent 模式对比

模式	描述	适用场景
Supervisor	一个中心 Agent 分配任务	任务类型明确可分类
串行流水线	Agent 按顺序依次处理	翻译 -> 校对 -> 润色
并行协作	多个 Agent 同时处理不同子任务	大型项目的并行研究
层级式	多层 Supervisor 嵌套	复杂的企业级任务
辩论式	多个 Agent 讨论并达成共识	需要多角度分析的问题

[!warning] 多 Agent 的权衡
多 Agent 系统虽然功能强大，但也意味着更高的 Token 消耗和更长的响应时间。不要为了使用多 Agent 而使用多 Agent — 简单任务用单一 Agent + 多工具往往更高效。

实战：构建研究助手Agent

下面是一个完整的研究助手 Agent 示例，它能够搜索互联网、读取文档并生成研究报告。

完整代码

"""
研究助手 Agent
功能：搜索互联网 + 读取文档 + 总结生成研究报告
"""

import os
from typing import Annotated
from typing_extensions import TypedDict

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver


# ============================================================
# 1. 定义工具
# ============================================================

search_tool = TavilySearchResults(max_results=5, search_depth="advanced")


@tool
def summarize_text(text: str, max_words: int = 200) -> str:
    """对给定的文本进行摘要总结。

    Args:
        text: 需要总结的文本内容
        max_words: 摘要的最大字数，默认200字
    """
    # 在实际项目中，可以使用 LLM 来生成摘要
    # 这里简化为截取前 max_words 个字符
    if len(text) <= max_words:
        return text
    return text[:max_words] + "..."


@tool
def format_report(title: str, sections: str) -> str:
    """将研究内容格式化为结构化的研究报告。

    Args:
        title: 报告标题
        sections: 报告内容，使用 Markdown 格式
    """
    report = f"""
# {title}

---
{sections}

---
*本报告由 AI 研究助手自动生成*
"""
    return report.strip()


tools = [search_tool, summarize_text, format_report]


# ============================================================
# 2. 定义 Agent 状态
# ============================================================

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


# ============================================================
# 3. 构建图
# ============================================================

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

SYSTEM_PROMPT = """你是一个专业的研究助手。你的任务是帮助用户研究特定主题。

工作流程：
1. 分析用户的研究需求
2. 使用 search 工具搜索相关信息
3. 如果搜索结果过长，使用 summarize_text 工具进行摘要
4. 整理所有收集到的信息
5. 使用 format_report 工具生成结构化的研究报告

注意事项：
- 从多个角度搜索信息，确保全面性
- 对关键数据进行交叉验证
- 报告结构清晰，包含引言、正文和结论
- 使用中文输出
"""


def research_agent(state: ResearchState):
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


# 构建图
builder = StateGraph(ResearchState)
builder.add_node("agent", research_agent)
builder.add_node("tools", ToolNode(tools=tools))

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

# 添加记忆
memory = MemorySaver()
research_agent_graph = builder.compile(checkpointer=memory)


# ============================================================
# 4. 运行研究助手
# ============================================================

def run_research(query: str, thread_id: str = "default"):
    """运行研究助手。"""
    config = {"configurable": {"thread_id": thread_id}}

    print(f"研究问题: {query}")
    print("=" * 60)

    result = research_agent_graph.invoke(
        {"messages": [{"role": "user", "content": query}]},
        config=config,
    )

    # 输出最终结果
    final_message = result["messages"][-1]
    print("\n" + "=" * 60)
    print("研究结果:")
    print("=" * 60)
    print(final_message.content)

    return result


# 运行示例
if __name__ == "__main__":
    run_research(
        "请研究 LangGraph 和 CrewAI 的区别，各自的优势和适用场景，"
        "并生成一份对比分析报告。"
    )

流式输出增强版

def stream_research(query: str, thread_id: str = "default"):
    """流式输出研究助手的执行过程。"""
    config = {"configurable": {"thread_id": thread_id}}

    print(f"[用户] {query}\n")

    for event in research_agent_graph.stream(
        {"messages": [{"role": "user", "content": query}]},
        config=config,
        stream_mode="values",
    ):
        if "messages" in event:
            last_msg = event["messages"][-1]

            # 工具调用
            if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
                for tc in last_msg.tool_calls:
                    print(f"  [Agent 思考] 需要调用工具: {tc['name']}")
                    print(f"  [工具输入] {tc['args']}")

            # 工具结果
            elif last_msg.type == "tool":
                print(f"  [工具结果] {last_msg.content[:100]}...")

            # 最终回答
            elif last_msg.type == "ai" and not getattr(last_msg, "tool_calls", []):
                if last_msg.content:
                    print(f"\n[最终回答]\n{last_msg.content}")

    print("\n" + "-" * 60)


# 运行
stream_research("2024年大语言模型的最新发展有哪些？")

Agent 最佳实践

何时使用 Agent vs Chain

场景	推荐方案	原因
固定格式的文档翻译	Chain	流程固定，无需决策
RAG 问答系统	Chain	检索 -> 生成流程固定
客服聊天机器人	Agent	需根据问题查询不同数据源
数据分析助手	Agent	需自主选择分析工具
代码生成助手	Agent	需根据需求选择不同工具
多轮研究任务	LangGraph Agent	需要状态管理和复杂流程

调试技巧

# 技巧1: 开启 verbose 模式
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印每一步的详细信息
)

# 技巧2: 使用 LangSmith 追踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

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

for step in result["intermediate_steps"]:
    action, observation = step
    print(f"工具: {action.tool}")
    print(f"输入: {action.tool_input}")
    print(f"输出: {observation[:200]}")
    print("---")

# 技巧4: LangGraph 回调
from langchain_core.callbacks import BaseCallbackHandler


class DebugCallback(BaseCallbackHandler):
    """自定义回调，打印 Agent 的每一步操作。"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM 调用] prompts: {prompts[0][:100]}...")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"[工具调用] {serialized.get('name')}({input_str})")

    def on_tool_end(self, output, **kwargs):
        print(f"[工具结果] {str(output)[:100]}...")


# 使用回调
result = agent_executor.invoke(
    {"input": "测试问题"},
    config={"callbacks": [DebugCallback()]},
)

成本控制

# 1. 使用更便宜的模型处理简单任务
from langchain_openai import ChatOpenAI

# 复杂决策用强模型
decision_llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 简单的文本处理用轻量模型
light_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 2. 限制最大迭代次数
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 限制最多5轮
    max_execution_time=30,  # 最多30秒
)

# 3. 缓存工具结果，避免重复调用
from functools import lru_cache


@tool
@lru_cache(maxsize=100)
def cached_search(query: str) -> str:
    """搜索信息（结果会被缓存）。"""
    search = TavilySearchResults(max_results=3)
    results = search.invoke(query)
    return str(results)

# 4. Token 预算控制
def estimate_tokens(text: str) -> int:
    """粗略估算 Token 数量（中文约1.5字/token）。"""
    return int(len(text) * 1.5)


MAX_TOKEN_BUDGET = 4000  # 单次请求最大 Token 预算

# 5. 选择合适的工具数量
# 工具过多会增加 LLM 的决策成本，建议控制在 3-8 个

Prompt 设计原则

# 好的 Agent System Prompt 示例
good_prompt = """你是一个专业的数据分析助手。

## 你的能力
- 查询数据库获取数据
- 执行数学计算
- 搜索互联网获取最新信息

## 工作流程
1. 理解用户需求
2. 判断需要哪些工具
3. 按需调用工具（每次只调用一个工具）
4. 基于工具结果回答问题

## 限制
- 如果不确定，请诚实说明
- 不要编造数据
- 数据来源需要标注

## 输出格式
使用中文回答，结构清晰，包含数据来源。
"""

# 差的 Prompt 示例 — 过于模糊
bad_prompt = "你是一个助手，可以帮助用户。"

[!tip] Agent 开发清单
明确定义每个工具的功能和参数（docstring 要清晰）
在 System Prompt 中说明 Agent 的能力和限制
设置合理的 max_iterations 防止无限循环
为每个工具添加错误处理和超时控制
使用 LangSmith 追踪和调试 Agent 行为
在生产环境中进行充分的边界测试
考虑添加人类审批（Human-in-the-loop）机制

常见问题与解决方案

问题	原因	解决方案
Agent 无限循环	工具结果未满足 LLM 的判断条件	设置 `max_iterations`，优化工具描述
Agent 不调用工具	Prompt 未明确指示何时使用工具	在 System Prompt 中说明工具使用场景
Agent 选错工具	工具描述不清晰，工具功能重叠	优化 docstring，减少工具数量
Token 消耗过高	多轮对话累积，工具返回过多内容	截断中间结果，控制工具输出长度
响应时间过长	工具执行慢，迭代次数多	优化工具性能，减少迭代次数
幻觉（编造数据）	LLM 在信息不足时自行编造	强调诚实原则，添加信息不足检测

本章小结

本章详细介绍了 LangChain Agent 智能代理与工具调用的核心知识：

Agent 概念 — Agent 是 LLM + 工具 + 自主决策的组合体，适用于步骤不确定的复杂任务
ReAct 模式 — Agent 通过"观察 - 思考 - 行动"的循环自主完成任务
工具定义 — 三种方式：@tool 装饰器、StructuredTool、BaseTool，优先使用装饰器
内置工具 — Tavily Search、Wikipedia、Python REPL 等，开箱即用
自定义工具 — 计算器、数据库查询、文件读取等实际业务工具的开发方法
create_react_agent — 高层 API，快速创建 ReAct Agent
AgentExecutor — Agent 的运行引擎，控制迭代次数、错误处理、超时等
LangGraph — 新一代 Agent 框架，支持复杂工作流、状态管理、人类在环
多Agent协作 — Supervisor 模式、串行流水线、并行协作等多种协作方式
最佳实践 — 合理选择 Agent vs Chain、调试技巧、成本控制、安全原则

[!tip] 学习建议
从简单的 create_react_agent 开始，逐步过渡到 LangGraph
先用 2-3 个工具构建基础 Agent，再逐步扩展
大量使用 LangSmith 进行调试和优化
在生产环境部署前，务必进行充分的安全评估和边界测试