概要

基于官方文档进行学习，但是会加注释和必要解释帮助理解，只会python基础语法也完全可以看懂！本人亲测！（暂时没写完 ！）

1. 入门指南

入门指南将学习如何使用LangChain创建一个既能回答问题又能调用工具的智能代理（Agent）。

---

第一部分：构建基础代理

这个基础代理使用Claude Sonnet 4.5作为语言模型，配备一个天气查询工具，并通过系统提示指导其行为。

from langchain.agents import create_agent

# 1. 定义工具函数
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"It's always sunny in {city}!"

# 2. 创建代理
agent = create_agent(
    model="claude-sonnet-4.5-20250929",  # 使用的语言模型
    tools=[get_weather],                  # 可用的工具列表
    system_prompt="You are a helpful assistant",  # 系统提示
)

# 3. 运行代理
response = agent.invoke({
    "messages": [{"role": "user", "content": "what is the weather in sf"}]
})

关键点：

• create_agent() 是创建代理的核心函数
• 代理 = 模型 + 工具 + 提示
• invoke() 方法用于执行代理

---

第二部分：构建实用的天气预报代理

我们将构建一个更复杂的代理，演示生产环境中的关键概念：

1. 详细的系统提示 - 获得更好的代理行为
2. 与外部数据集成的工具 - 实际调用API或查询数据库
3. 模型配置 - 获得一致的响应
4. 结构化输出 - 可预测的结果格式
5. 对话记忆 - 支持多轮聊天交互

分步实现

第1步：定义系统提示

系统提示定义了代理的角色和行为。要具体且有操作性。

SYSTEM_PROMPT = """
You are an expert weather forecaster, who speaks in puns.

You have access to two tools:
- get_weather_for_location: 用于获取特定位置的天气
- get_user_location: 用于获取用户的位置

如果用户询问天气，请确保您知道位置。如果从问题中可以看出他们指的是自己所在的位置，请使用 get_user_location 工具查找他们的位置。
"""

第2步：创建工具

工具允许模型通过调用函数与外部系统交互。工具可以依赖运行时上下文。

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime

# 天气查询工具
@tool
def get_weather_for_location(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"It's always sunny in {city}!"

# 定义运行时上下文
@dataclass
class Context:
    """自定义运行时上下文模式"""
    user_id: str

# 用户位置查询工具（依赖上下文）
@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
    """根据用户ID检索用户信息"""
    user_id = runtime.context.user_id
    return "Florida" if user_id == "1" else "SF"

工具开发要点：

• 使用 @tool 装饰器添加元数据
• 工具名称、描述和参数名称都会成为模型提示的一部分
• ToolRuntime 参数启用运行时依赖注入

什么是dataclass？

• dataclass 是 Python 3.7+ 引入的一个装饰器，用于简化类的创建。它自动为类生成一些特殊方法，让你可以用更少的代码定义数据容器类。
• @dataclass 装饰器会自动为 Context 类生成：
  • init() - 构造函数，接收 user_id 参数
  • repr() - 可读的字符串表示
  • eq() -相等性比较方法
  • 还有其他方法如 ne(), lt(), le() 等（如果设置了参数）

什么是 ToolRuntime？

ToolRuntime 是 LangChain 中的一个依赖注入容器。它允许工具函数访问运行时环境中的信息，而不需要工具函数自己管理状态。
想象一下，你的工具函数需要知道当前是哪个用户在使用代理。你有两种选择：

方法1：传统参数传递（繁琐）：

def tool(user_id: str, session: str, token: str): ...

方法2：使用 ToolRuntime（优雅）：

//定义好tool
# 1. 定义数据类（有哪些属性）
@dataclass
class UserContext:
    user_id: str      # ← 包含这个
    session: str      # ← 包含这个  

@tool
# 2. 使用工具时
@tool
def my_tool(runtime: ToolRuntime[UserContext]) -> str:
    ctx = runtime.context  # ← 这里得到完整的 UserContext 对象
    # 可以访问所有属性：
    print(ctx.user_id)   # "123"
    print(ctx.session)   # "session_abc"
    
    return f"User {ctx.user_id} in session {ctx.session}"

• [Context] 告诉系统：这个工具需要 Context 类型的数据
• runtime.context 获取实际数据
• LangChain 负责创建和管理 runtime，你不用管

第3步：配置模型

为您的用例设置具有合适参数的语言模型。

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4.5-20250929",  # 模型名称
    temperature=0.5,                # 创造性控制（0-1）
    timeout=10,                     # 超时时间（秒）
    max_tokens=1000                 # 最大生成令牌数
)

