LangChain 1.x 新特性代码速查(拿来可用 + 纠错指南)

LangChain 1.x 是一次重大重构,统一了 Agent 创建方式,引入了中间件体系,并提供了更标准化的内容表示。本文旨在提供一份可直接复制使用的代码速查,同时指出常见错误和注意事项,帮你快速上手并避免踩坑。


一、核心 Agent 创建

1.1 统一入口:create_agent

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

# 初始化模型(统一API)
model = init_chat_model("openai:gpt-4", temperature=0.3)

# 定义工具
@tool
def calculate(expression: str) -> str:
    """执行数学计算"""
    return str(eval(expression))

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[calculate],
    system_prompt="你是一个数学助手,只回答计算问题。"
)

# 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "123 * 456 = ?"}]
})
print(result["messages"][-1].content)

❗️常见错误

  • ❌ 误用旧版 AgentExecutor:1.x 中不需要 AgentExecutorcreate_agent 返回的对象已可直接执行。
  • ❌ 导入错误:init_chat_model 应从 langchain.chat_models 导入,而非 langchain_community
  • ❌ 忘记传 messages 键:invoke 参数必须是 {"messages": [...]} 格式。

二、中间件体系(Middleware)

中间件允许你在 Agent 执行流程中插入自定义逻辑。

2.1 内置中间件示例

from langchain.agents import create_agent
from langchain.agents.middleware import (
    HumanInTheLoopMiddleware,
    SummarizationMiddleware,
    PIIMiddleware,
    LLMToolSelectorMiddleware,
    TodoListMiddleware
)

agent = create_agent(
    model=model,
    tools=[send_email, read_file, delete_file, search_web],
    middleware=[
        # 敏感操作需人工审批
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
                "delete_file": True
            }
        ),
        # 长对话自动摘要(基于消息数)
        SummarizationMiddleware(model=model, trigger=("messages", 10), keep=("messages", 5)),
        # 隐藏敏感信息(如邮箱、电话)
        PIIMiddleware(patterns=["email", "phone"]),
        # 智能选择工具(最多选3个)
        LLMToolSelectorMiddleware(model=model, max_tools=3),
        # 自动拆解复杂任务为子任务列表
        TodoListMiddleware(),
    ],
    checkpointer=MemorySaver(),  # 记忆支持
)

❗️注意

  • 中间件顺序重要:HumanInTheLoopMiddleware 通常放在靠前位置,以便尽早拦截。
  • interrupt_on 的键是工具名,值可以是 True 或详细配置对象。
  • SummarizationMiddlewaretriggerkeep 支持 ("messages", n)("tokens", n)("fraction", f)

2.2 自定义中间件

from langchain.agents.middleware import AgentMiddleware, ModelRequest
from langchain_openai import ChatOpenAI

class DynamicModelMiddleware(AgentMiddleware):
    def wrap_model_call(self, request: ModelRequest, handler):
        # 根据消息数量动态切换模型
        if len(request.state["messages"]) > 10:
            request.model = ChatOpenAI(model="gpt-5")  # 高级模型
        else:
            request.model = ChatOpenAI(model="gpt-5-nano")  # 轻量模型
        return handler(request)

agent = create_agent(
    model=model,  # 默认模型
    tools=[...],
    middleware=[DynamicModelMiddleware()],
    context_schema=dict,  # 可选上下文
)

三、工具定义与文档理解

3.1 使用 @tool 装饰器 + Pydantic 参数校验

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

class FileWriteInput(BaseModel):
    filename: str = Field(description="文件名,只能包含字母、数字、下划线", regex=r"^[a-zA-Z0-9_.]+$")
    content: str = Field(description="文件内容", max_length=10000)
    encoding: Literal["utf-8", "gbk"] = Field(default="utf-8", description="编码格式")

@tool(args_schema=FileWriteInput)
def write_file(filename: str, content: str, encoding: str = "utf-8") -> str:
    """写入文件(敏感操作)"""
    with open(filename, "w", encoding=encoding) as f:
        f.write(content)
    return f"已写入 {filename}"

❗️要点

  • args_schema 让模型严格按照 Pydantic 模型传参,避免类型错误。
  • Field 的 description 会被模型读取,帮助理解参数含义。
  • 支持 Literal、正则等约束,实现“文档理解能力”。

3.2 工具内访问 Agent 状态(ToolRuntime)

from langchain_core.tools import tool, ToolRuntime

@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """获取当前用户信息(从状态中读取)"""
    user_name = runtime.state.get("user_name", "未知用户")
    return f"当前用户:{user_name}"

四、记忆系统

4.1 短期记忆(Checkpointer)

from langgraph.checkpoint.memory import MemorySaver  # 或 InMemorySaver

checkpointer = MemorySaver()

agent = create_agent(
    model=model,
    tools=...,
    checkpointer=checkpointer
)

# 使用 thread_id 区分不同对话
config = {"configurable": {"thread_id": "user_001"}}
agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config=config)
result = agent.invoke({"messages": [{"role": "user", "content": "我是谁?"}]}, config=config)
print(result["messages"][-1].content)  # 输出:你是张三

❗️导入注意BaseCheckpointSaver 的正确路径是 from langgraph.checkpoint.base import BaseCheckpointSaver,但通常直接用 MemorySaver

4.2 长期记忆(Store)

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

