代理中的上下文工程

概述

构建代理（或任何 LLM 应用程序）的难点在于使它们足够可靠。虽然它们在原型中可能有效，但在实际使用场景中往往会失败。

为什么代理会失败？

当代理失败时，通常是因为代理内部的 LLM 调用采取了错误的操作/没有按照我们的预期执行。LLM 失败的原因通常有两种：

底层 LLM 能力不足
没有将”正确的”上下文传递给 LLM

更多时候——实际上是第二个原因导致代理不可靠。 上下文工程是以正确的格式提供正确的信息和工具，以便 LLM 能够完成任务。这是 AI 工程师的首要工作。缺乏”正确的”上下文是更可靠代理的头号障碍，而 LangChain 的代理抽象专为促进上下文工程而独特设计。

上下文工程新手？从概念概述开始，了解不同类型的上下文以及何时使用它们。

代理循环

典型的代理循环由两个主要步骤组成：

模型调用 - 使用提示和可用工具调用 LLM，返回响应或执行工具的请求
工具执行 - 执行 LLM 请求的工具，返回工具结果

此循环持续进行，直到 LLM 决定完成。

您可以控制的内容

要构建可靠的代理，您需要控制代理循环中每个步骤发生的事情，以及步骤之间发生的事情。

上下文类型	您控制的内容	临时或持久
模型上下文	进入模型调用的内容（指令、消息历史、工具、响应格式）	临时
工具上下文	工具可以访问和产生的内容（对状态、存储、运行时上下文的读/写）	持久
生命周期上下文	模型和工具调用之间发生的事情（摘要、护栏、日志记录等）	持久

临时上下文

LLM 在单次调用中看到的内容。您可以修改消息、工具或提示，而无需更改状态中保存的内容。

持久上下文

跨回合保存在状态中的内容。生命周期钩子和工具写入会永久修改此内容。

数据源

在此过程中，您的代理访问（读/写）不同的数据源：

数据源	也称为	范围	示例
运行时上下文	静态配置	对话范围	用户 ID、API 密钥、数据库连接、权限、环境设置
状态	短期记忆	对话范围	当前消息、上传文件、认证状态、工具结果
存储	长期记忆	跨对话	用户偏好、提取的见解、记忆、历史数据

工作原理

LangChain 中间件是使上下文工程对使用 LangChain 的开发者变得实用的底层机制。中间件允许您钩入代理生命周期中的任何步骤并：

更新上下文
跳转到代理生命周期中的不同步骤

在本指南中，您将经常看到中间件 API 的使用，作为实现上下文工程的手段。

模型上下文

控制每次模型调用的内容——指令、可用工具、使用哪个模型以及输出格式。这些决策直接影响可靠性和成本。

系统提示词

开发者给 LLM 的基本指令。

消息

发送给 LLM 的完整消息列表（对话历史）。

工具

代理可用于执行操作的工具。

模型

要调用的实际模型（包括配置）。

响应格式

模型最终响应的模式规范。

所有这些类型的模型上下文都可以从状态（短期记忆）、存储（长期记忆）或运行时上下文（静态配置）中获取。

系统提示词

系统提示词设置 LLM 的行为和能力。不同的用户、上下文或对话阶段需要不同的指令。成功的代理利用记忆、偏好和配置，为对话的当前状态提供正确的指令。

状态
存储
运行时上下文

Access message count or conversation context from state:

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-4.1",
    tools=[...],
    middleware=[state_aware_prompt]
)

Access user preferences from long-term memory:

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

    # Read from Store: get user preferences
    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-4.1",
    tools=[...],
    middleware=[store_aware_prompt],
    context_schema=Context,
    store=InMemoryStore()
)

从运行时上下文访问用户 ID 或配置：

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:
    # 从运行时上下文读取：用户角色和环境
    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-4.1",
    tools=[...],
    middleware=[context_aware_prompt],
    context_schema=Context
)

消息

消息构成了发送给 LLM 的提示词。管理消息的内容至关重要，以确保 LLM 有正确的信息来良好地响应。

状态
存储
运行时上下文

