📚 AI 工程与智能体教程站

第 11 章 · 上下文工程：截断、压缩与 Token 预算 ​

本章目标：

  • 理解上下文窗口限制的应对策略
  • 掌握三种截断策略：FIFO、最近优先、摘要优先
  • 实现上下文压缩（LLM 摘要）
  • 设计 Token 预算控制系统

11.1 上下文窗口限制 ​

模型上下文窗口适用场景
gpt-4o-mini128K简单任务
gpt-4o128K通用任务
claude-3-opus200K复杂推理
claude-3.5-sonnet200K生产主力

11.2 截断策略 ​

python
from typing import List
from langchain_core.messages import BaseMessage

def truncate_messages(
    messages: List[BaseMessage],
    max_tokens: int,
    strategy: str = "recency"
) -> List[BaseMessage]:
    """上下文截断
    
    Args:
        messages: 消息列表
        max_tokens: 最大 token 数
        strategy: "fifo" | "recency" | "summary"
    """
    total_tokens = sum(len(m.content) // 2 for m in messages)
    
    if total_tokens <= max_tokens:
        return messages
    
    if strategy == "recency":
        # 保留最近的消息
        while total_tokens > max_tokens and len(messages) > 1:
            removed = messages.pop(0)
            total_tokens -= len(removed.content) // 2
        return messages
    
    elif strategy == "fifo":
        # 固定保留第一条（系统提示）
        return [messages[0]] + messages[-(max_tokens // 100):]
    
    elif strategy == "summary":
        # 用 LLM 摘要早期消息
        return summarize_and_keep(messages, max_tokens)
    
    return messages

def summarize_and_keep(
    messages: List[BaseMessage],
    max_tokens: int
) -> List[BaseMessage]:
    """使用 LLM 压缩早期消息"""
    from ai import createGateway
    gateway = createGateway({
        "apiKey": process.env.AI_GATEWAY_API_KEY
    })
    model = gateway("openai/gpt-4o-mini")
    
    early_msgs = messages[:-5]  # 保留最近 5 条
    late_msgs = messages[-5:]
    
    # 摘要早期对话
    summary_prompt = "请总结以下对话的关键内容："
    summary_result = model.generate([
        {"role": "system", "content": summary_prompt},
        {"role": "user", "content": str(early_msgs)}
    ])
    
    return [
        {"role": "assistant", "content": f"[历史摘要] {summary_result}"}
    ] + late_msgs

11.3 Token 预算控制 ​

python
import asyncio
from dataclasses import dataclass
from typing import Dict

@dataclass
class TokenBudget:
    max_tokens: int
    used_tokens: int = 0
    cost_per_1k: float = 0.00001  # $ per 1K tokens
    
    @property
    def remaining(self) -> int:
        return self.max_tokens - self.used_tokens
    
    @property
    def estimated_cost(self) -> float:
        return (self.used_tokens / 1000) * self.cost_per_1k
    
    def track(self, tokens: int) -> bool:
        """记录 token 使用，返回是否超出预算"""
        self.used_tokens += tokens
        return self.used_tokens <= self.max_tokens

# 使用示例
budget = TokenBudget(max_tokens=100_000)

async def call_llm(prompt: str, budget: TokenBudget) -> str:
    # 估算输入 token
    input_tokens = len(prompt) // 2
    
    if not budget.track(input_tokens):
        raise RuntimeError(f"Token 预算不足: 已用 {budget.used_tokens}/{budget.max_tokens}")
    
    # 调用 LLM
    result = await invoke_model(prompt)
    
    # 记录输出 token
    output_tokens = len(result) // 2
    budget.track(output_tokens)
    
    return result

11.4 跨状态持久化（Store） ​

python
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

# 长期记忆存储
store = InMemoryStore()

# 存储用户偏好
store.put(("users", "user_123"), "preferences", {
    "language": "zh",
    "timezone": "Asia/Shanghai",
    "project_context": "RAG 系统开发"
})

# 检索记忆
prefs = store.get(("users", "user_123"), "preferences")
print(prefs.value)  # {'language': 'zh', ...}

本章小结 ​

  • 截断策略：recency（保留最新）、fifo（保留最早）、summary（LLM 摘要）
  • Token 预算控制防止超支
  • Store 实现跨会话的长期记忆
  • 合理组合策略可在成本和质量间平衡

🛠️ 动手实践 ​

  1. 实现三种截断策略，对比效果差异
  2. 构建 Token 预算监控器，实时显示成本和剩余预算
  3. 实现带摘要压缩的长对话 Agent