第一篇，以翻译文档为主
系列文章：

• 构建基于 LangChain v1.0 的RAG语义搜索引擎(二)
• langchain V1.0使用指南 (一)
• langGraph V1.0使用指南 (三)

---

---

1 langchain v1.0 初始化

1.1 quickstart

GITHUB：langchain
文档：quickstart

python版本尽量高于等于py3.11

1.2 大模型qwen的使用

详情可以看依赖文档：
langchain-qwq

pip install -qU langchain-qwq

1.2.1 主要功能

langchain-qwq是一个Python包，提供了LangChain与阿里巴巴云BaiLian平台Qwen系列模型的无缝集成。它兼容OpenAI的API规范，支持多种模型及丰富功能。

• QwQ模型集成：全面支持QwQ模型，具备高级推理能力
• Qwen3模型集成：支持Qwen3系列，含混合推理模式
• 其他Qwen模型兼容：支持Qwen-Max、Qwen2.5等多种Qwen系列模型
• 视觉模型支持：原生支持Qwen-VL视觉语言模型
• 流式输出：支持同步与异步流式响应
• 工具调用：支持函数调用及并行执行
• 结构化输出：支持JSON模式和函数调用生成结构化结果
• 推理内容访问：可直接访问模型内部推理和思考内容

---

1.2.2. 安装与环境配置

pip install -U langchain-qwq

如果是从源码克隆后想安装额外依赖，可以选择：

pip install -U langchain-qwq[docs]
pip install -U langchain-qwq[test]
pip install -U langchain-qwq[codespell]
pip install -U langchain-qwq[lint]
pip install -U langchain-qwq[typing]

环境变量配置

• DASHSCOPE_API_KEY：必填，DashScope API密钥，用于身份认证
• DASHSCOPE_API_BASE：可选，API基础URL，默认国际端点为
  https://dashscope-intl.aliyuncs.com/compatible-mode/v1

注意：国内用户建议配置为国内端点，避免访问国际默认地址，提升速度和稳定性。

这里配置的方式只能是：

from langchain_qwq import ChatQwQ
apikey = 'sk-xxxxxxxxxx'
base_url='https://dashscope.aliyuncs.com/compatible-mode/v1'

import os
os.environ["DASHSCOPE_API_KEY"] = apikey
os.environ["DASHSCOPE_API_BASE"] = base_url

1.2.3 ChatQwQ：QwQ与QvQ模型聊天接口

ChatQwQ类专门用于调用QwQ和QvQ模型，支持模型推理内容访问及流式输出。

from langchain_qwq import ChatQwQ

model = ChatQwQ(model="qwq-plus")
response = model.invoke("Hello, how are you?")
print(response.content)

代码解读：

• ChatQwQ初始化时指定模型名，如qwq-plus。
• invoke方法发送消息给模型，返回response对象。
• response.content是模型回复的文本内容。

注意事项：

• 确保环境变量已正确配置API密钥。
• 模型名称需正确，否则会报错。

1.2.3 访问模型推理内容

response = model.invoke("Hello")
content = response.content
reasoning = response.additional_kwargs.get("reasoning_content", "")
print(f"Response: {content}")
print(f"Reasoning: {reasoning}")

代码解读：

• 除了回复文本，additional_kwargs中包含模型的推理细节（reasoning_content）。
• 这对于理解模型“思考”过程很有帮助。

注意事项：

• 并非所有模型或调用都返回推理内容，需确认模型支持。

1.2.4 流式输出示例:同步流式

model = ChatQwQ(model="qwq-plus")

is_first = True
is_end = True

for msg in model.stream("Hello"):
    if hasattr(msg, 'additional_kwargs') and "reasoning_content" in msg.additional_kwargs:
        if is_first:
            print("Starting to think...")
            is_first = False
        print(msg.additional_kwargs["reasoning_content"], end="", flush=True)
    elif hasattr(msg, 'content') and msg.content:
        if is_end:
            print("\nThinking ended")
            is_end = False
        print(msg.content, end="", flush=True)

代码解读：

• stream方法返回一个消息生成器，逐步输出推理和回答内容。
• 通过is_first和is_end控制打印逻辑，区分推理阶段和回答阶段。

1.2.5 异步流式

is_first = True
is_end = True

async for msg in model.astream("Hello"):
    if hasattr(msg, 'additional_kwargs') and "reasoning_content" in msg.additional_kwargs:
        if is_first:
            print("Starting to think...")
            is_first = False
        print(msg.additional_kwargs["reasoning_content"], end="", flush=True)
    elif hasattr(msg, 'content') and msg.content:
        if is_end:
            print("\nThinking ended")
            is_end = False
        print(msg.content, end="", flush=True)

代码解读：

• astream支持异步迭代，适合异步环境。
• 逻辑与同步流式类似。

注意事项：

• 使用异步流式时，需在异步函数或环境中调用。

1.2.6 Content Blocks支持

from langchain_qwq import ChatQwen, ChatQwQ

model = ChatQwQ(model="qwq-plus")
print(model.invoke("Hello").content_blocks)

说明：

• Content Blocks是Qwen模型支持的多模态内容格式，支持文本、图片等。
• content_blocks返回结构化内容块。