参数说明：

• temperature：控制输出的随机性（值越高越有创造性）
• timeout：API调用超时时间
• max_tokens：限制响应长度

第4步：定义响应格式（可选）

如果需要代理以特定模式响应，可以定义结构化响应格式。

from dataclasses import dataclass

@dataclass
class ResponseFormat:
    """代理的响应模式"""
    punny_response: str           # 包含双关语的响应（必填）
    weather_conditions: str | None = None  # 天气信息（可选）

第5步：添加记忆

为代理添加记忆以在交互过程中保持状态，使其能记住之前的对话。

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()  # 内存检查点

注意： 生产环境中应使用将数据保存到数据库的持久化检查点。

第6步：创建并运行代理

# 1. 创建完整代理
agent = create_agent(
    model=model,                                    # 配置好的模型
    system_prompt=SYSTEM_PROMPT,                    # 系统提示
    tools=[get_user_location, get_weather_for_location],  # 工具列表
    context_schema=Context,                         # 上下文模式
    response_format=ResponseFormat,                 # 响应格式
    checkpointer=checkpointer                       # 记忆检查点
)

# 2. 配置对话线程
config = {"configurable": {"thread_id": "1"}}  # 对话的唯一标识符

# 3. 首次调用（包含用户上下文）
response = agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
    config=config,                                  # 对话配置
    context=Context(user_id="1")                    # 用户上下文
)

# 4. 查看结构化响应
print(response['structured_response'])
# 输出示例：
# ResponseFormat(
#     punny_response="Florida is still having a 'sun-derful' day!...",
#     weather_conditions="It's always sunny in Florida!"
# )

# 5. 继续对话（使用相同的thread_id）
response = agent.invoke(
    {"messages": [{"role": "user", "content": "thank you!"}]},
    config=config,                                  # 相同的对话ID
    context=Context(user_id="1")                    # 相同的用户上下文
)

print(response['structured_response'])

---

完整示例代码

from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from dataclasses import dataclass

# 1. 系统提示
SYSTEM_PROMPT = """You are an expert weather forecaster..."""

# 2. 工具定义
@tool
def get_weather_for_location(city: str) -> str:
    return f"It's always sunny in {city}!"

@dataclass
class Context:
    user_id: str

@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
    user_id = runtime.context.user_id
    return "Florida" if user_id == "1" else "SF"

# 3. 模型配置
model = init_chat_model(
    "claude-sonnet-4.5-20250929",
    temperature=0.5,
    timeout=10,
    max_tokens=1000
)

# 4. 响应格式
@dataclass
class ResponseFormat:
    punny_response: str
    weather_conditions: str | None = None

# 5. 记忆
checkpointer = InMemorySaver()

# 6. 创建代理
agent = create_agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
    tools=[get_user_location, get_weather_for_location],
    context_schema=Context,
    response_format=ResponseFormat,
    checkpointer=checkpointer
)

# 7. 运行代理
config = {"configurable": {"thread_id": "1"}}

# 第一次调用
response1 = agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
    config=config,
    context=Context(user_id="1")
)

print("第一次响应:", response1['structured_response'])

# 第二次调用（继续对话）
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "thank you!"}]},
    config=config,
    context=Context(user_id="1")
)

print("第二次响应:", response2['structured_response'])

---

2. 智能体

概述

