LangChain Agent 完全指南：从 Tools 到 ReAct 循环的工程实践

发表于最近更新于最近

作者 KennethCheng

339~435 分钟 阅读

智能体 (Agent) = 大语言模型 + 工具 + 系统提示，让 LLM 拥有"手脚"和"行为准则"，能自主规划、调用工具、迭代求解复杂任务。

本文基于 LangChain v1.0，系统梳理 Agent 的核心概念、Tools 定义、Agent 构建、系统提示、结构化输出与多种调用方式，配套可直接拷贝运行的代码示例。

1. Agent 是什么

智能体 (Agent) 是将语言模型与工具相结合，创建可以推理任务、决定使用哪些工具并迭代寻求解决方案的智能系统。

1.1 关键要素拆解

要素	含义
语言模型	智能体的"大脑"，负责推理与决策
工具	智能体的"手脚"，负责与外部世界交互
推理任务	理解目标、拆解步骤、规划路径
决定使用哪些工具	根据任务动态选择最合适的工具
迭代寻求解决方案	多轮循环（观察 → 思考 → 行动 → 反思）直至完成

1.2 与纯 LLM 的区别

纯 LLM：仅依赖模型内部知识生成回答
Agent：在 LLM 基础上，自主选择并调用工具，可与外部数据 / API / 环境互动，并能多轮迭代

1.3 ReAct 工作循环

典型工作流：

Thought（思考）：分析当前任务，推理下一步行动
Action（行动）：选择并调用合适的工具
Observation（观察）：获取工具返回结果
Reflection（反思）：判断是否达成目标，必要时回到第 1 步继续循环

2. Tools：让 LLM 拥有"手脚"

Tools 是 model 与外部世界交互的接口，扩展模型能力边界，使大语言模型能够执行用户定义的动作，例如搜索网页、执行代码、访问数据库或调用其他服务。

2.1 工具的定义

from langchain_core.tools import tool

@tool
def calculate(expression: str) -> str:
    """Perform mathematical calculations and return the result.
    Args:
        expression: Mathematical expression to evaluate
        (e.g., "2 + 3 * 4", "sqrt(16)", "sin(pi/2)")
    Returns:
        The calculated result as a string
    """
    result = str(eval(expression))
    return result

三大要点：

利用 @tool 装饰器将函数转化为工具
入参类型要明确标注
工具描述（docstring）会帮助 model 理解何时 / 如何使用该工具

2.2 `@tool` 装饰器的属性自定义

@tool("calculator", description="执行算术计算。用于解决任何数学问题。")
def calculate(expression: str) -> str:
    """评估数学表达式。"""
    return str(eval(expression))

用 Pydantic 定义 args_schema：

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    """天气查询的输入参数。"""
    location: str = Field(description="城市名称或坐标")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位偏好"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> str:
    """获取当前天气和可选预报。"""
    temp = 22 if units == "celsius" else 72
    return f"{location}当前天气: {temp}度{units[0].upper()}"

2.3 StructuredTool 类的关键属性

通过 @tool 装饰器装饰，会将函数封装为结构化工具（StructuredTool 类）。

属性	类型	说明
`name`	`str`	工具的独特名称，命名应当直观且方便模型理解
`description`	`str`	告诉模型如何 / 何时 / 为何使用工具，可加入 few-shot 示例
`response_format`	`["content", "content_and_artifact"]`	工具的响应格式，默认为 `"content"`

方法	用途
`invoke` / `ainvoke`	直接对工具进行调用（同步 / 异步）
`get_input_schema`	获取工具的输入策略
`get_output_schema`	获取可用于工具已定义的输出 Pydantic 模型

2.4 多种工具类型

# 本地数据浏览
import csv, json
from typing import List, Dict

@tool
def search_csv(query: str, csv_path: str, limit: int = 10) -> str:
    """在本地 csv 文件中按关键词简单搜索，返回前 limit 条匹配记录（JSON 字符串）。"""
    query_lower = query.lower()
    results: List[Dict[str, str]] = []
    with open(csv_path, "r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            if any(query_lower in str(value).lower() for value in row.values()):
                results.append(row)
                if len(results) >= limit:
                    break
    return json.dumps(results, ensure_ascii=False, indent=2)

# 网络搜索 API
@tool
def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = False,
):
    """使用 Tavily 搜索引擎在互联网上搜索信息"""
    return tavily_client.search(
        query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )

tools = [internet_search, calculate, get_weather]

2.5 7 步工具集成流程

步骤	关键点
1. 函数 → 工具	`@tool` 装饰 + 类型注解 + 文档字符串
2. 命名描述	可选 `name=` / `description=`，动词开头
3. 参数模式	自动推导或 `args_schema=Pydantic`
4. 返回格式	默认 `content`；需额外数据用 `content_and_artifact`
5. 绑定模型	`model.bind_tools([tools])`
6. 触发调用	`invoke` 后查看 `ai_msg.tool_calls`
7. 解析请求	取 `name` / `args` / `id`，多条调用按 `id` 对应

3. 创建你的第一个 Agent

3.1 准备模型

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "openai:gpt-4.1",
    temperature=0.7,
    max_tokens=1000,
    timeout=30,
    max_retries=2,
    api_key="sk-......",
    base_url="https://api.deepseek.com/v1",  # 走 OpenAI 兼容协议
)

3.2 两种静态创建方法

from langchain.agents import create_agent

# 方式一：使用模型标识符字符串
agent = create_agent(
    "openai:gpt-4",
    tools=tools,
)

# 方式二：使用模型实例（推荐）
model = init_chat_model("openai:gpt-4o-mini")
agent = create_agent(model, tools=tools)

3.3 调用 Agent

res = agent.invoke({
    "messages": [
        {"role": "user",
         "content": "deepseek在2024-2025年之间,用户增长了百分多少??"}
    ]
})

消息轨迹（典型 ReAct 流程）：

[
    # 1. 用户提问
    HumanMessage(content="deepseek在2024-2025年之间,用户增长了百分多少??"),

    # 2. 模型决定调用搜索工具
    AIMessage(content="", additional_kwargs={'tool_calls': [{
        'id': 'call_niDwiOVimm1zkCcsG2DoDxQD',
        'function': {'arguments': '{"query": "deepseek user growth 2024 to 2025"}',
                     'name': 'internet_search'}
    }]}),

    # 3. 搜索工具返回结果
    ToolMessage(content='{"query": "...", "results": [...]}',
                name='internet_search',
                tool_call_id='call_niDwiOVimm1zkCcsG2DoDxQD'),

    # 4. 模型决定调用计算工具
    AIMessage(content="", additional_kwargs={'tool_calls': [{
        'id': 'call_lMtSVMcYCelvsFqzPZe348So',
        'function': {'arguments': '{"expression": "(125-61)/61*100"}',
                     'name': 'calculate'}
    }]}),

    # 5. 计算工具返回结果
    ToolMessage(content='104.91803278688525',
                name='calculate',
                tool_call_id='call_lMtSVMcYCelvsFqzPZe348So'),

    # 6. 模型总结答复
    AIMessage(content="在2024年到2025年之间，DeepSeek的用户增长了约104.92%。")
]

执行流程图：

4. 动态模型与中间件

动态模型在运行时根据当前状态和上下文进行选择，这使得复杂的路由逻辑和成本优化成为可能。

4.1 利用中间件定义动态模型 Agent

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.chat_models import init_chat_model

# 1️⃣ 定义两个备选模型
basic_model = init_chat_model("openai:gpt-4o-mini")
advanced_model = init_chat_model("openai:gpt-4o")

# 2️⃣ 定义模型选择逻辑
@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
    """Choose model based on conversation complexity."""
    message_count = len(request.state["messages"])

    if message_count > 5:
        # 使用高级模型进行更长时间的对话
        model = advanced_model
    else:
        model = basic_model

    request.model = model
    return handler(request)

# 3️⃣ 创建 Agent
agent = create_agent(
    model=basic_model,   # Default model
    tools=tools,
    middleware=[dynamic_model_selection]
)

4.2 调用效果

# 短消息（≤5 条）→ 走 gpt-4o-mini
agent.invoke({"messages": [{"role": "user", "content": "你好"}]})
# AIMessage(content='你好！有什么我可以帮助你的吗？', 
#           ...'model_name': 'gpt-4o-mini-2024-07-18', ...)

