智能体中的上下文工程

构建智能体（或任何大型语言模型应用）的难点在于使其足够可靠。虽然它们可能在原型中运行正常，但在实际使用场景中却常常失败。

概述

构建智能体（或任何大型语言模型应用）的难点在于使其足够可靠。虽然它们可能在原型中运行正常，但在实际使用场景中却常常失败。

智能体为何会失败？

当智能体失败时，通常是因为智能体内部的 LLM 调用采取了错误的行为 / 没有按预期执行。LLM 失败的原因有两种：

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

通常情况下——实际上是第二种原因导致智能体不可靠。

上下文工程是指以正确的方式提供合适的信息和工具，使 LLM 能够完成任务。这是 AI 工程师的首要工作。缺乏“正确的”上下文是构建更可靠智能体的首要障碍，而 LangChain 的智能体抽象设计独特，旨在促进上下文工程。

对上下文工程不熟悉？从 概念概述 开始，了解不同类型的上下文及其使用时机。

智能体循环

典型的智能体循环包含两个主要步骤：

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

这个循环会持续进行，直到 LLM 决定结束。

你可以控制什么

要构建可靠的智能体，你需要控制智能体循环每个步骤以及步骤之间的发生情况。

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

Transient context

LLM 单次调用时看到的内容。你可以修改消息、工具或提示词，而不会改变状态中保存的内容。

Persistent context

在回合之间保存的内容。生命周期钩子和工具写入会永久修改这些内容。

数据源

在整个过程中，你的智能体会访问（读取 / 写入）不同的数据源：

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

工作原理

LangChain 中间件 是底层机制，使上下文工程对使用 LangChain 的开发者变得实用。

中间件允许你钩入智能体生命周期的任何步骤，并：

• 更新上下文
• 跳转到智能体生命周期的其他步骤

在整个指南中，你会频繁看到中间件 API 的使用，作为实现上下文工程的手段。

模型上下文

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

System Prompt

开发者向 LLM 发出的基本指令。

Messages

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

Tools

智能体用于执行操作的实用工具。

Model

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

Response Format

模型最终响应的架构规范。

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

系统提示词

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

State

从状态中访问消息计数或对话上下文：

:::python

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 is a shortcut for 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-5.5",
    tools=[...],
    middleware=[state_aware_prompt]
)

:::

:::js

import { createAgent } from "langchain";

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  middleware: [
    dynamicSystemPromptMiddleware((state) => {
      // Read from State: check conversation length
      const messageCount = state.messages.length;

      let base = "You are a helpful assistant.";

      if (messageCount > 10) {
        base += "\nThis is a long conversation - be extra concise.";
      }

      return base;
    }),
  ],
});

:::

Store

从长期记忆中访问用户偏好：

:::python

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

:::

:::js

import * as z from "zod";
import { createAgent, dynamicSystemPromptMiddleware } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

type Context = z.infer<typeof contextSchema>;

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  contextSchema,
  middleware: [
    dynamicSystemPromptMiddleware<Context>(async (state, runtime) => {
      const userId = runtime.context.userId;

      // Read from Store: get user preferences
      const store = runtime.store;
      const userPrefs = await store.get(["preferences"], userId);

      let base = "You are a helpful assistant.";

      if (userPrefs) {
        const style = userPrefs.value?.communicationStyle || "balanced";
        base += `\nUser prefers ${style} responses.`;
      }

      return base;
    }),
  ],
});

:::

Runtime Context

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

:::python

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:
    # Read from Runtime Context: user role and environment
    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-5.5",
    tools=[...],
    middleware=[context_aware_prompt],
    context_schema=Context
)

:::

:::js

import * as z from "zod";
import { createAgent, dynamicSystemPromptMiddleware } from "langchain";

const contextSchema = z.object({
  userRole: z.string(),
  deploymentEnv: z.string(),
});

type Context = z.infer<typeof contextSchema>;

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  contextSchema,
  middleware: [
    dynamicSystemPromptMiddleware<Context>((state, runtime) => {
      // Read from Runtime Context: user role and environment
      const userRole = runtime.context.userRole;
      const env = runtime.context.deploymentEnv;

      let base = "You are a helpful assistant.";

      if (userRole === "admin") {
        base += "\nYou have admin access. You can perform all operations.";
      } else if (userRole === "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;
    }),
  ],
});

