Context Engineering深度解析：从RAG到下一代AI Agent记忆架构
2026年，AI Agent的战场已经从"模型能力"转向了"上下文能力"。谁能给模型最精准的信息，谁就能让Agent做出最靠谱的决策。这就是Context Engineering——正在取代Prompt Engineering成为AI工程师核心技能的新范式。
一、从Prompt Engineering到Context Engineering的范式跃迁
1.1 三阶段演化史
回顾AI应用工程的发展，我们可以清晰地看到三个阶段：
第一阶段：Prompt Engineering（2023-2024）
这个阶段的核心命题是"如何把话说清楚"。工程师们研究各种提示词技巧：角色设定、Few-shot示例、Chain-of-Thought思维链、ReAct推理框架。本质上是在解决一个问题——让模型更好地理解人类的意图。
# 典型的Prompt Engineering思维
prompt = """
你是一个有10年经验的Python高级工程师。
请按照以下步骤思考：
1. 分析问题
2. 设计方案
3. 编写代码
4. 测试验证
用户问题：{user_question}
"""
Prompt Engineering的问题在于：它假设所有需要的信息都已经在提示词里了。但现实中，Agent需要的上下文是动态变化的——用户的历史对话、外部系统的实时数据、工具调用的返回结果、当前任务的状态……这些都是静态提示词无法覆盖的。
第二阶段：Context Engineering（2025-2026）
Context Engineering的核心命题变成了"在关键时刻，让模型看到最精准的信息"。它不再只关注"怎么说"，而是关注"给什么"和"什么时候给"。
Context Engineering = f(指令, 用户信息, 外部数据, 工具结果, 任务状态, 历史记忆)
这个阶段的标志性事件包括：
- Anthropic推出Skills标准和MCP协议
- OpenAI发布Responses API支持更灵活的上下文管理
- 字节跳动开源DeerFlow 2.0，将Context Engineering作为Agent核心基础设施
- Manus团队发表Context Engineering六大核心策略
第三阶段：Harness Engineering（2026-）
最新的演化方向是Harness Engineering——不仅关注上下文本身，还关注Agent的整个运行环境：工具编排、安全护栏、错误恢复、可观测性。这一层解决的是"模型知道该做什么，但能不能安全稳定地做好"的问题。
1.2 为什么Context Engineering如此重要？
一个直观的类比：如果把LLM比作一个新入职的天才员工，那么：
- Prompt Engineering = 写一份清晰的JD（岗位描述）
- Context Engineering = 给他配备完整的工作台：参考资料、工具、历史案例、实时数据
- Harness Engineering = 给他搭建安全的工作流程：审批机制、异常处理、质量监控
Context Engineering之所以成为焦点，根本原因在于：
1. 上下文窗口再大也是稀缺资源
即使Claude支持200K tokens，GPT-4o支持128K tokens，真正可用的有效信息密度远没有那么高。研究表明，模型对长上下文中间位置信息的"注意力衰减"（Lost in the Middle）效应非常显著。塞入太多信息，反而会降低模型的推理质量。
2. 成本与延迟的硬约束
每次API调用的成本与token数成正比。一个100K token的上下文，光输入成本就可能超过$1。如果Agent需要多轮交互，成本会快速累积。Context Engineering的核心价值之一就是：用最少的token传递最有效的信息。
3. Agent可靠性的关键瓶颈
在实际生产环境中，Agent失败的最大原因往往不是模型能力不足，而是上下文出了问题：
- 给了过多无关信息，模型"抓不住重点"
- 缺少关键信息，模型"凭空猜测"
- 信息格式不对，模型"误解意图"
- 工具返回结果过大，模型"淹没在数据里"
二、Context Engineering的核心架构
2.1 上下文的四层模型
一个成熟的Context Engineering系统通常包含四层：
┌─────────────────────────────────────────────┐
│ Layer 4: 记忆层 (Memory) │
│ 长期记忆、用户画像、知识图谱、经验库 │
├─────────────────────────────────────────────┤
│ Layer 3: 工具层 (Tools) │
│ MCP工具集、API调用、数据库查询、文件系统 │
├─────────────────────────────────────────────┤
│ Layer 2: 检索层 (Retrieval) │
│ RAG、向量搜索、全文搜索、混合检索 │
├─────────────────────────────────────────────┤
│ Layer 1: 指令层 (Instruction) │
│ 系统提示词、角色设定、输出格式、约束规则 │
└─────────────────────────────────────────────┘
Layer 1：指令层 — 相对静态，定义Agent的行为边界。包括系统提示词、角色定义、输出格式要求、安全约束等。这层通常在会话开始时设定，期间不频繁变化。
Layer 2：检索层 — 动态响应用户查询。当用户提出问题时，系统从知识库中检索相关信息，注入上下文。这是传统RAG的核心层。
Layer 3：工具层 — 按需调用。当Agent需要执行操作（搜索网页、查询数据库、发送邮件）时，工具的描述和返回结果动态加入上下文。
Layer 4：记忆层 — 跨会话持久化。包括用户偏好、历史交互摘要、任务完成记录等。这层决定了Agent是否能"记住你是谁"。
2.2 上下文生命周期管理
Context Engineering不是一次性组装上下文，而是管理上下文的完整生命周期：
┌──────────┐
│ 构造 │ ← 组装初始上下文
│ Construct │
└────┬─────┘
│
┌────▼─────┐
│ 注入 │ ← 插入检索结果、工具描述
│ Inject │
└────┬─────┘
│
┌────▼─────┐
│ 压缩 │ ← 摘要、剪枝、优先级排序
│ Compress │
└────┬─────┘
│
┌────▼─────┐
│ 执行 │ ← 模型推理
│ Execute │
└────┬─────┘
│
┌────▼─────┐
│ 更新 │ ← 写回记忆、更新状态
│ Update │
└──────────┘
每个阶段都有关键的工程决策：
- 构造阶段：哪些信息必须包含？哪些可以省略？
- 注入阶段：检索结果如何排序？工具描述如何精简？
- 压缩阶段：历史对话何时摘要？如何保留关键信息？
- 执行阶段：输出是否需要后处理？是否触发安全护栏？
- 更新阶段：哪些信息值得记住？如何避免记忆污染？
三、RAG的进化：从朴素检索到智能上下文注入
3.1 传统RAG的局限
传统的RAG（Retrieval-Augmented Generation）流程非常直接：
# 朴素RAG实现
def naive_rag(query: str, vector_store, llm):
# 1. 向量检索
docs = vector_store.similarity_search(query, k=5)
# 2. 拼接上下文
context = "\n\n".join([doc.page_content for doc in docs])
# 3. 生成回答
prompt = f"根据以下资料回答问题：\n\n{context}\n\n问题：{query}"
return llm.generate(prompt)
这种模式存在几个严重问题：
问题1：检索质量不稳定
向量相似度≠语义相关度。用户问"Python如何处理并发"，检索出来的可能是"Python并发编程入门教程"（相关但太浅）和"Java并发包源码分析"（语义相似但不匹配）。
问题2：上下文噪声大
即使检索到了相关文档，文档中的大量内容可能与当前问题无关。一篇5000字的技术文章，可能只有200字是回答用户问题所需的。
问题3：缺乏上下文感知
传统RAG不考虑对话历史。用户先问"什么是Docker"，再问"如何部署"，RAG会重新检索"Docker部署"，但不知道用户可能已经在上一轮理解了Docker的基础概念。
3.2 高级RAG模式
现代Context Engineering引入了多种RAG增强技术：
混合检索（Hybrid Retrieval）
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma
def create_hybrid_retriever(docs, embedding_model, k=5):
"""结合向量检索和BM25关键词检索"""
# 向量检索器 - 擅长语义匹配
vector_store = Chroma.from_documents(docs, embedding_model)
vector_retriever = vector_store.as_retriever(search_kwargs={"k": k})
# BM25检索器 - 擅长精确关键词匹配
bm25_retriever = BM25Retriever.from_documents(docs, k=k)
# 混合检索，权重可调
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4] # 语义检索权重略高
)
return ensemble_retriever
混合检索的核心思想：向量检索捕获语义相似性，BM25捕获关键词精确匹配，两者互补可以显著提升检索质量。
查询改写（Query Rewriting）
def rewrite_query(original_query: str, chat_history: list, llm) -> list[str]:
"""根据对话历史改写查询，生成多个检索角度"""
history_text = "\n".join([f"{m['role']}: {m['content']}" for m in chat_history[-5:]])
prompt = f"""基于以下对话历史，将用户查询改写为3个不同角度的检索查询。
每个查询应该独立且完整，从不同维度搜索相关信息。
对话历史：
{history_text}
原始查询：{original_query}
请输出3个改写后的查询，每行一个："""
response = llm.generate(prompt)
queries = [q.strip() for q in response.split('\n') if q.strip()]
return [original_query] + queries[:3] # 原始查询 + 3个改写
查询改写解决了"用户说的话≠最佳检索词"的问题。比如用户说"那个部署的问题还没解决"，直接检索效果很差，但改写为"Docker容器部署失败排查方法"就能命中正确文档。
重排序（Reranking）
from sentence_transformers import CrossEncoder
class ContextReranker:
def __init__(self, model_name="BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: list[str], top_k: int = 3) -> list[str]:
"""使用交叉编码器对检索结果重排序"""
# 构造query-document对
pairs = [(query, doc) for doc in documents]
# 计算相关性分数
scores = self.model.predict(pairs)
# 按分数降序排列
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in ranked[:top_k]]
重排序是Context Engineering的关键一环：先用向量检索召回大量候选（比如top 20），再用更精确的模型重新排序，只保留最相关的top 3注入上下文。这样既保证了召回率，又控制了噪声。
3.3 分块压缩策略
当检索到的文档仍然过长时，需要智能压缩：
class ChunkCompressor:
"""上下文感知的文档压缩"""
def __init__(self, llm):
self.llm = llm
def extractive_compress(self, query: str, document: str, ratio: float = 0.3) -> str:
"""抽取式压缩：保留与查询最相关的句子"""
sentences = document.split('。')
# 计算每个句子与查询的相关性
scored_sentences = []
for i, sent in enumerate(sentences):
if len(sent.strip()) < 10:
continue
# 简单的BM25风格打分
score = self._compute_relevance(query, sent)
scored_sentences.append((i, score, sent))
# 按相关性排序，保留前ratio比例
scored_sentences.sort(key=lambda x: x[1], reverse=True)
keep_count = max(1, int(len(scored_sentences) * ratio))
kept = sorted(scored_sentences[:keep_count], key=lambda x: x[0])
return '。'.join([s for _, _, s in kept]) + '。'
def abstractive_compress(self, query: str, document: str) -> str:
"""生成式压缩：用LLM提炼关键信息"""
prompt = f"""请根据问题，从以下文档中提取最相关的关键信息。
只保留回答问题所需的最少信息，删除所有无关内容。
问题：{query}
文档：
{document}
关键信息摘要："""
return self.llm.generate(prompt)
3.4 Meta REFRAG：16倍上下文扩展
2026年Meta超级智能实验室提出的REFRAG框架，是RAG上下文工程的一个里程碑：
class REFRAGPipeline:
"""
REFRAG: REtrieval-Augmented Framework for Efficient Long Context Processing
核心思想：不是把检索到的完整文档注入上下文，
而是先将文档分块压缩为embedding表示，
只对最相关的块进行解压缩展开。
"""
def __init__(self, encoder, decoder, chunk_size=128):
self.encoder = encoder # 用于压缩文档块
self.decoder = decoder # 用于解压缩和生成
self.chunk_size = chunk_size
def process(self, query: str, retrieved_docs: list[str]) -> str:
# 1. 将文档分块
chunks = []
for doc in retrieved_docs:
chunks.extend(self._split_chunks(doc))
# 2. 将每个块压缩为embedding
chunk_embeddings = [self.encoder.encode(chunk) for chunk in chunks]
# 3. 计算query与每个chunk embedding的相关性分数
query_embedding = self.encoder.encode(query)
relevance_scores = [
self._cosine_similarity(query_embedding, emb)
for emb in chunk_embeddings
]
# 4. 只对top-k最相关的chunk进行解压缩（恢复原文）
top_k_indices = sorted(
range(len(relevance_scores)),
key=lambda i: relevance_scores[i],
reverse=True
)[:3] # 只展开最相关的3个块
# 5. 构造上下文：压缩的embedding + 展开的关键块
context_parts = []
for i, chunk in enumerate(chunks):
if i in top_k_indices:
context_parts.append(chunk) # 完整文本
# 其他块不注入，用embedding已有的信息
context = "\n\n".join(context_parts)
return self.decoder.generate(query=query, context=context)
REFRAG的核心突破在于：它证明了"智能压缩+选择性展开"可以在不损失质量的前提下，将有效上下文容量扩展16倍。这意味着原本只能处理20K token上下文的模型，可以处理320K token的信息量。
四、记忆架构：让Agent真正"记住你"
4.1 记忆的分层模型
人类记忆有感觉记忆、短期记忆、长期记忆之分。AI Agent的记忆系统也需要类似的分层：
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
@dataclass
class MemoryItem:
content: str
memory_type: str # "episodic" | "semantic" | "procedural"
importance: float # 0.0 - 1.0
created_at: datetime = field(default_factory=datetime.now)
last_accessed: datetime = field(default_factory=datetime.now)
access_count: int = 0
metadata: dict = field(default_factory=dict)
class AgentMemorySystem:
"""
三层记忆架构：
- 工作记忆（Working Memory）：当前会话的上下文窗口
- 短期记忆（Short-term Memory）：最近几次会话的摘要
- 长期记忆（Long-term Memory）：持久化的知识和经验
"""
def __init__(self, vector_store, summary_llm):
self.working_memory: list[dict] = [] # 当前会话消息
self.short_term: list[MemoryItem] = [] # 最近会话摘要
self.long_term_store = vector_store # 长期记忆向量库
self.summary_llm = summary_llm
self.max_working_tokens = 8000 # 工作记忆token上限
def add_message(self, role: str, content: str):
"""添加消息到工作记忆"""
self.working_memory.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
# 检查是否需要压缩工作记忆
if self._estimate_tokens() > self.max_working_tokens:
self._compress_working_memory()
def _compress_working_memory(self):
"""将旧消息压缩为摘要，保留最近的消息"""
# 保留最近4条消息
recent = self.working_memory[-4:]
to_compress = self.working_memory[:-4]
# 生成摘要
summary = self._generate_summary(to_compress)
# 存入短期记忆
self.short_term.append(MemoryItem(
content=summary,
memory_type="episodic",
importance=0.6,
metadata={"message_count": len(to_compress)}
))
# 工作记忆只保留最近消息
self.working_memory = recent
def _generate_summary(self, messages: list[dict]) -> str:
"""用LLM生成对话摘要"""
conversation = "\n".join([
f"{m['role']}: {m['content']}" for m in messages
])
prompt = f"""请将以下对话压缩为简洁的摘要，保留关键信息和决策：
{conversation}
摘要："""
return self.summary_llm.generate(prompt)
def recall(self, query: str, top_k: int = 5) -> list[MemoryItem]:
"""根据查询召回相关记忆"""
results = []
# 1. 从工作记忆中搜索
for msg in self.working_memory:
if self._is_relevant(query, msg["content"]):
results.append(MemoryItem(
content=msg["content"],
memory_type="working",
importance=0.9, # 工作记忆最重要
metadata={"source": "working_memory"}
))
# 2. 从短期记忆中搜索
for item in self.short_term:
if self._is_relevant(query, item.content):
results.append(item)
# 3. 从长期记忆中向量检索
long_term_results = self.long_term_store.similarity_search(query, k=top_k)
for doc in long_term_results:
results.append(MemoryItem(
content=doc.page_content,
memory_type="semantic",
importance=0.7,
metadata=doc.metadata
))
# 按重要性排序
results.sort(key=lambda x: x.importance, reverse=True)
return results[:top_k]
def consolidate(self):
"""记忆整合：将短期记忆提炼为长期记忆"""
if len(self.short_term) < 10:
return # 短期记忆不够多，不需要整合
# 按时间分组
recent_batch = self.short_term[-10:]
# 提炼为长期知识
summaries = [item.content for item in recent_batch]
consolidated = self._distill_knowledge(summaries)
# 存入长期记忆
self.long_term_store.add_texts(
texts=[consolidated],
metadatas=[{
"type": "consolidated_memory",
"source_items": len(recent_batch),
"created_at": datetime.now().isoformat()
}]
)
# 清理已整合的短期记忆
self.short_term = self.short_term[:-10]
def _is_relevant(self, query: str, content: str) -> bool:
"""简单相关性判断"""
query_terms = set(query.lower().split())
content_terms = set(content.lower().split())
overlap = query_terms & content_terms
return len(overlap) >= 2
def _estimate_tokens(self) -> int:
"""粗略估算token数"""
total_chars = sum(len(m["content"]) for m in self.working_memory)
return total_chars // 2 # 中文大约每2个字符1个token
def _distill_knowledge(self, summaries: list[str]) -> str:
"""从多条摘要中提炼长期知识"""
combined = "\n".join(summaries)
prompt = f"""从以下对话摘要中提取值得长期记住的知识、用户偏好和重要决策：
{combined}
提炼后的长期知识："""
return self.summary_llm.generate(prompt)
4.2 记忆衰减与遗忘机制
不是所有信息都值得永远记住。好的记忆系统需要遗忘机制：
import math
from datetime import datetime, timedelta
class MemoryDecay:
"""基于Ebbinghaus遗忘曲线的记忆衰减"""
@staticmethod
def compute_retention(memory: MemoryItem, current_time: datetime = None) -> float:
"""计算记忆的保留率（0.0 - 1.0）"""
if current_time is None:
current_time = datetime.now()
# 时间衰减：基于遗忘曲线
hours_elapsed = (current_time - memory.last_accessed).total_seconds() / 3600
# 艾宾浩斯遗忘曲线：R = e^(-t/S)
# S 为稳定性，由访问次数和重要性决定
stability = 1 + memory.access_count * 2 + memory.importance * 10
time_decay = math.exp(-hours_elapsed / stability)
# 访问次数加成
access_bonus = min(0.3, memory.access_count * 0.05)
# 重要性加成
importance_bonus = memory.importance * 0.2
retention = min(1.0, time_decay + access_bonus + importance_bonus)
return max(0.0, retention)
@staticmethod
def should_forget(memory: MemoryItem, threshold: float = 0.1) -> bool:
"""判断是否应该遗忘"""
return MemoryDecay.compute_retention(memory) < threshold
@staticmethod
def should_consolidate(memory: MemoryItem, threshold: float = 0.5) -> bool:
"""判断是否应该整合到长期记忆"""
retention = MemoryDecay.compute_retention(memory)
return retention > threshold and memory.importance > 0.6
4.3 知识图谱增强记忆
对于复杂Agent，纯文本记忆可能不够。知识图谱可以提供结构化的关联记忆：
class KnowledgeGraphMemory:
"""基于知识图谱的结构化记忆"""
def __init__(self):
self.entities = {} # entity_id -> Entity
self.relations = [] # (subject_id, predicate, object_id)
def add_entity(self, entity_id: str, entity_type: str,
properties: dict, embedding: list[float] = None):
"""添加实体"""
self.entities[entity_id] = {
"type": entity_type,
"properties": properties,
"embedding": embedding,
"created_at": datetime.now().isoformat()
}
def add_relation(self, subject: str, predicate: str, obj: str,
confidence: float = 1.0):
"""添加关系"""
self.relations.append({
"subject": subject,
"predicate": predicate,
"object": obj,
"confidence": confidence,
"created_at": datetime.now().isoformat()
})
def query_subgraph(self, entity_id: str, depth: int = 2) -> dict:
"""查询实体的关联子图"""
visited = set()
subgraph = {"entities": {}, "relations": []}
def dfs(current_id: str, current_depth: int):
if current_depth > depth or current_id in visited:
return
visited.add(current_id)
if current_id in self.entities:
subgraph["entities"][current_id] = self.entities[current_id]
for rel in self.relations:
if rel["subject"] == current_id:
subgraph["relations"].append(rel)
dfs(rel["object"], current_depth + 1)
elif rel["object"] == current_id:
subgraph["relations"].append(rel)
dfs(rel["subject"], current_depth + 1)
dfs(entity_id, 0)
return subgraph
def to_context_string(self, subgraph: dict) -> str:
"""将子图转换为可注入上下文的文本表示"""
lines = []
for eid, entity in subgraph["entities"].items():
props = ", ".join([f"{k}: {v}" for k, v in entity["properties"].items()])
lines.append(f"[{entity['type']}] {eid}: {props}")
for rel in subgraph["relations"]:
lines.append(f"{rel['subject']} --[{rel['predicate']}]--> {rel['object']}")
return "\n".join(lines)
五、工具上下文：MCP协议与Skills标准
5.1 MCP：工具连接的统一协议
Model Context Protocol（MCP）是Anthropic推出的开放标准，定义了LLM与外部工具/数据源的通信协议。它在Context Engineering中的角色是：标准化工具描述和结果的注入格式。
# MCP Server实现示例
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio
server = Server("my-knowledge-base")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""声明可用工具 - 这些描述会注入到模型上下文中"""
return [
Tool(
name="search_docs",
description="搜索内部知识库文档。当用户询问公司内部信息、技术文档、"
"产品规范时使用此工具。返回最相关的文档片段。",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询，可以是自然语言问题"
},
"category": {
"type": "string",
"enum": ["tech", "product", "hr", "all"],
"description": "文档类别过滤",
"default": "all"
},
"max_results": {
"type": "integer",
"description": "最大返回结果数",
"default": 5
}
},
"required": ["query"]
}
),
Tool(
name="get_doc_detail",
description="获取文档的完整内容。当search_docs返回的摘要不够时，"
"用此工具获取完整文档。",
inputSchema={
"type": "object",
"properties": {
"doc_id": {
"type": "string",
"description": "文档ID，从search_docs结果中获取"
}
},
"required": ["doc_id"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""执行工具调用"""
if name == "search_docs":
results = await search_knowledge_base(
query=arguments["query"],
category=arguments.get("category", "all"),
max_results=arguments.get("max_results", 5)
)
return [TextContent(
type="text",
text=format_search_results(results)
)]
elif name == "get_doc_detail":
doc = await get_document(arguments["doc_id"])
return [TextContent(
type="text",
text=doc.content
)]
MCP的关键Context Engineering意义：
- 标准化工具描述：工具的
description字段直接作为上下文注入模型，好的描述能让模型正确选择工具 - 结构化返回：工具结果以结构化方式返回，可以被智能压缩后再注入上下文
- 能力发现：模型可以动态发现可用工具，而不是硬编码在提示词中
5.2 Skills：可复用的能力模块
Skills标准更进一步——它将工具、知识、行为模式打包为一个完整的"能力单元"：
# Skill定义示例：代码审查专家
## 角色
你是一个资深代码审查专家，专注于发现代码中的问题和改进空间。
## 工具集
- `read_file`: 读取代码文件
- `search_code`: 搜索代码库
- `run_linter`: 运行静态分析
- `git_diff`: 获取代码变更
## 知识库
- 代码规范文档（自动注入）
- 常见反模式列表（自动注入）
- 团队历史Review记录（按需检索）
## 行为模式
1. 先读取变更文件列表
2. 逐文件审查，记录问题
3. 按严重程度分类：Critical / Warning / Suggestion
4. 生成结构化Review报告
## 输出格式
```markdown
## Code Review Report
### Critical Issues
- [file:line] 问题描述
### Warnings
- [file:line] 问题描述
### Suggestions
- [file:line] 问题描述
Skills标准的Context Engineering价值在于：它将上下文的组装逻辑模块化。不同的任务可以加载不同的Skill，每个Skill自带它需要的工具描述、知识库引用和行为模板。
## 六、Manus六大核心策略：Context Engineering的实战方法论
Manus团队在其技术分享中总结了六大Context Engineering策略，这些来自真实生产环境的经验极具参考价值：
### 策略一：围绕KV-Cache设计架构
```python
class KVCacheAwareContextManager:
"""
策略1：利用KV-Cache优化上下文结构
LLM推理时，KV-Cache可以缓存已计算的Key-Value对，
避免重复计算。如果上下文的前缀部分保持不变，
就可以复用KV-Cache，大幅降低延迟和成本。
核心原则：静态内容放前面，动态内容放后面。
"""
def __init__(self):
self.static_prefix = "" # 系统提示词、工具描述等静态内容
self.dynamic_suffix = "" # 用户消息、工具结果等动态内容
def build_context(self, system_prompt: str, tools_description: str,
chat_history: list, current_query: str) -> str:
"""构建KV-Cache友好的上下文"""
# 静态前缀 - 保持不变，可以复用KV-Cache
static = f"{system_prompt}\n\n{tools_description}"
# 检查静态前缀是否变化
if static != self.static_prefix:
self.static_prefix = static
# 前缀变化了，Cache失效，需要重新计算
# 这是昂贵的操作，应尽量避免
# 动态后缀 - 每次都不同
history = self._format_history(chat_history)
dynamic = f"{history}\n\nUser: {current_query}"
return f"{static}\n\n{dynamic}"
策略二：Logit Masking解决工具爆炸
class ToolSelector:
"""
策略2：当工具太多时，不要把所有工具描述都塞进上下文。
使用Logit Masking在生成阶段限制模型只能选择相关工具。
"""
def __init__(self, all_tools: list[dict]):
self.all_tools = all_tools
self.tool_index = self._build_index()
def select_relevant_tools(self, query: str, max_tools: int = 10) -> list[dict]:
"""根据查询选择最相关的工具子集"""
# 先用轻量级方法筛选
scores = []
for tool in self.all_tools:
score = self._compute_relevance(query, tool["description"])
scores.append((score, tool))
# 按相关性排序
scores.sort(reverse=True)
return [tool for _, tool in scores[:max_tools]]
def build_tool_context(self, query: str) -> tuple[str, list[str]]:
"""构建工具上下文，返回(工具描述文本, 允许的工具名列表)"""
relevant = self.select_relevant_tools(query)
descriptions = []
allowed_names = []
for tool in relevant:
descriptions.append(f"- {tool['name']}: {tool['description']}")
allowed_names.append(tool['name'])
return "\n".join(descriptions), allowed_names
当Agent系统有上百个工具时，把所有工具描述都放进上下文会消耗大量token，且会干扰模型选择。Logit Masking的核心思想是：先用检索/分类方法筛选出最相关的工具子集，只将这些工具的描述注入上下文。
策略三：文件系统作为外置显存
class ExternalizedMemory:
"""
策略3：用文件系统扩展模型的"显存"
模型的上下文窗口有限，但文件系统几乎无限。
让模型学会读写文件，就能突破上下文限制。
"""
def __init__(self, workspace_dir: str):
self.workspace = workspace_dir
self._ensure_dirs()
def _ensure_dirs(self):
"""创建工作区目录结构"""
import os
dirs = ["scratch", "memory", "artifacts", "cache"]
for d in dirs:
os.makedirs(f"{self.workspace}/{d}", exist_ok=True)
def save_to_scratch(self, key: str, content: str):
"""保存临时数据到草稿区"""
with open(f"{self.workspace}/scratch/{key}.txt", "w") as f:
f.write(content)
def load_from_scratch(self, key: str) -> str:
"""从草稿区加载数据"""
try:
with open(f"{self/workspace}/scratch/{key}.txt", "r") as f:
return f.read()
except FileNotFoundError:
return ""
def save_memory(self, category: str, content: str, metadata: dict = None):
"""保存到长期记忆区"""
import json
from datetime import datetime
entry = {
"content": content,
"metadata": metadata or {},
"created_at": datetime.now().isoformat()
}
filepath = f"{self.workspace}/memory/{category}.jsonl"
with open(filepath, "a") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
def get_context_summary(self) -> str:
"""生成工作区状态摘要，注入上下文"""
import os
summary_parts = []
# 列出草稿区文件
scratch_files = os.listdir(f"{self.workspace}/scratch")
if scratch_files:
summary_parts.append(f"Scratch pad: {', '.join(scratch_files)}")
# 列出记忆区分类
memory_files = os.listdir(f"{self.workspace}/memory")
if memory_files:
summary_parts.append(f"Memory categories: {', '.join(memory_files)}")
return "\n".join(summary_parts)
这个策略的精髓在于：模型不需要把所有信息都放在上下文里，而是可以先写入文件，需要时再读取。这让Agent能够处理远超上下文窗口的信息量。
策略四：背诵对抗注意力衰减
class RecitationStrategy:
"""
策略4：让模型定期"背诵"关键信息，对抗Lost-in-the-Middle效应
研究表明，LLM对上下文中间位置的信息关注度最低。
通过定期将关键信息重新放在上下文末尾，
可以显著提高模型对这些信息的利用率。
"""
def __init__(self):
self.key_facts: list[str] = []
self.recitation_interval = 5 # 每5轮对话背诵一次
self.turn_count = 0
def add_key_fact(self, fact: str):
"""记录关键事实"""
if fact not in self.key_facts:
self.key_facts.append(fact)
# 只保留最近的N个关键事实
if len(self.key_facts) > 10:
self.key_facts = self.key_facts[-10:]
def maybe_recite(self, current_context: str) -> str:
"""如果到了背诵时机，在上下文末尾重新强调关键信息"""
self.turn_count += 1
if self.turn_count % self.recitation_interval != 0:
return current_context
if not self.key_facts:
return current_context
recitation = "\n\n[Key Facts Reminder]\n"
for i, fact in enumerate(self.key_facts, 1):
recitation += f"{i}. {fact}\n"
return current_context + recitation
策略五：保留错误轨迹作为学习样本
class ErrorLearningContext:
"""
策略5：保留Agent的错误和修正过程，作为上下文中的"反面教材"
模型从错误中学习的能力往往被低估。
在上下文中保留之前的错误尝试和修正方法，
可以帮助模型避免重复犯同样的错误。
"""
def __init__(self, max_errors: int = 5):
self.error_history: list[dict] = []
self.max_errors = max_errors
def record_error(self, task: str, error: str, correction: str):
"""记录错误和修正"""
self.error_history.append({
"task": task,
"error": error,
"correction": correction,
"timestamp": datetime.now().isoformat()
})
if len(self.error_history) > self.max_errors:
self.error_history = self.error_history[-self.max_errors:]
def get_error_context(self) -> str:
"""生成错误学习上下文"""
if not self.error_history:
return ""
lines = ["## Previous Errors and Corrections\n"]
for i, err in enumerate(self.error_history, 1):
lines.append(f"### Error {i}")
lines.append(f"Task: {err['task']}")
lines.append(f"What went wrong: {err['error']}")
lines.append(f"Correct approach: {err['correction']}")
lines.append("")
return "\n".join(lines)
策略六：引入结构化噪声避免少样本陷阱
class DiverseExampleProvider:
"""
策略6：在Few-shot示例中引入多样性，避免模型陷入模式模仿
当上下文中的示例都遵循相同模式时，
模型容易机械模仿而忽略真正的推理。
适度引入"噪声"或变体，反而能提高泛化能力。
"""
def __init__(self, examples: list[dict]):
self.examples = examples
def select_diverse_examples(self, query: str, n: int = 3) -> list[dict]:
"""选择多样化的示例"""
# 按与查询的相关性排序
scored = [(self._relevance(query, ex), ex) for ex in self.examples]
scored.sort(reverse=True)
# 从top候选中选择多样化的子集
selected = []
candidates = [ex for _, ex in scored[:n * 3]] # 候选池
for _ in range(n):
if not candidates:
break
if not selected:
# 第一个选最相关的
selected.append(candidates.pop(0))
else:
# 后续选与已选示例差异最大的
best_idx = 0
best_diversity = -1
for i, cand in enumerate(candidates):
diversity = min([
self._diversity(cand, sel) for sel in selected
])
if diversity > best_diversity:
best_diversity = diversity
best_idx = i
selected.append(candidates.pop(best_idx))
return selected
def _relevance(self, query: str, example: dict) -> float:
"""计算示例与查询的相关性"""
# 简化实现
query_words = set(query.lower().split())
example_words = set(example.get("input", "").lower().split())
overlap = query_words & example_words
return len(overlap) / max(len(query_words), 1)
def _diversity(self, ex1: dict, ex2: dict) -> float:
"""计算两个示例之间的差异度"""
words1 = set(ex1.get("input", "").lower().split())
words2 = set(ex2.get("input", "").lower().split())
intersection = words1 & words2
union = words1 | words2
if not union:
return 1.0
return 1.0 - len(intersection) / len(union) # Jaccard距离
七、完整实战：构建一个生产级Context Engineering系统
让我们把前面的所有概念整合到一个完整的Agent上下文管理系统中：
"""
生产级Context Engineering系统
整合RAG、记忆、工具上下文、压缩策略
"""
from dataclasses import dataclass, field
from typing import Optional, Any
from datetime import datetime
import json
import hashlib
@dataclass
class ContextConfig:
"""上下文配置"""
max_context_tokens: int = 16000
max_retrieval_docs: int = 5
max_tools_in_context: int = 8
compression_ratio: float = 0.4
recitation_interval: int = 5
enable_kv_cache: bool = True
enable_error_learning: bool = True
class ProductionContextEngine:
"""
生产级上下文引擎
核心流程：
1. 构建静态前缀（系统提示 + 工具描述）
2. 注入检索结果（RAG）
3. 注入记忆（工作/短期/长期）
4. 压缩与优化
5. 拼装最终上下文
"""
def __init__(self, config: ContextConfig,
retriever, memory_system, tool_registry, llm):
self.config = config
self.retriever = retriever
self.memory = memory_system
self.tools = tool_registry
self.llm = llm
# KV-Cache优化：追踪静态前缀
self._cached_prefix_hash: Optional[str] = None
self._cached_prefix: str = ""
# 错误学习
self.error_history: list[dict] = []
# 背诵计数器
self.turn_count = 0
self.key_facts: list[str] = []
def build_context(self, user_query: str,
chat_history: list[dict],
system_prompt: str) -> str:
"""构建完整上下文"""
self.turn_count += 1
context_parts = []
# === Layer 1: 静态前缀（KV-Cache友好） ===
static_prefix = self._build_static_prefix(system_prompt)
context_parts.append(static_prefix)
# === Layer 2: 记忆注入 ===
memory_context = self._inject_memory(user_query)
if memory_context:
context_parts.append(f"## Relevant Memory\n{memory_context}")
# === Layer 3: RAG检索结果 ===
retrieval_context = self._inject_retrieval(user_query)
if retrieval_context:
context_parts.append(f"## Retrieved Knowledge\n{retrieval_context}")
# === Layer 4: 对话历史 ===
history_context = self._format_history(chat_history)
context_parts.append(f"## Conversation History\n{history_context}")
# === Layer 5: 错误学习 ===
if self.config.enable_error_learning and self.error_history:
error_context = self._format_error_history()
context_parts.append(f"## Previous Errors\n{error_context}")
# === Layer 6: 背诵提醒 ===
recitation = self._maybe_recite()
if recitation:
context_parts.append(recitation)
# === Layer 7: 当前查询 ===
context_parts.append(f"## Current Query\nUser: {user_query}")
# === 压缩与优化 ===
full_context = "\n\n".join(context_parts)
optimized = self._optimize_context(full_context)
return optimized
def _build_static_prefix(self, system_prompt: str) -> str:
"""构建静态前缀（可复用KV-Cache）"""
# 选择相关工具
relevant_tools = self.tools.select_relevant("", max_count=self.config.max_tools_in_context)
tools_desc = "\n".join([
f"- {t['name']}: {t['description']}" for t in relevant_tools
])
prefix = f"""# System Configuration
{system_prompt}
## Available Tools
{tools_desc}
## Output Requirements
- Respond in the same language as the user
- Use tools when needed, explain your reasoning
- Be concise but thorough"""
# 检查是否需要更新缓存
prefix_hash = hashlib.md5(prefix.encode()).hexdigest()
if prefix_hash != self._cached_prefix_hash:
self._cached_prefix_hash = prefix_hash
self._cached_prefix = prefix
# KV-Cache失效标记（实际实现中通知推理引擎）
return self._cached_prefix
def _inject_memory(self, query: str) -> str:
"""注入相关记忆"""
memories = self.memory.recall(query, top_k=3)
if not memories:
return ""
lines = []
for mem in memories:
lines.append(f"- [{mem.memory_type}] {mem.content}")
return "\n".join(lines)
def _inject_retrieval(self, query: str) -> str:
"""注入RAG检索结果"""
# 查询改写
queries = self._rewrite_query(query)
# 混合检索
all_docs = []
for q in queries:
docs = self.retriever.retrieve(q, top_k=self.config.max_retrieval_docs)
all_docs.extend(docs)
# 去重
unique_docs = self._deduplicate_docs(all_docs)
# 重排序
ranked = self._rerank(query, unique_docs)
# 压缩
compressed = self._compress_docs(query, ranked[:3])
return compressed
def _rewrite_query(self, query: str) -> list[str]:
"""查询改写"""
# 简化实现，实际应调用LLM
return [query]
def _deduplicate_docs(self, docs: list) -> list:
"""文档去重"""
seen = set()
unique = []
for doc in docs:
doc_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
if doc_hash not in seen:
seen.add(doc_hash)
unique.append(doc)
return unique
def _rerank(self, query: str, docs: list) -> list:
"""重排序（简化实现）"""
return docs # 实际应使用CrossEncoder
def _compress_docs(self, query: str, docs: list) -> str:
"""压缩文档"""
parts = []
for doc in docs:
content = doc.page_content
# 如果文档太长，截取关键部分
if len(content) > 1000:
content = content[:1000] + "..."
parts.append(content)
return "\n---\n".join(parts)
def _format_history(self, history: list[dict]) -> str:
"""格式化对话历史"""
lines = []
for msg in history[-6:]: # 只保留最近6条
lines.append(f"{msg['role']}: {msg['content']}")
return "\n".join(lines)
def _format_error_history(self) -> str:
"""格式化错误历史"""
lines = []
for err in self.error_history[-3:]: # 只保留最近3个错误
lines.append(f"- Task: {err['task']}")
lines.append(f" Error: {err['error']}")
lines.append(f" Fix: {err['correction']}")
return "\n".join(lines)
def _maybe_recite(self) -> str:
"""背诵关键信息"""
if self.turn_count % self.config.recitation_interval != 0:
return ""
if not self.key_facts:
return ""
lines = ["## Key Facts Reminder"]
for i, fact in enumerate(self.key_facts, 1):
lines.append(f"{i}. {fact}")
return "\n".join(lines)
def _optimize_context(self, context: str) -> str:
"""优化上下文，确保不超过token限制"""
# 估算token数
estimated_tokens = len(context) // 2 # 粗略估算
if estimated_tokens <= self.config.max_context_tokens:
return context
# 需要压缩：从后往前保留最重要的内容
# 策略：保留系统提示、最近的对话、当前查询，压缩中间内容
lines = context.split("\n")
# 找到关键段落的边界
preserved_sections = ["# System Configuration", "## Current Query", "## Conversation History"]
preserved = []
compressible = []
current_section = ""
current_lines = []
for line in lines:
if line.startswith("#"):
if current_section:
if any(ps in current_section for ps in preserved_sections):
preserved.extend(current_lines)
else:
compressible.append((current_section, current_lines[:]))
current_section = line
current_lines = [line]
else:
current_lines.append(line)
# 处理最后一段
if current_section:
if any(ps in current_section for ps in preserved_sections):
preserved.extend(current_lines)
else:
compressible.append((current_section, current_lines[:]))
# 压缩可压缩的部分
compressed = []
for section_title, section_lines in compressible:
compressed_text = "\n".join(section_lines)
if len(compressed_text) > 500:
compressed_text = compressed_text[:500] + "... [compressed]"
compressed.append(compressed_text)
return "\n\n".join(preserved + compressed)
def record_error(self, task: str, error: str, correction: str):
"""记录错误"""
self.error_history.append({
"task": task,
"error": error,
"correction": correction,
"timestamp": datetime.now().isoformat()
})
if len(self.error_history) > 10:
self.error_history = self.error_history[-10:]
def add_key_fact(self, fact: str):
"""添加关键事实"""
if fact not in self.key_facts:
self.key_facts.append(fact)
if len(self.key_facts) > 10:
self.key_facts = self.key_facts[-10:]
八、性能优化与成本控制
8.1 Token成本分析
Context Engineering直接影响API调用成本。以下是一个典型Agent会话的成本分析：
class CostAnalyzer:
"""上下文成本分析器"""
# 各主流模型的token价格（每百万token，美元）
PRICING = {
"claude-3.5-sonnet": {"input": 3.0, "output": 15.0},
"gpt-4o": {"input": 2.5, "output": 10.0},
"deepseek-v3": {"input": 0.27, "output": 1.1},
"qwen-max": {"input": 2.4, "output": 9.6},
}
@classmethod
def estimate_cost(cls, model: str, input_tokens: int, output_tokens: int) -> float:
"""估算单次调用成本（美元）"""
pricing = cls.PRICING.get(model, cls.PRICING["claude-3.5-sonnet"])
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
@classmethod
def compare_strategies(cls, model: str, base_context_tokens: int):
"""对比不同Context Engineering策略的成本"""
strategies = {
"No optimization": base_context_tokens,
"RAG only (50% reduction)": base_context_tokens * 0.5,
"RAG + Compression (70% reduction)": base_context_tokens * 0.3,
"Full CE pipeline (85% reduction)": base_context_tokens * 0.15,
}
output_tokens = 1000 # 假设输出1000 tokens
print(f"\nCost comparison for {model}")
print(f"Base context: {base_context_tokens} tokens")
print(f"Output: {output_tokens} tokens")
print("-" * 60)
for name, tokens in strategies.items():
cost = cls.estimate_cost(model, tokens, output_tokens)
print(f"{name:45} → ${cost:.4f}/call")
8.2 缓存策略
class ContextCache:
"""上下文缓存管理"""
def __init__(self, max_size: int = 100):
self.cache: dict[str, tuple[str, float]] = {}
self.max_size = max_size
def get(self, key: str) -> Optional[str]:
"""获取缓存的上下文"""
if key in self.cache:
content, timestamp = self.cache[key]
# 缓存有效期30分钟
if (datetime.now().timestamp() - timestamp) < 1800:
return content
else:
del self.cache[key]
return None
def set(self, key: str, content: str):
"""设置缓存"""
if len(self.cache) >= self.max_size:
# LRU淘汰：删除最旧的
oldest_key = min(self.cache, key=lambda k: self.cache[k][1])
del self.cache[oldest_key]
self.cache[key] = (content, datetime.now().timestamp())
@staticmethod
def make_key(prefix_hash: str, query_hash: str,
history_hash: str) -> str:
"""生成缓存key"""
combined = f"{prefix_hash}:{query_hash}:{history_hash}"
return hashlib.md5(combined.encode()).hexdigest()
8.3 异步并行优化
import asyncio
class AsyncContextBuilder:
"""异步上下文构建器，并行执行独立的上下文组装任务"""
def __init__(self, retriever, memory, tools):
self.retriever = retriever
self.memory = memory
self.tools = tools
async def build_context_parallel(self, query: str,
chat_history: list,
system_prompt: str) -> str:
"""并行构建上下文"""
# 这三个任务互相独立，可以并行执行
retrieval_task = asyncio.create_task(
self._async_retrieve(query)
)
memory_task = asyncio.create_task(
self._async_recall_memory(query)
)
tools_task = asyncio.create_task(
self._async_select_tools(query)
)
# 等待所有任务完成
retrieval_result, memory_result, tools_result = await asyncio.gather(
retrieval_task, memory_task, tools_task
)
# 组装上下文
context = self._assemble_context(
system_prompt=system_prompt,
tools_desc=tools_result,
memory_context=memory_result,
retrieval_context=retrieval_result,
chat_history=chat_history,
current_query=query
)
return context
async def _async_retrieve(self, query: str) -> str:
"""异步检索"""
# 模拟异步操作
await asyncio.sleep(0.1)
docs = self.retriever.retrieve(query)
return "\n\n".join([doc.page_content for doc in docs])
async def _async_recall_memory(self, query: str) -> str:
"""异步记忆召回"""
await asyncio.sleep(0.05)
memories = self.memory.recall(query)
return "\n".join([m.content for m in memories])
async def _async_select_tools(self, query: str) -> str:
"""异步工具选择"""
await asyncio.sleep(0.05)
tools = self.tools.select_relevant(query)
return "\n".join([f"- {t['name']}: {t['description']}" for t in tools])
九、Context Engineering的未来展望
9.1 从Context Engineering到Agentic Memory
2026年下半年，我们看到的趋势是：记忆系统正在从"被动存储"走向"主动思考"。未来的Agent记忆可能具备：
- 自主记忆管理：Agent自己决定什么值得记住，什么应该遗忘
- 记忆推理：基于记忆进行推理，而不仅仅是检索
- 记忆共享：多个Agent之间共享和同步记忆
- 记忆审计：可追溯的记忆来源和演变历史
9.2 上下文协议标准化
MCP、Skills、A2A（Agent-to-Agent）等协议的出现，标志着上下文管理正在走向标准化。未来可能出现：
- 统一的上下文交换格式：不同Agent框架之间可以共享上下文
- 上下文市场：可购买和销售的专业领域上下文包
- 上下文质量评估标准：量化评估上下文质量的行业标准
9.3 硬件级上下文优化
随着AI专用芯片的发展，未来可能出现：
- 硬件级KV-Cache管理：芯片直接优化上下文缓存
- 上下文预取：预测模型下一步需要的信息，提前加载
- 压缩协处理器：专用硬件加速上下文压缩
十、总结
Context Engineering不是RAG的简单升级，而是AI应用工程的范式转变。它要求工程师从"如何让模型理解我"转变为"如何在正确的时间给模型正确的信息"。
核心要点回顾：
- 四层架构：指令层、检索层、工具层、记忆层，各司其职
- RAG进化：从朴素检索到混合检索+重排序+智能压缩
- 记忆分层：工作记忆、短期记忆、长期记忆，配合遗忘机制
- 六大策略：KV-Cache优化、Logit Masking、外置显存、背诵、错误学习、结构化噪声
- 成本控制：压缩、缓存、并行，让Context Engineering在生产环境中可行
作为开发者，现在正是深入Context Engineering的最佳时机。这个领域还处于早期，谁先建立起完善的上下文管理体系，谁就能在AI Agent的竞赛中占据先机。
"The best AI engineer is not the one who writes the best prompts, but the one who builds the best context." — 2026年AI工程社区共识
参考资料：
- Anthropic: Model Context Protocol Specification
- Manus团队: Context Engineering六大核心策略
- Meta AI: REFRAG - Retrieval-Augmented Framework for Efficient Long Context
- LangChain: Advanced RAG Patterns
- OpenAI: Memory and Context Management Best Practices