在与当前查询相关时从状态注入上传文件的上下文：

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."""
    # 从状态中读取：获取上传文件的元数据
    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)}

在回答问题时参考这些文件。"""

        # 在最近的消息之前注入文件上下文
        messages = [
            *request.messages,
            {"role": "user", "content": file_context},
        ]
        request = request.override(messages=messages)

    return handler(request)

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[inject_file_context]
)

Inject user’s email writing style from Store to guide drafting:

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 = 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-4.1",
    tools=[...],
    middleware=[inject_writing_style],
    context_schema=Context,
    store=InMemoryStore()
)

Inject compliance rules from Runtime Context based on user’s jurisdiction:

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."""
    # 从运行时上下文读取：获取合规要求
    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-4.1",
    tools=[...],
    middleware=[inject_compliance_rules],
    context_schema=Context
)

Transient vs Persistent Message Updates:上面的示例使用 wrap_model_call 进行临时更新——修改发送给模型进行单次调用的消息，而不更改状态中保存的内容。要进行持久更新以修改状态，您可以：

Return a ExtendedModelResponse with a Command from wrap_model_call to inject state updates from the model call layer.
Use life-cycle hooks like before_model, after_model, or wrap_tool_call (for tool returns) to update the conversation history. See the middleware documentation for more details.

有关更多信息，请参阅状态更新。

Tools

工具允许模型与数据库、API 和外部系统交互。您如何定义和选择工具直接影响模型是否能有效地完成任务。

Defining tools

每个工具都需要清晰的名称、描述、参数名和参数描述。这些不仅仅是元数据——它们指导模型何时以及如何使用该工具的推理。

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
    """
    # 在此实现
    pass

Selecting tools

并非每个工具都适用于每种情况。过多的工具可能会使模型不堪重负（上下文过载）并增加错误；过少的工具则限制了能力。动态工具选择根据认证状态、用户权限、功能标志或对话阶段来调整可用的工具集。

状态
存储
运行时上下文

Enable advanced tools only after certain conversation milestones:

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 = 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-4.1",
    tools=[public_search, private_search, advanced_search],
    middleware=[state_based_tools]
)

Filter tools based on user preferences or feature flags in Store:

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

    # Read from Store: get user's enabled features
    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-4.1",
    tools=[search_tool, analysis_tool, export_tool],
    middleware=[store_based_tools],
    context_schema=Context,
    store=InMemoryStore()
)