:::

消息

消息构成发送给 LLM 的提示词。 管理消息内容至关重要，以确保 LLM 能够获得正确信息以良好响应。

State

在当前查询相关时，从 State 注入上传文件上下文：

:::python

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."""
    # Read from State: get uploaded files metadata
    uploaded_files = request.state.get("uploaded_files", [])  

    if uploaded_files:
        # Build context about available 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)}

Reference these files when answering questions."""

        # Inject file context before recent messages
        messages = [  
            *request.messages,
            {"role": "user", "content": file_context},
        ]
        request = request.override(messages=messages)  

    return handler(request)

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

:::

:::js

import { createMiddleware, HumanMessage } from "langchain";

const injectFileContext = createMiddleware({
  name: "InjectFileContext",
  wrapModelCall: (request, handler) => {
    // request.state is a shortcut for request.state.messages
    const uploadedFiles = request.state.uploadedFiles || [];  

    if (uploadedFiles.length > 0) {
      // Build context about available files
      const fileDescriptions = uploadedFiles.map(file =>
        `- ${file.name} (${file.type}): ${file.summary}`
      );

      const fileContext = `Files you have access to in this conversation:
${fileDescriptions.join("\n")}

Reference these files when answering questions.`;

      // Inject file context before recent messages
      const messages = [  
        ...request.messages,  // Rest of conversation
        new HumanMessage(fileContext)
      ];
      request = { ...request, messages };  
    }

    return handler(request);
  },
});

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  middleware: [injectFileContext],
});

:::

Store

从 Store 中注入用户的邮件写作风格以指导草拟：

:::python

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  

    # Read from Store: get user's writing style examples
    store = request.runtime.store  
    writing_style = store.get(("writing_style",), user_id)  

    if writing_style:
        style = writing_style.value
        # Build style guide from stored examples
        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', '')}"""

        # Append at end - models pay more attention to final messages
        messages = [
            *request.messages,
            {"role": "user", "content": style_context}
        ]
        request = request.override(messages=messages)  

    return handler(request)

agent = create_agent(
    model="gpt-5.5",
    tools=[...],
    middleware=[inject_writing_style],
    context_schema=Context,
    store=InMemoryStore()
)

:::

:::js

import * as z from "zod";
import { createMiddleware, HumanMessage } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

const injectWritingStyle = createMiddleware({
  name: "InjectWritingStyle",
  contextSchema,
  wrapModelCall: async (request, handler) => {
    const userId = request.runtime.context.userId;  

    // Read from Store: get user's writing style examples
    const store = request.runtime.store;  
    const writingStyle = await store.get(["writing_style"], userId);  

    if (writingStyle) {
      const style = writingStyle.value;
      // Build style guide from stored examples
      const styleContext = `Your writing style:
- Tone: ${style.tone || 'professional'}
- Typical greeting: "${style.greeting || 'Hi'}"
- Typical sign-off: "${style.signOff || 'Best'}"
- Example email you've written:
${style.exampleEmail || ''}`;

      // Append at end - models pay more attention to final messages
      const messages = [
        ...request.messages,
        new HumanMessage(styleContext)
      ];
      request = { ...request, messages };  
    }

    return handler(request);
  },
});

:::

Runtime Context

根据用户的司法管辖区从 Runtime Context 中注入合规规则：

:::python

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."""
    # Read from Runtime Context: get compliance requirements
    jurisdiction = request.runtime.context.user_jurisdiction  
    industry = request.runtime.context.industry  
    frameworks = request.runtime.context.compliance_frameworks  

    # Build compliance constraints
    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)}"""

        # Append at end - models pay more attention to final messages
        messages = [
            *request.messages,
            {"role": "user", "content": compliance_context}
        ]
        request = request.override(messages=messages)  

    return handler(request)

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

:::

:::js

import * as z from "zod";
import { createMiddleware, HumanMessage } from "langchain";