1.2.7 工具调用示例

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get the weather for a city"""
    return f"The weather in {city} is sunny."

bound_model = model.bind_tools([get_weather])
response = bound_model.invoke("What's the weather in New York?")
print(response.tool_calls)

代码解读：

• 使用@tool装饰器定义工具函数。
• bind_tools将工具绑定到模型，实现模型调用外部函数。
• response.tool_calls返回工具调用详情。

1.2.8 结构化输出: JSON模式

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

struct_model = model.with_structured_output(User, method="json_mode")
response = struct_model.invoke("Hello, I'm John and I'm 25 years old")
print(response)  # User(name='John', age=25)

1.2.9 函数调用模式

struct_model = model.with_structured_output(User, method="function_calling")
response = struct_model.invoke("My name is Alice and I'm 30")
print(response)  # User(name='Alice', age=30)

代码解读：

• 利用pydantic定义数据模型。
• 通过with_structured_output指定结构化输出方式。
• 方便直接得到格式化的对象。

1.2.10 与LangChain Agents集成

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_qwq import ChatQwQ

@tool
def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is sunny."

model = ChatQwQ(model="qwq-plus")
agent = create_agent(model, tools=[get_weather])
print(agent.invoke({"messages": [HumanMessage("What is the weather like in New York?")]}))

说明：

• 结合LangChain代理框架，实现复杂对话和工具调用自动化。

1.2.11 QvQ模型示例（支持图片输入）

from langchain_core.messages import HumanMessage

messages = [
    HumanMessage(
        content_blocks=[
            {
                "type": "image",
                "url": "https://www.example.com/image.jpg",
            },
            {"type": "text", "text": "describe the image"},
        ]
    )
]

model = ChatQwQ(model="qvq-plus")
print(model.invoke(messages))

说明：

• QvQ模型支持多模态输入，如图片+文字，适合视觉问答场景。
• 这里示例图片链接为https://www.example.com/image.jpg。

---

1.2.12 Qwen3及其他Qwen系列模型支持

ChatQwen类针对Qwen3及其他Qwen系列，支持思考模式和更多高级参数。

from langchain_qwq import ChatQwen

model = ChatQwen(model="qwen3-235b-a22b-instruct-2507")
response = model.invoke("Hello")
print(response.content)

model = ChatQwen(model="qwen3-235b-a22b-thinking-2507")
response = model.invoke("Hello")
reasoning = response.additional_kwargs.get("reasoning_content", "")
print(f"Reasoning: {reasoning}")

1.2.13 思考模式控制（仅Qwen3支持）

该功能仅适用于Qwen3系列模型（除最新版本外），如Qwen3-235b-a22b-thinking-2507等。

• 禁用思考模式：

model = ChatQwen(model="qwen3-32b", enable_thinking=False)
response = model.invoke("Hello")
print(response.content)  # 无推理内容

• 启用思考模式：

model = ChatQwen(model="qwen-plus-latest", enable_thinking=True)
response = model.invoke("Hello")
reasoning = response.additional_kwargs.get("reasoning_content", "")
print(f"Reasoning: {reasoning}")

• 控制思考长度：

model = ChatQwen(model="qwen3-32b", thinking_budget=20)
response = model.invoke("Hello")
reasoning = response.additional_kwargs.get("reasoning_content", "")
print(f"Limited reasoning: {reasoning}")

1.2.14 其他Qwen模型示例：Qwen-Max

model = ChatQwen(model="qwen-max-latest")
print(model.invoke("Hello").content)

bound_model = model.bind_tools([get_weather])
response = bound_model.invoke("Weather in Shanghai and Beijing?", parallel_tool_calls=True)
print(response.tool_calls)

struct_model = model.with_structured_output(User, method="json_mode")
result = struct_model.invoke("I'm Bob, 28 years old")
print(result)

1.2.15 Qwen2.5-72B

model = ChatQwen(model="qwen2.5-72b-instruct")
print(model.invoke("Hello").content)

bound_model = model.bind_tools([get_weather])
struct_model = model.with_structured_output(User, method="json_mode")

1.2.16 Content Blocks支持

model = ChatQwen(model="qwen-plus-latest", enable_thinking=True)
print(model.invoke("Hello").content_blocks)

1.2.17 与LangChain Agents集成

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_qwq import ChatQwen

@tool
def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is sunny."

model = ChatQwen(model="qwen-plus-latest")
agent = create_agent(model, tools=[get_weather])
print(agent.invoke({"messages": [HumanMessage("查询New York的天气")]}))

1.2.17 视觉模型示例

from langchain_core.messages import HumanMessage
from langchain_qwq import ChatQwen

messages = [
    HumanMessage(
        content_blocks=[
            {
                "type": "image",
                "url": "https://www.example.com/image.jpg",
            },
            {"type": "text", "text": "描述图片内容"},
        ]
    )
]

model = ChatQwen(model="qwen3-vl-plus")
print(model.invoke(messages))

1.2.18 模型功能对比表

功能 / 模型类型	ChatQwQ	ChatQwen
QwQ模型	✅ 主要支持	❌ 不支持
QvQ模型	✅ 主要支持	❌ 不支持
Qwen3模型	✅ 基础支持	✅ 增强支持
其他Qwen模型	❌ 不支持	✅ 完全支持
视觉模型	❌ 不支持	✅ 支持
思考模式	❌ 不支持	✅ 支持（仅Qwen3）
思考预算	❌ 不支持	✅ 支持（仅Qwen3）

---

1.2.19 使用建议

• 对于QwQ和QvQ模型，推荐使用ChatQwQ，支持推理和流式输出。
• 对于Qwen3系列模型（仅阿里云BaiLian平台提供，支持深度思考模式），使用ChatQwen，调用时自动启用流式。
• 对于其他Qwen系列模型（自部署或第三方部署的Qwen3等），使用ChatQwen，流式调用需用户自行实现。

1.3 其他大模型的使用

文档：model-basic-usage

1.3.1 大模型基础使用与功能

import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

model = init_chat_model("gpt-4.1")
response = model.invoke("Why do parrots talk?")

其中init_chat_model有很多参数

参数	说明
model	指定具体模型名称或ID
api_key	认证密钥，通常注册后获得
temperature	控制输出随机性，值越高越有创造力，越低越确定
timeout	最大等待时间，单位秒
max_tokens	限制输出最大长度
max_retries	请求失败时最大重试次数

model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    # Kwargs passed to the model:
    temperature=0.7,
    timeout=30,
    max_tokens=1000,
)

除了文本生成，许多模型还支持以下功能：

功能	说明
工具调用	调用外部工具（如数据库查询或API调用），并在回答中使用结果。
结构化输出	模型的回答遵循预定义格式，方便后续处理。
多模态	处理并返回除文本外的数据，如图片、音频和视频。
推理	模型执行多步骤推理以得出结论。

1.3.2 模型调用方法:invoke（调用）

最直接的调用方式是用 invoke()，传入单条消息或消息列表，模型生成完整回答。

示例：

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

可以传入消息列表，模拟对话历史。每条消息有角色（role），帮助模型理解对话身份：

conversation = [
    {"role": "system", "content": "You are a helpful assistant that translates English to French."},
    {"role": "user", "content": "Translate: I love programming."},
    {"role": "assistant", "content": "J'adore la programmation."},
    {"role": "user", "content": "Translate: I love building applications."}
]

response = model.invoke(conversation)
print(response)  # AIMessage("J'adore créer des applications.")

也可以用 LangChain 提供的消息类写法：

from langchain.messages import HumanMessage, AIMessage, SystemMessage

conversation = [
    SystemMessage("You are a helpful assistant that translates English to French."),
    HumanMessage("Translate: I love programming."),
    AIMessage("J'adore la programmation."),
    HumanMessage("Translate: I love building applications.")
]

response = model.invoke(conversation)
print(response)  # AIMessage("J'adore créer des applications.")

• invoke() 是同步调用，返回完整回答。
• 消息列表支持多轮对话，角色区分系统、用户和助手。
• 适合简单、完整的请求-响应场景。
• 传入的消息格式要符合要求，角色字段不可缺失。
• 对话历史越长，模型上下文消耗越大。

1.3.3 stream（流式）

大多数模型支持流式输出，边生成边返回内容，提升用户体验，尤其是长回答。

调用 stream() 返回一个生成器，逐块产出内容：

full = None  # None | AIMessageChunk
for chunk in model.stream("What color is the sky?"):
    full = chunk if full is None else full + chunk
    print(full.text)

print(full.content_blocks)
# [{'type': 'text', 'text': 'The sky is typically blue...'}]

流式返回的是多个 AIMessageChunk 对象，代表部分文本。可以累加成完整消息。

• 适合需要即时反馈的场景。

• 每个 chunk 是部分文本，最终可合并为完整回答。

• content_blocks 支持多模态内容。

• 流式处理需要异步或循环逻辑支持。

• 不是所有模型都支持流式。

---

1.3.4 batch（批量）

批量发送多个独立请求，提高效率，降低成本：

responses = model.batch([
    "Why do parrots have colorful feathers?",
    "How do airplanes fly?",
    "What is quantum computing?"
])
for response in responses:
    print(response)

默认 batch() 返回所有请求的最终回答。如果想边生成边接收单条回答，可以用 batch_as_completed()：

for response in model.batch_as_completed([
    "Why do parrots have colorful feathers?",
    "How do airplanes fly?",
    "What is quantum computing?"
]):
    print(response)

• 批量调用适合并行处理多请求。
• batch_as_completed 支持边处理边获取结果。
• 批量大小受模型和网络限制。
• 并行请求可能受限于API速率限制。

1.3.5 工具调用（Tool Calling）

模型可以请求调用外部工具，如数据库查询、网页搜索或代码执行。工具包括：

1. 一个定义工具名称、描述和参数的模式（通常是JSON Schema）
2. 一个执行工具功能的函数或方法

调用流程示意：

例如：

from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """Get the weather at a location."""
    return f"It's sunny in {location}."

model_with_tools = model.bind_tools([get_weather])

response = model_with_tools.invoke("What's the weather like in Boston?")
for tool_call in response.tool_calls:
    print(f"Tool: {tool_call['name']}")
    print(f"Args: {tool_call['args']}")

• @tool 装饰器定义工具函数。
• bind_tools 将工具绑定到模型，使模型能调用。
• response.tool_calls 返回模型请求调用的工具及参数。
• 使用模型外的工具调用时，需手动执行工具并将结果反馈给模型。
• 代理框架会自动处理工具调用循环。

1.3.6 结构化输出

还可以参考文档：Structured output

模型可以被要求按照特定结构返回结果，方便程序解析和后续处理。支持的结构化格式包括：

• Pydantic 模型
• TypedDict
• JSON Schema

结构化输出的关键考虑事项：

• 方法参数：一些提供商支持不同的方法（‘json_schema’、‘function_calling’、‘json_mode’）。其中，‘json_schema’ 通常指提供商专门提供的结构化输出功能；‘function_calling’ 通过强制按照给定的模式调用工具来生成结构化输出；‘json_mode’ 是某些提供商提供的 ‘json_schema’ 的前身——它生成有效的 JSON，但模式必须在提示中描述。
• 包含原始内容：使用 include_raw=True 可以同时获得解析后的输出和原始的 AI 消息。
• 验证：Pydantic 模型提供自动验证，而 TypedDict 和 JSON Schema 需要手动验证。

1.3.6.1 Pydantic

示例，使用 Pydantic 定义电影信息结构：

from pydantic import BaseModel, Field

class Movie(BaseModel):
    """A movie with details."""
    title: str = Field(..., description="The title of the movie")
    year: int = Field(..., description="The year the movie was released")
    director: str = Field(..., description="The director of the movie")
    rating: float = Field(..., description="The movie's rating out of 10")

model_with_structure = model.with_structured_output(Movie)
response = model_with_structure.invoke("Provide details about the movie Inception")
print(response)  # Movie(title="Inception", year=2010, director="Christopher Nolan", rating=8.8)

• with_structured_output 绑定结构化模型，约束回答格式。

• 返回值是对应的 Pydantic 实例，方便访问字段。

• 结构化模型定义需准确，避免解析错误。

• 可以设置 include_raw=True 同时返回原始消息和解析结果。

1.3.6.2 TypedDict

from typing_extensions import TypedDict, Annotated

class MovieDict(TypedDict):
    """A movie with details."""
    title: Annotated[str, ..., "The title of the movie"]
    year: Annotated[int, ..., "The year the movie was released"]
    director: Annotated[str, ..., "The director of the movie"]
    rating: Annotated[float, ..., "The movie's rating out of 10"]

model_with_structure = model.with_structured_output(MovieDict)
response = model_with_structure.invoke("Provide details about the movie Inception")
print(response)  # {'title': 'Inception', 'year': 2010, 'director': 'Christopher Nolan', 'rating': 8.8}

1.3.6.3 JSON Schema

import json

json_schema = {
    "title": "Movie",
    "description": "A movie with details",
    "type": "object",
    "properties": {
        "title": {
            "type": "string",
            "description": "The title of the movie"
        },
        "year": {
            "type": "integer",
            "description": "The year the movie was released"
        },
        "director": {
            "type": "string",
            "description": "The director of the movie"
        },
        "rating": {
            "type": "number",
            "description": "The movie's rating out of 10"
        }
    },
    "required": ["title", "year", "director", "rating"]
}

model_with_structure = model.with_structured_output(
    json_schema,
    method="json_schema",
)
response = model_with_structure.invoke("Provide details about the movie Inception")
print(response)  # {'title': 'Inception', 'year': 2010, ...}

1.3.7 支持多模态

部分模型能处理非文本数据（图片、音频、视频），通过内容块传递。

示例：

response = model.invoke("Create a picture of a cat")
print(response.content_blocks)
# [
#     {"type": "text", "text": "Here's a picture of a cat"},
#     {"type": "image", "base64": "...", "mime_type": "image/jpeg"},
# ]

速率限制与代理配置

1.3.8 速率限制

许多模型提供商限制调用频率，超限会报错。LangChain 支持传入速率限制器控制请求速率。

示例：

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,  # 每10秒1次请求
    check_every_n_seconds=0.1,
    max_bucket_size=10,
)

model = init_chat_model(
    model="gpt-5",
    model_provider="openai",
    rate_limiter=rate_limiter
)

1.3.9 基础URL与代理

支持配置 API 基础 URL，方便使用兼容 OpenAI 的其他服务或代理服务器。

示例：

model = init_chat_model(
    model="MODEL_NAME",
    model_provider="openai",
    base_url="BASE_URL",
    api_key="YOUR_API_KEY",
)

部分集成支持 HTTP 代理配置：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-4o",
    openai_proxy="http://proxy.example.com:8080"
)

日志概率与令牌使用

1.3.10 日志概率

部分模型可返回每个生成令牌的概率信息，需开启 logprobs 参数：

model = init_chat_model(
    model="gpt-4o",
    model_provider="openai"
).bind(logprobs=True)

response = model.invoke("Why do parrots talk?")
print(response.response_metadata["logprobs"])

令牌使用统计

多数模型提供令牌使用统计，可通过回调或上下文管理器追踪：

from langchain.chat_models import init_chat_model
from langchain_core.callbacks import UsageMetadataCallbackHandler

model_1 = init_chat_model(model="gpt-4o-mini")
model_2 = init_chat_model(model="claude-haiku-4-5-20251001")

callback = UsageMetadataCallbackHandler()
result_1 = model_1.invoke("Hello", config={"callbacks": [callback]})
result_2 = model_2.invoke("Hello", config={"callbacks": [callback]})

print(callback.usage_metadata)

1.3.11 调用配置与可配置模型

调用配置

调用时可传入 config 参数，控制执行行为、回调、元数据等：

response = model.invoke(
    "Tell me a joke",
    config={
        "run_name": "joke_generation",
        "tags": ["humor", "demo"],
        "metadata": {"user_id": "123"},
        "callbacks": [my_callback_handler],
    }
)

用途包括调试、日志、资源控制和调用追踪。

可配置模型

可以创建运行时可配置的模型，指定哪些字段可动态调整：

from langchain.chat_models import init_chat_model

configurable_model = init_chat_model(temperature=0)

configurable_model.invoke(
    "what's your name",
    config={"configurable": {"model": "gpt-5-nano"}},  # 使用 GPT-5-Nano
)
configurable_model.invoke(
    "what's your name",
    config={"configurable": {"model": "claude-sonnet-4-5-20250929"}},  # 使用 Claude
)

1.4 Messages 消息

在 LangChain 中，消息是模型上下文的基本单元。
它们代表模型的输入和输出，携带对话状态所需的内容和元数据。
当与大语言模型（LLM）交互时，消息对象包含以下内容：

• 角色（Role）：标识消息类型，比如 system（系统）、user（用户）等。
• 内容（Content）：消息的实际内容，可以是文本、图片、音频、文档等。
• 元数据（Metadata）：可选字段，如响应信息、消息ID和令牌使用情况。

LangChain 提供了统一的消息类型，适用于所有模型提供商，保证调用模型时行为一致。

消息类型	说明
系统消息SystemMessage	指定模型行为和对话上下文
用户消息HumanMessage	用户输入，支持文本、多模态内容
AI消息AIMessage	模型生成的回答，包含文本、工具调用和元数据
工具消息	工具调用的输出结果

1.4.1 消息的基本应用

最简单的使用方式是创建消息对象，并在调用模型时传入这些消息。

from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage, SystemMessage

model = init_chat_model("gpt-5-nano")

system_msg = SystemMessage("You are a helpful assistant.")
human_msg = HumanMessage("Hello, how are you?")

# 结合聊天模型使用
messages = [system_msg, human_msg]
response = model.invoke(messages)  # 返回 AIMessage

这里：

• SystemMessage 用于设置模型的行为和对话背景。
• HumanMessage 是用户输入。
• invoke() 方法调用模型，传入消息列表，返回模型生成的回答（AIMessage）。

1.4.2 文本提示与消息提示的区别

文本提示（Text Prompts）

文本提示是简单的字符串，适合单次生成，不需要保留对话历史的场景。

response = model.invoke("Write a haiku about spring")

适用场景：

• 单次独立请求
• 不需要对话上下文
• 代码简洁

消息提示（Message Prompts）

消息提示是消息对象的列表，适合多轮对话、包含系统指令或多模态内容。

from langchain.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage("You are a poetry expert"),
    HumanMessage("Write a haiku about spring"),
    AIMessage("Cherry blossoms bloom...")
]
response = model.invoke(messages)

适用场景：

• 多轮对话管理
• 处理多模态内容（图片、音频、文件）
• 需要系统指令引导模型行为

字典格式

也可以直接使用 OpenAI 聊天完成格式的字典列表：

messages = [
    {"role": "system", "content": "You are a poetry expert"},
    {"role": "user", "content": "Write a haiku about spring"},
    {"role": "assistant", "content": "Cherry blossoms bloom..."}
]
response = model.invoke(messages)

1.4.3 各类消息设置

系统消息（SystemMessage）

系统消息用于初始化模型的行为，设置语气、角色和回答准则。

from langchain.messages import SystemMessage, HumanMessage

system_msg = SystemMessage("""
You are a senior Python developer with expertise in web frameworks.
Always provide code examples and explain your reasoning.
Be concise but thorough in your explanations.
""")

messages = [
    system_msg,
    HumanMessage("How do I create a REST API?")
]
response = model.invoke(messages)

用户消息（HumanMessage）

代表用户输入，可以包含文本、图片、音频、文件等多模态内容。

human_msg = HumanMessage(
    content="Hello!",
    name="alice",  # 可选，标识不同用户
    id="msg_123"   # 可选，唯一标识便于追踪
)

AI消息（AIMessage）

代表模型的输出，可能包含多模态数据、工具调用和元数据。

response = model.invoke("Explain AI")
print(type(response))  # <class 'langchain_core.messages.ai.AIMessage'>

AI消息对象包含：

• 文本内容
• 原始内容
• 标准化的内容块（content blocks）
• 模型调用的工具调用信息（如果有）
• 唯一消息ID
• 令牌使用元数据
• 响应元数据

AIMessage：工具调用信息

当模型调用工具时，工具调用信息包含在 AIMessage 中：

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-5-nano")

def get_weather(location: str) -> str:
    """获取指定地点的天气"""
    ...

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("What's the weather in Paris?")

for tool_call in response.tool_calls:
    print(f"Tool: {tool_call['name']}")
    print(f"Args: {tool_call['args']}")
    print(f"ID: {tool_call['id']}")

此外，推理过程或引用等结构化数据也可能出现在消息内容中。

AIMessage：令牌使用示例

AIMessage 还可以包含令牌计数和使用元数据：

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-5-nano")
response = model.invoke("Hello!")
print(response.usage_metadata)

# 输出示例：
# {'input_tokens': 8,
#  'output_tokens': 304,
#  'total_tokens': 312,
#  'input_token_details': {'audio': 0, 'cache_read': 0},
#  'output_token_details': {'audio': 0, 'reasoning': 256}}

1.4.4 消息内容（Message Content）多模态输入

消息内容是发送给模型的数据载体。消息有一个 content 属性，支持多种数据类型：

• 字符串
• 提供商原生格式的内容块列表（如多模态数据）
• LangChain 标准内容块列表

图片URL

# From URL
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this image."},
        {"type": "image", "url": "https://example.com/path/to/image.jpg"},
    ]
}

# From base64 data
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this image."},
        {
            "type": "image",
            "base64": "AAAAIGZ0eXBtcDQyAAAAAGlzb21tcDQyAAACAGlzb2...",
            "mime_type": "image/jpeg",
        },
    ]
}

# From provider-managed File ID
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this image."},
        {"type": "image", "file_id": "file-abc123"},
    ]
}

PDF：

# From URL
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this document."},
        {"type": "file", "url": "https://example.com/path/to/document.pdf"},
    ]
}

# From base64 data
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this document."},
        {
            "type": "file",
            "base64": "AAAAIGZ0eXBtcDQyAAAAAGlzb21tcDQyAAACAGlzb2...",
            "mime_type": "application/pdf",
        },
    ]
}

# From provider-managed File ID
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this document."},
        {"type": "file", "file_id": "file-abc123"},
    ]
}

VIDEO

# From base64 data
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this video."},
        {
            "type": "video",
            "base64": "AAAAIGZ0eXBtcDQyAAAAAGlzb21tcDQyAAACAGlzb2...",
            "mime_type": "video/mp4",
        },
    ]
}

# From provider-managed File ID
message = {
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe the content of this video."},
        {"type": "video", "file_id": "file-abc123"},
    ]
}

1.4.5 跨提供商通用的消息内容

LangChain 提供跨提供商通用的消息内容标准表示。
消息对象实现了 content_blocks 属性，会延迟解析 content，转换成类型安全的标准格式。

例如，来自 ChatAnthropic 或 ChatOpenAI 的消息，可能包含思考（thinking）或推理（reasoning）内容块，但都能统一解析为 ReasoningContentBlock 格式：

from langchain.messages import AIMessage

message = AIMessage(
    content=[
        {"type": "thinking", "thinking": "...", "signature": "WaUjzkyp..."},
        {"type": "text", "text": "..."},
    ],
    response_metadata={"model_provider": "anthropic"}
)
print(message.content_blocks)

# 输出：
# [{'type': 'reasoning',
#   'reasoning': '...',
#   'extras': {'signature': 'WaUjzkyp...'}},
#  {'type': 'text', 'text': '...'}]

1.5 工具（Tools）

文档来源

许多 AI 应用通过自然语言与用户交互，但有些场景需要模型直接与外部系统（如 API、数据库或文件系统）交互，使用结构化输入。工具就是由代理（agents）调用以执行操作的组件。它们通过定义明确的输入和输出，扩展了模型的能力，使模型能与外部世界交互。

工具封装了一个可调用的函数及其输入模式（schema），可以传递给兼容的聊天模型，让模型决定是否调用工具以及使用哪些参数。在这些场景中，工具调用让模型生成符合指定输入模式的请求。

1.5.1 工具的基本定义

创建工具最简单的方式是使用 @tool 装饰器。默认情况下，函数的文档字符串（docstring）会成为工具的描述，帮助模型理解何时使用该工具。

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """
    Search the customer database for records matching the query.

    Args:
        query: Search terms to look for
        limit: Maximum number of results to return
    """
    return f"Found {limit} results for '{query}'"

代码解读：

• @tool 装饰器将普通函数转换为工具。
• 函数的类型提示（type hints）必须提供，因为它们定义了工具的输入模式。
• 函数的 docstring 应简洁明了，帮助模型理解工具用途。

注意事项：

• 类型提示是必需的，否则模型无法正确生成调用参数。
• 文档字符串应准确描述工具功能，便于模型判断调用时机。

1.5.2 自定义工具名称与描述

默认工具名称来自函数名，可通过装饰器参数覆盖，以提供更具描述性的名称。

@tool("web_search")  # 自定义名称
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

print(search.name)  # 输出：web_search

可覆盖自动生成的工具描述，为模型提供更清晰的指导。

@tool("calculator", description="Performs arithmetic calculations. Use this for any math problems.")
def calc(expression: str) -> str:
    """Evaluate mathematical expressions."""
    return str(eval(expression))

1.5.3 ToolRuntime

工具可以通过 ToolRuntime 参数访问运行时信息，包括：

访问项	说明
State	可变数据，执行过程中的状态（如消息、计数器、自定义字段）
Context	不可变配置，如用户ID、会话信息、应用特定配置
Store	持久化的长期记忆，跨会话保存数据
Stream Writer	工具执行时的实时流式更新
Config	执行配置（RunnableConfig）
Tool Call ID	当前工具调用的唯一ID

使用 ToolRuntime 可以在工具函数签名中直接添加 runtime: ToolRuntime 参数，框架会自动注入，且不会暴露给模型。

访问状态（State）

from langchain.tools import tool, ToolRuntime

@tool
def summarize_conversation(runtime: ToolRuntime) -> str:
    """Summarize the conversation so far."""
    messages = runtime.state["messages"]

    human_msgs = sum(1 for m in messages if m.__class__.__name__ == "HumanMessage")
    ai_msgs = sum(1 for m in messages if m.__class__.__name__ == "AIMessage")
    tool_msgs = sum(1 for m in messages if m.__class__.__name__ == "ToolMessage")

    return f"Conversation has {human_msgs} user messages, {ai_msgs} AI responses, and {tool_msgs} tool results"

访问自定义状态字段

@tool
def get_user_preference(pref_name: str, runtime: ToolRuntime) -> str:
    """Get a user preference value."""
    preferences = runtime.state.get("user_preferences", {})
    return preferences.get(pref_name, "Not set")

更新状态

使用 Command 来更新代理状态或控制执行流程。

from langgraph.types import Command
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.tools import tool

@tool
def clear_conversation() -> Command:
    """Clear the conversation history."""
    return Command(
        update={
            "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)],
        }
    )

@tool
def update_user_name(new_name: str, runtime: ToolRuntime) -> Command:
    """Update the user's name."""
    return Command(update={"user_name": new_name})