智能体（Agent）将语言模型与工具结合，创建能够对任务进行推理、决定使用哪些工具并迭代寻求解决方案的系统。智能体在循环中运行工具以实现目标，直到满足停止条件（模型输出最终结果或达到迭代次数限制）。

核心组件

1. 模型（Model）

模型是智能体的推理引擎，支持静态和动态模型选择。

静态模型

在创建智能体时配置一次，执行过程中保持不变。

方式1：使用模型标识符字符串

from langchain.agents import create_agent

agent = create_agent(
    "openai:gpt-5",  # 模型标识符
    tools=tools
)

模型标识符支持自动推断，如 "gpt-5" 会被推断为 "openai:gpt-5"。

方式2：直接使用模型实例（推荐）

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5",
    temperature=0.1,      # 控制创造性（0-1）
    max_tokens=1000,      # 最大生成令牌数
    timeout=30,           # 超时时间
    # ... 其他参数
)

agent = create_agent(model, tools=tools)

这种方式提供对配置的完全控制。

动态模型

在运行时根据状态和上下文选择模型，支持复杂路由逻辑和成本优化。

from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

# 创建两个模型实例
basic_model = ChatOpenAI(model="gpt-4o-mini")  # 轻量模型，成本低
advanced_model = ChatOpenAI(model="gpt-4o")     # 强大模型，性能好

# @wrap_model_call 是装饰器，表示这个函数要包装模型调用
@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
    """
    request: 模型请求对象，包含状态等信息
    handler: 实际的模型调用处理器
    -> ModelResponse: 返回模型响应
    """
    # 获取当前对话的消息数量
    message_count = len(request.state["messages"])
    
    # 根据消息数量决定使用哪个模型
    if message_count > 10:
        model = advanced_model  # 对话长，用强模型
    else:
        model = basic_model     # 对话短，用轻量模型
    
    # 修改请求中的模型
    request.model = model
    
    # 调用实际的模型处理器
    return handler(request)

# 创建智能体
agent = create_agent(
    model=basic_model,          # 默认模型（会被中间件覆盖）
    tools=tools,                # 工具列表（假设已定义）
    middleware=[dynamic_model_selection]  # 中间件列表
    # 执行时：用户输入 → middleware处理 → model推理 → 输出
)

注意：使用结构化输出时，不支持预绑定模型（已调用 bind_tools 的模型）。

2. 工具（Tools）

工具赋予智能体执行行动的能力，支持：

• 序列中的多个工具调用（由单个提示触发）
• 适当的并行工具调用
• 基于先前结果的动态工具选择
• 工具重试逻辑和错误处理
• 工具调用之间的状态持久化

定义工具

from langchain.tools import tool
from langchain.agents import create_agent

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"结果：{query}"

@tool
def get_weather(location: str) -> str:
    """获取位置的天气信息"""
    return f"{location} 的天气：晴朗，72°F"

agent = create_agent(model, tools=[search, get_weather])

如果提供空工具列表，智能体将仅包含 LLM 节点，不具备工具调用能力。

工具错误处理

自定义工具错误的处理方式：

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain_core.messages import ToolMessage

@wrap_tool_call
def handle_tool_errors(request, handler):
    """使用自定义消息处理工具执行错误"""
    try:
        return handler(request)
    except Exception as e:
        # 向模型返回自定义错误消息
        return ToolMessage(
            content=f"工具错误：请检查您的输入并重试。({str(e)})",
            tool_call_id=request.tool_call["id"]
        )

agent = create_agent(
    model="openai:gpt-4o",
    tools=[search, get_weather],
    middleware=[handle_tool_errors]
)

3. 系统提示（System Prompt）

通过提示塑造智能体的行为方式。

基本用法

agent = create_agent(
    model,
    tools,
    system_prompt="你是一个有帮助的助手。请简洁准确。"
)

动态系统提示

根据运行时上下文动态生成提示：

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

class Context(TypedDict):
    user_role: str