const contextSchema = z.object({
  userJurisdiction: z.string(),
  industry: z.string(),
  complianceFrameworks: z.array(z.string()),
});

type Context = z.infer<typeof contextSchema>;

const injectComplianceRules = createMiddleware<Context>({
  name: "InjectComplianceRules",
  contextSchema,
  wrapModelCall: (request, handler) => {
    // Read from Runtime Context: get compliance requirements
    const { userJurisdiction, industry, complianceFrameworks } = request.runtime.context;  

    // Build compliance constraints
    const rules = [];
    if (complianceFrameworks.includes("GDPR")) {
      rules.push("- Must obtain explicit consent before processing personal data");
      rules.push("- Users have right to data deletion");
    }
    if (complianceFrameworks.includes("HIPAA")) {
      rules.push("- Cannot share patient health information without authorization");
      rules.push("- Must use secure, encrypted communication");
    }
    if (industry === "finance") {
      rules.push("- Cannot provide financial advice without proper disclaimers");
    }

    if (rules.length > 0) {
      const complianceContext = `Compliance requirements for ${userJurisdiction}:
${rules.join("\n")}`;

      // Append at end - models pay more attention to final messages
      const messages = [
        ...request.messages,
        new HumanMessage(complianceContext)
      ];
      request = { ...request, messages };  
    }

    return handler(request);
  },
});

:::

瞬时与持久消息更新：

上述示例使用wrap_model_call进行瞬时更新 - 修改发送给模型的单次调用消息，而不改变保存的状态。

对于持久更新（修改状态），您可以：

:::python

• 返回一个@[ExtendedModelResponse]并从wrap_model_call获取@[Command]以从模型调用层注入状态更新。
• 使用生命周期钩子如before_model、after_model或wrap_tool_call（用于工具返回）来更新对话历史。有关更多详细信息，请参阅中间件文档。 :::

:::js

• 直接从wrapModelCall返回@[Command]以从模型调用层注入状态更新。
• 使用生命周期钩子如beforeModel、afterModel或wrapToolCall（用于工具返回）来更新对话历史。有关更多详细信息，请参阅中间件文档。 :::

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

工具

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

定义工具

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

:::python

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
    """
    # Implementation here
    pass

:::

:::js

import { tool } from "@langchain/core/tools";
import { z } from "zod";

const searchOrders = tool(
  async ({ userId, status, limit }) => {
    // Implementation here
  },
  {
    name: "search_orders",
    description: `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.`,
    schema: z.object({
      userId: z.string().describe("Unique identifier for the user"),
      status: z.enum(["pending", "shipped", "delivered"]).describe("Order status to filter by"),
      limit: z.number().default(10).describe("Maximum number of results to return"),
    }),
  }
);

:::

选择工具

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

State

在达到特定对话里程碑后仅启用高级工具：

:::python

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."""
    # Read from State: check if user has authenticated
    state = request.state  
    is_authenticated = state.get("authenticated", False)  
    message_count = len(state["messages"])

    # Only enable sensitive tools after authentication
    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:
        # Limit tools early in conversation
        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-5.5",
    tools=[public_search, private_search, advanced_search],
    middleware=[state_based_tools]
)

:::

:::js

import { createMiddleware } from "langchain";

const stateBasedTools = createMiddleware({
  name: "StateBasedTools",
  wrapModelCall: (request, handler) => {
    // Read from State: check authentication and conversation length
    const state = request.state;  
    const isAuthenticated = state.authenticated || false;  
    const messageCount = state.messages.length;

    let filteredTools = request.tools;

    // Only enable sensitive tools after authentication
    if (!isAuthenticated) {
      filteredTools = request.tools.filter(t => t.name.startsWith("public_"));  
    } else if (messageCount < 5) {
      filteredTools = request.tools.filter(t => t.name !== "advanced_search");  
    }

    return handler({ ...request, tools: filteredTools });  
  },
});

:::

Store

根据用户偏好或 Store 中的功能标志过滤工具：

:::python

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", [])
        # Only include tools that are enabled for this user
        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-5.5",
    tools=[search_tool, analysis_tool, export_tool],
    middleware=[store_based_tools],
    context_schema=Context,
    store=InMemoryStore()
)