访问上下文（Context）

通过 runtime.context 访问不可变的配置和上下文数据，如用户ID、会话信息或应用配置。

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

USER_DATABASE = {
    "user123": {
        "name": "Alice Johnson",
        "account_type": "Premium",
        "balance": 5000,
        "email": "alice@example.com"
    },
    "user456": {
        "name": "Bob Smith",
        "account_type": "Standard",
        "balance": 1200,
        "email": "bob@example.com"
    }
}

@dataclass
class UserContext:
    user_id: str

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Get the current user's account information."""
    user_id = runtime.context.user_id

    if user_id in USER_DATABASE:
        user = USER_DATABASE[user_id]
        return f"Account holder: {user['name']}\nType: {user['account_type']}\nBalance: ${user['balance']}"
    return "User not found"

示例中，工具根据上下文中的用户ID查询数据库，返回账户信息。

访问记忆（Store）

通过 runtime.store 访问跨会话持久化存储，保存和检索用户或应用数据。

from typing import Any
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent

# 读取用户信息
@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
    """Look up user info."""
    store = runtime.store
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "Unknown user"

# 保存用户信息
@tool
def save_user_info(user_id: str, user_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """Save user info."""
    store = runtime.store
    store.put(("users",), user_id, user_info)
    return "Successfully saved user info."

store = InMemoryStore()
agent = create_agent(
    model,
    tools=[get_user_info, save_user_info],
    store=store
)

# 第一次会话：保存用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev"}]
})

# 第二次会话：获取用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Get user info for user with id 'abc123'"}]
})
# 返回用户信息：
# - Name: Foo
# - Age: 25
# - Email: foo@langchain.dev

1.8 快速入门：用 LangChain 构建智能代理

下面演示如何用最简单的代码，创建一个能回答天气问题的智能代理。

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[get_weather],
    system_prompt="You are a helpful assistant",
)

# 调用代理，询问旧金山天气
agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)

• get_weather 是一个简单的工具函数，模拟返回城市天气。
• create_agent 创建一个智能代理，指定使用 Claude Sonnet 4.5 语言模型，绑定天气查询工具。
• system_prompt 用来告诉模型它的身份和行为准则。
• agent.invoke 用于发送用户消息并获取回复。

系统提示（system prompt）用来指导代理的行为，示例中设计为：

• 代理是一个“幽默的天气预报专家”，说话带双关语（puns）。
• 代理有两个工具：
  • get_weather_for_location：获取指定地点天气
  • get_user_location：获取用户所在位置
• 如果用户问天气但没指定地点，代理会自动用 get_user_location 工具获取用户位置。

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

# 系统提示，指导代理行为
SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.

You have access to two tools:

- get_weather_for_location: use this to get the weather for a specific location
- get_user_location: use this to get the user's location

