5.3 Context Engineering

本节介绍上下文工程，这是构建可靠 Agent 的核心技能。

什么是 Context Engineering？

上下文工程（Context Engineering） 是为 LLM 提供正确的信息和工具，使其能够完成任务的技术。

"Context engineering is providing the right information and tools in the right format so the LLM can accomplish a task."

这是 AI 工程师构建可靠 Agent 的首要职责。

为什么 Agent 会失败？

Agent 失败通常有两个原因：

模型能力不足 - 模型本身无法完成任务
上下文缺失或错误 - 更常见的原因

"Context problems are the number one blocker for more reliable agents."

上下文问题是阻碍 Agent 可靠性的头号障碍。

三种上下文类型

1. 模型上下文（Model Context）

控制每次模型调用的输入，瞬态（每次调用时构建）：

元素	说明	可动态调整
System Prompt	基础指令	根据对话状态、用户偏好调整
Messages	对话历史	动态修改、摘要、修剪
Tools	可用工具	根据权限、阶段过滤
Model	使用的模型	根据复杂度切换
Response Format	输出格式	根据阶段调整 Schema

2. 工具上下文（Tool Context）

控制工具的访问和修改，持久（跨调用保存）：

操作	说明
读取	工具访问 state、store、runtime context
写入	工具通过 Command 更新状态

3. 生命周期上下文（Life-cycle Context）

管理 Agent 步骤之间发生的事情，通过中间件实现：

长对话摘要
护栏和验证
日志和监控
数据转换

三种数据源

来源	别名	作用域	示例
Runtime Context	静态配置	每次会话	API 密钥、权限、用户 ID
State	短期记忆	每次会话	消息、工具结果、认证状态
Store	长期记忆	跨会话	用户偏好、历史洞察

动态系统提示

基于状态的提示

python

from langchain.agents import dynamic_prompt, ModelRequest

@dynamic_prompt
def state_aware_prompt(request: ModelRequest) -> str:
    """根据对话状态调整提示"""
    message_count = len(request.messages)

    base = "你是一个有帮助的助手。"

    if message_count > 10:
        base += "\n这是一个较长的对话，请尽量简洁。"

    if message_count > 20:
        base += "\n对话已很长，考虑总结之前的内容。"

    return base

基于用户的提示

python

@dynamic_prompt
def user_aware_prompt(request: ModelRequest) -> str:
    """根据用户信息调整提示"""
    user_name = request.runtime.context.user_name
    user_role = request.runtime.context.role

    prompt = f"你是一个助手。当前用户是 {user_name}。"

    if user_role == "expert":
        prompt += "\n用户是专家，可以使用技术术语。"
    elif user_role == "beginner":
        prompt += "\n用户是初学者，请使用简单易懂的语言。"

    return prompt

动态工具选择

根据状态动态过滤可用工具：

python

from langchain.agents import wrap_model_call, ModelRequest, ModelResponse

@wrap_model_call
def filter_tools_by_auth(request: ModelRequest, handler) -> ModelResponse:
    """根据认证状态过滤工具"""
    is_authenticated = request.state.get("authenticated", False)

    if not is_authenticated:
        # 未认证用户只能使用公开工具
        public_tools = [
            t for t in request.tools
            if t.name.startswith("public_")
        ]
        request = request.override(tools=public_tools)

    return handler(request)

@wrap_model_call
def filter_tools_by_stage(request: ModelRequest, handler) -> ModelResponse:
    """根据对话阶段过滤工具"""
    stage = request.state.get("conversation_stage", "initial")

    stage_tools = {
        "initial": ["greet", "get_info"],
        "processing": ["search", "calculate", "fetch_data"],
        "finalizing": ["summarize", "send_report"],
    }

    allowed_tool_names = stage_tools.get(stage, [])
    filtered_tools = [
        t for t in request.tools
        if t.name in allowed_tool_names
    ]

    if filtered_tools:
        request = request.override(tools=filtered_tools)

    return handler(request)

工具中的状态访问

读取状态

python

from langchain_core.tools import tool
from langchain.agents import ToolRuntime