:::

:::js

import * as z from "zod";
import { createMiddleware } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

const storeBasedTools = createMiddleware({
  name: "StoreBasedTools",
  contextSchema,
  wrapModelCall: async (request, handler) => {
    const userId = request.runtime.context.userId;  

    // Read from Store: get user's enabled features
    const store = request.runtime.store;  
    const featureFlags = await store.get(["features"], userId);  

    let filteredTools = request.tools;

    if (featureFlags) {
      const enabledFeatures = featureFlags.value?.enabledTools || [];
      filteredTools = request.tools.filter(t => enabledFeatures.includes(t.name));  
    }

    return handler({ ...request, tools: filteredTools });  
  },
});

:::

Runtime Context

根据运行时上下文中的用户权限过滤工具：

:::python

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":
        # Admins get all tools
        pass
    elif user_role == "editor":
        # Editors can't delete
        tools = [t for t in request.tools if t.name != "delete_data"]
        request = request.override(tools=tools)
    else:
        # Viewers get read-only tools
        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-5.5",
    tools=[read_data, write_data, delete_data],
    middleware=[context_based_tools],
    context_schema=Context
)

:::

:::js

import * as z from "zod";
import { createMiddleware } from "langchain";

const contextSchema = z.object({
  userRole: z.string(),
});

const contextBasedTools = createMiddleware({
  name: "ContextBasedTools",
  contextSchema,
  wrapModelCall: (request, handler) => {
    // Read from Runtime Context: get user role
    const userRole = request.runtime.context.userRole;  

    let filteredTools = request.tools;

    if (userRole === "admin") {
      // Admins get all tools
    } else if (userRole === "editor") {
      filteredTools = request.tools.filter(t => t.name !== "delete_data");  
    } else {
      filteredTools = request.tools.filter(t => t.name.startsWith("read_"));  
    }

    return handler({ ...request, tools: filteredTools });  
  },
});

:::

参见 动态工具，用于过滤预注册的工具以及在运行时注册工具（例如，从 MCP 服务器）。

模型

不同的模型具有不同的优势、成本和上下文窗口。为当前任务选择合适的模型，这可能在智能体运行期间发生变化。

State

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

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-5.5")
efficient_model = init_chat_model("gpt-5.4-mini")

@wrap_model_call
def state_based_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """Select model based on State conversation length."""
    # request.messages is a shortcut for 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-5.4-mini",
    tools=[...],
    middleware=[state_based_model]
)

import { createMiddleware, initChatModel } from "langchain";

// Initialize models once outside the middleware
const largeModel = initChatModel("claude-sonnet-4-6");
const standardModel = initChatModel("gpt-5.5");
const efficientModel = initChatModel("gpt-5.4-mini");

const stateBasedModel = createMiddleware({
  name: "StateBasedModel",
  wrapModelCall: (request, handler) => {
    // request.messages is a shortcut for request.state.messages
    const messageCount = request.messages.length;  
    let model;

    if (messageCount > 20) {
      model = largeModel;
    } else if (messageCount > 10) {
      model = standardModel;
    } else {
      model = efficientModel;
    }

    return handler({ ...request, model });  
  },
});

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-5.5": init_chat_model("gpt-5.5"),
    "gpt-5.4-mini": init_chat_model("gpt-5.4-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-5.5",
    tools=[...],
    middleware=[store_based_model],
    context_schema=Context,
    store=InMemoryStore()
)

import * as z from "zod";
import { createMiddleware, initChatModel } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

// Initialize available models once
const MODEL_MAP = {
  "gpt-5.5": initChatModel("gpt-5.5"),
  "gpt-5.4-mini": initChatModel("gpt-5.4-mini"),
  "claude-sonnet": initChatModel("claude-sonnet-4-6"),
};