agent = create_agent(
    model=model,
    tools=...,
    store=store,
    # 可选:定义自定义状态字段
    state_schema=UserState  # 自定义 TypedDict
)

# 跨线程共享长期记忆
agent.invoke(..., config={"configurable": {"thread_id": "session1"}})
agent.invoke(..., config={"configurable": {"thread_id": "session2"}})  # 可访问 store 中的共享数据

五、结构化输出

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

class Weather(BaseModel):
    temperature: float
    condition: str

def weather_tool(city: str) -> str:
    return f"{city} 天气晴,22°C"

# 使用工具策略(通用)
agent = create_agent(
    "openai:gpt-4o-mini",
    tools=[weather_tool],
    response_format=ToolStrategy(Weather)
)

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气如何?"}]})
print(result["structured_response"])  # Weather(temperature=22.0, condition='sunny')

❗️说明

  • ToolStrategy 通过调用工具生成结构化输出,兼容所有模型。
  • ProviderStrategy 利用模型原生结构化能力(如 OpenAI 的 response_format),需模型支持。

六、标准化内容块 content_blocks

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-5")
response = model.invoke("法国首都是什么?")

# 统一访问不同类型的内容
for block in response.content_blocks:
    if block["type"] == "reasoning":
        print(f"推理过程:{block['reasoning']}")
    elif block["type"] == "text":
        print(f"最终答案:{block['text']}")
    elif block["type"] == "tool_call":
        print(f"工具调用:{block['name']}({block['args']})")
    elif block["type"] == "citation":
        print(f"引用来源:{block['sources']}")

❗️注意

  • content_blocks 仅在支持的集成包中可用:anthropic、openai、google-genai、aws、ollama。
  • 旧版代码中直接使用 response.content 仍然兼容,但推荐迁移到 content_blocks 以获得更丰富的结构化信息。

七、子 Agent(SubAgent)

from deepagents import create_deep_agent

# 定义子 Agent 配置
search_subagent = {
    "name": "search_agent",
    "description": "擅长网络搜索,用于查找实时信息",
    "system_prompt": "你是一个搜索专家,只返回搜索结果。",
    "tools": [internet_search_tool],
    "model": model
}

# 主 Agent 集成子 Agent
agent = create_deep_agent(
    model=model,
    tools=[calculate],  # 主 Agent 自己的工具
    subagents=[search_subagent]
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "搜索最新的 AI 新闻,然后计算 2^10"}]
})

❗️注意deepagents 包需要单独安装:pip install deepagents==0.3.0


八、文件系统中间件

from langchain.agents import create_agent
from deepagents.middleware.filesystem import FilesystemMiddleware
from deepagents.backends import StateBackend, StoreBackend, CompositeBackend

# 内存状态后端(短期)
agent = create_agent(
    model=model,
    middleware=[
        FilesystemMiddleware(backend=StateBackend)  # 文件存在状态中
    ]
)

# 跨会话持久化后端(长期)
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
agent = create_agent(
    model=model,
    store=store,
    middleware=[
        FilesystemMiddleware(backend=lambda runtime: StoreBackend(runtime))
    ]
)

❗️注意:文件系统中间件默认提供 ls, read_file, write_file, delete_file 等工具,可通过 custom_tool_descriptions 自定义描述。


九、LangSmith 集成(可观测性)

import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGSMITH_ENDPOINT"] = "https://api.smith.langchain.com"

# 之后所有 Agent 调用自动被追踪
agent.invoke(...)

❗️注意:LangSmith 环境变量需在程序启动前设置,或在 .env 文件中定义。


十、v1.1 / v1.2 补充特性

10.1 模型配置文件

model = init_chat_model("claude-sonnet-4-5-20250929")
print(model.profile)  # 输出模型能力:max tokens、支持结构化输出等

10.2 严格模式(仅 ProviderStrategy)

agent = create_agent(
    model="openai:gpt-5",
    response_format=ProviderStrategy(Weather, strict=True)  # 强制符合 schema
)

10.3 工具 extras 属性

@tool(extras={"defer_loading": True})
def special_tool(...):
    """支持特定提供商的额外参数"""
    ...

十一、常见错误与解决方案

错误现象 原因 解决方案
ImportError: cannot import name 'BaseCheckpointSaver' 导入路径错误 from langgraph.checkpoint.base import BaseCheckpointSaver
AttributeError: 'AIMessage' object has no attribute 'content_blocks' 使用的模型包不支持 content_blocks 升级到支持的集成包(如 langchain-anthropic)
TypeError: create_agent() got an unexpected keyword argument 'executor' 混用旧版 API 移除 executor,直接使用 create_agent
ValueError: interrupt_on keys must be tool names interrupt_on 中使用了不存在的工具名 检查工具名称是否与定义一致
ModuleNotFoundError: No module named 'deepagents' 未安装 deepagents pip install deepagents==0.3.0
KeyError: 'messages' invoke 参数缺少 messages 确保输入为 {"messages": [...]}
ValidationError: 1 validation error for ... 工具参数不符合 Pydantic schema 检查模型传入的参数类型或值

结语

LangChain 1.x 通过 create_agent 统一了 Agent 构建,用中间件体系取代了繁琐的 hook,并提供了标准化的内容表示和记忆系统。以上代码片段覆盖了大部分核心特性,可直接复制使用。建议结合 LangChain 官方文档 和实际项目需求灵活调整。如果在使用中遇到其他问题,欢迎继续交流!

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