@tool
def check_authentication(runtime: ToolRuntime) -> str:
    """检查认证状态"""
    current_state = runtime.state
    is_authenticated = current_state.get("authenticated", False)

    if is_authenticated:
        return "用户已认证"
    else:
        return "用户未认证，请先登录"

@tool
def get_conversation_summary(runtime: ToolRuntime) -> str:
    """获取对话摘要"""
    messages = runtime.state.get("messages", [])

    if len(messages) < 5:
        return "对话刚开始，暂无摘要"

    # 返回最近消息的摘要
    recent = messages[-5:]
    summary = "\n".join([m.content[:50] for m in recent])
    return f"最近对话摘要:\n{summary}"

写入状态

使用 Command 对象更新状态：

python

from langchain_core.tools import tool
from langchain.agents import ToolRuntime, Command

@tool
def authenticate_user(
    username: str,
    password: str,
    runtime: ToolRuntime
) -> Command:
    """用户认证"""
    # 验证逻辑（示例）
    if username == "admin" and password == "correct":
        return Command(
            update={
                "authenticated": True,
                "user_role": "admin"
            }
        )
    elif password == "correct":
        return Command(
            update={
                "authenticated": True,
                "user_role": "user"
            }
        )
    else:
        return Command(
            update={
                "authenticated": False,
                "auth_attempts": runtime.state.get("auth_attempts", 0) + 1
            }
        )

@tool
def set_conversation_stage(
    stage: str,
    runtime: ToolRuntime
) -> Command:
    """设置对话阶段"""
    valid_stages = ["initial", "processing", "finalizing"]

    if stage not in valid_stages:
        return Command(update={})  # 不更新

    return Command(
        update={"conversation_stage": stage}
    )

摘要中间件

管理长对话的上下文：

python

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

agent = create_agent(
    model="gpt-4o",
    tools=[my_tools],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",  # 用于摘要的模型
            trigger={"tokens": 4000},  # 触发阈值
            keep={"messages": 20},  # 保留最近消息数
        ),
    ],
)

自定义摘要逻辑

python

from langchain.agents import before_model

@before_model
def custom_summarization(state, runtime):
    """自定义摘要逻辑"""
    messages = state["messages"]

    if len(messages) > 50:
        # 保留系统消息和最近 20 条
        system_msgs = [m for m in messages if m.type == "system"]
        recent_msgs = messages[-20:]

        # 创建摘要消息
        old_msgs = messages[len(system_msgs):-20]
        summary_content = f"[之前对话摘要: 共 {len(old_msgs)} 条消息]"

        state["messages"] = (
            system_msgs +
            [SystemMessage(content=summary_content)] +
            recent_msgs
        )

    return None

Agent 循环架构

┌─────────────────────────────────────────────────────────┐
│                     Agent Loop                           │
├─────────────────────────────────────────────────────────┤
│                                                          │
│   ┌──────────────┐    ┌──────────────┐    ┌──────────┐  │
│   │  Model Call  │───▶│ Tool Execute │───▶│ Continue │  │
│   └──────────────┘    └──────────────┘    └──────────┘  │
│          │                   │                  │        │
│          ▼                   ▼                  ▼        │
│   ┌──────────────────────────────────────────────────┐  │
│   │                  Context Sources                  │  │
│   ├──────────────┬──────────────┬───────────────────┤  │
│   │   Runtime    │    State     │      Store        │  │
│   │   Context    │ (短期记忆)   │   (长期记忆)       │  │
│   └──────────────┴──────────────┴───────────────────┘  │
│                                                          │
└─────────────────────────────────────────────────────────┘

完整示例

python

from dataclasses import dataclass
from langchain.agents import (
    create_agent,
    ToolRuntime,
    Command,
    dynamic_prompt,
    wrap_model_call,
    ModelRequest,
    ModelResponse,
)
from langchain.agents.middleware import SummarizationMiddleware
from langchain_core.tools import tool
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver

# 上下文定义
@dataclass
class UserContext:
    user_id: str
    user_name: str
    expertise_level: str  # "beginner" | "intermediate" | "expert"