const storeBasedModel = createMiddleware({
  name: "StoreBasedModel",
  contextSchema,
  wrapModelCall: async (request, handler) => {
    const userId = request.runtime.context.userId;  

    // Read from Store: get user's preferred model
    const store = request.runtime.store;  
    const userPrefs = await store.get(["preferences"], userId);  

    let model = request.model;

    if (userPrefs) {
      const preferredModel = userPrefs.value?.preferredModel;
      if (preferredModel && MODEL_MAP[preferredModel]) {
        model = MODEL_MAP[preferredModel];  
      }
    }

    return handler({ ...request, model });  
  },
});

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-5.5")
budget_model = init_chat_model("gpt-5.4-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-5.5",
    tools=[...],
    middleware=[context_based_model],
    context_schema=Context
)

:::

:::js

import * as z from "zod";
import { createMiddleware, initChatModel } from "langchain";

const contextSchema = z.object({
  costTier: z.string(),
  environment: z.string(),
});

// Initialize models once outside the middleware
const premiumModel = initChatModel("claude-sonnet-4-6");
const standardModel = initChatModel("gpt-5.5");
const budgetModel = initChatModel("gpt-5.4-mini");

const contextBasedModel = createMiddleware({
  name: "ContextBasedModel",
  contextSchema,
  wrapModelCall: (request, handler) => {
    // Read from Runtime Context: cost tier and environment
    const costTier = request.runtime.context.costTier;  
    const environment = request.runtime.context.environment;  

    let model;

    if (environment === "production" && costTier === "premium") {
      model = premiumModel;
    } else if (costTier === "budget") {
      model = budgetModel;
    } else {
      model = standardModel;
    }

    return handler({ ...request, model });  
  },
});

:::

See Dynamic model for more examples.

响应格式

结构化输出将非结构化文本转换为经过验证的结构化数据。当提取特定字段或为下游系统返回数据时，自由文本是不够的。

工作原理： 当你提供响应格式作为模式时，模型的最终响应将保证符合该模式。智能体运行模型/工具调用循环，直到模型完成调用工具，然后最终响应将被强制转换为提供的格式。

定义格式

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

:::python

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

:::

:::js

import { z } from "zod";

const customerSupportTicket = z.object({
  category: z.enum(["billing", "technical", "account", "product"]).describe(
    "Issue category"
  ),
  priority: z.enum(["low", "medium", "high", "critical"]).describe(
    "Urgency level"
  ),
  summary: z.string().describe(
    "One-sentence summary of the customer's issue"
  ),
  customerSentiment: z.enum(["frustrated", "neutral", "satisfied"]).describe(
    "Customer's emotional tone"
  ),
}).describe("Structured ticket information extracted from customer message");

:::

选择格式

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

State

根据对话状态配置结构化输出：

:::python

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 is a shortcut for 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-5.5",
    tools=[...],
    middleware=[state_based_output]
)

:::

:::js

import { createMiddleware } from "langchain";
import { z } from "zod";

const simpleResponse = z.object({
  answer: z.string().describe("A brief answer"),
});

const detailedResponse = z.object({
  answer: z.string().describe("A detailed answer"),
  reasoning: z.string().describe("Explanation of reasoning"),
  confidence: z.number().describe("Confidence score 0-1"),
});

const stateBasedOutput = createMiddleware({
  name: "StateBasedOutput",
  wrapModelCall: (request, handler) => {
    // request.state is a shortcut for request.state.messages
    const messageCount = request.messages.length;  

    let responseFormat;
    if (messageCount < 3) {
      // Early conversation - use simple format
      responseFormat = simpleResponse; 
    } else {
      // Established conversation - use detailed format
      responseFormat = detailedResponse; 
    }

    return handler({ ...request, responseFormat });
  },
});

:::

Store

根据用户偏好配置输出格式：

:::python

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

:::

:::js

import * as z from "zod";
import { createMiddleware } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

const verboseResponse = z.object({
  answer: z.string().describe("Detailed answer"),
  sources: z.array(z.string()).describe("Sources used"),
});

const conciseResponse = z.object({
  answer: z.string().describe("Brief answer"),
});