If a user asks you for the weather, make sure you know the location. If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location."""

# 定义上下文数据结构，保存用户相关信息
@dataclass
class Context:
    """自定义运行时上下文结构。"""
    user_id: str

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

# 定义工具：根据用户ID获取用户位置
@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
    """根据用户ID获取用户位置。"""
    user_id = runtime.context.user_id
    return "Florida" if user_id == "1" else "SF"

# 初始化聊天模型，设置温度为0保证稳定输出
model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    temperature=0
)

# 定义代理响应格式，包含双关语回复和天气信息
@dataclass
class ResponseFormat:
    """代理响应结构。"""
    punny_response: str  # 搞笑双关语回复，必填
    weather_conditions: str | None = None  # 天气详情，可选

# 设置内存保存器，用于会话记忆
checkpointer = InMemorySaver()

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

# 配置唯一会话ID，支持多轮对话
config = {"configurable": {"thread_id": "1"}}

# 运行代理，用户询问天气
response = agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
    config=config,
    context=Context(user_id="1")
)

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

# 继续对话，用户表示感谢
response = agent.invoke(
    {"messages": [{"role": "user", "content": "thank you!"}]},
    config=config,
    context=Context(user_id="1")
)

print(response['structured_response'])
# 输出示例：
# ResponseFormat(
#     punny_response="You're 'thund-erfully' welcome! ...",
#     weather_conditions=None
# )

• 使用 @dataclass 定义上下文和响应格式，方便管理数据结构。
• @tool 装饰器定义两个工具函数，分别获取城市天气和用户位置。
• init_chat_model 初始化聊天模型，设置温度为0保证响应稳定。
• ToolStrategy 结合响应格式，确保代理输出结构化数据，便于后续处理。
• InMemorySaver 实现简单的会话记忆，支持多轮对话。
• agent.invoke 支持传入上下文和配置，实现个性化和连续对话。

2 重要组件

2.1 agent

来自文档：
https://docs.langchain.com/oss/python/langchain/agents

智能代理（Agent）是将语言模型与工具结合的系统，能够：

• 理解任务并进行推理
• 决定使用哪些工具
• 通过多轮调用工具，逐步完成目标

LangChain 提供了 create_agent 函数，帮助开发者快速创建生产级智能代理。

智能代理的工作流程遵循 ReAct（Reasoning + Acting）模式：模型先进行推理，然后调用工具，观察结果，再继续推理，直到满足停止条件（如模型输出最终答案或达到最大迭代次数）。

组件	作用说明
模型	代理的推理引擎，负责理解和生成内容
工具	代理执行的辅助操作，如查询、计算等
输入	用户传入的消息或请求
输出	代理的最终回答或中间观察结果

静态与动态模型的差异：

特性	静态模型	动态模型
定义时机	在创建代理时固定配置，一旦设定执行过程中不变	根据当前对话状态和上下文动态选择模型实例
灵活性	低，整个会话使用同一个模型	高，可根据对话复杂度、任务需求等条件切换不同模型
使用场景	适合简单、单一模型需求的场景	适合复杂对话、多模型路由、成本优化等高级场景
实现方式	直接传入模型名称字符串或模型实例	通过中间件（middleware）拦截请求，动态修改模型参数
代码示例	agent = create_agent("gpt-5", tools=tools)	利用 @wrap_model_call 中间件动态切换模型
优点	简单易用，配置直接	灵活智能，能根据上下文优化资源和响应质量
缺点	不适合多样化需求，可能资源利用不佳	实现复杂，需要编写中间件逻辑，维护成本较高

2.1.1 静态模型（static Model）

模型是代理的“思考大脑”，LangChain 支持多种方式配置模型：
静态模型在代理创建时指定，执行过程中不变，是最常见的用法。

直接使用模型实例（如 OpenAI 的 ChatOpenAI），可以设置温度、最大令牌数等参数：

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5",
    temperature=0.1,
    max_tokens=1000,
    timeout=30
)

agent = create_agent(model, tools=tools)

• ChatOpenAI 是一个模型实例，支持详细配置。
• 通过实例传入 create_agent，可自定义模型行为。
• 根据需求调整参数，如温度控制回答随机性，max_tokens 控制回答长度。
• 模型名称需与实际可用模型匹配。

---

2.1.2 动态模型

动态模型根据当前对话状态动态选择，适合复杂场景和成本优化。

示例通过中间件实现模型切换：

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
def dynamic_model_selection(request: ModelRequest, 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]
)

• 使用 @wrap_model_call 装饰器定义中间件函数。
• 根据对话消息数动态切换模型实例。
• 中间件插入代理执行流程，实现灵活调度。
• 动态模型适合对话复杂度或任务需求变化较大的场景。
• 需保证所有模型接口兼容。

---

2.1.3 工具（Tools）

工具是代理执行动作的手段，比如搜索信息、查询天气等。
通过 @tool 装饰器定义工具函数，并传入代理：

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

@tool
def search(query: str) -> str:
    """搜索信息。"""
    return f"Results for: {query}"

@tool
def get_weather(location: str) -> str:
    """获取指定地点的天气。"""
    return f"Weather in {location}: Sunny, 72°F"

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

• @tool 装饰器标记函数为代理可调用的工具。
• 代理可根据上下文自动调用合适工具。
• 工具函数应尽量简洁明了。
• 如果工具列表为空，代理仅使用语言模型，不调用工具。

中间件处理报错：
可以通过中间件捕获工具执行异常，返回自定义错误消息：

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

@wrap_tool_call
def handle_tool_errors(request, handler):
    """自定义工具错误处理。"""
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({str(e)})",
            tool_call_id=request.tool_call["id"]
        )

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

• @wrap_tool_call 用于定义工具调用中间件。
• 捕获异常并返回格式化错误信息给模型。
• 错误处理提升用户体验，避免程序崩溃。
• 自定义错误消息可根据业务需求调整。

---

2.1.4 系统提示（System Prompt）

系统提示用来塑造代理行为。可直接传入字符串：

agent = create_agent(
    model,
    tools,
    system_prompt="You are a helpful assistant. Be concise and accurate."
)

如果不传，代理会根据消息自动推断任务。

2.1.5 动态系统prompt

通过中间件@dynamic_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", "user")
    base_prompt = "You are a helpful assistant."

    if user_role == "expert":
        return f"{base_prompt} Provide detailed technical responses."
    elif user_role == "beginner":
        return f"{base_prompt} Explain concepts simply and avoid jargon."

    return base_prompt

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

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Explain machine learning"}]},
    context={"user_role": "expert"}
)

• @dynamic_prompt 装饰器定义动态系统提示中间件。
• 根据上下文中的用户角色返回不同提示。
• 使代理回答更符合用户需求和身份。

---

2.1.6 代理调用（Invocation）

调用代理时，传入消息更新状态：

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)

代理支持流式输出，方便显示中间步骤：

for chunk in agent.stream({
    "messages": [{"role": "user", "content": "Search for AI news and summarize the findings"}]
}, stream_mode="values"):
    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]}")

2.1.7 结构化输出

通过 response_format 参数，代理可返回结构化数据。

• ToolStrategy：通过模拟工具调用生成结构化输出，适用于支持工具调用的模型。
• ProviderStrategy：利用模型提供商原生结构化输出功能（如 OpenAI）。

如果你用的是支持结构化输出的模型（比如OpenAI的某些版本），用 ProviderStrategy 更靠谱；
如果模型不支持，就用 ToolStrategy 来“凑合”着用。

ToolStrategy示例：

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

class ContactInfo(BaseModel):
    name: str
    email: str
    phone: str

agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]
})

print(result["structured_response"])
# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

@ProviderStrategy，示例：

from langchain.agents.structured_output import ProviderStrategy

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

---

2.1.8 会话记忆（Memory）

代理会通过消息状态自动维护对话历史。你也可以配置代理使用自定义状态模式，在对话过程中记住额外的信息。存储在状态中的信息可以看作是代理的短期记忆。
定义自定义状态有两种方式：

• 通过中间件（推荐）
• 通过 create_agent 上的 state_schema

from langchain.agents import AgentState

class CustomState(AgentState):
    user_preferences: dict

通过中间件或 state_schema 传入自定义状态，代理即可记住更多上下文。

通过中间件middleware定义状态记忆

from langchain.agents import AgentState
from langchain.agents.middleware import AgentMiddleware
from typing import Any


class CustomState(AgentState):
    user_preferences: dict

class CustomMiddleware(AgentMiddleware):
    state_schema = CustomState
    tools = [tool1, tool2]

    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:
        ...

agent = create_agent(
    model,
    tools=tools,
    middleware=[CustomMiddleware()]
)

# The agent can now track additional state beyond messages
result = agent.invoke({
    "messages": [{"role": "user", "content": "I prefer technical explanations"}],
    "user_preferences": {"style": "technical", "verbosity": "detailed"},
})

• CustomState(AgentState)：让代理除了默认的消息历史外，还能额外保存一些自定义的信息。
• CustomMiddleware(AgentMiddleware)：在模型调用之前执行一些自定义逻辑
  • state: CustomState：当前代理的状态，包括消息历史和自定义的
  • user_preferences 等信息。
  • runtime：当前运行时环境或上下文，包含调用相关的信息。

通过 state_schema定义state

from langchain.agents import AgentState

class CustomState(AgentState):
    user_preferences: dict

agent = create_agent(
    model,
    tools=[tool1, tool2],
    state_schema=CustomState
)
# The agent can now track additional state beyond messages
result = agent.invoke({
    "messages": [{"role": "user", "content": "I prefer technical explanations"}],
    "user_preferences": {"style": "technical", "verbosity": "detailed"},
})

2.2 短期记忆（Short-term Memory）

文档地址
短期记忆指的是应用在单个对话线程或会话中，记住之前的交互内容。最常见的短期记忆形式是对话历史。

然而，长时间的对话会给当前的语言模型（LLM）带来挑战：

• 完整的对话历史可能超出模型的上下文窗口限制，导致上下文丢失或错误。
• 即使模型支持完整上下文，长上下文会导致模型表现下降，容易被过时或无关内容干扰，同时响应速度变慢，成本增加。

聊天模型通过消息来接收上下文，包括系统指令（system message）和用户输入（human messages）。消息在对话中交替出现，消息列表会随着对话增长而变长。由于上下文窗口有限，很多应用需要采用“遗忘”或删除过时信息的技术。

2.2.1 使用短期记忆

要为代理添加短期记忆（线程级持久化），需要在创建代理时指定一个 checkpointer（检查点管理器）。

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver  

agent = create_agent(
    "gpt-5",
    [get_user_info],
    checkpointer=InMemorySaver(),  
)

agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    {"configurable": {"thread_id": "1"}},  
)

代码解读：

• InMemorySaver 是一个内存中的检查点管理器，适合测试和开发。
• thread_id 标识对话线程，保证同一线程的记忆被保存和调用。

2.2.2 生产环境中的短期记忆

在生产环境，应使用数据库支持的检查点管理器，比如基于 PostgreSQL 的 PostgresSaver。

pip install langgraph-checkpoint-postgres

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver  

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 自动创建数据库表
    agent = create_agent(
        "gpt-5",
        [get_user_info],
        checkpointer=checkpointer,  
    )

代码解读：

• PostgresSaver 连接 PostgreSQL 数据库，持久化存储对话状态。
• setup() 会自动创建所需表结构，方便部署。

2.2.3 自定义代理记忆

默认情况下，代理通过 AgentState 管理短期记忆，尤其是通过 messages 字段保存对话历史。你可以扩展 AgentState 添加更多自定义字段，并通过 state_schema 参数传入 create_agent。

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver

class CustomAgentState(AgentState):  
    user_id: str
    preferences: dict

agent = create_agent(
    "gpt-5",
    [get_user_info],
    state_schema=CustomAgentState,  
    checkpointer=InMemorySaver(),
)

# 调用时传入自定义状态字段
result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],
        "user_id": "user_123",  
        "preferences": {"theme": "dark"}  
    },
    {"configurable": {"thread_id": "1"}}
)

代码解读：

• 通过继承 AgentState 定义自定义状态结构。
• 代理会自动管理这些状态字段，方便后续工具和中间件访问。

2.2.4 常见短期记忆管理策略

由于长对话可能超出模型上下文窗口，常用以下策略来管理消息历史：

策略	说明
修剪消息	根据令牌数限制，保留最近的若干消息，删除或忽略较早消息
删除消息	主动删除特定消息或全部消息，控制消息数量
总结消息	使用模型自动摘要对话历史，保留关键信息，减少上下文长度

---

修剪消息

大多数 LLM 有最大上下文窗口（以令牌计）。修剪消息即统计消息令牌数，接近限制时删除部分旧消息。

在 LangChain 中，可以用 @before_model 中间件实现：

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """只保留最近几条消息以适应上下文窗口。"""
    messages = state["messages"]

    if len(messages) <= 3:
        return None  # 无需修改

    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages

    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            *new_messages
        ]
    }

agent = create_agent(
    model,
    tools=[],
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

输出示例：

Your name is Bob. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.

---

删除消息

可以通过删除消息管理历史，删除特定消息或全部消息。

删除特定消息示例：

from langchain.messages import RemoveMessage  

def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 删除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}  

删除全部消息示例：

from langgraph.graph.message import REMOVE_ALL_MESSAGES

def delete_messages(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}  

结合 @after_model 中间件自动删除旧消息：

from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """删除旧消息，保持对话可控。"""
    messages = state["messages"]
    if len(messages) > 2:
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None

agent = create_agent(
    "gpt-5-nano",
    tools=[],
    system_prompt="Please be concise and to the point.",
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}

for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

---

总结消息

修剪或删除可能丢失重要信息，部分应用通过模型对消息历史进行总结，保留关键信息，减少上下文长度。

使用内置的 SummarizationMiddleware：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig

checkpointer = InMemorySaver()

agent = create_agent(
    model="gpt-4o",
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            max_tokens_before_summary=4000,  # 触发总结的令牌数
            messages_to_keep=20,  # 总结后保留的消息数
        )
    ],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

输出示例：

Your name is Bob!

2.2.5 在工具中读取短期记忆

工具函数通过隐藏的 ToolRuntime 参数访问状态。

from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime

class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """查询用户信息。"""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)

result = agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# 输出：User is John Smith.

2.2.6 在工具中写入短期记忆

工具可以返回状态更新，用于持久化中间结果或供后续工具访问。

from langchain.tools import tool, ToolRuntime
from langchain_core.runnables import RunnableConfig
from langchain.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel

class CustomState(AgentState):  
    user_name: str

class CustomContext(BaseModel):
    user_id: str

@tool
def update_user_info(runtime: ToolRuntime[CustomContext, CustomState]) -> Command:
    """查询并更新用户信息。"""
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={  
        "user_name": name,
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })

@tool
def greet(runtime: ToolRuntime[CustomContext, CustomState]) -> str:
    """找到用户信息后，用于问候用户。"""
    user_name = runtime.state.get("user_name", None)
    if user_name is None:
       return Command(update={
            "messages": [
                ToolMessage(
                    "Please call the 'update_user_info' tool it will get and update the user's name.",
                    tool_call_id=runtime.tool_call_id
                )
            ]
        })
    return f"Hello {user_name}!"

agent = create_agent(
    model="gpt-5-nano",
    tools=[update_user_info, greet],
    state_schema=CustomState, 
    context_schema=CustomContext,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    context=CustomContext(user_id="user_123"),
)

2.2.7 在中间件中访问短期记忆（动态提示）

通过中间件动态生成提示，基于对话历史或自定义状态字段。

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

class CustomContext(TypedDict):
    user_name: str

def get_weather(city: str) -> str:
    """获取城市天气。"""
    return f"The weather in {city} is always sunny!"

@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest) -> str:
    user_name = request.runtime.context["user_name"]
    system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
    return system_prompt

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[dynamic_system_prompt],
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

示例输出：

================================ Human Message =================================

What is the weather in SF?
================================== Ai Message ==================================

Tool Calls:
  get_weather (call_WFQlOGn4b2yoJrv7cih342FG)
 Call ID: call_WFQlOGn4b2yoJrv7cih342FG
  Args:
    city: San Francisco
================================= Tool Message =================================
Name: get_weather

The weather in San Francisco is always sunny!
================================== Ai Message ==================================

Hi John Smith, the weather in San Francisco is always sunny!

2.2.8 @before_model 中间件访问短期记忆

在模型调用前处理消息。

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """只保留最近几条消息以适应上下文窗口。"""
    messages = state["messages"]

    if len(messages) <= 3:
        return None

    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages

    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            *new_messages
        ]
    }

agent = create_agent(
    model,
    tools=[],
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

---

2.2.9 @after_model 中间件访问短期记忆

在模型调用后处理消息。

from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime

@stop_words = ["password", "secret"]

@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """删除包含敏感词的消息。"""
    last_message = state["messages"][-1]
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}
    return None

agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[validate_response],
    checkpointer=InMemorySaver(),
)

2.3 LangChain 内置中间件详解

Built-in middleware

本文为初学者科普介绍 LangChain 提供的内置中间件（middleware）。中间件是指在模型调用和工具执行之间插入的功能模块，用于增强代理（agent）的能力，满足各种生产环境需求。本文内容基于 LangChain 官方文档，涵盖了通用（Provider-agnostic）和特定模型供应商的中间件，详细介绍了它们的功能、应用场景和代码示例。

---

2.3.1 通用中间件（Provider-agnostic middleware）

这些中间件适用于所有大型语言模型（LLM）供应商，具备通用性。

中间件名称	功能描述
Summarization	在接近上下文窗口限制时，自动总结对话历史
Human-in-the-loop	在执行工具调用前暂停，等待人工审批
Model call limit	限制模型调用次数，防止成本过高或无限循环
Tool call limit	限制工具调用次数，控制资源使用
Model fallback	主模型失败时自动切换备用模型
PII detection	检测并处理个人身份信息（PII）
To-do list	为代理提供任务规划和跟踪能力
LLM tool selector	使用 LLM 智能选择相关工具，减少无关工具调用
Tool retry	失败的工具调用自动重试，支持指数退避
LLM tool emulator	使用 LLM 模拟工具执行，便于测试
Context editing	管理对话上下文，清理旧工具调用结果
Shell tool	为代理暴露持久的 shell 会话，执行系统命令
File search	提供基于文件系统的 Glob 和 Grep 搜索工具

2.3.2 Summarization（自动总结）

当对话接近模型上下文窗口限制时，自动总结旧对话，保留最近消息，压缩较早内容。适合长时间对话、多轮复杂对话或需要完整上下文保存的场景。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[weather_tool, calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            trigger={"tokens": 4000},
            keep={"messages": 20},
        ),
    ],
)

代码解读

• model="gpt-4o"：主模型
• tools：代理可调用的工具
• SummarizationMiddleware：使用更小模型生成对话摘要
• trigger：触发条件，这里是当对话超过4000个token
• keep：总结后保留最近20条消息

注意事项

• 触发条件支持多种配置（token数、消息数、上下文比例）
• 保留条件需指定一种（消息数、token数或比例）
• 适用于长时间多轮对话，防止上下文溢出

---

2.3.3 Human-in-the-loop（人工审核）

在执行工具调用前暂停，等待人工审批、编辑或拒绝。适用于高风险操作（数据库写入、财务交易）、合规流程或需要人工反馈的对话。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="gpt-4o",
    tools=[read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email_tool": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "read_email_tool": False,
            }
        ),
    ],
)

代码解读

• interrupt_on 配置哪些工具调用需要人工干预
• 可以定义允许的决策（批准、编辑、拒绝）
• 结合状态保存器（checkpointer）保持对话状态

注意事项

• 适合敏感操作，防止自动错误执行
• 需要人工参与，可能影响响应时效

---

2.3.4 Model call limit（模型调用限制）

限制模型调用次数，防止无限循环和过高成本。

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=10,
            run_limit=5,
            exit_behavior="end",
        ),
    ],
)

代码解读

• thread_limit：同一对话线程最大调用数
• run_limit：单次调用最大调用数
• exit_behavior：达到限制后的行为（结束或抛错）

注意事项

• 适合生产环境成本控制
• 可防止代理进入死循环

---

2.3.5 Tool call limit（工具调用限制）

限制工具调用次数，可针对所有工具或单个工具。

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware

global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)
search_limiter = ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)
database_limiter = ToolCallLimitMiddleware(tool_name="query_database", thread_limit=10)
strict_limiter = ToolCallLimitMiddleware(tool_name="scrape_webpage", run_limit=2, exit_behavior="error")

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, database_tool, scraper_tool],
    middleware=[global_limiter, search_limiter, database_limiter, strict_limiter],
)

代码解读

• thread_limit：对话线程内调用限制
• run_limit：单次调用限制
• exit_behavior：限制触发时行为（继续、抛错、结束）

注意事项

• 至少需指定一个限制（线程或单次）
• 可针对单个工具细粒度控制

---

2.3.6 Model fallback（模型回退）

主模型失败时自动切换备用模型，提升鲁棒性和成本效率。

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        ModelFallbackMiddleware(
            "gpt-4o-mini",
            "claude-3-5-sonnet-20241022",
        ),
    ],
)

代码解读

• 主模型失败时依次尝试备用模型
• 支持多级回退

注意事项

• 适合对模型稳定性有要求的应用
• 可节约成本使用更便宜模型作为备选

---

2.3.7 PII detection（个人身份信息检测）

检测并处理对话中的个人身份信息（PII），如邮箱、信用卡等，支持多种处理策略。

策略	说明
block	发现即抛异常，阻止继续执行
redact	替换为 [REDACTED_TYPE]
mask	部分掩码（如信用卡只显示后4位）
hash	用确定性哈希替换

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
import re

# 使用正则表达式检测API Key
agent1 = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",
        ),
    ],
)

# 使用自定义函数检测SSN
def detect_ssn(content: str) -> list[dict[str, str | int]]:
    import re
    matches = []
    pattern = r"\d{3}-\d{2}-\d{4}"
    for match in re.finditer(pattern, content):
        ssn = match.group(0)
        first_three = int(ssn[:3])
        if first_three not in [0, 666] and not (900 <= first_three <= 999):
            matches.append({
                "text": ssn,
                "start": match.start(),
                "end": match.end(),
            })
    return matches

agent3 = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        PIIMiddleware(
            "ssn",
            detector=detect_ssn,
            strategy="hash",
        ),
    ],
)

代码解读

• 支持内置PII类型和自定义检测器（正则或函数）
• 自定义函数需返回匹配文本及位置

注意事项

• 适合合规要求高的行业，如金融、医疗
• 可配置检测时机（输入、输出、工具结果）

---

2.3.8 To-do list（任务清单）

为代理提供任务规划和跟踪能力，适合复杂多步骤任务。

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[read_file, write_file, run_tests],
    middleware=[TodoListMiddleware()],
)

代码解读

• 自动管理任务列表，帮助代理有序完成复杂工作

注意事项

• 可自定义系统提示和任务工具说明

---

2.3.9 LLM tool selector（工具选择器）

用 LLM 智能选择最相关工具，减少无关调用，降低token消耗。

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolSelectorMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[tool1, tool2, tool3, tool4, tool5, ...],
    middleware=[
        LLMToolSelectorMiddleware(
            model="gpt-4o-mini",
            max_tools=3,
            always_include=["search"],
        ),
    ],
)

代码解读

• 先用小模型筛选工具，再调用主模型
• 可设定最大工具数和必选工具

注意事项

• 适合工具数量多的场景

---

2.3.10 Tool retry（工具重试）

对失败的工具调用自动重试，支持指数退避，提升可靠性。

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, database_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=60.0,
            jitter=True,
            tools=["api_tool"],
            retry_on=(ConnectionError, TimeoutError),
            on_failure="return_message",
        ),
    ],
)

代码解读

• max_retries：最大重试次数
• backoff_factor：指数退避倍数
• initial_delay：初始等待时间（秒）
• jitter：是否加随机抖动，防止请求集中
• retry_on：指定重试异常类型
• on_failure：重试失败后的处理方式

注意事项

• 适合网络不稳定或外部API易暂时失败的场景

---

2.3.11 LLM tool emulator（工具模拟）

用 LLM 生成模拟工具响应，便于测试和开发。

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

@tool
def get_weather(location: str) -> str:
    return f"Weather in {location}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    return "Email sent"

agent = create_agent(
    model="gpt-4o",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator()],
)

代码解读

• 所有工具调用都由 LLM 生成响应，不执行真实操作
• 支持指定部分工具模拟或自定义模拟模型

注意事项

• 适合开发阶段、工具不可用或成本高昂时使用

---

2.3.12 Context editing（上下文编辑）

管理对话上下文，自动清理旧工具调用结果，防止上下文过大。

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool, database_tool],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=2000,
                    keep=3,
                    clear_tool_inputs=False,
                    exclude_tools=[],
                    placeholder="[cleared]",
                ),
            ],
        ),
    ],
)

代码解读

• ClearToolUsesEdit：当对话token数超过阈值时，清除较早的工具输出
• keep：保留最近调用数
• placeholder：被清理内容的占位符

注意事项

• 有助于长对话降低token消耗
• 可排除特定工具不被清理

---

2.3.13 Shell tool（Shell 工具）

为代理提供持久 shell 会话，支持执行系统命令。

from langchain.agents import create_agent
from langchain.agents.middleware import ShellToolMiddleware, HostExecutionPolicy

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            execution_policy=HostExecutionPolicy(),
        ),
    ],
)

代码解读

• workspace_root：shell 会话的根目录
• execution_policy：执行策略（本地主机、Docker隔离、Codex沙箱）

注意事项

• 适合需要执行文件系统操作、脚本或自动化命令的代理
• 注意安全隔离策略

---

2.3.14 File search（文件搜索）

提供基于文件系统的 Glob 和 Grep 搜索工具，支持文件名和内容正则搜索。

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",
            use_ripgrep=True,
            max_file_size_mb=10,
        ),
    ],
)

代码解读

• root_path：搜索根目录
• use_ripgrep：优先使用 ripgrep 工具，性能更好
• max_file_size_mb：跳过大文件

注意事项

• 适合代码库浏览、分析和文件查找
• 支持多种输出模式和过滤

---

2.4 特定供应商中间件

Built-in middleware

2.4.1 Anthropic 专用中间件

中间件名称	功能描述
Prompt caching	缓存固定提示内容，降低调用成本和延迟
Bash tool	使用 Claude 原生 bash 工具执行本地命令
Text editor	提供 Claude 的文本编辑工具
Memory	提供 Claude 的持久记忆工具
File search	支持基于状态的文件搜索

2.4.2 Prompt caching（提示缓存）

缓存对话中静态或重复内容，减少重复计算和API调用成本。

from langchain_anthropic import ChatAnthropic
from langchain_anthropic.middleware import AnthropicPromptCachingMiddleware
from langchain.agents import create_agent

LONG_PROMPT = """
Please be a helpful assistant.