Filter tools based on user permissions from Runtime Context:

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."""
    # Read from Runtime Context: get user role
    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-4.1",
    tools=[read_data, write_data, delete_data],
    middleware=[context_based_tools],
    context_schema=Context
)

有关过滤预注册工具和运行时注册工具（例如从 MCP 服务器）的信息，请参阅动态工具。

模型

不同的模型具有不同的优势、成本和上下文窗口。为手头的任务选择合适的模型，这在代理运行过程中可能会发生变化。

状态
存储
运行时上下文

根据状态中的对话长度使用不同的模型：

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

# Initialize models once outside the middleware
large_model = init_chat_model("claude-sonnet-4-6")
standard_model = init_chat_model("gpt-4.1")
efficient_model = init_chat_model("gpt-4.1-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:
        # Long conversation - use model with larger context window
        model = large_model
    elif message_count > 10:
        # Medium conversation
        model = standard_model
    else:
        # Short conversation - use efficient model
        model = efficient_model

    request = request.override(model=model)

    return handler(request)

agent = create_agent(
    model="gpt-4.1-mini",
    tools=[...],
    middleware=[state_based_model]
)

Use user’s preferred model from Store:

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

# Initialize available models once
MODEL_MAP = {
    "gpt-4.1": init_chat_model("gpt-4.1"),
    "gpt-4.1-mini": init_chat_model("gpt-4.1-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

    # Read from Store: get user's preferred model
    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-4.1",
    tools=[...],
    middleware=[store_based_model],
    context_schema=Context,
    store=InMemoryStore()
)

Select model based on cost limits or environment from Runtime Context:

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

# Initialize models once outside the middleware
premium_model = init_chat_model("claude-sonnet-4-6")
standard_model = init_chat_model("gpt-4.1")
budget_model = init_chat_model("gpt-4.1-mini")

@wrap_model_call
def context_based_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """Select model based on Runtime Context."""
    # Read from Runtime Context: cost tier and environment
    cost_tier = request.runtime.context.cost_tier
    environment = request.runtime.context.environment

    if environment == "production" and cost_tier == "premium":
        # Production premium users get best model
        model = premium_model
    elif cost_tier == "budget":
        # Budget tier gets efficient model
        model = budget_model
    else:
        # Standard tier
        model = standard_model

    request = request.override(model=model)

    return handler(request)

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[context_based_model],
    context_schema=Context
)

有关更多示例，请参阅动态模型。

响应格式

结构化输出将非结构化文本转换为经过验证的结构化数据。当提取特定字段或为下游系统返回数据时，自由格式的文本是不够的。 How it works: When you provide a schema as the response format, the model’s final response is guaranteed to conform to that schema. The agent runs the model / tool calling loop until the model is done calling tools, then the final response is coerced into the provided format.

Defining formats

模式定义指导模型。字段名、类型和描述精确指定了输出应遵循的格式。

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'"
    )

Selecting formats

动态响应格式选择根据用户偏好、对话阶段或角色来调整模式——在早期返回简单格式，随着复杂性的增加返回详细格式。

状态
存储
运行时上下文

Configure structured output based on conversation state:

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:
        # Early conversation - use simple format
        request = request.override(response_format=SimpleResponse)
    else:
        # Established conversation - use detailed format
        request = request.override(response_format=DetailedResponse)

    return handler(request)

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[state_based_output]
)

Configure output format based on user preferences in Store:

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

    # Read from Store: get user's preferred response style
    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-4.1",
    tools=[...],
    middleware=[store_based_output],
    context_schema=Context,
    store=InMemoryStore()
)

Configure output format based on Runtime Context like user role or environment:

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."""
    # 从运行时上下文读取：用户角色和环境
    user_role = request.runtime.context.user_role
    environment = request.runtime.context.environment

    if user_role == "admin" and environment == "production":
        # Admins in production get detailed output
        request = request.override(response_format=AdminResponse)
    else:
        # Regular users get simple output
        request = request.override(response_format=UserResponse)

    return handler(request)

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[context_based_output],
    context_schema=Context
)

工具上下文

工具的特殊之处在于它们既读取又写入上下文。在最基本的情况下，当工具执行时，它接收 LLM 的请求参数并返回工具消息。工具完成其工作并产生结果。工具还可以为模型获取重要信息，使其能够执行和完成任务。

读取

大多数现实世界的工具需要的不仅仅是 LLM 的参数。它们需要用于数据库查询的用户 ID、用于外部服务的 API 密钥，或当前的会话状态来做出决策。工具从状态、存储和运行时上下文中读取以访问这些信息。

状态
存储
运行时上下文

Read from State to check current session information:

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent

@tool
def check_authentication(
    runtime: ToolRuntime
) -> str:
    """Check if user is authenticated."""
    # Read from State: check current auth status
    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-4.1",
    tools=[check_authentication]
)

Read from Store to access persisted user preferences:

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

    # Read from Store: get existing preferences
    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-4.1",
    tools=[get_preference],
    context_schema=Context,
    store=InMemoryStore()
)

Read from Runtime Context for configuration like API keys and user IDs:

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."""
    # Read from Runtime Context: get API key and DB connection
    user_id = runtime.context.user_id
    api_key = runtime.context.api_key
    db_connection = runtime.context.db_connection

    # Use configuration to fetch data
    results = perform_database_query(db_connection, query, api_key)

    return f"Found {len(results)} results for user {user_id}"

agent = create_agent(
    model="gpt-4.1",
    tools=[fetch_user_data],
    context_schema=Context
)

# Invoke with runtime context
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Get my data"}]},
    context=Context(
        user_id="user_123",
        api_key="sk-...",
        db_connection="postgresql://..."
    )
)

写入

工具结果可用于帮助代理完成给定任务。工具既可以直接向模型返回结果，也可以更新代理的记忆，使重要的上下文在后续步骤中可用。

状态
存储

使用 Command 写入状态以跟踪特定于会话的信息：

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."""
    # Perform authentication (simplified)
    if password == "correct":
        # Write to State: mark as authenticated using Command
        return Command(
            update={"authenticated": True},
        )
    else:
        return Command(update={"authenticated": False})