@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """根据用户角色生成系统提示"""
    user_role = request.runtime.context.get("user_role")
    base_prompt = "你是一个有帮助的助手。"
    
    if user_role == "expert":
        return f"{base_prompt} 提供详细的技术响应。"
    elif user_role == "beginner":
        return f"{base_prompt} 简单解释概念，避免使用行话。"
    
    return base_prompt

agent = create_agent(
    model="openai:gpt-4o",
    tools=[web_search],
    middleware=[user_role_prompt],
    context_schema=Context
)

# 根据上下文动态设置系统提示
result = agent.invoke(
    {"messages": [{"role": "user", "content": "解释机器学习"}]},
    context={"user_role": "expert"}
)

调用智能体

基本调用

result = agent.invoke(
    {"messages": [{"role": "user", "content": "旧金山天气如何？"}]}
)

智能体遵循 LangGraph Graph API，支持所有相关方法。

高级概念

结构化输出（Structured Output）

智能体可以特定格式返回输出，有两种策略：

1. ToolStrategy

使用人工工具调用生成结构化输出，适用于任何支持工具调用的模型：

# 导入必要的库
from pydantic import BaseModel  # Pydantic用于数据验证和设置
from langchain.agents import create_agent  # 创建智能体的主函数
from langchain.agents.structured_output import ToolStrategy  # 结构化输出策略：通过工具调用实现

# 第一步：定义数据结构模板
# BaseModel是Pydantic的基类，用于创建数据模型
class ContactInfo(BaseModel):
    # 定义三个必填字段，都是字符串类型
    name: str   # 姓名
    email: str  # 邮箱
    phone: str  # 电话
   
# 第二步：创建使用ToolStrategy的智能体
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search_tool],  
    response_format=ToolStrategy(ContactInfo)
)

# 第三步：调用智能体并获取结构化结果
result = agent.invoke({
    # 传入用户消息
    "messages": [{
        "role": "user", 
        "content": "提取联系信息：John Doe, john@example.com, (555) 123-4567"
    }]
})

# 第四步：获取结构化响应
result["structured_response"]
# 输出：ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')
# 这是一个ContactInfo对象，不是普通字符串

# 使用这个对象：
contact = result["structured_response"]
print(contact.name)   # "John Doe"
print(contact.email)  # "john@example.com"
print(contact.phone)  # "(555) 123-4567"

2. ProviderStrategy

使用模型提供商的原生结构化输出（response_format），更可靠但仅适用于支持的提供商：

from langchain.agents.structured_output import ProviderStrategy

agent = create_agent(
    model="openai:gpt-4o",  
    response_format=ProviderStrategy(ContactInfo)
)

注意：从 LangChain 1.0 开始，必须明确使用 ToolStrategy 或 ProviderStrategy，不再支持直接传递模式。

记忆（Memory）

智能体通过消息状态自动维护对话历史，还可以配置自定义状态模式。

通过中间件定义状态（推荐）

# 导入必要的类和模块
from langchain.agents import AgentState  # 智能体状态基类
from langchain.agents.middleware import AgentMiddleware  # 中间件基类

# 第一步：定义自定义状态类
# 继承 AgentState（所有智能体状态的基类）
class CustomState(AgentState):
    """
    自定义智能体状态
    智能体默认只有 messages 字段（对话历史）
    这里添加额外的 user_preferences 字段来存储用户偏好
    """
    user_preferences: dict  # 新增字段：存储用户偏好设置
    # AgentState 已经包含了 messages 字段
    # 所以 CustomState 实际有：messages + user_preferences 两个字段

