Runtime 与上下文

Runtime（运行时）是 LangChain Agent 执行环境的核心抽象，它管理 Agent 的生命周期、工具执行、上下文传递和中间件。理解 Runtime 是构建生产级 Agent 的关键。

什么是 Runtime

Runtime 是 Agent 的运行时容器，负责：

• 协调模型和工具的调用
• 维护执行上下文（context）
• 执行中间件链
• 管理状态和生命周期

from langchain import create_agent
from langchain.chat_models import init_chat_model

# create_agent 内部会自动创建并管理 Runtime
agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="你是一个助手。",
)

ToolRuntime

ToolRuntime 是工具执行的运行时环境，让工具能够访问上下文信息：

from langchain.tools import ToolRuntime, tool
from langchain import create_agent
from langchain.chat_models import init_chat_model

@tool
def get_user_info() -> str:
    """获取当前用户信息。"""
    # 自动注入的 ToolRuntime 可以访问上下文
    user_id = ToolRuntime.get_context("user_id")
    user_name = ToolRuntime.get_context("user_name")
    return f"用户 {user_name}(ID: {user_id})"

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[get_user_info],
    system_prompt="查询用户信息。",
    context_schema={
        "user_id": str,
        "user_name": str,
    },
)

# 传入上下文
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我是谁？"}]},
    context={"user_id": "U12345", "user_name": "张三"},
)

ToolRuntime 核心方法

from langchain.tools import ToolRuntime

# 读取上下文
user_id = ToolRuntime.get_context("user_id")
all_context = ToolRuntime.get_all_context()  # 获取所有上下文

# 设置/更新上下文（在工具执行中）
ToolRuntime.set_context("processed_count", 42)

# 检查上下文是否存在
if ToolRuntime.has_context("user_tier"):
    tier = ToolRuntime.get_context("user_tier")

# 获取运行时元信息
agent_name = ToolRuntime.get_agent_name()
conversation_id = ToolRuntime.get_conversation_id()

context_schema

context_schema 定义了 Agent 接受的上下文数据的结构，提供类型检查和文档：

from langchain import create_agent
from typing import Optional

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="个性化助手。",
    context_schema={
        "user_id": str,                    # 必填
        "user_name": str,                  # 必填
        "user_tier": Optional[str],        # 可选，默认 None
        "session_id": str,                 # 必填
        "preferences": Optional[dict],     # 可选用户偏好
    },
)

# 传入上下文
result = agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    context={
        "user_id": "U001",
        "user_name": "李四",
        "session_id": "session_abc",
    },
)

Pydantic 模型作为 context_schema

from pydantic import BaseModel, Field
from typing import Optional

class AgentContext(BaseModel):
    """Agent 运行时上下文。"""
    user_id: str = Field(description="用户唯一标识")
    user_name: str = Field(description="用户姓名")
    user_tier: str = Field(default="普通", description="用户等级")
    language: str = Field(default="zh-CN", description="语言偏好")
    max_tokens_budget: Optional[int] = Field(default=None, description="Token 预算")

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="助手。",
    context_schema=AgentContext,
)

在工具中访问上下文

from langchain.tools import ToolRuntime, tool
from langchain import create_agent
from langchain.chat_models import init_chat_model

@tool
def personalized_greeting() -> str:
    """根据用户信息生成个性化问候。"""
    context = ToolRuntime.get_all_context()
    user_name = context.get("user_name", "朋友")
    hour = context.get("hour", 12)
    
    greeting = "早上好" if 6 <= hour < 12 else \
               "下午好" if 12 <= hour < 18 else \
               "晚上好"
    
    return f"{greeting}，{user_name}！有什么可以帮助您的？"

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[personalized_greeting],
    system_prompt="生成个性化问候。",
    context_schema={"user_name": str, "hour": int},
)

在中间件中访问上下文

中间件可以读取和修改上下文：

from langchain.middleware import Middleware

class AuditMiddleware(Middleware):
    """审计中间件，记录所有操作。"""
    
    async def before_execution(self, context: dict):
        # 读取上下文
        user_id = context.get("user_id")
        action = context.get("current_action")
        
        # 记录审计日志
        print(f"[AUDIT] 用户 {user_id} 执行操作：{action}")
        
        # 可以修改上下文
        context["audit_timestamp"] = __import__("time").time()
        return context
    
    async def after_execution(self, context: dict, result: any):
        print(f"[AUDIT] 操作完成，结果长度：{len(str(result))}")
        return result

# 添加中间件
agent.add_middleware(AuditMiddleware())

内置上下文字段

Agent 运行时自动维护以下上下文字段：

字段	类型	说明
conversation_id	str	对话唯一标识
user_input	str	当前用户输入
messages	list	对话消息历史
tools	list	可用工具列表
tool_results	dict	已执行工具的结果缓存

上下文生命周期

上下文在每次 invoke 调用时创建，调用结束后可持久化：

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[],
    system_prompt="助手。",
    context_schema={"user_id": str, "session_data": dict},
)

# 每次 invoke 都是独立的上下文
# 如果需要跨调用保持状态，需自行管理
session_store = {}

def handle_conversation(user_message: str, user_id: str):
    # 恢复会话数据
    session = session_store.get(user_id, {"history": [], "count": 0})
    
    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_message}]},
        context={"user_id": user_id, "session_data": session},
    )
    
    # 更新会话数据
    session["count"] += 1
    session["history"].append(user_message)
    session_store[user_id] = session
    
    return result

context 与 @dynamic_prompt

上下文也可以传递给动态提示词：

from langchain.prompts import dynamic_prompt

@dynamic_prompt
def smart_prompt(context: dict) -> str:
    user_name = context.get("user_name", "用户")
    user_tier = context.get("user_tier", "普通")
    tool_count = len(context.get("tools", []))
    
    return f"""你是 {user_name} 的个人助手。
用户等级：{user_tier}
可用工具数：{tool_count}

请注意这个用户的偏好和历史。"""

最佳实践

1. context_schema 始终定义清楚：明确哪些上下文是必需的
2. 敏感信息不过 context：密码、Token 等敏感数据不要传入上下文
3. 上下文尽量轻量：避免传入大数据集，需要时通过工具加载
4. 中间件处理横切关注点：日志、审计、限流等用中间件实现
5. 跨调用状态外部管理：Runtime 上下文是单次调用的，持久状态需要在外部维护
6. 使用 Pydantic 模型：提供更好的类型检查和 IDE 支持

下一步

• 学习动态提示词如何利用上下文
• 查看Guardrails 安全护栏中的上下文安全检查
• 了解MCP 接入中的上下文传递