agent = create_agent(
    model="gpt-4.1",
    tools=[authenticate_user]
)

Write to Store to persist data across sessions:

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

    # Read existing preferences
    store = runtime.store
    existing_prefs = store.get(("preferences",), user_id)

    # Merge with new preference
    prefs = existing_prefs.value if existing_prefs else {}
    prefs[preference_key] = preference_value

    # Write to Store: save updated preferences
    store.put(("preferences",), user_id, prefs)

    return f"Saved preference: {preference_key} = {preference_value}"

agent = create_agent(
    model="gpt-4.1",
    tools=[save_preference],
    context_schema=Context,
    store=InMemoryStore()
)

有关在工具中访问状态、存储和运行时上下文的全面示例，请参阅工具。

生命周期上下文

控制核心代理步骤之间发生的事情——拦截数据流以实现摘要、护栏和日志记录等横切关注点。正如您在模型上下文和工具上下文中所见，中间件是使上下文工程变得实用的机制。中间件允许您钩入代理生命周期中的任何步骤，并且可以：

Update context - Modify state and store to persist changes, update conversation history, or save insights
Jump in the lifecycle - Move to different steps in the agent cycle based on context (e.g., skip tool execution if a condition is met, repeat model call with modified context)

Example: Summarization

最常见的生命周期模式之一是在对话历史变得过长时自动压缩它。与模型上下文中显示的临时消息修剪不同，摘要持久更新状态——永久地用摘要替换旧消息，该摘要将保存用于所有后续回合。 LangChain 为此提供了内置中间件：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4.1-mini",
            trigger={"tokens": 4000},
            keep={"messages": 20},
        ),
    ],
)

当对话超过 token 限制时，SummarizationMiddleware 会自动：

使用单独的 LLM 调用摘要较早的消息
在状态中用摘要消息替换它们（永久性）
保持最近的消息完整以提供上下文

摘要后的对话历史会被永久更新——后续回合将看到摘要而不是原始消息。

有关内置中间件的完整列表、可用的钩子以及如何创建自定义中间件，请参阅中间件文档。

最佳实践

从简单开始 - 从静态提示和工具开始，仅在需要时添加动态功能
增量测试 - 一次添加一个上下文工程功能
监控性能 - 跟踪模型调用、token 使用量和延迟
使用内置中间件 - 利用 SummarizationMiddleware、LLMToolSelectorMiddleware 等
记录您的上下文策略 - 明确传递了什么上下文以及为什么传递
理解临时与持久：模型上下文更改是临时的（每次调用），而生命周期上下文更改会持久保存到状态

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

概述

为什么代理会失败？

代理循环

您可以控制的内容

临时上下文

持久上下文

数据源

工作原理

模型上下文

系统提示词

消息

工具

模型

响应格式

系统提示词

消息

Tools

Defining tools

Selecting tools

模型

响应格式

Defining formats

Selecting formats

工具上下文

读取

写入

生命周期上下文

Example: Summarization

最佳实践

相关资源

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

​概述

​为什么代理会失败？

​代理循环

​您可以控制的内容

临时上下文

持久上下文

​数据源

​工作原理

​模型上下文

系统提示词

消息

工具

模型

响应格式

​系统提示词

​消息

​Tools

​Defining tools

​Selecting tools

​模型

​响应格式

​Defining formats

​Selecting formats

​工具上下文

​读取

​写入

​生命周期上下文

​Example: Summarization

​最佳实践

​相关资源

概述

为什么代理会失败？

代理循环

您可以控制的内容

数据源

工作原理

模型上下文

系统提示词

消息

Tools

Defining tools

Selecting tools

模型

响应格式

Defining formats

Selecting formats

工具上下文

读取

写入

生命周期上下文

Example: Summarization

最佳实践

相关资源