<Lots more context ...>
"""

agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    system_prompt=LONG_PROMPT,
    middleware=[AnthropicPromptCachingMiddleware(ttl="5m")],
)

# 调用示例
agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Bob"}]})
agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]})

代码解读

• 缓存系统提示、工具定义和历史消息
• ttl 指缓存时间（如5分钟）
• 后续请求重用缓存，节省成本和延迟

---

2.4.3 OpenAI 专用中间件

中间件名称	功能描述
Content moderation	使用 OpenAI 审核端点检测不安全内容

2.4.4 Content moderation（内容审核）

检测用户输入、模型输出和工具结果中的不安全内容，支持多种处理策略。

from langchain_openai import ChatOpenAI
from langchain_openai.middleware import OpenAIModerationMiddleware
from langchain.agents import create_agent

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool, customer_data_tool],
    middleware=[
        OpenAIModerationMiddleware(
            model="omni-moderation-latest",
            check_input=True,
            check_output=True,
            exit_behavior="end",
        ),
    ],
)

代码解读

• 审核模型选项多样
• 可审核输入、输出和工具结果
• 违规时可结束、抛错或替换内容

2.5 自定义中间件开发：Custom middleware

本文面向初学者，介绍如何在 LangChain 框架中构建自定义中间件。中间件是插入在代理（agent）执行流程中的钩子（hooks），用于拦截和控制代理的行为。通过实现钩子函数，可以在特定执行点插入自定义逻辑，如日志记录、重试机制、缓存等。

2.5.1 钩子（Hooks）:Node-style 钩子

中间件提供两种样式的钩子函数，用于拦截代理执行过程中的不同阶段：

钩子类型	作用方式	典型用途	可用钩子名称
Node-style	顺序执行，在特定节点运行	日志、验证、状态更新	before_agent, before_model, after_model, after_agent
Wrap-style	拦截执行，控制是否调用处理函数，支持重试等	重试、缓存、请求转换	wrap_model_call, wrap_tool_call

Node-style 钩子

这些钩子会按顺序执行，适合在执行流程中插入观察或简单处理逻辑。

• before_agent：代理开始前执行（每次调用一次）
• before_model：每次模型调用前执行
• after_model：每次模型响应后执行
• after_agent：代理完成后执行（每次调用一次）

from langchain.agents.middleware import before_model, after_model, AgentState
from langchain.messages import AIMessage
from langgraph.runtime import Runtime
from typing import Any

@before_model(can_jump_to=["end"])
def check_message_limit(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    if len(state["messages"]) >= 50:
        return {
            "messages": [AIMessage("Conversation limit reached.")],
            "jump_to": "end"
        }
    return None

@after_model
def log_response(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"Model returned: {state['messages'][-1].content}")
    return None

代码解读

• check_message_limit：在模型调用前检查消息数量，超过50条则跳转结束
• log_response：模型调用后打印最新回复内容

注意事项

• can_jump_to=["end"] 允许钩子通过返回 jump_to 指令提前结束执行

---

2.5.2 第二种：Wrap-style 钩子

这类钩子拦截执行过程，可以决定是否调用后续处理函数，支持多次调用（如重试）。

• wrap_model_call：包裹每次模型调用
• wrap_tool_call：包裹每次工具调用

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            if attempt == 2:
                raise
            print(f"Retry {attempt + 1}/3 after error: {e}")

代码解读

• retry_model 钩子为模型调用添加最多3次重试机制
• 通过调用 handler(request) 实际执行模型调用

注意事项

• 可根据异常类型或其他条件自定义重试逻辑

---

2.5.3 创建中间件的方法: 装饰器方式（Decorator-based）

适合快速实现单个钩子函数，无需复杂配置。

可用装饰器如下：

装饰器名称	作用描述
@before_agent	代理开始前执行（每次调用一次）
@before_model	每次模型调用前执行
@after_model	每次模型调用后执行
@after_agent	代理完成后执行（每次调用一次）
@wrap_model_call	包裹每次模型调用
@wrap_tool_call	包裹每次工具调用
@dynamic_prompt	生成动态系统提示

from langchain.agents.middleware import before_model, wrap_model_call
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse
from langchain.agents import create_agent
from langgraph.runtime import Runtime
from typing import Any, Callable

@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    print(f"About to call model with {len(state['messages'])} messages")
    return None

@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            if attempt == 2:
                raise
            print(f"Retry {attempt + 1}/3 after error: {e}")

agent = create_agent(
    model="gpt-4o",
    middleware=[log_before_model, retry_model],
    tools=[...],
)

代码解读

• log_before_model：模型调用前打印消息数
• retry_model：模型调用失败时重试最多3次
• 通过 middleware 参数将钩子传入代理

注意事项

• 装饰器方式适合简单、单钩子场景
• 不适合复杂配置或多钩子组合

---

2.5.4 创建中间件的方法: 类方式（Class-based）

适合需要多个钩子、同步异步支持或复杂配置的中间件。

from langchain.agents.middleware import AgentMiddleware, AgentState
from langgraph.runtime import Runtime
from typing import Any

class LoggingMiddleware(AgentMiddleware):
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"About to call model with {len(state['messages'])} messages")
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        print(f"Model returned: {state['messages'][-1].content}")
        return None

agent = create_agent(
    model="gpt-4o",
    middleware=[LoggingMiddleware()],
    tools=[...],
)

代码解读

• LoggingMiddleware 类实现两个钩子
• 通过实例化类传入代理作为中间件

注意事项

• 适合需要多钩子、异步支持和配置的场景
• 方便跨项目复用

---

2.5.5 扩展状态结构:agent

中间件可以扩展代理状态，添加自定义字段以支持更复杂逻辑。

from langchain.agents.middleware import AgentState, before_model, after_model
from typing_extensions import NotRequired
from typing import Any
from langgraph.runtime import Runtime

class CustomState(AgentState):
    model_call_count: NotRequired[int]
    user_id: NotRequired[str]

@before_model(state_schema=CustomState, can_jump_to=["end"])
def check_call_limit(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    count = state.get("model_call_count", 0)
    if count > 10:
        return {"jump_to": "end"}
    return None

@after_model(state_schema=CustomState)
def increment_counter(state: CustomState, runtime: Runtime) -> dict[str, Any] | None:
    return {"model_call_count": state.get("model_call_count", 0) + 1}

agent = create_agent(
    model="gpt-4o",
    middleware=[check_call_limit, increment_counter],
    tools=[...],
)

# 调用时传入自定义状态字段
result = agent.invoke({
    "messages": [HumanMessage("Hello")],
    "model_call_count": 0,
    "user_id": "user-123",
})

代码解读

• 通过继承 AgentState 添加 model_call_count 和 user_id 字段
• 钩子函数使用扩展状态类型，方便访问和更新自定义字段
• 代理调用时可以传入自定义状态

注意事项

• 扩展状态时需保证字段类型正确
• 方便实现复杂逻辑，如调用次数统计、用户身份识别

---

2.5.6 多中间件执行顺序

当代理配置多个中间件时，执行顺序如下：

执行阶段	执行顺序说明
before_* hook	按中间件列表顺序依次执行（先注册先执行）
wrap_* hook	嵌套调用，最先注册的中间件最外层包裹，依次向内调用
after_* hook	逆序执行（后注册先执行）

执行流程示意：

Before hooks run in order:
middleware1.before_agent()
middleware2.before_agent()
middleware3.before_agent()

Agent loop starts
middleware1.before_model()
middleware2.before_model()
middleware3.before_model()

Wrap hooks nest like function calls:
middleware1.wrap_model_call() → middleware2.wrap_model_call() → middleware3.wrap_model_call() → model

After hooks run in reverse order:
middleware3.after_model()
middleware2.after_model()
middleware1.after_model()

Agent loop ends
middleware3.after_agent()
middleware2.after_agent()
middleware1.after_agent()

---

2.5.7 代理跳转（Agent jumps）

中间件可以通过返回一个带 jump_to 字段的字典，控制代理流程跳转，提前结束或跳转到指定节点。

跳转目标	说明
'end'	跳转到代理执行结束（触发第一个 after_agent 钩子）
'tools'	跳转到工具调用节点
'model'	跳转到模型调用节点（触发第一个 before_model 钩子）

from langchain.agents.middleware import after_model, hook_config, AgentState
from langchain.messages import AIMessage
from langgraph.runtime import Runtime
from typing import Any

@after_model
@hook_config(can_jump_to=["end"])
def check_for_blocked(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    last_message = state["messages"][-1]
    if "BLOCKED" in last_message.content:
        return {
            "messages": [AIMessage("I cannot respond to that request.")],
            "jump_to": "end"
        }
    return None

代码解读

• 检查模型回复中是否包含“BLOCKED”关键词
• 若包含，返回跳转结束指令，并发送拒绝消息

3 高阶： LangChain 安全护栏（Guardrails）

在构建安全合规的 AI 应用时，护栏（Guardrails）是非常重要的一环。它们通过在代理（agent）执行的关键节点验证和过滤内容，防止敏感信息泄露、不安全行为以及不合规的操作，保障应用的安全稳定运行。本文介绍了 LangChain 中护栏的概念、内置功能、实现方式及示例，帮助初学者理解并应用护栏机制。

护栏主要通过验证和过滤内容，防止问题发生在源头。常见的应用包括：

• 防止个人身份信息（PII）泄露
• 检测并阻止提示注入攻击（prompt injection）
• 屏蔽不适当或有害内容
• 执行业务规则和合规要求
• 验证输出质量和准确性

护栏通常通过中间件（middleware）实现，拦截代理执行流程中的关键点，如代理启动前、执行后、模型调用和工具调用周围。

图示：中间件执行流程示意图，来源 LangChain 官方

---

3 LangChain 护栏实现方式

guardrails

LangChain 支持两种互补的护栏实现方式：

方式	说明	适用场景
内置护栏（Built-in guardrails）	LangChain 自带的中间件功能，如 PII 检测、人工审核等	常见安全需求，快速集成
自定义护栏（Custom guardrails）	用户根据需求编写自定义中间件，灵活控制验证和过滤逻辑	复杂业务规则、个性化安全策略

---

3.1 内置护栏详解

3.1.1 个人身份信息（PII）检测

LangChain 提供内置的 PII 检测中间件，用于识别和处理对话中的敏感个人信息，如邮箱、信用卡号、IP 地址等。适用于医疗、金融等合规要求高的行业，或需要保护用户隐私的客服系统。

处理策略	说明	示例
redact	替换为 [REDACTED_TYPE]	[REDACTED_EMAIL]
mask	部分掩码（如信用卡只显示后4位）	****-****-****-1234
hash	用确定性哈希替换	a8f5f167...
block	发现即抛异常，阻止继续执行	抛出错误

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[customer_service_tool, email_tool],
    middleware=[
        # 用户输入中邮箱用 redact 替换
        PIIMiddleware(
            "email",
            strategy="redact",
            apply_to_input=True,
        ),
        # 信用卡用 mask 部分掩码
        PIIMiddleware(
            "credit_card",
            strategy="mask",
            apply_to_input=True,
        ),
        # API Key 检测并阻止
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",
            strategy="block",
            apply_to_input=True,
        ),
    ],
)

# 当用户输入包含 PII 时，按策略处理
result = agent.invoke({
    "messages": [{"role": "user", "content": "My email is john.doe@example.com and card is 5105-1051-0510-5100"}]
})

代码解读

• 三个 PII 中间件分别处理邮箱、信用卡和 API Key
• apply_to_input=True 表示检测用户输入内容
• detector 支持正则表达式或自定义检测器
• 根据策略自动替换、掩码或阻止含敏感信息的请求

注意事项

• 内置支持邮箱、信用卡、IP、MAC 地址、URL 等多种 PII 类型
• 可自定义检测器满足特殊需求
• 可配置检测时机（输入、模型输出、工具结果）

---

3.2 另一大内置护栏：人工审核（Human-in-the-loop）

人工审核中间件在执行敏感操作前暂停，等待人工批准。适合金融交易、生产数据修改、对外发送通讯等高风险操作。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, send_email_tool, delete_database_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # 敏感操作需人工审批
                "send_email": True,
                "delete_database": True,
                # 安全操作自动通过
                "search": False,
            }
        ),
    ],
    # 状态持久化，支持中断恢复
    checkpointer=InMemorySaver(),
)

# 需要线程ID保持上下文
config = {"configurable": {"thread_id": "some_id"}}

# 调用代理，敏感操作暂停等待人工批准
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Send an email to the team"}]},
    config=config
)

# 人工批准后恢复执行
result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config
)

代码解读

• interrupt_on 配置哪些工具调用需人工审批
• 使用 InMemorySaver 持久化对话状态
• 代理调用时传入线程 ID 以支持暂停后恢复

注意事项

• 适用于高风险业务场景
• 需人工介入，可能影响响应时效

---

3.3 自定义护栏

除了内置护栏，用户还可以编写自定义中间件，在代理执行前后插入任意验证、过滤逻辑。

护栏类型	作用点	典型用途
Before agent	代理启动前运行一次	鉴权、限流、请求过滤
After agent	代理完成后运行一次	输出质量校验、安全扫描

多层护栏可以组合使用，形成多重保护。

3.3.1 Before agent 拦截

使用“代理前”钩子在每次调用开始时对请求进行一次验证。这对于会话级别的检查非常有用，比如身份验证、速率限制，或在任何处理开始之前阻止不当请求。

from typing import Any

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime

class ContentFilterMiddleware(AgentMiddleware):
    """Deterministic guardrail: Block requests containing banned keywords."""

    def __init__(self, banned_keywords: list[str]):
        super().__init__()
        self.banned_keywords = [kw.lower() for kw in banned_keywords]

    @hook_config(can_jump_to=["end"])
    def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        # Get the first user message
        if not state["messages"]:
            return None

        first_message = state["messages"][0]
        if first_message.type != "human":
            return None

        content = first_message.content.lower()

        # Check for banned keywords
        for keyword in self.banned_keywords:
            if keyword in content:
                # Block execution before any processing
                return {
                    "messages": [{
                        "role": "assistant",
                        "content": "I cannot process requests containing inappropriate content. Please rephrase your request."
                    }],
                    "jump_to": "end"
                }

        return None

# Use the custom guardrail
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool],
    middleware=[
        ContentFilterMiddleware(
            banned_keywords=["hack", "exploit", "malware"]
        ),
    ],
)

# This request will be blocked before any processing
result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I hack into a database?"}]
})

3.3.2 After agent 拦截

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langgraph.runtime import Runtime
from langchain.messages import AIMessage
from langchain.chat_models import init_chat_model
from typing import Any

class SafetyGuardrailMiddleware(AgentMiddleware):
    """Model-based guardrail: Use an LLM to evaluate response safety."""

    def __init__(self):
        super().__init__()
        self.safety_model = init_chat_model("gpt-4o-mini")

    @hook_config(can_jump_to=["end"])
    def after_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        # Get the final AI response
        if not state["messages"]:
            return None

        last_message = state["messages"][-1]
        if not isinstance(last_message, AIMessage):
            return None

        # Use a model to evaluate safety
        safety_prompt = f"""Evaluate if this response is safe and appropriate.
        Respond with only 'SAFE' or 'UNSAFE'.

        Response: {last_message.content}"""

        result = self.safety_model.invoke([{"role": "user", "content": safety_prompt}])

        if "UNSAFE" in result.content:
            return {
                "messages": [{
                    "role": "assistant",
                    "content": "I cannot provide that response. Please rephrase your request."
                }],
                "jump_to": "end"
            }

        return None

# Use the safety guardrail
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4o",
    tools=[search_tool, calculator_tool],
    middleware=[SafetyGuardrailMiddleware()],
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "How do I make explosives?"}]
})

4 Agent 上下文工程

构建可靠的代理（agent）或任何大型语言模型（LLM）应用是件不容易的事。虽然原型可能跑得通，但在真实场景中经常会失败。本文将详细介绍导致代理失败的原因、代理的核心执行流程，以及如何通过上下文工程提升代理的可靠性。

代理失败通常是因为代理内部的 LLM 调用做出了错误的操作，或者没有按预期执行。LLM 失败主要有两个原因：

1. 底层的 LLM 能力不足
2. 没有向 LLM 传递“正确”的上下文信息

多数情况下，第二点才是导致代理不可靠的主要原因。所谓上下文工程，就是以正确的格式向 LLM 提供恰当的信息和工具，帮助它完成任务。这是 AI 工程师的核心工作。缺乏“正确”上下文是阻碍代理稳定性的最大障碍，而 LangChain 的代理抽象设计正是为支持上下文工程而生。

一个典型的代理执行循环包含两个主要步骤：

1. 模型调用（Model call）
   使用提示（prompt）和可用工具调用 LLM，返回模型响应或请求执行工具的指令。
2. 工具执行（Tool execution）
   执行 LLM 请求的工具操作，返回工具结果。

图示：LangChain 核心代理循环，来源 LangChain 官方

这个循环会一直执行，直到 LLM 决定结束。

为了构建可靠的代理，你需要控制代理循环中每一步以及步骤之间的行为。

上下文类型	你能控制的内容	持久性
模型上下文（Model Context）	模型调用时使用的内容（指令、消息历史、工具、响应格式）	瞬态（每次调用）
工具上下文（Tool Context）	工具可以访问和产生的内容（读写状态、存储、运行时上下文）	持久
生命周期上下文（Life-cycle Context）	模型调用和工具调用之间发生的操作（摘要、护栏、日志等）	持久

---

4.1 上下文数据获取来源

代理在执行过程中会访问多种数据源，既包括读取也包括写入：

数据源	其他称呼	作用范围	示例
运行时上下文	静态配置	会话范围	用户ID、API密钥、数据库连接、权限、环境配置
状态	短期记忆	会话范围	当前消息、上传文件、认证状态、工具结果
存储	长期记忆	跨会话	用户偏好、提取的洞察、记忆、历史数据

控制每次模型调用时输入的内容，包括指令、可用工具、所用模型和输出格式。这些直接影响代理的可靠性和成本。

模型上下文可以从以下三类数据源获取信息：

• 状态（短期记忆）
• 存储（长期记忆）
• 运行时上下文（静态配置）

4.2 注入上传文件的上下文信息

消息组成了发送给 LLM 的提示内容。合理管理消息内容，确保 LLM 拥有正确的信息回应非常关键。

示例：在相关查询时注入上传文件的上下文信息

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable

@wrap_model_call
def inject_file_context(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """注入用户本会话上传文件的上下文信息。"""
    # 从状态中读取上传文件元数据
    uploaded_files = request.state.get("uploaded_files", [])

    if uploaded_files:
        # 构建文件描述
        file_descriptions = []
        for file in uploaded_files:
            file_descriptions.append(
                f"- {file['name']} ({file['type']}): {file['summary']}"
            )

        file_context = f"""Files you have access to in this conversation:
{chr(10).join(file_descriptions)}

