LangChain 的 create_agent 在底层运行在 LangGraph 的运行时之上。
LangGraph 暴露了一个 Runtime 对象,包含以下信息:
- 上下文:静态信息,如用户 ID、数据库连接或代理调用的其他依赖项
- 存储:用于长期记忆的 BaseStore 实例
- 流式写入器:用于通过
"custom" 流式传输模式流式传输信息的对象
- 执行信息:当前执行的标识和重试信息(线程 ID、运行 ID、尝试次数)
- 服务器信息:在 LangGraph Server 上运行时的服务器特定元数据(助手 ID、图形 ID、认证用户)
运行时上下文为您的工具和中间件提供了依赖注入。您可以在调用代理时注入运行时依赖项(如数据库连接、用户 ID 或配置),而不是硬编码值或使用全局状态。这使您的工具更容易测试、可重用和灵活。
您可以在工具和中间件中访问运行时信息。
使用 create_agent 创建代理时,您可以指定 context_schema 来定义代理 Runtime 中存储的 context 的结构。
调用代理时,传递带有相关运行配置的 context 参数:
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")
)
在工具内部
您可以在工具内部访问运行时信息以:
- 访问上下文
- 读取或写入长期记忆
- 写入自定义流(例如,工具进度/更新)
使用 ToolRuntime 参数在工具内部访问 Runtime 对象。
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.execution_info 访问执行标识(线程 ID、运行 ID),通过 runtime.server_info 访问服务器特定元数据(助手 ID、认证用户)(在 LangGraph Server 上运行时):
from langchain.tools import tool, ToolRuntime
@tool
def context_aware_tool(runtime: ToolRuntime) -> str:
"""A tool that uses execution and server info."""
# 访问线程和运行 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。
访问 runtime.execution_info 和 runtime.server_info 需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。
在中间件内部
您可以在中间件中访问运行时信息,以基于用户上下文创建动态提示词、修改消息或控制代理行为。
使用 Runtime 参数在节点风格钩子内部访问 Runtime 对象。对于包装风格钩子,Runtime 对象在 ModelRequest 参数内部可用。
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.execution_info 和 runtime.server_info:
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
需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。