# 第二步：创建自定义中间件
class CustomMiddleware(AgentMiddleware):
    """
    自定义中间件
    中间件可以在智能体执行的不同阶段插入自定义逻辑
    """
    
    # 指定这个中间件使用的状态模式
    state_schema = CustomState
    # 告诉系统："我这个中间件需要 CustomState 类型的状态"
    
    # 指定这个中间件管理的工具（可选）
    tools = [tool1, tool2]
    # 这些工具可以访问 CustomState 中的额外字段
    
    # 定义在模型调用之前执行的方法
    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        """
        在模型推理之前被调用
        state: 当前智能体状态（包含 messages 和 user_preferences）
        runtime: 运行时上下文
        返回值：可以返回一个字典来修改状态，或返回 None 表示不修改
        """
        # 示例：根据用户偏好修改系统提示
        if "user_preferences" in state:
            preferences = state["user_preferences"]
            if preferences.get("style") == "technical":
                # 如果是技术型用户，可以在这里添加技术性提示
                # 例如修改消息或添加技术细节
                pass
        
        # 如果不修改状态，返回 None
        return None
    
    # 还可以定义其他钩子方法：
    # after_model() - 模型调用后执行
    # before_tool() - 工具调用前执行
    # after_tool() - 工具调用后执行

# 第三步：创建带有自定义中间件的智能体
agent = create_agent(
    model,  # 模型实例
    tools=tools,  # 智能体的主要工具列表
    middleware=[CustomMiddleware()]  # 添加自定义中间件
    # 执行顺序：用户输入 → middleware处理 → 模型推理 → 输出
)

# 第四步：调用智能体并传入自定义状态
result = agent.invoke({
    # 必需：消息历史（智能体默认状态）
    "messages": [{
        "role": "user", 
        "content": "我更喜欢技术性解释"
    }],
    
    # 自定义状态字段（CustomState 中定义的）
    "user_preferences": {
        "style": "technical",      # 偏好技术性解释
        "verbosity": "detailed",   # 偏好详细解释
        # 可以添加更多自定义字段
        "language": "zh",         # 语言偏好
        "expertise_level": "advanced"  # 专业水平
    }
    
    # 状态传递机制：
    # 1. 这里传入的字典会被转换为 CustomState 对象
    # 2. 中间件可以读取和修改这些状态
    # 3. 状态会在整个对话过程中保持
})

# 第五步：后续调用保持状态
# 下一次调用时，状态会自动包含之前的 user_preferences
result2 = agent.invoke({
    "messages": [{
        "role": "user", 
        "content": "请解释神经网络"
    }]
    # 不需要再传 user_preferences，智能体会记住
    # 中间件会根据已有的偏好调整解释风格
})

通过 state_schema 定义状态

from langchain.agents import AgentState

class CustomState(AgentState):
    user_preferences: dict

agent = create_agent(
    model,
    tools=[tool1, tool2],
    state_schema=CustomState
)

注意：从 LangChain 1.0 开始，自定义状态必须是 TypedDict 类型，不再支持 Pydantic 模型和数据类。

流式传输（Streaming）

显示智能体执行的中间进度：

for chunk in agent.stream({
    "messages": [{"role": "user", "content": "搜索 AI 新闻并总结发现"}]
}, stream_mode="values"):
    # 每个块包含该时间点的完整状态
    latest_message = chunk["messages"][-1]
    
    if latest_message.content:
        print(f"智能体：{latest_message.content}")
    elif latest_message.tool_calls:
        print(f"正在调用工具：{[tc['name'] for tc in latest_message.tool_calls]}")

中间件（Middleware）

中间件为自定义智能体行为提供强大的扩展性，可以在执行的不同阶段拦截和修改数据流。

中间件功能：

• 在调用模型之前处理状态（消息裁剪、上下文注入）
• 修改或验证模型的响应（防护栏、内容过滤）
• 使用自定义逻辑处理工具执行错误
• 基于状态或上下文实现动态模型选择
• 添加自定义日志、监控或分析

常用装饰器：

• @before_model：模型调用前执行
• @after_model：模型调用后执行
• @wrap_tool_call：包装工具调用
• @dynamic_prompt：动态生成提示
• @wrap_model_call：包装模型调用

示例：自定义日志中间件

from langchain.agents.middleware import before_model

@before_model
def log_request(state, runtime):
    """记录模型请求"""
    print(f"请求模型，消息数：{len(state.get('messages', []))}")
    print(f"当前状态：{state}")
    return None  # 不修改状态

agent = create_agent(
    model="openai:gpt-4o",
    tools=tools,
    middleware=[log_request]
)