Reference these files when answering questions."""

        # 在最近消息之前注入文件上下文
        messages = [
            *request.messages,
            {"role": "user", "content": file_context},
        ]
        request = request.override(messages=messages)

    return handler(request)

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[inject_file_context]
)

代码解读

• 读取当前会话上传的文件信息
• 将文件信息作为新消息注入给模型，辅助回答相关问题

注意事项

• 这种注入是临时的，针对当前模型调用生效

4.3 模型选择（Model）

不同模型有不同优势、成本和上下文窗口大小。根据任务选择合适的模型，且模型选择可能随代理运行过程变化。

示例：根据对话长度选择不同模型

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

# 初始化模型实例
large_model = init_chat_model("claude-sonnet-4-5-20250929")
standard_model = init_chat_model("gpt-4o")
efficient_model = init_chat_model("gpt-4o-mini")

@wrap_model_call
def state_based_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    # 根据消息数量选择模型
    message_count = len(request.messages)

    if message_count > 20:
        model = large_model
    elif message_count > 10:
        model = standard_model
    else:
        model = efficient_model

    request = request.override(model=model)

    return handler(request)

agent = create_agent(
    model="gpt-4o-mini",
    tools=[...],
    middleware=[state_based_model]
)

代码解读

• 对话长时使用上下文窗口更大的模型
• 对话短时用轻量模型节省资源

注意事项

• 模型切换需平滑，避免上下文丢失

更多示例见：动态模型选择。

---

4.4 响应格式（Response Format）

结构化输出将非结构化文本转成经过验证的结构化数据。提取字段或供下游系统使用时，自由文本不够用。

工作原理：当你提供一个响应格式的 schema，模型最终输出会保证符合该 schema。代理会循环调用模型和工具，直到模型结束调用工具，然后将最终结果转换为指定格式。

4.5 工具上下文（Tool Context）

工具既读取也写入上下文。最基本的情况是工具接收 LLM 的请求参数，执行工作后返回结果。工具还可以获取关键信息，帮助模型完成任务。

大多数工具需要的不只是 LLM 参数，还需要用户 ID、API 密钥或当前会话状态等信息。工具通过读取状态、存储和运行时上下文获取这些信息。

4.5.1 读入

示例：读取状态判断用户是否认证

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

@tool
def check_authentication(
    runtime: ToolRuntime
) -> str:
    """检查用户是否已认证。"""
    current_state = runtime.state
    is_authenticated = current_state.get("authenticated", False)

    if is_authenticated:
        return "User is authenticated"
    else:
        return "User is not authenticated"

agent = create_agent(
    model="gpt-4o",
    tools=[check_authentication]
)

4.5.2 写入

工具结果可以直接返回给模型，也可以更新代理的记忆，使未来步骤能访问重要上下文。

示例：工具认证用户并更新状态

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
from langgraph.types import Command

@tool
def authenticate_user(
    password: str,
    runtime: ToolRuntime
) -> Command:
    """认证用户并更新状态。"""
    if password == "correct":
        return Command(
            update={"authenticated": True},
        )
    else:
        return Command(update={"authenticated": False})

agent = create_agent(
    model="gpt-4o",
    tools=[authenticate_user]
)

更多示例见：工具文档。

---

4.6 生命周期上下文（Life-cycle Context）

控制核心代理步骤之间发生的事情——截取数据流，实现跨切面功能，如摘要、护栏、日志等。正如你在模型上下文和工具上下文中看到的，中间件是实现上下文工程的关键。

中间件允许你：

1. 更新上下文 — 修改状态和存储，持久化更改，更新对话历史或保存洞察
2. 跳转生命周期 — 根据上下文跳转到代理循环的不同步骤（例如满足条件时跳过工具执行，或修改上下文后重复模型调用）

图示：代理循环中的中间件钩子，来源 LangChain 官方

最常见的生命周期模式之一是当对话历史过长时自动压缩。与模型上下文中短暂的消息截断不同，摘要中间件持久更新状态，用摘要永久替换旧消息，供未来所有对话使用。LangChain 提供了内置的摘要中间件：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            max_tokens_before_summary=4000,  # 超过4000令牌触发摘要
            messages_to_keep=20,  # 摘要后保留最近20条消息
        ),
    ],
)

当对话超过令牌限制时，SummarizationMiddleware 会自动：

1. 使用独立模型调用总结旧消息
2. 用摘要消息永久替换状态中的旧消息
3. 保留近期消息供上下文使用

未来对话将基于摘要而非原始消息。

5 多Agent模式

multi-agent

多代理系统（Multi-agent systems）将复杂应用拆分成多个专业化代理协同工作。与依赖单一代理处理所有步骤不同，多代理架构允许你将多个小而专注的代理组合成协调的工作流。

多代理系统适合以下场景：

• 单个代理拥有过多工具时，容易做出错误的使用决策。
• 上下文或记忆信息过大，单个代理难以有效跟踪。
• 任务需要专业化分工（例如计划者、研究员、数学专家等）。

模式	工作原理	控制流	典型应用场景
工具调用 (Tool Calling)	主管理代理（supervisor）将其他代理当作“工具”调用。这些“工具”代理不直接与用户交互，只执行任务并返回结果。	集中式：所有路由通过调用代理进行。	任务编排，有结构化工作流。
控制权转移 (Handoffs)	当前代理决定将控制权转给另一个代理。活跃代理发生变化，用户可直接与新代理互动。	去中心化：代理可变更活跃代理身份。	多领域对话，专家接管场景。

选择合适的多代理模式

问题	工具调用	控制权转移
需要对工作流进行集中控制吗？	✅ 是	❌ 否
想让代理直接与用户交互吗？	❌ 否	✅ 是
需要复杂、类人专家间的对话吗？	❌ 有限	✅ 强

---

5.1 工具调用模式详解

在工具调用模式中，一个代理（称为“控制器”）将其他代理视为需要时调用的“工具”。控制器负责协调，工具代理执行具体任务并返回结果。流程如下：

1. 控制器接收输入，决定调用哪个工具（子代理）。
2. 工具代理根据控制器指令执行任务。
3. 工具代理将结果返回给控制器。
4. 控制器决定下一步操作或结束。

用户 → 控制器代理 → 工具代理1 / 工具代理2 → 用户响应

下面是一个最简示例，主代理通过工具定义访问单个子代理：

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

subagent1 = create_agent(model="...", tools=[...])

@tool(
    "subagent1_name",
    description="subagent1_description"
)
def call_subagent1(query: str):
    result = subagent1.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content

agent = create_agent(model="...", tools=[call_subagent1])

代码解读：

• 创建一个子代理 subagent1，负责特定任务。
• 定义一个工具函数 call_subagent1，将查询传给子代理并返回结果。
• 主代理通过工具调用子代理，完成任务分配。

注意事项：

• 工具名称和描述影响主代理调用决策。
• 子代理独立运行，返回结果给主代理。

---

5.2 可定制的上下文传递点

你可以在多个环节控制主代理与子代理间的上下文传递：

控制点	说明
子代理名称	主代理如何称呼子代理，影响提示和调用判断。
子代理描述	主代理对该子代理的认知，影响调用时机和方式。
传入子代理的输入	可定制传给子代理的输入内容，帮助其更好理解任务。
子代理返回的输出	可调整返回内容，控制主代理如何解读结果。

主要有两种方式：

• 修改提示：调整主代理的提示或工具元数据（子代理名和描述），更好地指导调用时机和方式。
• 上下文注入：通过代码从主代理状态中提取更多上下文（如完整消息历史、之前结果、任务元数据）传给子代理。

示例：用自定义状态和上下文注入调用子代理

from langchain.agents import AgentState
from langchain.tools import tool, ToolRuntime

class CustomState(AgentState):
    example_state_key: str

@tool(
    "subagent1_name",
    description="subagent1_description"
)
def call_subagent1(query: str, runtime: ToolRuntime[None, CustomState]):
    # 利用自定义逻辑处理输入消息和状态
    subagent_input = some_logic(query, runtime.state["messages"])
    result = subagent1.invoke({
        "messages": subagent_input,
        "example_state_key": runtime.state["example_state_key"]
    })
    return result["messages"][-1].content

代码解读：

• 定义自定义状态类，扩展状态字段。
• 工具函数中通过 runtime.state 访问主代理状态，注入更多上下文给子代理。

注意事项：

• 确保主代理和子代理状态模式一致。
• 上下文注入提升子代理理解能力。

---

5.3 控制子代理的输出

常用策略：

• 修改提示：优化子代理提示，明确应返回的内容，避免遗漏或冗长。提醒子代理所有重要信息必须包含在最终消息中，因为控制器和用户只看到最终输出。
• 自定义输出格式：在代码中调整或丰富子代理响应，再返回给主代理。可以返回额外状态或元数据，需用 Command 对象封装，方便合并状态。

示例：返回带状态更新的自定义输出

from typing import Annotated
from langchain.agents import AgentState
from langchain.tools import InjectedToolCallId
from langgraph.types import Command, ToolMessage

@tool(
    "subagent1_name",
    description="subagent1_description"
)
def call_subagent1(
    query: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    result = subagent1.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return Command(update={
        "example_state_key": result["example_state_key"],
        "messages": [
            ToolMessage(
                content=result["messages"][-1].content,
                tool_call_id=tool_call_id
            )
        ]
    })

代码解读：

• 通过 Command 返回状态更新和消息，保持调用一致性。
• 使用 tool_call_id 确保工具调用结果正确匹配。

注意事项：

• 需要明确返回格式，避免信息丢失。
• 适合复杂交互和状态管理。

---

5.4 控制权转移模式（Handoffs）

在控制权转移模式中，代理可以直接将控制权传递给其他代理。当前活跃代理变化，用户直接与新活跃代理交互。流程：

1. 当前代理判断需要其他代理协助。
2. 将控制权和状态传给下一个代理。
3. 新代理直接与用户交互，直到再次转移或结束。

该模式适合多领域对话和专家接管场景。

6 Retrieval - 检索

retrieval

大型语言模型（LLMs）非常强大，但它们有两个主要限制：

• 有限的上下文 — 它们不能一次性处理整个语料库。
• 静态的知识 — 它们的训练数据是固定在某个时间点的。

检索技术通过在查询时获取相关的外部知识来解决这些问题。这也是**检索增强生成（Retrieval-Augmented Generation，简称RAG）**的基础：用上下文相关的信息增强LLM的回答能力。

6.1 构建知识库

知识库是用于检索的文档或结构化数据的仓库。

如果你需要自定义知识库，可以使用LangChain的文档加载器（document loaders）和向量存储（vector stores）从自己的数据构建。

如果你已经有了知识库（例如SQL数据库、客户关系管理系统CRM或内部文档系统），不需要重新构建。你可以：

• 将它作为Agentic RAG中的工具连接给代理（agent）。
• 查询它并将检索到的内容作为上下文提供给LLM（即2-Step RAG）。

下面是一个教程，教你如何构建一个可搜索的知识库和最简RAG流程：

教程：语义搜索
学习如何使用LangChain的文档加载器、嵌入模型和向量存储，从自己的数据创建一个可搜索的知识库。
这个教程会教你如何基于PDF构建搜索引擎，实现根据查询检索相关段落，并在此基础上实现最简RAG工作流，展示如何将外部知识整合进LLM推理中。

6.2 从检索到RAG

检索允许LLM在运行时访问相关上下文，但大多数实际应用更进一步：它们将检索与生成结合，产生有依据、上下文感知的答案。

这就是**检索增强生成（RAG）**的核心思想。检索流程成为结合搜索和生成的基础系统。

6.2.1 检索流程

一个典型的检索工作流程如下：

每个组件都是模块化的：你可以替换加载器、分割器、嵌入模型或向量存储，而无需重写应用逻辑。

6.2.2 构建模块介绍

模块	作用说明
文档加载器	从外部来源（Google Drive、Slack、Notion等）导入数据，返回标准化的Document对象。
文本分割器	将大文档拆分成更小的块，使得每个块可以单独检索并能适配模型的上下文窗口大小。
嵌入模型	将文本转换成数字向量，使含义相似的文本在向量空间中距离更近。
向量存储	专门用于存储和搜索嵌入向量的数据库。
检索器	接收非结构化查询，返回相关文档的接口。

6.3 RAG架构

RAG可以有多种实现方式，取决于系统需求。下面是几种主要架构的对比：

架构	描述	控制力	灵活性	延迟	典型应用场景
2-Step RAG	先检索再生成，简单且预测性强	✅ 高	❌ 低	⚡ 快	FAQ、文档机器人
Agentic RAG	由LLM驱动的代理决定何时及如何检索，边推理边检索	❌ 低	✅ 高	⏳ 不定	需要访问多工具的研究助理
Hybrid	结合两者特点，加入验证步骤，平衡控制力和灵活性	⚖️ 中等	⚖️ 中等	⏳ 不定	领域特定问答，带质量验证

延迟说明：
2-Step RAG的延迟通常更可预测，因为LLM调用次数固定且有限。假设LLM推理时间为主导因素。但实际延迟也会受检索步骤性能影响，比如API响应、网络延迟、数据库查询等。

6.3.1 2-Step RAG

在2-Step RAG中，检索步骤总是先于生成执行。这种架构简单且可预测，适合检索相关文档是生成答案前提的应用。

教程推荐：检索增强生成（RAG）
学习如何构建一个基于数据的问答聊天机器人。教程覆盖两种方法：

• 灵活使用工具运行搜索的RAG代理，适合通用用途。
• 只需一次LLM调用的2-Step RAG链，适合简单任务，快速高效。

6.3.2 Agentic RAG

Agentic RAG结合了RAG和基于代理的推理优势。代理（由LLM驱动）在交互过程中逐步推理，并决定何时以及如何检索信息。

代理实现RAG行为的唯一需求是能访问一个或多个工具，这些工具能获取外部知识，如文档加载器、Web API或数据库查询。

下面是一个简单的Agentic RAG示例代码，演示如何用Python创建一个能通过网络抓取网页内容的工具：

import requests
from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

@tool
def fetch_url(url: str) -> str:
    """Fetch text content from a URL"""
    response = requests.get(url, timeout=10.0)
    response.raise_for_status()
    return response.text

system_prompt = """\
Use fetch_url when you need to fetch information from a web-page; quote relevant snippets.
"""

agent = create_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[fetch_url],  # 这是一个检索工具
    system_prompt=system_prompt,
)

代码解读：

• fetch_url 是一个工具函数，用于从指定URL获取网页文本内容。
• create_agent 创建了一个代理，使用指定的语言模型和工具，系统提示告诉代理何时使用该工具。
• 代理可在推理时调用fetch_url工具动态获取信息。

注意事项：

• 网络请求有超时设置，防止长时间挂起。
• 需要处理请求异常，确保程序健壮。

6.4 扩展示例：LangGraph文档的Agentic RAG

这个例子实现了一个Agentic RAG系统，用于帮助用户查询LangGraph文档。代理先加载一个包含文档URL列表的文件llms.txt，然后根据用户问题动态调用fetch_documentation工具检索相关内容。

import requests
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool
from markdownify import markdownify

ALLOWED_DOMAINS = ["https://langchain-ai.github.io/"]
LLMS_TXT = 'https://langchain-ai.github.io/langgraph/llms.txt'

@tool
def fetch_documentation(url: str) -> str:
    """Fetch and convert documentation from a URL"""
    if not any(url.startswith(domain) for domain in ALLOWED_DOMAINS):
        return (
            "Error: URL not allowed. "
            f"Must start with one of: {', '.join(ALLOWED_DOMAINS)}"
        )
    response = requests.get(url, timeout=10.0)
    response.raise_for_status()
    return markdownify(response.text)

# 预先获取llms.txt内容，无需每次调用时请求
llms_txt_content = requests.get(LLMS_TXT).text

system_prompt = f"""
You are an expert Python developer and technical assistant.
Your primary role is to help users with questions about LangGraph and related tools.

