09 | Context Engineering——比 Prompt 更重要的「给模型看什么」

问题: "Python 的 asyncio 怎么用在 Web 爬虫中？"

System Prompt: 你是一个 Python 专家。
用户问题: Python 的 asyncio 怎么用在 Web 爬虫中？

System Prompt: 你是一个 Python 专家。
[附带 50 篇随机 Python 文章的摘要]
用户问题: Python 的 asyncio 怎么用在 Web 爬虫中？

System Prompt: 你是一个 Python 专家，专注异步编程方向。

[参考文档 1] aiohttp 官方文档: 异步 HTTP 客户端示例...
[参考文档 2] Python 3.12 asyncio 新特性: TaskGroup 用法...
[参考文档 3] 实际案例: 用 asyncio + aiohttp 爬取 1000 个页面的性能对比...

用户问题: Python 的 asyncio 怎么用在 Web 爬虫中？

128K tokens 总量
├── System Prompt:      ~500 tokens    (0.4%)
├── 历史对话:           ~2,000 tokens  (1.6%)
├── 检索到的文档:       ~120,000 tokens (93.8%)  ← 大头在这里
├── 用户当前问题:       ~200 tokens    (0.2%)
└── 留给模型输出:       ~5,300 tokens  (4.1%)

# RAG 检索返回了 20 个文档片段，每个都有相关度分数
results = [
    {"text": "...", "score": 0.95},
    {"text": "...", "score": 0.87},
    {"text": "...", "score": 0.82},
    {"text": "...", "score": 0.45},  # 相关度低
    {"text": "...", "score": 0.38},  # 相关度低
    ...
]

# 策略：只取 top-3 且分数 > 0.7 的
selected = [r for r in results[:5] if r["score"] > 0.7]

# 构建上下文
context = "\n\n".join([
    f"[参考文档 {i+1}] (相关度: {r['score']:.2f})\n{r['text']}"
    for i, r in enumerate(selected)
])

System Prompt + 用户问题 → 不需要检索，直接回答
窗口占用：极少

System Prompt + RAG 检索结果 + 用户问题 → 基于文档回答
窗口占用：中等

System Prompt + RAG 检索 + 历史推理链 + CoT 提示 + 用户问题
窗口占用：大量

def build_context(user_query, conversation_history):
    """根据任务复杂度动态构建上下文"""

    # 1. 基础层：始终包含
    context = [SYSTEM_PROMPT]

    # 2. 判断是否需要 RAG
    if needs_external_knowledge(user_query):
        docs = rag_search(user_query, top_k=3)
        context.append(format_documents(docs))

    # 3. 判断是否需要历史对话
    if len(conversation_history) > 0:
        # 不是全部历史，只取最近 3 轮
        recent = conversation_history[-6:]  # 3 轮 = 6 条消息
        context.append(format_history(recent))

    # 4. 判断是否需要 CoT 引导
    if is_complex_reasoning(user_query):
        context.append("请按步骤分析这个问题...")

    return "\n\n".join(context)

def trim_history_sliding(history, max_turns=5):
    """保留最近 N 轮对话（每轮 = user + assistant 共 2 条）"""
    return history[-(max_turns * 2):]

def compress_history(history, max_tokens=2000):
    """超长历史时压缩早期对话为摘要"""
    if estimate_tokens(history) <= max_tokens:
        return history

    # 分割：早期 + 近期
    split_point = len(history) // 2
    early = history[:split_point]
    recent = history[split_point:]

    # 用 LLM 压缩早期对话
    summary = llm.invoke(
        f"将以下对话历史压缩为一段简洁的摘要（保留关键信息和结论）：\n"
        f"{format_messages(early)}"
    )

    # 返回：摘要 + 近期对话
    return [
        {"role": "system", "content": f"[历史对话摘要] {summary.content}"},
        *recent
    ]

# 不是保留原始对话
history_raw = [
    {"role": "user", "content": "我叫张三，在北京做 AI 开发"},
    {"role": "assistant", "content": "你好张三..."},
    {"role": "user", "content": "我们公司在做 RAG 相关产品"},
    ...
]

# 而是提取为结构化记忆
memory_structured = {
    "user_name": "张三",
    "location": "北京",
    "profession": "AI 开发",
    "company_focus": "RAG 产品"
}

# 注入上下文时只需要一小段
context = f"用户信息：{memory_structured['user_name']}，{memory_structured['location']}，{memory_structured['profession']}"

AGENT_SYSTEM_PROMPT = """
## 角色
你是一个专业的技术文档助手，专注于 AI/LLM 领域。

## 能力
你可以搜索互联网、查询知识库、执行 Python 代码进行数据分析。

## 行为规范
1. 回答必须基于事实，引用具体来源
2. 如果不确定，明确说"我不确定"而不是编造
3. 代码示例用 Python，除非用户指定其他语言
4. 使用中文回答，技术术语保留英文

## 工具使用规则
- 遇到实时性问题（价格、新闻、天气），必须先搜索
- 遇到计算问题，使用 calculator 而不是心算
- 不要在不确定时猜测，用工具验证

## 输出格式
- 简洁直接，不要"很高兴为您解答"等废话
- 技术内容用 Markdown 格式化
- 关键结论加粗
"""

初始上下文: System Prompt + 用户任务
第 1 轮: + Thought₁ + Action₁ + Observation₁
第 2 轮: + Thought₂ + Action₂ + Observation₂
第 3 轮: + Thought₃ + Action₃ + Observation₃
...
第 N 轮: 上下文已经很长了

def compress_agent_context(history, current_loop):
    """压缩 Agent 的历史循环信息"""

    if current_loop <= 5:
        # 前 5 轮：保留完整信息
        return history

    # 超过 5 轮：压缩早期循环
    early = history[:5]   # 前 5 轮
    recent = history[5:]  # 最近几轮

    # 把早期循环压缩为摘要
    early_summary = summarize_agent_loops(early)

    return [
        {"role": "system", "content": f"[已完成的操作摘要] {early_summary}"},
        *recent  # 最近几轮保持完整
    ]