5.2 Runtime
本节介绍 LangChain 的运行时系统，实现工具和中间件的依赖注入。
什么是 Runtime？
LangChain 的 create_agent 基于 LangGraph 的运行时基础设施。Runtime 对象 提供三个核心组件：
| 组件 | 说明 | 用途 |
|---|---|---|
| Context | 静态配置信息 | 用户 ID、数据库连接、依赖项 |
| Store | BaseStore 实例 | 长期记忆管理 |
| Stream Writer | 流写入器 | 自定义流式输出 |
"Runtime context provides dependency injection for your tools and middleware."
核心优势
- 消除硬编码 - 配置与代码分离
- 消除全局状态 - 每次调用独立
- 提高可测试性 - 便于模拟和测试
- 增强复用性 - 工具可跨项目使用
定义 Context Schema
使用 dataclass 定义上下文结构：
python
from dataclasses import dataclass
from langchain.agents import create_agent
@dataclass
class Context:
"""运行时上下文"""
user_id: str
user_name: str
api_key: str
is_admin: bool = False
agent = create_agent(
model="gpt-4o",
tools=[my_tools],
context_schema=Context
)
# 调用时传入上下文
result = agent.invoke(
{"messages": [{"role": "user", "content": "你好"}]},
context=Context(
user_id="user_123",
user_name="张三",
api_key="sk-xxx",
is_admin=False
)
)在工具中访问 Runtime
使用 ToolRuntime 参数访问运行时信息：
python
from langchain_core.tools import tool
from langchain.agents import ToolRuntime
@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
"""获取当前用户信息"""
user_name = runtime.context.user_name
user_id = runtime.context.user_id
return f"当前用户: {user_name} (ID: {user_id})"
@tool
def fetch_user_preferences(runtime: ToolRuntime[Context]) -> str:
"""获取用户偏好设置"""
user_id = runtime.context.user_id
# 从 Store 获取长期记忆
if runtime.store:
memory = runtime.store.get(("users",), user_id)
if memory:
return f"用户偏好: {memory.value['preferences']}"
return "未找到用户偏好设置"
@tool
def admin_operation(action: str, runtime: ToolRuntime[Context]) -> str:
"""管理员操作（需要权限）"""
if not runtime.context.is_admin:
return "错误: 需要管理员权限"
return f"已执行管理员操作: {action}"在中间件中访问 Runtime
通过 request.runtime 访问运行时：
python
from langchain.agents import dynamic_prompt, ModelRequest
@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
"""根据用户信息生成个性化系统提示"""
user_name = request.runtime.context.user_name
is_admin = request.runtime.context.is_admin
base_prompt = f"你是一个助手。当前用户是 {user_name}。"
if is_admin:
base_prompt += "\n该用户是管理员，可以执行高级操作。"
else:
base_prompt += "\n该用户是普通用户，请限制敏感操作。"
return base_prompt在其他钩子中使用
python
from langchain.agents import before_model, after_model
@before_model
def log_request(state, runtime):
"""记录请求日志"""
user_id = runtime.context.user_id
print(f"[{user_id}] 发起模型调用")
return None
@after_model
def track_usage(state, response, runtime):
"""追踪使用情况"""
user_id = runtime.context.user_id
if hasattr(response, "response_metadata"):
usage = response.response_metadata.get("usage", {})
tokens = usage.get("total_tokens", 0)
print(f"[{user_id}] 使用了 {tokens} tokens")
return None访问 Store（长期记忆）
Runtime 提供对 Store 的访问，用于跨会话持久化数据：
python
from langchain_core.tools import tool
from langchain.agents import ToolRuntime
@tool
def save_user_preference(
key: str,
value: str,
runtime: ToolRuntime[Context]
) -> str:
"""保存用户偏好"""
user_id = runtime.context.user_id
if runtime.store:
runtime.store.put(
("user_preferences", user_id),
key,
{"value": value}
)
return f"已保存偏好: {key} = {value}"
return "Store 不可用"
@tool
def get_user_preference(
key: str,
runtime: ToolRuntime[Context]
) -> str:
"""获取用户偏好"""
user_id = runtime.context.user_id
if runtime.store:
result = runtime.store.get(("user_preferences", user_id), key)
if result:
return f"{key} = {result.value['value']}"
return f"未找到偏好: {key}"使用 Stream Writer
Runtime 提供流写入器，用于自定义流式输出：
python
from langchain.agents import get_stream_writer, ToolRuntime
from langchain_core.tools import tool
@tool
def long_running_task(task_name: str, runtime: ToolRuntime) -> str:
"""执行长时间任务"""
writer = get_stream_writer()
writer(f"开始任务: {task_name}")
# 模拟多步骤处理
import time
for i in range(5):
time.sleep(1)
writer(f"进度: {(i + 1) * 20}%")
writer("任务完成!")
return f"任务 '{task_name}' 执行成功"接收自定义流：
python
for event in agent.stream(
{"messages": [...]},
stream_mode="custom"
):
print(f"[进度] {event}")完整示例
python
from dataclasses import dataclass
from langchain.agents import (
create_agent,
ToolRuntime,
dynamic_prompt,
ModelRequest,
before_model,
)
from langchain_core.tools import tool
from langgraph.store.memory import InMemoryStore
# 1. 定义上下文结构
@dataclass
class AppContext:
user_id: str
user_name: str
organization: str
role: str # "admin" | "user" | "guest"
api_quota: int
# 2. 定义工具
@tool
def get_organization_data(runtime: ToolRuntime[AppContext]) -> str:
"""获取组织数据"""
org = runtime.context.organization
return f"组织 '{org}' 的数据: ..."
@tool
def use_api_quota(amount: int, runtime: ToolRuntime[AppContext]) -> str:
"""消耗 API 配额"""
quota = runtime.context.api_quota
if amount > quota:
return f"配额不足: 需要 {amount}，剩余 {quota}"
return f"已使用 {amount} 配额，剩余 {quota - amount}"
# 3. 定义动态提示
@dynamic_prompt
def role_based_prompt(request: ModelRequest) -> str:
ctx = request.runtime.context
base = f"你是 {ctx.organization} 的助手。"
base += f"\n当前用户: {ctx.user_name} ({ctx.role})"
if ctx.role == "admin":
base += "\n用户是管理员，可以执行所有操作。"
elif ctx.role == "user":
base += "\n用户是普通成员，限制敏感操作。"
else:
base += "\n用户是访客，只能查看公开信息。"
return base
# 4. 定义中间件
@before_model
def check_quota(state, runtime):
if runtime.context.api_quota <= 0:
return {"jump_to": "end"}
return None
# 5. 创建 Agent
store = InMemoryStore()
agent = create_agent(
model="gpt-4o",
tools=[get_organization_data, use_api_quota],
context_schema=AppContext,
middleware=[role_based_prompt, check_quota],
store=store,
)
# 6. 调用 Agent
result = agent.invoke(
{"messages": [{"role": "user", "content": "查看组织数据"}]},
context=AppContext(
user_id="user_456",
user_name="李四",
organization="Acme Corp",
role="admin",
api_quota=100
)
)
print(result["messages"][-1].content)Runtime 组件关系
┌─────────────────────────────────────────────────────┐
│ Runtime │
├─────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Context │ │ Store │ │ Stream │ │
│ │ │ │ │ │ Writer │ │
│ │ - user_id │ │ - get() │ │ │ │
│ │ - api_key │ │ - put() │ │ - write() │ │
│ │ - config │ │ - search() │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Tools / Middleware │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────┘最佳实践
| 实践 | 说明 |
|---|---|
| 最小化 Context | 只包含必要的信息 |
| 类型安全 | 使用 dataclass 定义结构 |
| 避免敏感信息 | 不要在 Context 中存储密码 |
| 使用 Store | 持久化数据用 Store，临时数据用 State |
| 日志记录 | 记录 Context 使用情况（脱敏） |
上一节：5.1 Guardrails