外观
概述
构建智能体(或任何 LLM 应用)的难点在于让它们足够可靠。尽管它们在原型阶段可能表现良好,但在真实世界用例中往往会失败。
智能体为什么会失败?
当智能体失败时,通常是因为智能体内部的 LLM 调用采取了错误的动作 / 没有按我们的预期行事。LLM 会因以下两种原因之一而失败:
- 底层 LLM 的能力不足
- 没有将"正确的"上下文传递给 LLM
在大多数情况下,真正导致智能体不可靠的其实是第二个原因。
上下文工程(context engineering)就是以正确的格式提供正确的信息和工具,使 LLM 能够完成任务。这是 AI 工程师的首要职责。缺少"正确的"上下文,是打造更可靠智能体的头号障碍,而 LangChain 的智能体抽象在设计上就特别适合促进上下文工程。
TIP
上下文工程新手?先从概念概述开始,了解不同类型的上下文以及何时使用它们。
智能体循环
一个典型的智能体循环由两个主要步骤组成:
- 模型调用——使用提示词和可用工具调用 LLM,返回响应或执行工具的请求
- 工具执行——执行 LLM 请求的工具,返回工具结果

这个循环会一直持续,直到 LLM 决定结束为止。
你能控制什么
要构建可靠的智能体,你需要控制智能体循环中每一步发生的内容,以及步骤之间发生的内容。
| 上下文类型 | 你能控制的内容 | 临时或持久化 |
|---|---|---|
| 模型上下文 | 进入模型调用的内容(指令、消息历史、工具、响应格式) | 临时 |
| 工具上下文 | 工具可以访问和产生的内容(对状态、store、运行时上下文的读/写) | 持久化 |
| 生命周期上下文 | 模型调用与工具调用之间发生的内容(摘要、护栏、日志记录等) | 持久化 |
- 临时上下文 — 单个调用中 LLM 看到的内容。你可以在不改变状态中保存内容的前提下修改消息、工具或提示词。
- 持久化上下文 — 在多个回合之间保存到状态中的内容。生命周期钩子和工具写入会永久修改这部分内容。
数据源
在整个过程中,你的智能体会访问(读取 / 写入)不同的数据源:
| 数据源 | 也称为 | 作用域 | 示例 |
|---|---|---|---|
| 运行时上下文 | 静态配置 | 会话作用域 | 用户 ID、API 密钥、数据库连接、权限、环境设置 |
| 状态 | 短期记忆 | 会话作用域 | 当前消息、已上传的文件、身份验证状态、工具结果 |
| Store | 长期记忆 | 跨会话 | 用户偏好、提取的见解、记忆、历史数据 |
工作原理
LangChain 的中间件是底层的机制,它让使用 LangChain 的开发者能够实际开展上下文工程。
中间件允许你挂接到智能体生命周期中的任何步骤,并:
- 更新上下文
- 跳转到智能体生命周期中的不同步骤
在本指南中,你会频繁看到中间件 API 被用作实现上下文工程这一目标的手段。
模型上下文
控制进入每次模型调用的内容——指令、可用工具、使用哪个模型以及输出格式。这些决策会直接影响可靠性和成本。
- 系统提示词 — 开发者提供给 LLM 的基础指令。
- 消息 — 发送给 LLM 的完整消息列表(对话历史)。
- 工具 — 智能体可用于执行动作的工具。
- 模型 — 将要调用的实际模型(包括配置)。
- 响应格式 — 模型最终响应的 schema 规范。
所有这些类型的模型上下文都可以取自状态(短期记忆)、Store(长期记忆)或运行时上下文(静态配置)。
系统提示词
系统提示词设定 LLM 的行为和能力。不同的用户、上下文或对话阶段需要不同的指令。成功的智能体会借助记忆、偏好和配置,为对话的当前状态提供正确的指令。
状态
从状态中访问消息计数或对话上下文:
python
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
@dynamic_prompt
def state_aware_prompt(request: ModelRequest) -> str:
# request.messages 是 request.state["messages"] 的快捷方式
message_count = len(request.messages)
base = "You are a helpful assistant."
if message_count > 10:
base += "\nThis is a long conversation - be extra concise."
return base
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[state_aware_prompt]
)typescript
import { createAgent } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [...],
middleware: [
dynamicSystemPromptMiddleware((state) => {
// 从状态(State)中读取:检查对话长度
const messageCount = state.messages.length;
let base = "You are a helpful assistant.";
if (messageCount > 10) {
base += "\nThis is a long conversation - be extra concise.";
}
return base;
}),
],
});Store
从长期记忆中访问用户偏好:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
@dynamic_prompt
def store_aware_prompt(request: ModelRequest) -> str:
user_id = request.runtime.context.user_id
# 从 Store 中读取:获取用户偏好
store = request.runtime.store
user_prefs = store.get(("preferences",), user_id)
base = "You are a helpful assistant."
if user_prefs:
style = user_prefs.value.get("communication_style", "balanced")
base += f"\nUser prefers {style} responses."
return base
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[store_aware_prompt],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createAgent, dynamicSystemPromptMiddleware } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
type Context = z.infer<typeof contextSchema>;
const agent = createAgent({
model: "gpt-5.5",
tools: [...],
contextSchema,
middleware: [
dynamicSystemPromptMiddleware<Context>(async (state, runtime) => {
const userId = runtime.context.userId;
// 从 Store 中读取:获取用户偏好
const store = runtime.store;
const userPrefs = await store.get(["preferences"], userId);
let base = "You are a helpful assistant.";
if (userPrefs) {
const style = userPrefs.value?.communicationStyle || "balanced";
base += `\nUser prefers ${style} responses.`;
}
return base;
}),
],
});运行时上下文
从运行时上下文中访问用户 ID 或配置:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
@dataclass
class Context:
user_role: str
deployment_env: str
@dynamic_prompt
def context_aware_prompt(request: ModelRequest) -> str:
# 从运行时上下文(Runtime Context)中读取:用户角色和环境
user_role = request.runtime.context.user_role
env = request.runtime.context.deployment_env
base = "You are a helpful assistant."
if user_role == "admin":
base += "\nYou have admin access. You can perform all operations."
elif user_role == "viewer":
base += "\nYou have read-only access. Guide users to read operations only."
if env == "production":
base += "\nBe extra careful with any data modifications."
return base
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[context_aware_prompt],
context_schema=Context
)typescript
import * as z from "zod";
import { createAgent, dynamicSystemPromptMiddleware } from "langchain";
const contextSchema = z.object({
userRole: z.string(),
deploymentEnv: z.string(),
});
type Context = z.infer<typeof contextSchema>;
const agent = createAgent({
model: "gpt-5.5",
tools: [...],
contextSchema,
middleware: [
dynamicSystemPromptMiddleware<Context>((state, runtime) => {
// 从运行时上下文(Runtime Context)中读取:用户角色和环境
const userRole = runtime.context.userRole;
const env = runtime.context.deploymentEnv;
let base = "You are a helpful assistant.";
if (userRole === "admin") {
base += "\nYou have admin access. You can perform all operations.";
} else if (userRole === "viewer") {
base += "\nYou have read-only access. Guide users to read operations only.";
}
if (env === "production") {
base += "\nBe extra careful with any data modifications.";
}
return base;
}),
],
});消息
消息构成了发送给 LLM 的提示词。 管理消息的内容至关重要,这样才能确保 LLM 拥有正确的信息并良好地回应。
状态
当上传的文件与当前查询相关时,从状态中注入文件上下文:
python
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:
"""Inject context about files user has uploaded this session."""
# 从状态(State)中读取:获取已上传文件的元数据
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-5.5",
tools=[...],
middleware=[inject_file_context]
)typescript
import { createMiddleware } from "langchain";
const injectFileContext = createMiddleware({
name: "InjectFileContext",
wrapModelCall: (request, handler) => {
// request.state 是 request.state.messages 的快捷方式
const uploadedFiles = request.state.uploadedFiles || [];
if (uploadedFiles.length > 0) {
// 构建关于可用文件的上下文
const fileDescriptions = uploadedFiles.map(file =>
`- ${file.name} (${file.type}): ${file.summary}`
);
const fileContext = `Files you have access to in this conversation:
${fileDescriptions.join("\n")}
Reference these files when answering questions.`;
// 在最近的消息之前注入文件上下文
const messages = [
...request.messages, // 对话的其余部分
{ role: "user", content: fileContext }
];
request = request.override({ messages });
}
return handler(request);
},
});
const agent = createAgent({
model: "gpt-5.5",
tools: [...],
middleware: [injectFileContext],
});Store
从 Store 中注入用户的邮件写作风格,以指导草稿的撰写:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
@wrap_model_call
def inject_writing_style(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Inject user's email writing style from Store."""
user_id = request.runtime.context.user_id
# 从 Store 中读取:获取用户的邮件写作风格示例
store = request.runtime.store
writing_style = store.get(("writing_style",), user_id)
if writing_style:
style = writing_style.value
# 根据存储的示例构建风格指南
style_context = f"""Your writing style:
- Tone: {style.get('tone', 'professional')}
- Typical greeting: "{style.get('greeting', 'Hi')}"
- Typical sign-off: "{style.get('sign_off', 'Best')}"
- Example email you've written:
{style.get('example_email', '')}"""
# 在末尾追加:模型会更关注最后的消息
messages = [
*request.messages,
{"role": "user", "content": style_context}
]
request = request.override(messages=messages)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[inject_writing_style],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
const injectWritingStyle = createMiddleware({
name: "InjectWritingStyle",
contextSchema,
wrapModelCall: async (request, handler) => {
const userId = request.runtime.context.userId;
// 从 Store 中读取:获取用户的邮件写作风格示例
const store = request.runtime.store;
const writingStyle = await store.get(["writing_style"], userId);
if (writingStyle) {
const style = writingStyle.value;
// 根据存储的示例构建风格指南
const styleContext = `Your writing style:
- Tone: ${style.tone || 'professional'}
- Typical greeting: "${style.greeting || 'Hi'}"
- Typical sign-off: "${style.signOff || 'Best'}"
- Example email you've written:
${style.exampleEmail || ''}`;
// 在末尾追加:模型会更关注最后的消息
const messages = [
...request.messages,
{ role: "user", content: styleContext }
];
request = request.override({ messages });
}
return handler(request);
},
});运行时上下文
根据用户的司法管辖区域,从运行时上下文注入合规规则:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@dataclass
class Context:
user_jurisdiction: str
industry: str
compliance_frameworks: list[str]
@wrap_model_call
def inject_compliance_rules(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Inject compliance constraints from Runtime Context."""
# 从运行时上下文(Runtime Context)中读取:获取合规要求
jurisdiction = request.runtime.context.user_jurisdiction
industry = request.runtime.context.industry
frameworks = request.runtime.context.compliance_frameworks
# 构建合规约束
rules = []
if "GDPR" in frameworks:
rules.append("- Must obtain explicit consent before processing personal data")
rules.append("- Users have right to data deletion")
if "HIPAA" in frameworks:
rules.append("- Cannot share patient health information without authorization")
rules.append("- Must use secure, encrypted communication")
if industry == "finance":
rules.append("- Cannot provide financial advice without proper disclaimers")
if rules:
compliance_context = f"""Compliance requirements for {jurisdiction}:
{chr(10).join(rules)}"""
# 在末尾追加:模型会更关注最后的消息
messages = [
*request.messages,
{"role": "user", "content": compliance_context}
]
request = request.override(messages=messages)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[inject_compliance_rules],
context_schema=Context
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userJurisdiction: z.string(),
industry: z.string(),
complianceFrameworks: z.array(z.string()),
});
type Context = z.infer<typeof contextSchema>;
const injectComplianceRules = createMiddleware<Context>({
name: "InjectComplianceRules",
contextSchema,
wrapModelCall: (request, handler) => {
// 从运行时上下文(Runtime Context)中读取:获取合规要求
const { userJurisdiction, industry, complianceFrameworks } = request.runtime.context;
// 构建合规约束
const rules = [];
if (complianceFrameworks.includes("GDPR")) {
rules.push("- Must obtain explicit consent before processing personal data");
rules.push("- Users have right to data deletion");
}
if (complianceFrameworks.includes("HIPAA")) {
rules.push("- Cannot share patient health information without authorization");
rules.push("- Must use secure, encrypted communication");
}
if (industry === "finance") {
rules.push("- Cannot provide financial advice without proper disclaimers");
}
if (rules.length > 0) {
const complianceContext = `Compliance requirements for ${userJurisdiction}:
${rules.join("\n")}`;
// 在末尾追加:模型会更关注最后的消息
const messages = [
...request.messages,
{ role: "user", content: complianceContext }
];
request = request.override({ messages });
}
return handler(request);
},
});INFO
临时 vs 持久化消息更新:
上面的示例使用 wrap_model_call 进行临时更新——修改发送给模型的单次调用消息,而不改变状态中保存的内容。
对于修改状态的持久化更新,你可以:
从
wrap_model_call返回一个携带Command的ExtendedModelResponse,以从模型调用层注入状态更新。使用生命周期钩子(如
before_model、after_model或wrap_tool_call(用于工具返回))来更新对话历史。更多细节请参阅中间件文档。直接从
wrapModelCall返回一个Command,以从模型调用层注入状态更新。使用生命周期钩子(如
beforeModel、afterModel或wrapToolCall(用于工具返回))来更新对话历史。更多细节请参阅中间件文档。
更多信息请参阅状态更新。
工具
工具让模型可以与数据库、API 和外部系统交互。你如何定义和选择工具,会直接影响模型能否有效地完成任务。
定义工具
每个工具都需要清晰的名称、描述、参数名称和参数描述。这些不只是元数据——它们会指导模型推理在何时以及如何使用该工具。
python
from langchain.tools import tool
@tool(parse_docstring=True)
def search_orders(
user_id: str,
status: str,
limit: int = 10
) -> str:
"""Search for user orders by status.
Use this when the user asks about order history or wants to check
order status. Always filter by the provided status.
Args:
user_id: Unique identifier for the user
status: Order status: 'pending', 'shipped', or 'delivered'
limit: Maximum number of results to return
"""
# 在此处实现
passtypescript
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const searchOrders = tool(
async ({ userId, status, limit }) => {
// 在此处实现
},
{
name: "search_orders",
description: `Search for user orders by status.
Use this when the user asks about order history or wants to check
order status. Always filter by the provided status.`,
schema: z.object({
userId: z.string().describe("Unique identifier for the user"),
status: z.enum(["pending", "shipped", "delivered"]).describe("Order status to filter by"),
limit: z.number().default(10).describe("Maximum number of results to return"),
}),
}
);选择工具
并非每个工具都适合每种场景。工具过多可能会让模型不堪重负(使上下文过载)并增加错误;工具过少则会限制能力。动态工具选择会根据身份验证状态、用户权限、功能开关或对话阶段来调整可用的工具集。
状态
只有在达到某些对话里程碑之后才启用高级工具:
python
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@wrap_model_call
def state_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Filter tools based on conversation State."""
# 从状态(State)中读取:检查用户是否已认证
state = request.state
is_authenticated = state.get("authenticated", False)
message_count = len(state["messages"])
# 仅在认证之后启用敏感工具
if not is_authenticated:
tools = [t for t in request.tools if t.name.startswith("public_")]
request = request.override(tools=tools)
elif message_count < 5:
# 在对话早期限制工具
tools = [t for t in request.tools if t.name != "advanced_search"]
request = request.override(tools=tools)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[public_search, private_search, advanced_search],
middleware=[state_based_tools]
)typescript
import { createMiddleware } from "langchain";
const stateBasedTools = createMiddleware({
name: "StateBasedTools",
wrapModelCall: (request, handler) => {
// 从状态(State)中读取:检查认证状态和对话长度
const state = request.state;
const isAuthenticated = state.authenticated || false;
const messageCount = state.messages.length;
let filteredTools = request.tools;
// 仅在认证之后启用敏感工具
if (!isAuthenticated) {
filteredTools = request.tools.filter(t => t.name.startsWith("public_"));
} else if (messageCount < 5) {
filteredTools = request.tools.filter(t => t.name !== "advanced_search");
}
return handler({ ...request, tools: filteredTools });
},
});Store
根据 Store 中的用户偏好或功能开关过滤工具:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
@wrap_model_call
def store_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Filter tools based on Store preferences."""
user_id = request.runtime.context.user_id
# 从 Store 中读取:获取用户已启用的功能
store = request.runtime.store
feature_flags = store.get(("features",), user_id)
if feature_flags:
enabled_features = feature_flags.value.get("enabled_tools", [])
# 只包含为该用户启用的工具
tools = [t for t in request.tools if t.name in enabled_features]
request = request.override(tools=tools)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[search_tool, analysis_tool, export_tool],
middleware=[store_based_tools],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
const storeBasedTools = createMiddleware({
name: "StoreBasedTools",
contextSchema,
wrapModelCall: async (request, handler) => {
const userId = request.runtime.context.userId;
// 从 Store 中读取:获取用户已启用的功能
const store = request.runtime.store;
const featureFlags = await store.get(["features"], userId);
let filteredTools = request.tools;
if (featureFlags) {
const enabledFeatures = featureFlags.value?.enabledTools || [];
filteredTools = request.tools.filter(t => enabledFeatures.includes(t.name));
}
return handler({ ...request, tools: filteredTools });
},
});运行时上下文
根据运行时上下文中的用户权限过滤工具:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@dataclass
class Context:
user_role: str
@wrap_model_call
def context_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Filter tools based on Runtime Context permissions."""
# 从运行时上下文(Runtime Context)中读取:获取用户角色
user_role = request.runtime.context.user_role
if user_role == "admin":
# 管理员获得所有工具
pass
elif user_role == "editor":
# 编辑者不能删除
tools = [t for t in request.tools if t.name != "delete_data"]
request = request.override(tools=tools)
else:
# 查看者获得只读工具
tools = [t for t in request.tools if t.name.startswith("read_")]
request = request.override(tools=tools)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[read_data, write_data, delete_data],
middleware=[context_based_tools],
context_schema=Context
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userRole: z.string(),
});
const contextBasedTools = createMiddleware({
name: "ContextBasedTools",
contextSchema,
wrapModelCall: (request, handler) => {
// 从运行时上下文(Runtime Context)中读取:获取用户角色
const userRole = request.runtime.context.userRole;
let filteredTools = request.tools;
if (userRole === "admin") {
// 管理员获得所有工具
} else if (userRole === "editor") {
filteredTools = request.tools.filter(t => t.name !== "delete_data");
} else {
filteredTools = request.tools.filter(t => t.name.startsWith("read_"));
}
return handler({ ...request, tools: filteredTools });
},
});关于过滤已注册工具以及在运行时注册工具(例如来自 MCP 服务器),请参阅动态工具。
模型
不同的模型有不同的优势、成本和上下文窗口。要为手头的任务选择正确的模型,而任务在智能体运行期间可能会发生变化。
状态
根据状态中的对话长度使用不同的模型:
python
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-6")
standard_model = init_chat_model("gpt-5.5")
efficient_model = init_chat_model("gpt-5.4-mini")
@wrap_model_call
def state_based_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select model based on State conversation length."""
# request.messages 是 request.state["messages"] 的快捷方式
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-5.4-mini",
tools=[...],
middleware=[state_based_model]
)typescript
import { createMiddleware, initChatModel } from "langchain";
// 在中间件外部只初始化一次模型
const largeModel = initChatModel("claude-sonnet-4-6");
const standardModel = initChatModel("gpt-5.5");
const efficientModel = initChatModel("gpt-5.4-mini");
const stateBasedModel = createMiddleware({
name: "StateBasedModel",
wrapModelCall: (request, handler) => {
// request.messages 是 request.state.messages 的快捷方式
const messageCount = request.messages.length;
let model;
if (messageCount > 20) {
model = largeModel;
} else if (messageCount > 10) {
model = standardModel;
} else {
model = efficientModel;
}
return handler({ ...request, model });
},
});Store
使用 Store 中用户的偏好模型:
python
from dataclasses import dataclass
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
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
# 只初始化一次可用的模型
MODEL_MAP = {
"gpt-5.5": init_chat_model("gpt-5.5"),
"gpt-5.4-mini": init_chat_model("gpt-5.4-mini"),
"claude-sonnet": init_chat_model("claude-sonnet-4-6"),
}
@wrap_model_call
def store_based_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select model based on Store preferences."""
user_id = request.runtime.context.user_id
# 从 Store 中读取:获取用户偏好的模型
store = request.runtime.store
user_prefs = store.get(("preferences",), user_id)
if user_prefs:
preferred_model = user_prefs.value.get("preferred_model")
if preferred_model and preferred_model in MODEL_MAP:
request = request.override(model=MODEL_MAP[preferred_model])
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[store_based_model],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createMiddleware, initChatModel } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
// 只初始化一次可用的模型
const MODEL_MAP = {
"gpt-5.5": initChatModel("gpt-5.5"),
"gpt-5.4-mini": initChatModel("gpt-5.4-mini"),
"claude-sonnet": initChatModel("claude-sonnet-4-6"),
};
const storeBasedModel = createMiddleware({
name: "StoreBasedModel",
contextSchema,
wrapModelCall: async (request, handler) => {
const userId = request.runtime.context.userId;
// 从 Store 中读取:获取用户偏好的模型
const store = request.runtime.store;
const userPrefs = await store.get(["preferences"], userId);
let model = request.model;
if (userPrefs) {
const preferredModel = userPrefs.value?.preferredModel;
if (preferredModel && MODEL_MAP[preferredModel]) {
model = MODEL_MAP[preferredModel];
}
}
return handler({ ...request, model });
},
});运行时上下文
根据运行时上下文中的成本限制或环境选择模型:
python
from dataclasses import dataclass
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
@dataclass
class Context:
cost_tier: str
environment: str
# 在中间件外部只初始化一次模型
premium_model = init_chat_model("claude-sonnet-4-6")
standard_model = init_chat_model("gpt-5.5")
budget_model = init_chat_model("gpt-5.4-mini")
@wrap_model_call
def context_based_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select model based on Runtime Context."""
# 从运行时上下文(Runtime Context)中读取:成本层级和环境
cost_tier = request.runtime.context.cost_tier
environment = request.runtime.context.environment
if environment == "production" and cost_tier == "premium":
# 生产环境中的付费用户获得最佳模型
model = premium_model
elif cost_tier == "budget":
# 低成本层级使用高效模型
model = budget_model
else:
# 标准层级
model = standard_model
request = request.override(model=model)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[context_based_model],
context_schema=Context
)typescript
import * as z from "zod";
import { createMiddleware, initChatModel } from "langchain";
const contextSchema = z.object({
costTier: z.string(),
environment: z.string(),
});
// 在中间件外部只初始化一次模型
const premiumModel = initChatModel("claude-sonnet-4-6");
const standardModel = initChatModel("gpt-5.5");
const budgetModel = initChatModel("gpt-5.4-mini");
const contextBasedModel = createMiddleware({
name: "ContextBasedModel",
contextSchema,
wrapModelCall: (request, handler) => {
// 从运行时上下文(Runtime Context)中读取:成本层级和环境
const costTier = request.runtime.context.costTier;
const environment = request.runtime.context.environment;
let model;
if (environment === "production" && costTier === "premium") {
model = premiumModel;
} else if (costTier === "budget") {
model = budgetModel;
} else {
model = standardModel;
}
return handler({ ...request, model });
},
});更多示例请参阅动态模型。
响应格式
结构化输出将非结构化文本转换为经过验证的结构化数据。当要提取特定字段或为下游系统返回数据时,自由格式的文本是不够的。
**工作原理:**当你将 schema 作为响应格式提供时,模型的最终响应保证会符合该 schema。智能体会运行模型 / 工具调用循环,直到模型调用完工具,然后将最终响应强制转换为所提供的格式。
定义格式
schema 定义会指导模型。字段名、类型和描述会精确指定输出应该遵循的格式。
python
from pydantic import BaseModel, Field
class CustomerSupportTicket(BaseModel):
"""Structured ticket information extracted from customer message."""
category: str = Field(
description="Issue category: 'billing', 'technical', 'account', or 'product'"
)
priority: str = Field(
description="Urgency level: 'low', 'medium', 'high', or 'critical'"
)
summary: str = Field(
description="One-sentence summary of the customer's issue"
)
customer_sentiment: str = Field(
description="Customer's emotional tone: 'frustrated', 'neutral', or 'satisfied'"
)typescript
import { z } from "zod";
const customerSupportTicket = z.object({
category: z.enum(["billing", "technical", "account", "product"]).describe(
"Issue category"
),
priority: z.enum(["low", "medium", "high", "critical"]).describe(
"Urgency level"
),
summary: z.string().describe(
"One-sentence summary of the customer's issue"
),
customerSentiment: z.enum(["frustrated", "neutral", "satisfied"]).describe(
"Customer's emotional tone"
),
}).describe("Structured ticket information extracted from customer message");选择格式
动态响应格式选择会根据用户偏好、对话阶段或角色来调整 schema——在早期返回简单格式,随着复杂度增加再返回详细格式。
状态
根据对话状态配置结构化输出:
python
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from pydantic import BaseModel, Field
from typing import Callable
class SimpleResponse(BaseModel):
"""Simple response for early conversation."""
answer: str = Field(description="A brief answer")
class DetailedResponse(BaseModel):
"""Detailed response for established conversation."""
answer: str = Field(description="A detailed answer")
reasoning: str = Field(description="Explanation of reasoning")
confidence: float = Field(description="Confidence score 0-1")
@wrap_model_call
def state_based_output(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select output format based on State."""
# request.messages 是 request.state["messages"] 的快捷方式
message_count = len(request.messages)
if message_count < 3:
# 对话早期:使用简单格式
request = request.override(response_format=SimpleResponse)
else:
# 已进行较久的对话:使用详细格式
request = request.override(response_format=DetailedResponse)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[state_based_output]
)typescript
import { createMiddleware } from "langchain";
import { z } from "zod";
const simpleResponse = z.object({
answer: z.string().describe("A brief answer"),
});
const detailedResponse = z.object({
answer: z.string().describe("A detailed answer"),
reasoning: z.string().describe("Explanation of reasoning"),
confidence: z.number().describe("Confidence score 0-1"),
});
const stateBasedOutput = createMiddleware({
name: "StateBasedOutput",
wrapModelCall: (request, handler) => {
// request.state 是 request.state.messages 的快捷方式
const messageCount = request.messages.length;
let responseFormat;
if (messageCount < 3) {
// 对话早期:使用简单格式
responseFormat = simpleResponse;
} else {
// 已进行较久的对话:使用详细格式
responseFormat = detailedResponse;
}
return handler({ ...request, responseFormat });
},
});Store
根据 Store 中的用户偏好配置输出格式:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from pydantic import BaseModel, Field
from typing import Callable
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
class VerboseResponse(BaseModel):
"""Verbose response with details."""
answer: str = Field(description="Detailed answer")
sources: list[str] = Field(description="Sources used")
class ConciseResponse(BaseModel):
"""Concise response."""
answer: str = Field(description="Brief answer")
@wrap_model_call
def store_based_output(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select output format based on Store preferences."""
user_id = request.runtime.context.user_id
# 从 Store 中读取:获取用户偏好的响应风格
store = request.runtime.store
user_prefs = store.get(("preferences",), user_id)
if user_prefs:
style = user_prefs.value.get("response_style", "concise")
if style == "verbose":
request = request.override(response_format=VerboseResponse)
else:
request = request.override(response_format=ConciseResponse)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[store_based_output],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
const verboseResponse = z.object({
answer: z.string().describe("Detailed answer"),
sources: z.array(z.string()).describe("Sources used"),
});
const conciseResponse = z.object({
answer: z.string().describe("Brief answer"),
});
const storeBasedOutput = createMiddleware({
name: "StoreBasedOutput",
wrapModelCall: async (request, handler) => {
const userId = request.runtime.context.userId;
// 从 Store 中读取:获取用户偏好的响应风格
const store = request.runtime.store;
const userPrefs = await store.get(["preferences"], userId);
const style = userPrefs?.value?.responseStyle || "concise";
const responseFormat =
style === "verbose" ? verboseResponse : conciseResponse;
return handler({
...request,
responseFormat,
});
},
});运行时上下文
根据运行时上下文(如用户角色或环境)配置输出格式:
python
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from pydantic import BaseModel, Field
from typing import Callable
@dataclass
class Context:
user_role: str
environment: str
class AdminResponse(BaseModel):
"""Response with technical details for admins."""
answer: str = Field(description="Answer")
debug_info: dict = Field(description="Debug information")
system_status: str = Field(description="System status")
class UserResponse(BaseModel):
"""Simple response for regular users."""
answer: str = Field(description="Answer")
@wrap_model_call
def context_based_output(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Select output format based on Runtime Context."""
# 从运行时上下文(Runtime Context)中读取:用户角色和环境
user_role = request.runtime.context.user_role
environment = request.runtime.context.environment
if user_role == "admin" and environment == "production":
# 生产环境中的管理员获得详细输出
request = request.override(response_format=AdminResponse)
else:
# 普通用户获得简单输出
request = request.override(response_format=UserResponse)
return handler(request)
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[context_based_output],
context_schema=Context
)typescript
import * as z from "zod";
import { createMiddleware } from "langchain";
const contextSchema = z.object({
userRole: z.string(),
environment: z.string(),
});
const adminResponse = z.object({
answer: z.string().describe("Answer"),
debugInfo: z.record(z.any()).describe("Debug information"),
systemStatus: z.string().describe("System status"),
});
const userResponse = z.object({
answer: z.string().describe("Answer"),
});
const contextBasedOutput = createMiddleware({
name: "ContextBasedOutput",
wrapModelCall: (request, handler) => {
// 从运行时上下文(Runtime Context)中读取:用户角色和环境
const userRole = request.runtime.context.userRole;
const environment = request.runtime.context.environment;
let responseFormat;
if (userRole === "admin" && environment === "production") {
responseFormat = adminResponse;
} else {
responseFormat = userResponse;
}
return handler({ ...request, responseFormat });
},
});工具上下文
工具的特殊之处在于它们既会读取也会写入上下文。
在最基本的情况下,当工具执行时,它会接收 LLM 的请求参数并返回一条工具消息。工具完成其工作并产生结果。
工具还可以为模型获取重要的信息,使模型能够执行并完成任务。
读取
大多数真实世界的工具需要的不仅仅是 LLM 的参数。它们需要用户 ID 来进行数据库查询、需要 API 密钥来调用外部服务,或者需要当前的会话状态来做出决策。工具会从状态、store 和运行时上下文中读取信息来访问这些内容。
状态
从状态中读取以检查当前的会话信息:
python
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
@tool
def check_authentication(
runtime: ToolRuntime
) -> str:
"""Check if user is authenticated."""
# 从状态(State)中读取:检查当前认证状态
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-5.5",
tools=[check_authentication]
)typescript
import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";
const checkAuthentication = tool(
async (_, runtime: ToolRuntime) => {
// 从状态(State)中读取:检查当前认证状态
const currentState = runtime.state;
const isAuthenticated = currentState.authenticated || false;
if (isAuthenticated) {
return "User is authenticated";
} else {
return "User is not authenticated";
}
},
{
name: "check_authentication",
description: "Check if user is authenticated",
schema: z.object({}),
}
);Store
从 Store 中读取以访问持久化的用户偏好:
python
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
@tool
def get_preference(
preference_key: str,
runtime: ToolRuntime[Context]
) -> str:
"""Get user preference from Store."""
user_id = runtime.context.user_id
# 从 Store 中读取:获取已有的偏好
store = runtime.store
existing_prefs = store.get(("preferences",), user_id)
if existing_prefs:
value = existing_prefs.value.get(preference_key)
return f"{preference_key}: {value}" if value else f"No preference set for {preference_key}"
else:
return "No preferences found"
agent = create_agent(
model="gpt-5.5",
tools=[get_preference],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";
const contextSchema = z.object({
userId: z.string(),
});
const getPreference = tool(
async ({ preferenceKey }, runtime: ToolRuntime) => {
const userId = runtime.context.userId;
// 从 Store 中读取:获取已有的偏好
const store = runtime.store;
const existingPrefs = await store.get(["preferences"], userId);
if (existingPrefs) {
const value = existingPrefs.value?.[preferenceKey];
return value ? `${preferenceKey}: ${value}` : `No preference set for ${preferenceKey}`;
} else {
return "No preferences found";
}
},
{
name: "get_preference",
description: "Get user preference from Store",
schema: z.object({
preferenceKey: z.string(),
}),
}
);运行时上下文
从运行时上下文中读取配置,如 API 密钥和用户 ID:
python
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
@dataclass
class Context:
user_id: str
api_key: str
db_connection: str
@tool
def fetch_user_data(
query: str,
runtime: ToolRuntime[Context]
) -> str:
"""Fetch data using Runtime Context configuration."""
# 从运行时上下文(Runtime Context)中读取:获取 API 密钥和数据库连接
user_id = runtime.context.user_id
api_key = runtime.context.api_key
db_connection = runtime.context.db_connection
# 使用配置获取数据
results = perform_database_query(db_connection, query, api_key)
return f"Found {len(results)} results for user {user_id}"
agent = create_agent(
model="gpt-5.5",
tools=[fetch_user_data],
context_schema=Context
)
# 使用运行时上下文调用
result = agent.invoke(
{"messages": [{"role": "user", "content": "Get my data"}]},
context=Context(
user_id="user_123",
api_key="sk-...",
db_connection="postgresql://..."
)
)typescript
import * as z from "zod";
import { tool } from "@langchain/core/tools";
import { createAgent } from "langchain";
const contextSchema = z.object({
userId: z.string(),
apiKey: z.string(),
dbConnection: z.string(),
});
const fetchUserData = tool(
async ({ query }, runtime: ToolRuntime<any, typeof contextSchema>) => {
// 从运行时上下文(Runtime Context)中读取:获取 API 密钥和数据库连接
const { userId, apiKey, dbConnection } = runtime.context;
// 使用配置获取数据
const results = await performDatabaseQuery(dbConnection, query, apiKey);
return `Found ${results.length} results for user ${userId}`;
},
{
name: "fetch_user_data",
description: "Fetch data using Runtime Context configuration",
schema: z.object({
query: z.string(),
}),
}
);
const agent = createAgent({
model: "gpt-5.5",
tools: [fetchUserData],
contextSchema,
});写入
工具结果可用于帮助智能体完成给定任务。工具既可以将结果直接返回给模型,也可以更新智能体的记忆,使重要的上下文可供后续步骤使用。
状态
使用 `Command` 写入状态以跟踪特定会话的信息:
python
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:
"""Authenticate user and update State."""
# 执行身份认证(简化版)
if password == "correct":
# 写入状态(State):使用 Command 标记为已认证
return Command(
update={"authenticated": True},
)
else:
return Command(update={"authenticated": False})
agent = create_agent(
model="gpt-5.5",
tools=[authenticate_user]
)typescript
import * as z from "zod";
import { tool } from "@langchain/core/tools";
import { createAgent } from "langchain";
import { Command } from "@langchain/langgraph";
const authenticateUser = tool(
async ({ password }) => {
// 执行身份认证
if (password === "correct") {
// 写入状态(State):使用 Command 标记为已认证
return new Command({
update: { authenticated: true },
});
} else {
return new Command({ update: { authenticated: false } });
}
},
{
name: "authenticate_user",
description: "Authenticate user and update State",
schema: z.object({
password: z.string(),
}),
}
);Store
写入 Store 以在多次会话之间持久化数据:
python
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
@tool
def save_preference(
preference_key: str,
preference_value: str,
runtime: ToolRuntime[Context]
) -> str:
"""Save user preference to Store."""
user_id = runtime.context.user_id
# 读取已有的偏好
store = runtime.store
existing_prefs = store.get(("preferences",), user_id)
# 与新偏好合并
prefs = existing_prefs.value if existing_prefs else {}
prefs[preference_key] = preference_value
# 写入 Store:保存更新后的偏好
store.put(("preferences",), user_id, prefs)
return f"Saved preference: {preference_key} = {preference_value}"
agent = create_agent(
model="gpt-5.5",
tools=[save_preference],
context_schema=Context,
store=InMemoryStore()
)typescript
import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";
const savePreference = tool(
async ({ preferenceKey, preferenceValue }, runtime: ToolRuntime<any, typeof contextSchema>) => {
const userId = runtime.context.userId;
// 读取已有的偏好
const store = runtime.store;
const existingPrefs = await store.get(["preferences"], userId);
// 与新偏好合并
const prefs = existingPrefs?.value || {};
prefs[preferenceKey] = preferenceValue;
// 写入 Store:保存更新后的偏好
await store.put(["preferences"], userId, prefs);
return `Saved preference: ${preferenceKey} = ${preferenceValue}`;
},
{
name: "save_preference",
description: "Save user preference to Store",
schema: z.object({
preferenceKey: z.string(),
preferenceValue: z.string(),
}),
}
);关于在工具中访问状态、store 和运行时上下文的完整示例,请参阅工具。
生命周期上下文
控制核心智能体步骤之间发生的内容——拦截数据流以实现摘要、护栏和日志记录等横切关注点。
正如你在模型上下文和工具上下文中所见,中间件是让上下文工程切实可行的机制。中间件允许你挂接到智能体生命周期中的任何步骤,并且可以:
- 更新上下文——修改状态和 store 以持久化更改、更新对话历史或保存见解
- 在生命周期中跳转——根据上下文移动到智能体循环中的不同步骤(例如,在满足某个条件时跳过工具执行,或在修改上下文后重复模型调用)

示例:摘要
最常见的生命周期模式之一,是在对话历史变得过长时自动将其压缩。与模型上下文中展示的临时消息裁剪不同,摘要会持久化地更新状态——用摘要永久替换旧消息,并将该摘要保存下来供所有后续回合使用。
LangChain 为此提供了内置中间件:
python
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[
SummarizationMiddleware(
model="gpt-5.4-mini",
trigger={"tokens": 4000},
keep=("messages", 20),
),
],
)typescript
import { createAgent, summarizationMiddleware } from "langchain";
const agent = createAgent({
model: "gpt-5.5",
tools: [...],
middleware: [
summarizationMiddleware({
model: "gpt-5.4-mini",
trigger: { tokens: 4000 },
keep: { messages: 20 },
}),
],
});当对话超过 token 限制时,SummarizationMiddleware 会自动:
- 使用一次单独的 LLM 调用对较旧的消息进行摘要
- 用摘要消息永久替换它们(保存在状态中)
- 保留最近的消息以提供上下文
摘要后的对话历史会被永久更新——后续回合将看到摘要,而不是原始消息。
INFO
关于内置中间件的完整列表、可用的钩子以及如何创建自定义中间件,请参阅中间件文档。
最佳实践
- 从简单开始——先从静态的提示词和工具入手,仅在需要时再添加动态特性
- 增量测试——一次只添加一个上下文工程特性
- 监控性能——跟踪模型调用、token 用量和延迟
- 使用内置中间件——善用
SummarizationMiddleware、LLMToolSelectorMiddleware等 - 记录你的上下文策略——清楚地说明传递的是什么上下文以及为什么
- 理解临时与持久化的区别:模型上下文的更改是临时的(按调用计),而生命周期上下文的更改会持久化到状态中