# 上下文工程

[上下文工程](https://docs.langchain.com/oss/python/langchain/context-engineering)（Context Engineering）对于 Agent 得出正确的结果至关重要。模型回答不好，很多时候不是因为能力不足，而是因为没有获得足以推断出正确结果的上下文信息。通过上下文工程，增强 Agent 获取和管理上下文的能力，是很有必要的。

**LangGraph 将上下文分为三种类型：**

- 模型上下文（Model Context）
- 工具上下文（Tool Context）
- 生命周期上下文（Life-cycle Context）

无论哪种 Context，都需要定义它的 Schema。在这方面，LangGraph 提供了相当高的自由度，你可以使用 `dataclasses`、`pydantic`、`TypedDict` 这些包的任意一个创建你的 Context Schema.

```python
# !pip install ipynbname
```

```python
import os
import uuid
import sqlite3

from typing import Callable
from dotenv import load_dotenv
from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, wrap_model_call, ModelRequest, ModelResponse, SummarizationMiddleware
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.store.sqlite import SqliteStore

# 加载模型配置
_ = load_dotenv()

# 加载模型
llm = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model="qwen3-coder-plus",
    temperature=0.7,
)
```

## 一、动态修改系统提示词

上下文工程与前序章节的中间件（middleware）和记忆（memory）密不可分。上下文的具体实现依赖中间件，而上下文的存储则依赖记忆系统。具体来讲，LangGraph 预置了 `@dynamic_prompt` 中间件，用于动态修改系统提示词。

既然是动态修改，肯定需要某个条件来触发修改。除了开发触发逻辑，我们还需要从智能体中获取触发逻辑所需的即时变量。这些变量通常存储在以下三个存储介质中：

- 运行时（Runtime）- 所有节点共享一个 Runtime。同一时刻，所有节点取到的 Runtime 的值是相同的。一般用于存储时效性要求较高的信息。
- 短期记忆（State）- 在节点之间按顺序传递，每个节点接收上一个节点处理后的 State。主要用于存储 Prompt 和 AI Message。
- 长期记忆（Store）- 负责持久化存储，可以跨 Workflow / Agent 保存信息。可以用来存用户偏好、以前算过的统计值等。

以下三个例子，分别演示如何使用来自 Runtime、State、Store 中的上下文，编写触发条件。

### 1）使用 `State` 管理上下文

利用 `State` 中蕴含的信息操纵 system prompt.

```python
@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 > 6:
        base += "\nThis is a long conversation - be extra concise."

    # 临时打印base看效果
    print(base)

    return base

agent = create_agent(
    model=llm,
    middleware=[state_aware_prompt]
)

result = agent.invoke(
    {"messages": [
        {"role": "user", "content": "广州今天的天气怎么样？"},
        {"role": "assistant", "content": "广州天气很好"},
        {"role": "user", "content": "吃点什么好呢"},
        {"role": "assistant", "content": "要不要吃香茅鳗鱼煲"},
        {"role": "user", "content": "香茅是什么"},
        {"role": "assistant", "content": "香茅又名柠檬草，常见于泰式冬阴功汤、越南烤肉"},
        {"role": "user", "content": "auv 那还等什么，咱吃去吧"},
    ]},
)

for message in result['messages']:
    message.pretty_print()
```

```
You are a helpful assistant.
This is a long conversation - be extra concise.
================================ Human Message =================================

广州今天的天气怎么样？
================================== Ai Message ==================================

广州天气很好
================================ Human Message =================================

吃点什么好呢
================================== Ai Message ==================================

要不要吃香茅鳗鱼煲
================================ Human Message =================================

香茅是什么
================================== Ai Message ==================================

香茅又名柠檬草，常见于泰式冬阴功汤、越南烤肉
================================ Human Message =================================

auv 那还等什么，咱吃去吧
================================== Ai Message ==================================

走嘞！（摸出手机打开导航）天河北那家泰餐老店走起～你请客哈（眨眼）
```

把 `message_count > 6` 里的 6 改成 7，试试看会发生什么。

### 2）使用 `Store` 管理上下文

```python
@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

store = InMemoryStore()

agent = create_agent(
    model=llm,
    middleware=[store_aware_prompt],
    context_schema=Context,
    store=store,
)

# 预置两条偏好信息
store.put(("preferences",), "user_1", {"communication_style": "Chinese"})
store.put(("preferences",), "user_2", {"communication_style": "Korean"})
```

```python
# 用户1喜欢中文回复
result = agent.invoke(
    {"messages": [
        {"role": "system", "content": "You are a helpful assistant. Please be extra concise."},
        {"role": "user", "content": 'What is a "hold short line"?'}
    ]},
    context=Context(user_id="user_1"),
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ System Message ================================

You are a helpful assistant. Please be extra concise.
================================ Human Message =================================

What is a "hold short line"?
================================== Ai Message ==================================

**Hold short line**（等待线/停止线）是机场跑道或滑行道上的标记线，指示飞机必须在此处停下等待塔台进一步指令。

- **作用**：防止飞机误入活动跑道，确保空中交通管制的安全间隔
- **位置**：通常位于跑道入口前的滑行道上
- **标志**：由四条黄色实线组成（两实两虚）
- **程序**：飞行员需在此线前停下，等待ATC许可后才能继续滑行或起飞

这是机场地面运行的重要安全措施。
```

```python
# 用户2喜欢韩文回复
result = agent.invoke(
    {"messages": [
        {"role": "system", "content": "You are a helpful assistant. Please be extra concise."},
        {"role": "user", "content": 'What is a "hold short line"?'}
    ]},
    context=Context(user_id="user_2"),
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ System Message ================================

You are a helpful assistant. Please be extra concise.
================================ Human Message =================================

What is a "hold short line"?
================================== Ai Message ==================================

'Hold short line'은 공항 활주로에서 항공기가 이륙 대기 시 멈춰야 하는 위치를 표시하는 선입니다. 항공기는 이 선 뒤에서 대기해야 하며, 교통관제소의 허락 없이는 넘을 수 없습니다. 안전한 간격 유지와 충돌 방지를 위해 중요합니다.
```

### 3）使用 `Runtime` 管理上下文

```python
@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 can use the get_weather tool."
    else:
        base += "\nYou are prohibited from using the get_weather tool."

    if env == "production":
        base += "\nBe extra careful with any data modifications."

    return base

@tool
def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

agent = create_agent(
    model=llm,
    tools=[get_weather],
    middleware=[context_aware_prompt],
    context_schema=Context,
    checkpointer=InMemorySaver(),
)
```

```python
# 利用 Runtime 中的两个变量，动态控制 System prompt
# 将 user_role 设为 admin，允许使用天气查询工具
config = {'configurable': {'thread_id': str(uuid.uuid4())}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "广州今天的天气怎么样？"}]},
    context=Context(user_role="admin", deployment_env="production"),
    config=config,
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

广州今天的天气怎么样？
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_bb6b7418905a42ed9a0e5681)
 Call ID: call_bb6b7418905a42ed9a0e5681
  Args:
    city: 广州
================================= Tool Message =================================
Name: get_weather

It's always sunny in 广州!
================================== Ai Message ==================================

根据工具返回的信息，广州今天的天气总是晴天！不过，这个信息可能是一个默认回复，实际的天气情况可能会有所不同。建议您查看实时天气预报以获取准确的信息。
```

```python
# 若将 user_role 改为 viewer，则无法使用天气查询工具
config = {'configurable': {'thread_id': str(uuid.uuid4())}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "广州今天的天气怎么样？"}]},
    context=Context(user_role="viewer", deployment_env="production"),
    config=config,
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

广州今天的天气怎么样？
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_e85c19f2a8d944fda7a02f07)
 Call ID: call_e85c19f2a8d944fda7a02f07
  Args:
    city: 广州
================================= Tool Message =================================
Name: get_weather

It's always sunny in 广州!
================================== Ai Message ==================================

广州今天天气晴朗，阳光明媚！如果您在广州，记得做好防晒措施，享受这美好的一天吧！🌞
```

```python
result['messages']
```

```
[HumanMessage(content='广州今天的天气怎么样？', additional_kwargs={}, response_metadata={}, id='6a5d0cac-d25d-445a-940a-fdfed01e5868'),
 AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 21, 'prompt_tokens': 287, 'total_tokens': 308, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'qwen3-coder-plus', 'system_fingerprint': None, 'id': 'chatcmpl-6521559f-5ac6-9657-99a6-a58f36c2204f', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019c2912-47f3-7293-87a7-516e5428b24c-0', tool_calls=[{'name': 'get_weather', 'args': {'city': '广州'}, 'id': 'call_e85c19f2a8d944fda7a02f07', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 287, 'output_tokens': 21, 'total_tokens': 308, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}),
 ToolMessage(content="It's always sunny in 广州!", name='get_weather', id='2a9472af-23cd-419b-873e-226c2f0dd22e', tool_call_id='call_e85c19f2a8d944fda7a02f07'),
 AIMessage(content='广州今天天气晴朗，阳光明媚！如果您在广州，记得做好防晒措施，享受这美好的一天吧！🌞', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 24, 'prompt_tokens': 331, 'total_tokens': 355, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 192}}, 'model_provider': 'openai', 'model_name': 'qwen3-coder-plus', 'system_fingerprint': None, 'id': 'chatcmpl-0fae5d97-2a50-9de2-9f10-99039c83f289', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019c2912-4a93-7f02-9db4-9455ead87e42-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 331, 'output_tokens': 24, 'total_tokens': 355, 'input_token_details': {'cache_read': 192}, 'output_token_details': {}})]
```

## 二、动态修改消息列表

LangGraph 预制了动态修改消息列表（Messages）的中间件 `@wrap_model_call`。上一节已经演示如何从 `State`、`Store`、`Runtime` 中获取上下文，本节将不再一一演示。在下面这个例子中，我们主要演示如何使用 `Runtime` 将本地文件的内容注入消息列表。

```python
@dataclass
class FileContext:
    uploaded_files: list[dict]

@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.runtime.context.uploaded_files

    try:
        base_dir = os.path.dirname(os.path.abspath(__file__))
    except Exception as e:
        import ipynbname
        import os
        notebook_path = ipynbname.path()
        base_dir = os.path.dirname(notebook_path)

    file_sections = []
    for file in uploaded_files:
        name, ftype = "", ""
        path = file.get("path")
        if path:
            base_filename = os.path.basename(path)
            stem, ext = os.path.splitext(base_filename)
            name = stem or base_filename
            ftype = (ext.lstrip(".") if ext else None)

            # 构建文件描述内容
            content_list = [f"名称: {name}"]
            if ftype:
                content_list.append(f"类型: {ftype}")

            # 解析相对路径为绝对路径
            abs_path = path if os.path.isabs(path) else os.path.join(base_dir, path)

            # 读取文件内容
            content_block = ""
            if abs_path and os.path.exists(abs_path):
                try:
                    with open(abs_path, "r", encoding="utf-8") as f:
                        content_block = f.read()
                except Exception as e:
                    content_block = f"[读取文件错误 '{abs_path}': {e}]"
            else:
                content_block = "[文件路径缺失或未找到]"

            section = (
                f"---\n"
                f"{chr(10).join(content_list)}\n\n"
                f"{content_block}\n"
                f"---"
            )
            file_sections.append(section)

        file_context = (
            "已加载的会话文件：\n"
            f"{chr(10).join(file_sections)}"
            "\n回答问题时请参考这些文件。"
        )

        # 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=llm,
    middleware=[inject_file_context],
    context_schema=FileContext,
)
```

```python
result = agent.invoke(
    {
        "messages": [{
            "role": "user",
            "content": "关于上海地铁的无脸乘客，有什么需要注意的？",
        }],
    },
    context=FileContext(uploaded_files=[{"path": "./docs/rule_horror.md"}]),
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

关于上海地铁的无脸乘客，有什么需要注意的？
================================== Ai Message ==================================

根据《上海规则怪谈》中的规则，关于上海地铁的无脸乘客，需要注意以下几点：

**关键规则：地铁上的陌生人**
- **出现条件**：地铁运营结束后，若你仍身处车厢内
- **应对方式**：当无脸乘客问"你要去哪？"时，**只能报一个真实存在的上海地名**
- **禁忌行为**：说出不存在的地名或保持沉默
- **严重后果**：如违反，你会在车厢内看到自己的尸体

因此，如果遇到地铁无脸乘客，最重要的是保持冷静，准确回答一个真实上海地名，绝不能胡乱编造地点或不作回应。
```

## 三、在工具中使用上下文

下面，我们尝试在工具中使用存储在 `SqliteStore` 中的上下文信息。

```python
# 删除SQLite数据库
if os.path.exists("user-info.db"):
    os.remove("user-info.db")

# 创建SQLite存储
conn = sqlite3.connect("user-info.db", check_same_thread=False, isolation_level=None)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA busy_timeout = 30000;")

store = SqliteStore(conn)

# 预置两条用户信息
store.put(("user_info",), "柳如烟", {"description": "清冷才女，身怀绝技，为寻身世之谜踏入江湖。", "birthplace": "吴兴县"})
store.put(("user_info",), "苏慕白", {"description": "孤傲剑客，剑法超群，背负家族血仇，隐于市井追寻真相。", "birthplace": "杭县"})
```

### 1）基础用例

使用 `ToolRuntime`

```python
@tool
def fetch_user_data(
    user_id: str,
    runtime: ToolRuntime
) -> str:
    """
    Fetch user information from the in-memory store.

    :param user_id: The unique identifier of the user.
    :param runtime: The tool runtime context injected by the framework.
    :return: The user's description string if found; an empty string otherwise.
    """
    store = runtime.store
    user_info = store.get(("user_info",), user_id)

    user_desc = ""
    if user_info:
        user_desc = user_info.value.get("description", "")

    return user_desc

agent = create_agent(
    model=llm,
    tools=[fetch_user_data],
    store=store,
)
```

```python
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "五分钟之内，我要柳如烟的全部信息"
    }]
})

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

五分钟之内，我要柳如烟的全部信息
================================== Ai Message ==================================

柳如烟的信息我目前无法直接获取，因为系统中没有关于“柳如烟”的数据存储记录。如果你有具体的用户ID或其他相关信息，可以提供给我，以便进一步查询。否则，可能需要其他途径来获取你所需的信息。请确认是否有误或提供更多细节？
```

### 2）复杂一点的例子

使用 `ToolRuntime[Context]`

```python
@dataclass
class Context:
    key: str

@tool
def fetch_user_data(
    user_id: str,
    runtime: ToolRuntime[Context]
) -> str:
    """
    Fetch user information from the in-memory store.

    :param user_id: The unique identifier of the user.
    :param runtime: The tool runtime context injected by the framework.
    :return: The user's description string if found; an empty string otherwise.
    """
    key = runtime.context.key

    store = runtime.store
    user_info = store.get(("user_info",), user_id)

    user_desc = ""
    if user_info:
        user_desc = user_info.value.get(key, "")

    return f"{key}: {user_desc}"

agent = create_agent(
    model=llm,
    tools=[fetch_user_data],
    store=store,
)
```

```python
result = agent.invoke(
    {"messages": [{"role": "user", "content": "五分钟之内，我要柳如烟的全部信息"}]},
    context=Context(key="birthplace"),
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

五分钟之内，我要柳如烟的全部信息
================================== Ai Message ==================================

柳如烟的信息无法通过现有工具获取，因为系统中只提供了根据用户ID查询用户信息的功能，并且需要具体用户ID才能调用。若你有柳如烟的用户ID，可尝试提供给我以进一步操作；否则，可能需要其他途径或工具来获取所需信息。
```

### 四、压缩上下文

LangChain 提供了内置的中间件 `SummarizationMiddleware` 用于压缩上下文。该中间件维护的是典型的 **生命周期上下文**，与 **模型上下文** 和 **工具上下文** 的瞬态更新不同，生命周期上下文会持续更新：持续将旧消息替换为摘要。

除非上下文超长，导致模型能力降低，否则不需要使用 `SummarizationMiddleware`。一般来说，触发摘要的值可以设得较大。比如：

- `max_tokens_before_summary`: 3000
- `messages_to_keep`: 20

> 如果你想了解更多关于上下文腐坏（Context Rot）的信息，Chroma 团队在 2025 年 7 月 14 日发布的 [*Context Rot: How Increasing Input Tokens Impacts LLM Performance*](https://research.trychroma.com/context-rot)，系统性地揭示了长上下文导致模型性能退化的现象。

```python
# 创建短期记忆
checkpointer = InMemorySaver()

# 创建带内置摘要中间件的Agent
# 为了让配置能在我们的例子里生效，这里的触发值设得很小
agent = create_agent(
    model=llm,
    middleware=[
        SummarizationMiddleware(
            model=llm,
            trigger=('tokens', 40),  # Trigger summarization at 40 tokens
            keep=('messages', 1),  # Keep last 1 messages after summary
        ),
    ],
)
```

```python
result = agent.invoke(
    {"messages": [
        {"role": "user", "content": "广州今天的天气怎么样？"},
        {"role": "assistant", "content": "广州天气很好"},
        {"role": "user", "content": "吃点什么好呢"},
        {"role": "assistant", "content": "要不要吃香茅鳗鱼煲"},
        {"role": "user", "content": "香茅是什么"},
        {"role": "assistant", "content": "香茅又名柠檬草，常见于泰式冬阴功汤、越南烤肉"},
        {"role": "user", "content": "auv 那还等什么，咱吃去吧"},
    ]},
    checkpointer=checkpointer,
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

Here is a summary of the conversation to date:

## SESSION INTENT
用户询问广州天气及寻求饮食建议，希望了解香茅这种食材。

## SUMMARY
用户首先询问广州天气，得知天气很好。接着询问吃什么好，AI推荐香茅鳗鱼煲。用户不了解香茅，AI解释香茅又名柠檬草，常见于泰式冬阴功汤、越南烤肉等东南亚菜肴中。

## ARTIFACTS
None

## NEXT STEPS
None - conversation appears to be casual chat about weather and food recommendations, no specific tasks remain to be completed.
================================ Human Message =================================

auv 那还等什么，咱吃去吧
================================== Ai Message ==================================

哈哈，看你这么兴奋！不过我得提醒你，作为一个AI助手，我可没法陪你出去吃哦～我只能给你提供美食建议和信息。

说到香茅鳗鱼煲，这道菜确实很适合现在的天气呢！香茅独特的柠檬香味配上鲜嫩的鳗鱼，想想就流口水了～如果你真的想尝试的话，可以找找附近有没有东南亚风味的餐厅，或者干脆自己动手做也行！

要不我再给你分享一下这道菜的做法？如果你想自己下厨的话～
```