Agent_06_ContextEngineering
创建于 更新于
前面五个专题在讲 Agent 架构——怎么组织 LLM、记忆、工具、规划。但所有这些架构都有一个共同的瓶颈：Context Window。
Context Window 是 LLM 能同时看到的所有信息：system prompt + 对话历史 + 工具描述 + 用户输入。这个窗口大小有限（GPT-4 是 128K，Claude 是 200K），但更关键的是，LLM 的注意力不是均匀分布的。
几个已知事实：
- 首尾效应：LLM 对上下文开头和结尾的信息更敏感，中间部分容易丢失
- 信息过载：上下文越长，指令遵循能力越差。32K 上下文的准确率比 4K 低 15-20%
- 信噪比：无关信息越多，LLM 对关键指令的注意力越被稀释
先看一个反面例子：
python
# 反例：一个巨大的 system prompt 塞所有东西
system_prompt = """
你是一个数据分析助手。
你的职责：
1. 读取用户指定的数据文件
2. 清洗数据（去重、处理缺失值、异常值检测）
3. 进行描述性统计分析
4. 根据数据特征选择合适的可视化图表
5. 生成分析报告
可用工具：
- read_csv: 读取 CSV 文件
- read_excel: 读取 Excel 文件
- clean_data: 清洗数据
- analyze: 统计分析
- plot_chart: 生成图表
- save_report: 保存报告
- send_email: 发送邮件
- search_knowledge_base: 搜索知识库
- get_user_preferences: 获取用户偏好
- ... 还有 20 个工具
用户偏好：
- 喜欢柱状图
- 报告用中文
- 数据保留两位小数
- 图表用 seaborn 风格
- 标题用加粗
- 表格用 markdown 格式
- ...
历史对话：
[2025-01-01] 用户说："我主要做销售数据分析"
[2025-01-05] 用户说："我们的财年是 4 月到次年 3 月"
[2025-01-10] 用户说："异常值的定义是超过 3 倍标准差"
... 还有 50 条历史
请根据以上所有信息，执行用户的请求。
"""
# 问题：
# 1. 太长——占用了 5000+ token
# 2. 信息层级混乱——LLM 不知道该优先关注什么
# 3. 大部分信息在大多数对话中用不到
# 4. 历史对话是噪声，不是信号解法是把上下文拆成分层结构，每层有不同的注入策略：
text
Context 分层架构
Layer 1: 核心指令（始终注入）—— 200 tok
- Agent 角色定义
- 核心行为规则
- 输出格式要求
Layer 2: 技能描述（按需注入）—— 500 tok
- 当前任务相关的工具描述
- 工具的参数 schema
- 工具的使用示例
Layer 3: 用户上下文（检索注入）—— 1000 tok
- 用户偏好（长期记忆）
- 相关历史对话
- 领域知识（按需检索）
Layer 4: 任务数据（动态注入）—— 可变
- 用户当前请求
- 工具执行结果
- 中间变量
总计：~1700 tok（固定）+ 可变数据
vs 反例：5000+ tok（全部塞在一起）分层 Context 注入引擎
这个例子实现了一个分层 Context 管理系统，每层有不同的注入策略：
python
"""
分层 Context 注入引擎
Context 不是"一个字符串"，而是有层级的信息结构。
每层有不同的生命周期、大小限制、和注入策略。
核心分层：
1. Core（核心指令）—— 始终注入，极少变化
2. Skills（技能）—— 按需注入，只包含当前需要的工具
3. Memory（记忆）—— 检索注入，只注入相关的部分
4. Task（任务）—— 动态注入，当前请求和数据
"""
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ContextLayer:
"""
Context 层 —— 定义一层信息的注入策略
name: 层级名称
content: 内容（字符串或可调用对象）
strategy: 注入策略
- "always": 始终注入（核心指令）
- "conditional": 条件注入（按需加载技能）
- "retrieval": 检索注入（从记忆库检索相关内容）
- "dynamic": 动态注入（每轮更新的任务数据）
max_tokens: 最大 token 限制
priority: 优先级（决定注入顺序和冲突时的保留策略）
"""
name: str
content: str | Any
strategy: str = "always"
max_tokens: int = 1000
priority: int = 1 # 1 = 最高（最先注入，最后被截断）
class ContextInjector:
"""
Context 注入管理器
核心逻辑：
1. 按优先级顺序注入各层
2. 检查总 token 是否超出限制
3. 超出时从低优先级层开始截断
4. 返回最终的完整 context
"""
def __init__(self, max_context_tokens: int = 8000):
self.max_context_tokens = max_context_tokens
self.layers: dict[str, ContextLayer] = {}
def add_layer(self, name: str, layer: ContextLayer) -> None:
self.layers[name] = layer
def remove_layer(self, name: str) -> None:
self.layers.pop(name, None)
def build_context(
self,
task_input: str = "",
retrieval_fn: dict[str, Any] = None,
condition_fn: dict[str, Any] = None,
) -> str:
"""
构建完整的上下文——按优先级注入各层
task_input: 当前任务/用户请求
retrieval_fn: 检索函数（用于 retrieval 策略层）
condition_fn: 条件函数（用于 conditional 策略层）
"""
retrieval_fn = retrieval_fn or {}
condition_fn = condition_fn or {}
# 按优先级排序（数字小的先注入）
sorted_layers = sorted(self.layers.values(), key=lambda l: l.priority)
parts = []
used_tokens = 0
for layer in sorted_layers:
content = self._get_layer_content(
layer, task_input, retrieval_fn, condition_fn
)
if not content:
continue
layer_content = self._truncate(content, layer.max_tokens)
layer_tokens = self._estimate_tokens(layer_content)
if used_tokens + layer_tokens > self.max_context_tokens:
remaining = self.max_context_tokens - used_tokens
if remaining <= 0:
break
layer_content = self._truncate(content, remaining)
parts.append(f"## {layer.name}\n{layer_content}")
used_tokens += self._estimate_tokens(layer_content)
return "\n\n".join(parts)
def _get_layer_content(
self,
layer: ContextLayer,
task_input: str,
retrieval_fn: dict,
condition_fn: dict,
) -> str | None:
"""根据注入策略获取层内容"""
if layer.strategy == "always":
return layer.content if isinstance(layer.content, str) else str(layer.content)
elif layer.strategy == "conditional":
checker = condition_fn.get(layer.name)
if checker and not checker(task_input):
return None
return layer.content if isinstance(layer.content, str) else str(layer.content)
elif layer.strategy == "retrieval":
retriever = retrieval_fn.get(layer.name)
if retriever:
return retriever(task_input)
return None
elif layer.strategy == "dynamic":
return task_input
return None
def _truncate(self, text: str, max_tokens: int) -> str:
"""按 token 限制截断文本"""
estimated_chars = max_tokens * 3
if len(text) > estimated_chars:
return text[:estimated_chars] + "\n...（已截断）"
return text
def _estimate_tokens(self, text: str) -> int:
"""估算 token 数"""
return len(text) // 3
# ==================== 使用示例 ====================
injector = ContextInjector(max_context_tokens=8000)
# Layer 1: 核心指令（始终注入）
injector.add_layer("core", ContextLayer(
name="核心指令",
content=(
"你是一个数据分析助手。你必须：\n"
"1. 先理解用户需求\n"
"2. 选择合适的工具执行\n"
"3. 输出结构化的分析报告\n"
"始终用中文回复。"
),
strategy="always",
max_tokens=200,
priority=1,
))
# Layer 2: 工具描述（条件注入——只在需要工具时注入）
injector.add_layer("tools", ContextLayer(
name="可用工具",
content=(
"- read_csv(path): 读取 CSV 文件\n"
"- analyze(data, metrics): 统计分析\n"
"- plot_chart(data, type): 生成图表\n"
"- save_report(content, path): 保存报告"
),
strategy="conditional",
max_tokens=500,
priority=2,
))
# Layer 3: 用户记忆（检索注入——只注入与当前任务相关的记忆）
injector.add_layer("memory", ContextLayer(
name="用户偏好",
content=None,
strategy="retrieval",
max_tokens=500,
priority=3,
))
# Layer 4: 当前任务（动态注入）
injector.add_layer("task", ContextLayer(
name="当前任务",
content="",
strategy="dynamic",
max_tokens=1000,
priority=4,
))
# 构建上下文
def memory_retriever(query: str) -> str:
"""模拟记忆检索——返回与查询相关的用户偏好"""
preferences = {
"chart": "用户偏好柱状图，不喜欢饼图",
"language": "用户要求所有报告用中文",
"decimal": "数据保留两位小数",
}
if "图表" in query or "图" in query:
return preferences.get("chart", "")
if "语言" in query:
return preferences.get("language", "")
return ""
context = injector.build_context(
task_input="分析这份销售数据，用图表展示趋势",
retrieval_fn={"memory": memory_retriever},
condition_fn={"tools": lambda x: "工具" in x or "分析" in x or "数据" in x},
)
print(context)每一层的"注入策略"决定了它什么时候出现、出现多少。核心指令永远在，工具按需加载，记忆只检索相关的，任务数据每轮更新。这样既保证了必要信息不缺，又避免了信息过载。
System Prompt 的指令设计模式
Context Engineering 不只是"怎么组织信息"，更重要的是怎么写指令让 LLM 真正遵守。下面对比几种 System Prompt 写法的差异：
json\n{json.dumps(schema, indent=2, ensureascii=False)}\n
text
"
@staticmethod
def thinking_framework(steps: list[dict]) -> str:
"""
模式四：思维框架
给 LLM 一个思考流程，让它按步骤推理。
这就是 Chain of Thought 的变体。
让 LLM 先想清楚再回答，提高准确性。
"""
steps_text = "\n".join(
f"步骤 {i+1} ({s['name']}): {s['description']}"
for i, s in enumerate(steps)
)
return f"## 思考流程\n请按以下步骤分析问题：\n{steps_text}"
@staticmethod
def tool_usage_guide(tools: list[dict], examples: list[str]) -> str:
"""
模式五：工具使用指南
不仅列出工具，还告诉 LLM 什么时候用什么工具。
工具描述只说"工具能做什么"，但没有说"什么时候该用"。
使用指南补了这个 gap。
"""
tool_text = "\n".join(
f"- {t['name']}: {t['description']}\n"
f" 使用场景: {t.get('when_to_use', '不确定时')}\n"
f" 避免场景: {t.get('when_to_avoid', '无')}"
)
examples_text = "\n".join(f"- {e}" for e in examples)
return f"## 工具\n{tool_text}\n\n## 使用示例\n{examples_text}"
@staticmethod
def progressive_disclosure(layers: list[dict]) -> str:
"""
模式六：渐进式披露
不一次性告诉 LLM 所有规则，而是在需要时才披露。
减少初始上下文的噪声。
LLM 只在相关时才看到相关规则，注意力更集中。
"""
return layers
# ==================== 组合使用 ====================
def build_production_system_prompt(
role: str = "数据分析助手",
constraints: list[str] = None,
tools: list[dict] = None,
output_format: dict = None,
) -> str:
"""
生产级 System Prompt —— 组合多种模式
不是所有模式都需要用。根据场景选择 2-3 种组合即可。
"""
constraints = constraints or [
"不要编造数据或引用不存在的来源",
"如果信息不足，明确说明而不是推测",
"所有数据引用必须标注来源",
]
parts = []
# 1. 角色定义
parts.append(SystemPromptPatterns.role_definition(
role=role,
responsibilities=[
"理解用户的数据分析需求",
"选择合适的分析工具和可视化方式",
"生成结构化的分析报告",
],
))
# 2. 约束 + 示例
parts.append(SystemPromptPatterns.constraint_setting(
constraints=constraints,
examples=[
("分析销售数据", "根据实际数据：Q4 销售额 500 万，同比增长 15%"),
("信息不足时", "当前数据集中缺少地区信息，无法进行地区对比分析"),
],
))
# 3. 输出 Schema（如果需要结构化输出）
if output_format:
parts.append(SystemPromptPatterns.output_schema(output_format))
return "\n\n".join(parts)
# 使用示例
prompt = build_production_system_prompt(
role="数据分析助手",
output_format={
"type": "object",
"properties": {
"summary": {"type": "string", "description": "数据摘要"},
"insights": {
"type": "array",
"items": {"type": "string"},
"description": "关键发现（最多 5 条）",
},
"recommendations": {
"type": "array",
"items": {"type": "string"},
"description": "建议（最多 3 条）",
},
},
"required": ["summary", "insights"],
},
)
print(prompt)这几种模式不需要全部用上。简单任务一个角色定义就够了；需要控制 LLM 行为时加约束和示例；需要结构化输出时加 Schema。模式是工具，不是 checklist。
信噪比优化——上下文压缩与精炼
Context Engineering 最实战的一个技能：怎么在有限的上下文窗口内，最大化有用信息的密度。
python
"""
信噪比优化 —— 上下文压缩与精炼
问题：Agent 运行过程中，上下文不断增长。
工具输出的大量数据中，只有一小部分是 LLM 真正需要的。
如果直接把原始数据塞进上下文，LLM 的注意力会被噪声淹没。
解决方案：在注入上下文之前，对数据进行"压缩与精炼"——
只保留关键信息，去掉无关细节。
"""
class ContextRefiner:
"""
上下文精炼器
四种策略：
1. 摘要压缩——用 LLM 把长文本压缩成短摘要
2. 关键信息提取——提取结构化的关键信息
3. 结构化转换——把非结构化数据转为结构化格式
4. 去重合并——去除重复信息
"""
@staticmethod
def summarize(text: str, max_length: int = 200, llm=None) -> str:
"""
策略一：摘要压缩
如果有 LLM，让 LLM 做摘要。否则用简单的启发式方法。
"""
if llm and len(text) > max_length:
prompt = f"""
请将以下文本压缩为一段摘要（不超过 {max_length} 字）。
只保留关键信息：数据、结论、异常。
去掉细节、例子和冗余描述。
{text[:2000]}
"""
return llm.generate(prompt, system="你是文本摘要专家。")
# 降级：简单的句子提取
sentences = text.split("。")
# 保留包含数字的句子（通常是最重要的）
important = [s for s in sentences if any(c.isdigit() for c in s)]
result = "。".join(important[:5]) + "。"
return result if result else text[:max_length]
@staticmethod
def extract_key_info(data: dict, schema: dict) -> str:
"""
策略二：关键信息提取
从结构化数据中提取关键字段。
schema 定义了哪些字段是"关键的"。
"""
lines = []
for key, info in schema.items():
if key in data:
value = data[key]
if isinstance(value, str) and len(value) > info.get("max_len", 100):
value = value[:info.get("max_len", 100)] + "..."
lines.append(f"{info.get('label', key)}: {value}")
return "\n".join(lines)
@staticmethod
def deduplicate_contexts(contexts: list[str], similarity_threshold: float = 0.8) -> list[str]:
"""
策略三：去重合并
多个工具输出的上下文可能有重复信息。
提取唯一的信息集合。
"""
if not contexts:
return []
unique = [contexts[0]]
for ctx in contexts[1:]:
is_duplicate = False
for existing in unique:
ctx_words = set(ctx.split())
existing_words = set(existing.split())
if not ctx_words or not existing_words:
continue
overlap = len(ctx_words & existing_words) / len(ctx_words | existing_words)
if overlap > similarity_threshold:
is_duplicate = True
break
if not is_duplicate:
unique.append(ctx)
return unique
# ==================== 使用示例 ====================
# 场景：工具返回了大量原始数据，需要精炼后注入上下文
# 原始数据（假设从 API 返回）
raw_data = {
"sales": [
{"date": "2025-01", "revenue": 100, "cost": 60, "region": "华北", "product": "A", "units": 50},
{"date": "2025-02", "revenue": 120, "cost": 72, "region": "华北", "product": "A", "units": 60},
{"date": "2025-03", "revenue": 150, "cost": 90, "region": "华东", "product": "B", "units": 75},
# ... 还有 1000 条
],
"metadata": {
"source": "sales_db",
"export_time": "2025-04-01 10:00:00",
"total_records": 1000,
"exported_by": "admin",
"format_version": "2.1",
# ... 很多元数据字段
},
}
# 精炼后的上下文
summary = f"数据量: {raw_data['metadata']['total_records']} 条\n"
summary += f"时间范围: {raw_data['sales'][0]['date']} ~ {raw_data['sales'][-1]['date']}\n"
summary += f"收入趋势: Q1 从 100 增长到 150（+50%）\n"
summary += f"涉及地区: 华北、华东"
print(summary)
# 输出：
# 数据量: 1000 条
# 时间范围: 2025-01 ~ 2025-03
# 收入趋势: Q1 从 100 增长到 150（+50%）
# 涉及地区: 华北、华东
# 原始数据可能 5000+ token，精炼后只需 ~100 token工具返回的原始数据中，90% 以上对 LLM 推理没有直接价值。全部塞进上下文不仅浪费 token，还会稀释注意力。在注入上下文之前先做数据精炼，只给 LLM 真正需要的信息。
上下文窗口的位置效应
研究发现 LLM 对上下文中不同位置的信息敏感度不同：
- 首尾效应：LLM 对上下文开头和结尾的信息 recall 率最高，中间最低
- "lost in the middle"：关键信息放在上下文中间时，准确率下降 10-30%
- 指令位置：指令放在开头比放在中间的效果好 20%
python
"""
上下文位置效应与信息布局
布局策略：
- 最重要的指令 -> 上下文开头（system prompt 前几行）
- 工具描述 -> 中间（这是参考信息，不需要高 recall）
- 当前任务 -> 结尾（最接近生成位置，recall 最高）
- 用户偏好 -> 开头或结尾，不要放中间
"""
class PositionAwareContextBuilder:
"""
感知位置的上下文构建器
根据信息的重要程度，安排其在上下文中的位置。
"""
def __init__(self):
self.beginning: list[str] = []
self.middle: list[str] = []
self.end: list[str] = []
def add(self, content: str, position: str = "middle") -> None:
"""
添加内容到指定位置
position:
- "beginning": 高优先级指令、角色定义、核心约束
- "middle": 工具描述、参考文档、历史数据
- "end": 用户当前请求、最新工具结果
"""
if position == "beginning":
self.beginning.append(content)
elif position == "middle":
self.middle.append(content)
elif position == "end":
self.end.append(content)
def build(self) -> str:
"""构建最终上下文——按位置排列"""
parts = []
if self.beginning:
parts.append("### 核心指令\n" + "\n".join(self.beginning))
if self.middle:
parts.append("### 参考信息\n" + "\n".join(self.middle))
if self.end:
parts.append("### 当前任务\n" + "\n".join(self.end))
return "\n\n".join(parts)
# 使用示例
builder = PositionAwareContextBuilder()
# 开头：最重要的指令
builder.add(
"你是数据分析助手。你必须基于实际数据回答，不得编造。"
"如果数据不足，明确说明。",
position="beginning",
)
# 中间：参考信息（工具描述）
builder.add(
"可用工具：\n"
"- read_csv(path): 读取 CSV\n"
"- analyze(data): 统计分析\n"
"- plot(data, type): 生成图表",
position="middle",
)
# 中间：历史数据
builder.add(
"用户偏好：喜欢柱状图，报告用中文。",
position="middle",
)
# 结尾：当前任务（离生成最近）
builder.add(
"用户请求：分析 sales.csv 的季度趋势，用图表展示。",
position="end",
)
context = builder.build()
print(context)
# 输出：
# ### 核心指令
# 你是数据分析助手。你必须基于实际数据回答，不得编造。如果数据不足，明确说明。
#
# ### 参考信息
# 可用工具：...
# 用户偏好：...
#
# ### 当前任务
# 用户请求：分析 sales.csv 的季度趋势，用图表展示。如果只有一个地方放最重要的指令，放开头。如果只能放一个任务描述，放结尾。不要在上下文的中间位置放关键信息——那是 LLM 注意力最低的区域。
实际考虑
Context Engineering 不是所有场景都需要。任务简单、一个 500 字的 system prompt 就能搞定，或者上下文窗口只用了不到 20%、工具不超过 3 个、没有跨轮次记忆需求，那就不需要搞这些。
但 system prompt 超过 2000 字、工具有 10 个以上、Agent 有长期记忆需求、工具输出很长需要精炼——这些场景下分层注入能带来 5-6 倍的 token 节省和更高的信噪比。
代价也是有的：需要维护多层信息的注入逻辑而不是一个 prompt 字符串；retrieval 策略会增加延迟；分层之后出问题要排查是哪一层出了状况；条件注入如果判断不准可能漏掉必要信息。所以总上下文使用量不到窗口 30% 的时候，没必要引入这套机制。
各主流框架其实都在做类似的事：Anthropic 官方博客推荐分层 system prompt 加渐进式工具披露；OpenAI 的 Assistant API 用 Instructions/Tools/Thread 天然分层；LangChain 的 PromptTemplate 负责指令层、ChatMessageHistory 负责记忆层；LlamaIndex 的核心能力就是从海量数据中检索少量注入上下文。
从一个 prompt 到 Context Engineering
Context Engineering 不是一开始就该上的东西，而是演进出来的。
起步通常是一个简单的 prompt 字符串。然后发现需要工具描述，拼在一起。然后发现上下文太长了，开始粗暴截断。粗暴截断不行，改成按需加载工具。再然后发现需要记忆，引入检索注入。最后发现管理这些注入逻辑越来越乱，才抽出了分层注入引擎。
每一步都有明确的需求驱动，不是一开始就搞一个复杂的引擎。
python
# Step 1: 起点
prompt = "你是一个助手，帮我分析数据"
# Step 2: 发现需要工具
prompt = system_prompt + "\n\n可用工具:\n" + tool_descriptions
# Step 3: 发现上下文爆了，粗暴截断
if len(prompt) > max_tokens:
prompt = prompt[:max_tokens]
# Step 4: 截断不行，按需加载
relevant_tools = select_tools_for_task(task, all_tools)
prompt = system_prompt + "\n\n可用工具:\n" + relevant_tools
# Step 5: 需要记忆，检索注入
relevant_memory = retrieve_memory(task, memory_store)
prompt = system_prompt + "\n\n" + relevant_memory + "\n\n任务: " + task
# Step 6: 管理混乱，分层注入引擎
injector = ContextInjector(max_tokens=8000)
injector.add_layer("core", core_layer)
injector.add_layer("skills", skills_layer)
injector.add_layer("memory", memory_layer)
context = injector.build_context(task_input=task)理解了 Context Engineering，就不会再抱怨"LLM 怎么忽略了指令"——通常不是写得不够多，而是放的位置不对、层次不清、信噪比太低。