LangChain 1.x 语法新特性代码速查(拿来可用 + 纠错指南)
·
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 中不需要AgentExecutor,create_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或详细配置对象。SummarizationMiddleware的trigger和keep支持("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 官方文档 和实际项目需求灵活调整。如果在使用中遇到其他问题,欢迎继续交流!
更多推荐


所有评论(0)