外观
概述
LangChain 的 create_agent 在底层运行于 LangGraph 的运行时之上。 LangChain 的 createAgent 在底层运行于 LangGraph 的运行时之上。 LangGraph 暴露一个 Runtime 对象,其中包含以下信息:
- Context(上下文):静态信息,如用户 ID、数据库连接,或一次智能体调用所需的其他依赖
- Store(存储):用于长期记忆的 BaseStore 实例
- Stream writer(流写入器):用于通过
"custom"流式输出模式流式传输信息的对象 - Execution info(执行信息):当前执行的标识与重试信息(线程 ID、运行 ID、尝试次数)
- Server info(服务器信息):在 LangGraph Server 上运行时服务器特定的元数据(assistant ID、graph ID、已认证用户)
TIP
运行时上下文为你的工具和中间件提供依赖注入。你无需硬编码值或使用全局状态,而可以在调用智能体时注入运行时依赖(如数据库连接、用户 ID 或配置)。这使得你的工具更易于测试、更可复用、更灵活。
TIP
运行时上下文是你将数据贯穿智能体进行传递的方式。与其把内容存储在全局状态中,不如把值——如数据库连接、用户会话或配置——附加到 context 上,并在工具和中间件内部访问它们。这保持了无状态、可测试性和可复用性。
访问
使用 create_agent 创建智能体时,你可以指定 context_schema 来定义存储在智能体 Runtime 中的 context 的结构。 使用 createAgent 创建智能体时,你可以指定 contextSchema 来定义存储在智能体 Runtime 中的 context 的结构。
调用智能体时,传入包含本次运行相关配置的 context 参数:
python
from dataclasses import dataclass
from langchain.agents import create_agent
@dataclass
class Context:
user_name: str
agent = create_agent(
model="gpt-5-nano",
tools=[...],
context_schema=Context
)
agent.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
context=Context(user_name="John Smith")
)ts
import * as z from "zod";
import { createAgent } from "langchain";
const contextSchema = z.object({
userName: z.string(),
});
const agent = createAgent({
model: "gpt-5.5",
tools: [
/* ... */
],
contextSchema,
});
const result = await agent.invoke(
{ messages: [{ role: "user", content: "What's my name?" }] },
{ context: { userName: "John Smith" } }
);在工具内部
你可以在工具内部访问运行时信息,以:
- 访问上下文
- 读取或写入长期记忆
- 写入自定义流(例如工具进度/更新)
在工具内部使用 ToolRuntime 参数来访问 Runtime 对象。
python
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
@dataclass
class Context:
user_id: str
@tool
def fetch_user_email_preferences(runtime: ToolRuntime[Context]) -> str:
"""Fetch the user's email preferences from the store."""
user_id = runtime.context.user_id
preferences: str = "The user prefers you to write a brief and polite email."
if runtime.store:
if memory := runtime.store.get(("users",), user_id):
preferences = memory.value["preferences"]
return preferences在工具内部使用 runtime 参数来访问 Runtime 对象。
ts
import * as z from "zod";
import { tool } from "langchain";
import { type ToolRuntime } from "@langchain/core/tools";
const contextSchema = z.object({
userName: z.string(),
});
const fetchUserEmailPreferences = tool(
async (_, runtime: ToolRuntime<any, typeof contextSchema>) => {
const userName = runtime.context?.userName;
if (!userName) {
throw new Error("userName is required");
}
let preferences = "The user prefers you to write a brief and polite email.";
if (runtime.store) {
const memory = await runtime.store?.get(["users"], userName);
if (memory) {
preferences = memory.value.preferences;
}
}
return preferences;
},
{
name: "fetch_user_email_preferences",
description: "Fetch the user's email preferences.",
schema: z.object({}),
}
);工具内部的执行信息和服务器信息
当在 LangGraph Server 上运行时,可通过 runtime.execution_info 访问执行标识(线程 ID、运行 ID),并通过 runtime.server_info 访问服务器特定的元数据(assistant ID、已认证用户):
python
from langchain.tools import tool, ToolRuntime
@tool
def context_aware_tool(runtime: ToolRuntime) -> str:
"""A tool that uses execution and server info."""
# 访问线程 ID 和运行 ID
info = runtime.execution_info
print(f"Thread: {info.thread_id}, Run: {info.run_id}")
# 访问服务器信息(仅在 LangGraph Server 上可用)
server = runtime.server_info
if server is not None:
print(f"Assistant: {server.assistant_id}")
if server.user is not None:
print(f"User: {server.user.identity}")
return "done"当不在 LangGraph Server 上运行时(例如本地开发期间),server_info 为 None。
当在 LangGraph Server 上运行时,可通过 runtime.executionInfo 访问执行标识(线程 ID、运行 ID),并通过 runtime.serverInfo 访问服务器特定的元数据(assistant ID、已认证用户):
ts
import { tool } from "langchain";
import * as z from "zod";
const contextAwareTool = tool(
async (_input, runtime) => {
// 访问线程 ID 和运行 ID
const info = runtime.executionInfo;
console.log(`Thread: ${info.threadId}, Run: ${info.runId}`);
// 访问服务器信息(仅在 LangGraph Server 上可用)
const server = runtime.serverInfo;
if (server != null) {
console.log(`Assistant: ${server.assistantId}`);
if (server.user != null) {
console.log(`User: ${server.user.identity}`);
}
}
return "done";
},
{
name: "context_aware_tool",
description: "A tool that uses execution and server info.",
schema: z.object({}),
}
);当不在 LangGraph Server 上运行时(例如本地开发期间),serverInfo 为 null。
INFO
runtime.execution_info 和 runtime.server_info 需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。
INFO
runtime.executionInfo 和 runtime.serverInfo 需要 deepagents>=1.9.0(或 @langchain/langgraph>=1.2.8)。
在中间件内部
你可以在中间件中访问运行时信息,以创建动态提示词、修改消息,或根据用户上下文控制智能体行为。
在节点式钩子中使用 Runtime 参数访问 Runtime 对象。对于包装式钩子,Runtime 对象在 ModelRequest 参数内可用。
python
from dataclasses import dataclass
from langchain.messages import AnyMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import dynamic_prompt, ModelRequest, before_model, after_model
from langgraph.runtime import Runtime
@dataclass
class Context:
user_name: str
# 动态提示词
@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
# 模型前钩子
@before_model
def log_before_model(state: AgentState, runtime: Runtime[Context]) -> dict | None:
print(f"Processing request for user: {runtime.context.user_name}")
return None
# 模型后钩子
@after_model
def log_after_model(state: AgentState, runtime: Runtime[Context]) -> dict | None:
print(f"Completed request for user: {runtime.context.user_name}")
return None
agent = create_agent(
model="gpt-5-nano",
tools=[...],
middleware=[dynamic_system_prompt, log_before_model, log_after_model],
context_schema=Context
)
agent.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
context=Context(user_name="John Smith")
)在中间件内部使用 runtime 参数访问 Runtime 对象。
ts
import * as z from "zod";
import { createAgent, createMiddleware, SystemMessage } from "langchain";
const contextSchema = z.object({
userName: z.string(),
});
// 动态提示词中间件
const dynamicPromptMiddleware = createMiddleware({
name: "DynamicPrompt",
contextSchema,
beforeModel: (state, runtime) => {
const userName = runtime.context?.userName;
if (!userName) {
throw new Error("userName is required");
}
const systemMsg = `You are a helpful assistant. Address the user as ${userName}.`;
return {
messages: [new SystemMessage(systemMsg), ...state.messages],
};
},
});
// 日志中间件
const loggingMiddleware = createMiddleware({
name: "Logging",
contextSchema,
beforeModel: (state, runtime) => {
console.log(`Processing request for user: ${runtime.context?.userName}`);
return;
},
afterModel: (state, runtime) => {
console.log(`Completed request for user: ${runtime.context?.userName}`);
return;
},
});
const agent = createAgent({
model: "gpt-5.5",
tools: [
/* ... */
],
middleware: [dynamicPromptMiddleware, loggingMiddleware],
contextSchema,
});
const result = await agent.invoke(
{ messages: [{ role: "user", content: "What's my name?" }] },
{ context: { userName: "John Smith" } }
);中间件内部的执行信息和服务器信息
中间件钩子也可以访问 runtime.execution_info 和 runtime.server_info:
python
from langchain.agents import AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
@before_model
def auth_gate(state: AgentState, runtime: Runtime) -> dict | None:
"""Block unauthenticated users when running on LangGraph Server."""
server = runtime.server_info
if server is not None and server.user is None:
raise ValueError("Authentication required")
print(f"Thread: {runtime.execution_info.thread_id}")
return None中间件钩子也可以访问 runtime.executionInfo 和 runtime.serverInfo:
ts
import { createMiddleware } from "langchain";
const authGate = createMiddleware({
name: "AuthGate",
beforeModel: (state, runtime) => {
const server = runtime.serverInfo;
if (server != null && server.user == null) {
throw new Error("Authentication required");
}
console.log(`Thread: ${runtime.executionInfo.threadId}`);
return;
},
});INFO
需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。
INFO
需要 deepagents>=1.9.0(或 @langchain/langgraph>=1.2.8)。