# 长消息（>5 条）→ 走 gpt-4o
res["messages"].append({"role": "user", 
                        "content": "你认为接下来deepseek的发展前景如何会如何??"})
agent.invoke(res)
# AIMessage(content='DeepSeek的未来发展前景在业内被…', 
#           ...'model_name': 'gpt-4o', …)

4.3 适用场景

成本优化：简单任务用小模型，复杂任务用大模型
路由分发：按主题 / 语言 / 用户等级路由到不同模型
A/B 测试：不同用户走不同模型，对比效果
降级兜底：主模型不可用时切到备用模型

5. 系统提示：静态与动态

我们可以通过提供系统提示来确定 Agent 处理任务的方式。system_prompt 参数可以以字符串形式提供。

5.1 静态系统提示

agent = create_agent(
    model,
    tools,
    system_prompt="你是我的得力助手。要做到言简意赅，准确无误的回答我的问题。"
)

💡 字符串形式直接传入即可，适合固定不变的角色设定。

5.2 动态系统提示词

让系统提示根据上下文动态生成，从而适配不同的用户角色、场景等。

from typing import TypedDict
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest


# 1️⃣ 定义上下文结构
class Context(TypedDict):
    user_role: str


# 2️⃣ 编写动态提示函数
@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """Generate system prompt based on user role."""
    user_role = request.runtime.context.get("user_role", "user")
    base_prompt = "你是我的得力助手。"

    if user_role == "expert":
        return f"{base_prompt} 提供详细的技术答复。"
    elif user_role == "beginner":
        return f"{base_prompt} 简单解释概念，避免行话。"

    return base_prompt


# 3️⃣ 创建 Agent（绑定 context_schema）
agent = create_agent(
    model="openai:gpt-4o",
    tools=tools,
    middleware=[user_role_prompt],
    context_schema=Context
)

# 4️⃣ 调用时显式传入 context
result = agent.invoke(
    {"messages": [{"role": "user", "content": "为我解释机器学习这个概念。"}]},
    context={"user_role": "expert"}   # 👈 关键
)

5.3 核心机制

5.4 静态 vs 动态对比

维度	静态 `system_prompt`	动态 `@dynamic_prompt`
写法	字符串直接传参	装饰器 + 函数返回字符串
变化性	创建时固定	每次调用按 context 重新生成
上下文感知	❌	✅（`request.runtime.context`）
上下文传递	无	`context_schema=Context` + `invoke(context=...)`
适用场景	角色固定	多角色 / 多场景 / 多语言

6. 结构化输出

在某些情况下，我们可能希望 Agent 以特定格式来返回输出。LangChain 通过 response_format 参数提供了结构化输出的配置。ToolStrategy 使用人工工具调用来生成结构化输出，适用于任何支持工具调用的模型。

6.1 定义输出格式

from pydantic import BaseModel
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

# 1️⃣ 用 Pydantic 定义期望的输出结构
class ContactInfo(BaseModel):
    name: str
    email: str
    phone: str

# 2️⃣ 在创建 Agent 时通过 response_format 注入
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=tools,
    response_format=ToolStrategy(ContactInfo)   # 👈 关键
)