const storeBasedOutput = createMiddleware({
  name: "StoreBasedOutput",
  wrapModelCall: async (request, handler) => {
    const userId = request.runtime.context.userId;  

    // Read from Store: get user's preferred response style
    const store = request.runtime.store;  
    const userPrefs = await store.get(["preferences"], userId);  

    const style = userPrefs?.value?.responseStyle || "concise";
    const responseFormat =
      style === "verbose" ? verboseResponse : conciseResponse;  

    return handler({
      ...request,
      responseFormat,
    });
  },
});

:::

Runtime Context

根据运行时上下文（如用户角色或环境）配置输出格式：

:::python

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."""
    # Read from Runtime Context: user role and environment
    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-5.5",
    tools=[...],
    middleware=[context_based_output],
    context_schema=Context
)

:::

:::js

import * as z from "zod";
import { createMiddleware } from "langchain";

const contextSchema = z.object({
  userRole: z.string(),
  environment: z.string(),
});

const adminResponse = z.object({
  answer: z.string().describe("Answer"),
  debugInfo: z.record(z.any()).describe("Debug information"),
  systemStatus: z.string().describe("System status"),
});

const userResponse = z.object({
  answer: z.string().describe("Answer"),
});

const contextBasedOutput = createMiddleware({
  name: "ContextBasedOutput",
  wrapModelCall: (request, handler) => {
    // Read from Runtime Context: user role and environment
    const userRole = request.runtime.context.userRole;  
    const environment = request.runtime.context.environment;  

    let responseFormat;
    if (userRole === "admin" && environment === "production") {
      responseFormat = adminResponse;  
    } else {
      responseFormat = userResponse;  
    }

    return handler({ ...request, responseFormat });
  },
});

:::

工具上下文

工具的特殊之处在于它们既能读取也能写入上下文。

在最基本的情况下，当工具执行时，它接收 LLM 的请求参数并返回一个工具消息。工具执行其工作并产生结果。

工具还可以为模型获取重要信息，使模型能够执行并完成任务。

读取

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

State

从状态读取以检查当前会话信息：

:::python

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-5.5",
    tools=[check_authentication]
)

:::

:::js

import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";

const checkAuthentication = tool(
  async (_, runtime: ToolRuntime) => {
    // Read from State: check current auth status
    const currentState = runtime.state;
    const isAuthenticated = currentState.authenticated || false;

    if (isAuthenticated) {
      return "User is authenticated";
    } else {
      return "User is not authenticated";
    }
  },
  {
    name: "check_authentication",
    description: "Check if user is authenticated",
    schema: z.object({}),
  }
);

:::

Store

从存储读取以访问持久化的用户偏好设置：

:::python

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-5.5",
    tools=[get_preference],
    context_schema=Context,
    store=InMemoryStore()
)

:::

:::js

import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
});

const getPreference = tool(
  async ({ preferenceKey }, runtime: ToolRuntime) => {
    const userId = runtime.context.userId;

    // Read from Store: get existing preferences
    const store = runtime.store;
    const existingPrefs = await store.get(["preferences"], userId);

    if (existingPrefs) {
      const value = existingPrefs.value?.[preferenceKey];
      return value ? `${preferenceKey}: ${value}` : `No preference set for ${preferenceKey}`;
    } else {
      return "No preferences found";
    }
  },
  {
    name: "get_preference",
    description: "Get user preference from Store",
    schema: z.object({
      preferenceKey: z.string(),
    }),
  }
);

:::

Runtime Context

从运行时上下文读取配置，如 API 密钥和用户 ID：

:::python

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-5.5",
    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://..."
    )
)

:::

:::js

import * as z from "zod";
import { tool } from "@langchain/core/tools";
import { createAgent } from "langchain";

const contextSchema = z.object({
  userId: z.string(),
  apiKey: z.string(),
  dbConnection: z.string(),
});

const fetchUserData = tool(
  async ({ query }, runtime: ToolRuntime<any, typeof contextSchema>) => {
    // Read from Runtime Context: get API key and DB connection
    const { userId, apiKey, dbConnection } = runtime.context;

    // Use configuration to fetch data
    const results = await performDatabaseQuery(dbConnection, query, apiKey);

    return `Found ${results.length} results for user ${userId}`;
  },
  {
    name: "fetch_user_data",
    description: "Fetch data using Runtime Context configuration",
    schema: z.object({
      query: z.string(),
    }),
  }
);

const agent = createAgent({
  model: "gpt-5.5",
  tools: [fetchUserData],
  contextSchema,
});

:::

写入

工具结果可用于帮助智能体完成指定任务。工具可以直接向模型返回结果，同时更新智能体的记忆，以便在后续步骤中提供重要上下文。

State

使用命令写入状态以跟踪会话特定信息：

:::python

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-5.5",
    tools=[authenticate_user]
)

:::

:::js

import * as z from "zod";
import { tool } from "@langchain/core/tools";
import { createAgent } from "langchain";
import { Command } from "@langchain/langgraph";

const authenticateUser = tool(
  async ({ password }) => {
    // Perform authentication
    if (password === "correct") {
      // Write to State: mark as authenticated using Command
      return new Command({
        update: { authenticated: true },
      });
    } else {
      return new Command({ update: { authenticated: false } });
    }
  },
  {
    name: "authenticate_user",
    description: "Authenticate user and update State",
    schema: z.object({
      password: z.string(),
    }),
  }
);

:::

Store

写入存储以跨会话持久化数据：

:::python

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-5.5",
    tools=[save_preference],
    context_schema=Context,
    store=InMemoryStore()
)

:::

:::js

import * as z from "zod";
import { createAgent, tool, type ToolRuntime } from "langchain";

const savePreference = tool(
  async ({ preferenceKey, preferenceValue }, runtime: ToolRuntime<any, typeof contextSchema>) => {
    const userId = runtime.context.userId;

    // Read existing preferences
    const store = runtime.store;
    const existingPrefs = await store.get(["preferences"], userId);

    // Merge with new preference
    const prefs = existingPrefs?.value || {};
    prefs[preferenceKey] = preferenceValue;

    // Write to Store: save updated preferences
    await store.put(["preferences"], userId, prefs);

    return `Saved preference: ${preferenceKey} = ${preferenceValue}`;
  },
  {
    name: "save_preference",
    description: "Save user preference to Store",
    schema: z.object({
      preferenceKey: z.string(),
      preferenceValue: z.string(),
    }),
  }
);

:::

参见 工具，了解访问状态、存储和运行时上下文在工具中的完整示例。

生命周期上下文

控制智能体核心步骤之间的行为——拦截数据流以实现跨领域关注点，如摘要、约束和日志记录。

如你在 模型上下文 和 工具上下文 中所见，中间件 是使上下文工程实践可行的机制。中间件允许你钩入智能体生命周期的任何步骤，并执行以下操作：

1. 更新上下文 - 修改状态和存储以持久化更改，更新对话历史记录或保存见解
2. 跳入生命周期 - 根据上下文在智能体周期中的不同步骤之间移动（例如，如果满足条件则跳过工具执行，使用修改后的上下文重复模型调用）

示例：摘要

最常见的生命周期模式之一是在对话历史记录过长时自动压缩。与在 模型上下文 中显示的临时消息修剪不同，摘要会持久更新状态 - 永久替换旧消息，并保存供所有未来回合使用。

LangChain 提供了内置中间件：

:::python

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

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

:::

:::js

import { createAgent, summarizationMiddleware } from "langchain";

const agent = createAgent({
  model: "gpt-5.5",
  tools: [...],
  middleware: [
    summarizationMiddleware({
      model: "gpt-5.4-mini",
      trigger: { tokens: 4000 },
      keep: { messages: 20 },
    }),
  ],
});

:::

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

1. 使用单独的 LLM 调用摘要旧消息
2. 将其替换为状态中的摘要消息（永久）
3. 保留最近的消息以供上下文

摘要后的对话历史记录会永久更新 - 未来的回合将看到摘要而不是原始消息。

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

最佳实践

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

相关资源

• 上下文概念概述 - 了解上下文类型和何时使用它们
• 中间件 - 完整的中间件指南
• 工具 - 工具创建和上下文访问
• 记忆 - 短期和长期记忆模式
• 智能体 - 核心智能体概念

---

官方原文：src/oss/langchain/context-engineering.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。