# 动态提示
@dynamic_prompt
def expertise_based_prompt(request: ModelRequest) -> str:
    level = request.runtime.context.expertise_level
    name = request.runtime.context.user_name

    prompts = {
        "beginner": f"你是一个耐心的老师，用简单的语言向 {name} 解释概念。",
        "intermediate": f"你是一个助手，向 {name} 提供详细但清晰的回答。",
        "expert": f"你是一个专业顾问，可以与 {name} 使用技术术语深入讨论。",
    }

    return prompts.get(level, prompts["intermediate"])

# 动态工具过滤
@wrap_model_call
def filter_by_expertise(request: ModelRequest, handler) -> ModelResponse:
    level = request.runtime.context.expertise_level

    if level == "beginner":
        # 初学者只能使用简单工具
        simple_tools = [t for t in request.tools if not t.name.startswith("advanced_")]
        request = request.override(tools=simple_tools)

    return handler(request)

# 工具定义
@tool
def search_docs(query: str) -> str:
    """搜索文档"""
    return f"搜索结果: {query} 相关内容..."

@tool
def advanced_analysis(data: str, runtime: ToolRuntime) -> str:
    """高级分析（仅限专家）"""
    return f"高级分析结果: {data}"

@tool
def track_learning_progress(
    topic: str,
    status: str,
    runtime: ToolRuntime[UserContext]
) -> Command:
    """追踪学习进度"""
    user_id = runtime.context.user_id

    # 保存到长期记忆
    if runtime.store:
        runtime.store.put(
            ("learning_progress", user_id),
            topic,
            {"status": status, "timestamp": "now"}
        )

    return Command(
        update={"last_topic": topic}
    )

# 创建 Agent
agent = create_agent(
    model="gpt-4o",
    tools=[search_docs, advanced_analysis, track_learning_progress],
    context_schema=UserContext,
    middleware=[
        expertise_based_prompt,
        filter_by_expertise,
        SummarizationMiddleware(
            model="gpt-4o-mini",
            trigger={"tokens": 3000},
            keep={"messages": 15},
        ),
    ],
    store=InMemoryStore(),
    checkpointer=InMemorySaver(),
)

# 使用
result = agent.invoke(
    {"messages": [{"role": "user", "content": "解释一下机器学习"}]},
    context=UserContext(
        user_id="user_789",
        user_name="小明",
        expertise_level="beginner"
    ),
    config={"configurable": {"thread_id": "session_1"}}
)

最佳实践

实践	说明
从静态开始	先使用静态配置，再逐步添加动态特性
逐个测试	每次只测试一个功能
监控 Token	跟踪 token 使用和延迟
使用内置	优先使用内置中间件
区分瞬态/持久	明确哪些数据需要保存

上下文类型对比

类型	持久性	作用域	修改方式
模型上下文	瞬态	单次调用	中间件
工具上下文	持久	跨调用	Command
生命周期上下文	持久	跨步骤	中间件

上一节：5.2 Runtime

下一节：5.4 Model Context Protocol MCP

5.3 Context Engineering ​

什么是 Context Engineering？ ​

为什么 Agent 会失败？ ​

三种上下文类型 ​

1. 模型上下文（Model Context） ​

2. 工具上下文（Tool Context） ​

3. 生命周期上下文（Life-cycle Context） ​

三种数据源 ​

动态系统提示 ​

基于状态的提示 ​

基于用户的提示 ​

动态工具选择 ​

工具中的状态访问 ​

读取状态 ​

写入状态 ​

摘要中间件 ​

自定义摘要逻辑 ​

Agent 循环架构 ​

完整示例 ​

最佳实践 ​

上下文类型对比 ​

5.3 Context Engineering

什么是 Context Engineering？

为什么 Agent 会失败？

三种上下文类型

1. 模型上下文（Model Context）

2. 工具上下文（Tool Context）

3. 生命周期上下文（Life-cycle Context）

三种数据源

动态系统提示

基于状态的提示

基于用户的提示

动态工具选择

工具中的状态访问

读取状态

写入状态

摘要中间件

自定义摘要逻辑

Agent 循环架构

完整示例

最佳实践

上下文类型对比