6.2 调用并取出结果

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": """从以下内容中提取信息：大家都觉得非常靠谱的小明，
        他的具体联系渠道我这里可以给你。他总是能及时回复邮件，
        所以如果你有文件或详细问题，发邮件到 xiaoming@gmail.com 这个邮箱地址。
        当然，如果事情比较紧急，或者需要实时语音沟通确认细节，
        你也可以直接拨打他的个人办公电话号码是 (04xx) 123-4567，
        这个号码通常在工作日的办公时间内都能接通"""}
    ]
})

print(result["structured_response"])
# ContactInfo(
#     name='小明',
#     email='xiaoming@gmail.com',
#     phone='(04xx) 123-4567'
# )

6.3 `ToolStrategy` 名字由来

Tool：借助"工具调用"机制让模型按 schema 输出
Strategy：它是一种输出策略（区别于直接 prompt 让模型生成 JSON）
优点：不依赖模型原生结构化输出能力 → 任何支持 tool calling 的模型都能用

7. 多种调用方式

所有 Agent 都有自身状态 (state)，其中包含一连串信息；对 Agent 的调用，实际上是在"状态"中增加新的信息，并更新代理状态。除了基本的 invoke 方法，与 model 类似，我们还可以通过其他方法对 agent 进行调用以满足不同需求。

7.1 同步调用 `invoke`

response = model.invoke("Why do parrots have colorful feathers?")

7.2 批量调用 `batch`

responses = model.batch([
    "为什么鹦鹉有彩色羽毛？",
    "飞机是如何飞行的？",
    "什么是量子计算？"
])

7.3 流式调用 `stream`

for chunk in model.stream("为什么鹦鹉有彩色羽毛？"):
    print(chunk.content, end="", flush=True)

# 构建完整的流式响应消息
full_response = None
for chunk in model.stream("天空是什么颜色的?"):
    full_response = chunk if full_response is None else full_response + chunk
print(full_response.content)

7.4 异步调用 `ainvoke` / `astream`

async def async_invoke():
    response = await model.ainvoke("为什么鹦鹉会模仿人类说话？")
    return response

async def async_stream():
    async for chunk in model.astream("解释机器学习"):
        print(chunk.content, end="", flush=True)

7.5 Agent 的流式调用

for chunk in agent.stream(
    {"messages": [{"role": "user", 
                   "content": "deepseek在2024-2025年之间,用户增长了百分之多少??"}]},
    stream_mode="values",   # 每个 chunk 是"到目前为止的完整 state"
):
    latest_message = chunk["messages"][-1]

    if latest_message.content:
        print(f"Agent: {latest_message.content}")
    elif latest_message.tool_calls:
        print(f"Calling tools: {[tc['name'] for tc in latest_message.tool_calls]}")

7.6 调用方式速查表

调用方式	适用场景	性能特点	用户体验
同步调用	简单应用、测试调试、低并发	响应时间中等，资源占用低	需要等待完整响应
批量调用	数据处理、报表生成、批量任务	高吞吐量，适合密集任务	延迟较高，适合后台任务
流式调用	实时对话、长文本生成、演示	首字延迟低，渐进式输出	体验最佳，过程透明
异步调用	Web 服务、高并发应用、实时系统	高并发，资源利用率高	取决于具体实现

方法	同步/异步	用途
`invoke`	同步	单次调用，等待完整结果
`ainvoke`	异步	Web 服务 / 高并发
`stream`	同步流式	实时观察 AI 思考与工具调用
`astream`	异步流式	异步环境 + 流式体验
`batch`	同步批量	离线批量任务
`abatch`	异步批量	高并发批量任务

7.7 多轮对话：复用 state 追加新消息

# 第一次调用
res = agent.invoke({"messages": [{"role": "user", "content": "你好"}]})

# 追加新问题（保留上下文）
res["messages"].append({
    "role": "user",
    "content": "你认为接下来deepseek的发展前景如何会如何??"
})
res = agent.invoke(res)

💡 关键点：传入的 res 同时也是"新的输入"，Agent 会自动追加消息并更新 state。

8. 总结速记

8.1 Agent 一句话公式

Agent = create_agent(model, tools, middleware, system_prompt, response_format)
调用 = invoke / ainvoke / stream / astream / batch

8.2 5 大主题回顾

主题	关键 API	一句话
基本创建	`create_agent(model, tools)`	静态绑定模型与工具
动态模型	`wrap_model_call` + `middleware=[...]`	按消息数 / 复杂度路由模型
系统提示	`system_prompt` 或 `@dynamic_prompt`	静态 / 动态角色设定
结构化输出	`response_format=ToolStrategy(Schema)`	让模型按 Pydantic 字段输出
多种调用	`invoke` / `stream` / `ainvoke` ...	同步 / 异步 / 流式 / 批量

8.3 5 大主题思维导图

8.4 选型决策树

参考资料

如果本文对你有帮助，欢迎点赞 / 收藏 / 分享 💖

Ai, LangChain

LangChain Ai

许可协议: CC BY 4.0

目录