Instructions:

1. If a user asks a question you're unsure about — or one that likely involves API usage,
   behavior, or configuration — you MUST use the `fetch_documentation` tool to consult the relevant docs.
2. When citing documentation, summarize clearly and include relevant context from the content.
3. Do not use any URLs outside of the allowed domain.
4. If a documentation fetch fails, tell the user and proceed with your best expert understanding.

You can access official documentation from the following approved sources:

{llms_txt_content}

You MUST consult the documentation to get up to date documentation
before answering a user's question about LangGraph.

Your answers should be clear, concise, and technically accurate.
"""

tools = [fetch_documentation]

model = init_chat_model("claude-sonnet-4-0", max_tokens=32000)

agent = create_agent(
    model=model,
    tools=tools,  # 这是检索工具
    system_prompt=system_prompt,  # 系统提示
    name="Agentic RAG",
)

response = agent.invoke({
    'messages': [
        HumanMessage(content=(
            "Write a short example of a langgraph agent using the "
            "prebuilt create react agent. the agent should be able "
            "to look up stock pricing information."
        ))
    ]
})

print(response['messages'][-1].content)

代码解读：

• fetch_documentation工具限制只能访问指定域名，防止不安全请求。
• 使用markdownify将HTML文档转换成Markdown格式，便于处理和展示。
• 代理在回答用户问题前，会调用工具动态查询最新文档，保证答案时效和准确。
• 通过系统提示明确代理的行为规范和权限。

注意事项：

• 域名白名单保护安全。
• 文档转换和网络请求可能有性能开销。
• 需要合理控制模型最大token数，避免上下文过大。

6.5 Hybrid RAG

Hybrid RAG结合了2-Step和Agentic RAG的特点，加入了查询预处理、检索验证和答案验证等中间步骤，既比固定流程更灵活，又保持一定的执行控制。

典型组件包括：

• 查询增强：修改输入问题以提升检索质量，比如改写不清楚的查询，生成多种变体，或加上下文扩展。
• 检索验证：评估检索结果是否相关且充分，不够时可重新调整查询并再次检索。
• 答案验证：检查生成答案的准确性、完整性和与源内容的一致性，必要时重新生成或修正。

该架构通常支持多轮迭代：

这种架构适合：

• 查询模糊或信息不足的应用
• 需要质量控制或验证的系统
• 多来源、多轮次精炼的工作流

教程推荐：Agentic RAG与自我纠错
一个结合代理推理、检索和自我纠错的Hybrid RAG示例。

7 studio部署

本文介绍如何使用 Studio 来本地可视化、交互和调试你的智能代理（agent）。Studio 是一个免费且功能强大的智能代理集成开发环境（IDE），它与 LangSmith 集成，支持追踪、评估和提示工程。你可以清楚地看到代理的思考过程，追踪每一个决策，从而构建更智能、更可靠的代理。

下方是 Studio 的介绍视频：

7.1 准备工作

开始之前，请确保你已经准备好以下内容：

• 拥有一个 LangSmith 的 API 密钥（注册免费）。

7.2 本地代理服务器搭建流程

7.2.1 安装 LangGraph 命令行工具

确保你的 Python 版本不低于 3.11，然后运行以下命令安装 LangGraph CLI：

# Python >= 3.11 是必需的
pip install --upgrade "langgraph-cli[inmem]"

7.2.2 准备你的代理

以下示例代码展示了一个简单的邮件发送代理：

from langchain.agents import create_agent

def send_email(to: str, subject: str, body: str):
    """发送邮件的工具函数"""
    email = {
        "to": to,
        "subject": subject,
        "body": body
    }
    # 这里可以添加实际的邮件发送逻辑

    return f"Email sent to {to}"

agent = create_agent(
    "gpt-4o",
    tools=[send_email],
    system_prompt="You are an email assistant. Always use the send_email tool.",
)

代码解读：

• send_email 函数模拟了邮件发送操作，接收收件人、主题和正文参数，返回发送成功的字符串。
• create_agent 函数用来创建一个基于 GPT-4o 的智能代理，这个代理会使用 send_email 工具，并且系统提示（system_prompt）告诉代理它的角色是“邮件助手”，总是调用 send_email 工具。

注意事项：

• 这里的邮件发送逻辑是示意，实际应用中需要接入邮件服务。
• create_agent 返回的是一个编译好的 LangGraph 图，可以直接用于后续配置。

7.2.3 配置环境变量

在项目根目录下新建 .env 文件，填写你的 LangSmith API 密钥：

LANGSMITH_API_KEY=lsv2...

提醒： 请勿将 .env 文件提交到 Git 等版本控制系统，避免泄露密钥。

7.2.4 创建 LangGraph 配置文件

在应用目录下创建 langgraph.json 文件，内容如下：

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent.py:agent"
  },
  "env": ".env"
}

• dependencies 指明依赖路径。
• graphs 指定代理对象的位置，格式为 文件路径:对象名。
• env 指定环境变量文件。

create_agent 自动返回编译好的 LangGraph 图，直接赋值给 graphs 中的 agent。

更多配置细节可参考 LangGraph 配置文件参考。

7.2.5 项目结构示例

目前你的项目结构应类似如下：

my-app/
├── src
│   └── agent.py
├── .env
└── langgraph.json

7.2.6 安装依赖

在项目根目录执行：

pip install -e .
uv sync

• pip install -e . 安装当前项目依赖。
• uv sync 用于同步依赖。

7.2.7 启动代理服务器并访问 Studio

启动本地代理服务器：

langgraph dev

注意： Safari 浏览器可能会阻止 localhost 连接 Studio，遇到此问题请加上 --tunnel 参数启动，使用安全隧道访问 Studio。

服务器启动后，你的代理接口地址为：

http://127.0.0.1:2024

Studio UI 可通过以下地址访问：

https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

下图展示了 Studio 中代理视图的界面示例：

7.3 Studio 功能特点

• Studio 让代理的每一步都可视化，方便观察。
• 可以回放任何输入，查看具体的提示词（prompt）、工具参数、返回值以及令牌和延迟指标。
• 如果工具出现异常，Studio 会记录异常及其上下文，帮助快速定位问题。
• 你可以保持开发服务器运行，随时编辑提示词或工具接口，Studio 会热重载并实时反映变化。
• 支持从任意步骤重新运行对话线程，验证行为变化。
• 随着代理变复杂，Studio 可以扩展支持多节点图，保持决策过程清晰且可复现。