系列导航
– 第四篇：Context层设计 ← 你在这里
一、引言：上下文是Agent的”工作记忆”
想象你是一位外科医生，正在做一台复杂的手术。你不可能把所有的医学教科书都背在脑子里，但你需要在手术过程中随时调取相关知识——病人的病历、影像资料、手术方案、药物禁忌……这些即时可及的知识，就是你的”工作记忆”。
对于AI Agent来说，Context（上下文）就是它的工作记忆。
一个没有良好Context管理的Agent，就像一个失忆的外科医生——每次对话都要从头开始，对用户的历史一无所知，对专业知识只能靠”猜”。而一个Context层设计精良的Agent，则像一位准备充分的专家：它知道你是谁，了解你的历史需求，掌握相关领域的专业知识，并且能够精准地在需要的时候把正确的信息放到正确的位置。
1.1 为什么Context层如此关键？
在Harness Engineering的六层架构中，Context层位于Prompt层和Tool层之上，是连接”模型能力”与”知识供给”的桥梁。
`
┌─────────────────────────────────────────────┐
│ Orchestration 编排层 │
│ ┌─────────────────────────┐ │
│ │ Context 上下文层 │ │ ← 本篇主题
│ │ ┌──────────┐ ┌────────┐│ │
│ │ │Prompt层 │ │Tool层 ││ │
│ │ └──────────┘ └────────┘│ │
│ └─────────────────────────┘ │
│ ┌─────────────────────────┐ │
│ │ Memory 记忆层 │ │
│ └─────────────────────────┘ │
│ ┌─────────────────────────┐ │
│ │ Model 模型层 │ │
│ └─────────────────────────┘ │
└─────────────────────────────────────────────┘
`
类比：如果Prompt层是”指令”，Tool层是”工具”，那么Context层就是”材料”。就像做菜一样——你有菜谱（Prompt），有厨具（Tool），但没有食材（Context），什么也做不出来。
1.2 Context层的核心职责
| 职责 | 描述 | 示例 |
|---|
|——|——|——|
| 知识供给 | 为模型提供回答问题所需的知识 | 文档检索、知识库查询 |
|---|---|---|
| 历史管理 | 维护对话历史和用户状态 | 多轮对话上下文 |
| 信息筛选 | 从海量数据中提取相关信息 | RAG检索、重排序 |
| 窗口管理 | 在有限窗口内最大化信息密度 | 上下文压缩、滑动窗口 |
| 动态注入 | 根据任务需求动态组装上下文 | 任务感知的上下文选择 |
二、上下文窗口的本质
2.1 Token限制与信息密度
所有大语言模型都有一个上下文窗口（Context Window）——它是模型在单次推理中能够”看到”的最大文本长度。
类比：上下文窗口就像你的桌面面积。桌面越大，你能同时摊开的文件越多。但桌面再大也有边界——你不可能把整个图书馆摊在一张桌子上。
| 模型 | 上下文窗口 | 大约等于 |
|---|
|——|———–|———|
| GPT-4o | 128K tokens | ~300页A4文档 |
|---|---|---|
| Claude 3.5 Sonnet | 200K tokens | ~500页A4文档 |
| Gemini 1.5 Pro | 1M tokens | ~2500页A4文档 |
| GPT-4o mini | 128K tokens | ~300页A4文档 |
| DeepSeek-V3 | 128K tokens | ~300页A4文档 |
| Llama 3.1 | 128K tokens | ~300页A4文档 |
但窗口大小 ≠ 有效信息量。这里有两个关键概念：
① 信息密度（Information Density）
`
信息密度 = 有效信息量 / Token数量
`
一段精心组织的技术文档，信息密度可能达到0.8；而一段冗余重复的对话，信息密度可能只有0.1。
② 注意力衰减（Attention Decay）
即使在窗口范围内，模型对不同位置信息的关注程度也不均匀：
`
注意力分布示意图（非线性）：
注意力强度
高 │ ██ ██
│ ████ ████
│ ██████ ██████
│ ████████ ████████
低 │ ████████████████████████████████████████
└──────────────────────────────────────────→
开头(指令) 中间(上下文) 末尾(最近对话)
`
这就是著名的 “Lost in the Middle” 现象——模型对上下文开头和结尾的信息注意力最强，对中间部分的信息容易忽略。
2.2 上下文的三个层次
一个完整的Agent上下文通常由三个层次组成：
`python
context = {
“system”: “…”, # 系统指令层：角色定义、行为规范
“retrieved”: “…”, # 检索知识层：从知识库中检索的文档
“conversation”: “…” # 对话历史层：用户与Agent的交互记录
}
`
| 层次 | 特征 | 生命周期 | 典型占比 |
|---|
|——|——|———|———|
| System（系统层） | 相对固定，定义Agent身份和行为 | 整个会话期间 | 5-15% |
|---|---|---|---|
| Retrieved（检索层） | 动态变化，根据用户问题检索 | 单次推理 | 30-60% |
| Conversation（对话层） | 随对话累积增长 | 滑动窗口 | 25-50% |
类比：这就像一个人的工作状态——System是你的职业身份（”我是医生”），Retrieved是你查阅的资料（”病人的CT报告”），Conversation是你和病人的对话记录（”上次你说头痛对吗？”）。
三、Context管理的三大策略
面对上下文窗口有限、知识需求无限的矛盾，工程界发展出了三大策略：
`
┌─────────────────────────────────────────────────────┐
│ Context管理策略 │
├─────────────┬─────────────┬─────────────────────────┤
│ RAG │ 长上下文 │ 动态注入 │
│ 检索增强 │ 压缩扩展 │ 智能组装 │
├─────────────┼─────────────┼─────────────────────────┤
│ 从外部知识库 │ 利用超长上下 │ 根据任务类型 │
│ 检索相关文档 │ 文窗口塞入 │ 动态选择和组装 │
│ 注入上下文 │ 更多信息 │ 最优上下文 │
├─────────────┼─────────────┼─────────────────────────┤
│ 适合：知识库 │ 适合：长文档 │ 适合：复杂任务 │
│ 问答、客服 │ 分析、总结 │ 多步骤推理 │
└─────────────┴─────────────┴─────────────────────────┘
`
在实际应用中，这三种策略往往混合使用。接下来我们逐一深入。
四、RAG：检索增强生成
4.1 RAG架构设计
RAG（Retrieval-Augmented Generation）是目前最主流的上下文管理方案。它的核心思想是：不把所有知识都塞进上下文，而是在推理时根据用户问题，从外部知识库中检索最相关的文档片段，然后注入上下文。
类比：RAG就像一个经验丰富的律师——他不会把所有法律条文都背下来，但他知道如何快速检索到相关的法条，然后结合这些法条来回答你的问题。
标准RAG架构如下：
`
用户问题
│
▼
┌──────────────┐ ┌──────────────────┐
│ Query处理 │────▶│ 向量检索 │
│ (Embedding) │ │ (Vector Search) │
└──────────────┘ └────────┬─────────┘
│
┌─────────▼─────────┐
│ 重排序 │
│ (Reranking) │
└─────────┬─────────┘
│
┌──────────────┐ ┌────────▼──────────┐
│ LLM推理 │◀────│ 上下文组装 │
│ 生成回答 │ │ (Context Build) │
└──────────────┘ └───────────────────┘
▲
│
┌─────────┴─────────┐
│ 向量数据库 │
│ (Vector Store) │
└───────────────────┘
▲
│
┌─────────┴─────────┐
│ 文档分块 │
│ (Chunking) │
└───────────────────┘
▲
│
原始文档
`
4.2 文档分块策略
文档分块是RAG的第一步，也是最影响最终效果的一步。分块的质量直接决定了检索的质量。
类比：分块就像图书馆的编目方式。如果把整本书当成一个条目，你很难找到具体的段落；如果把每个字当成一个条目，又失去了上下文。好的分块就像好的编目——既精确又完整。
4.2.1 固定大小分块（Fixed-size Chunking）
最简单的方式，按固定Token数量切分：
`python
def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
“””固定大小分块，带重叠”””
tokens = tokenize(text)
chunks = []
start = 0
while start < len(tokens):
end = start + chunk_size
chunk = detokenize(tokens[start:end])
chunks.append(chunk)
start = end – overlap # 重叠部分保证上下文连贯
return chunks
`
优点：实现简单，分块大小可预测
缺点：可能在句子中间切断，破坏语义完整性
4.2.2 语义分块（Semantic Chunking）
根据语义边界切分，比如按段落、章节、或语义相似度：
`python
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=[“nn”, “n”, “。”, “！”, “？”, “；”, ” “],
length_function=len,
)
chunks = splitter.split_text(document)
`
4.2.3 递归分块（Recursive Chunking）
LangChain推荐的默认策略，按层级分隔符递归切分：
`python
# 优先按段落切分 → 段落太长则按句子切分 → 句子太长则按词切分
separators = [“nn”, “n”, “。”, “.”, ” “, “”]
`
分块策略对比
| 策略 | 复杂度 | 语义完整性 | 适用场景 |
|---|
|——|——–|———–|———|
| 固定大小 | ⭐ | ⭐⭐ | 结构化程度低的文档 |
|---|---|---|---|
| 语义分块 | ⭐⭐⭐ | ⭐⭐⭐⭐ | 文章、报告等有明确段落的文档 |
| 递归分块 | ⭐⭐ | ⭐⭐⭐ | 通用场景（推荐默认选择） |
| 按文档结构 | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | 技术文档、Wiki等结构化内容 |
实践建议：chunk_size建议设置为256-1024 tokens，overlap建议为chunk_size的10%-15%。
4.3 Embedding模型选择
Embedding模型将文本转换为向量表示，是语义检索的基础。选择合适的Embedding模型对检索质量至关重要。
类比：Embedding就像给每段文本分配一个”GPS坐标”。意思相近的文本，它们的”坐标”就接近；意思不同的文本，”坐标”就远离。
| 模型 | 维度 | 中文支持 | 价格 | 推荐场景 |
|---|
|——|——|———|——|———|
text-embedding-3-small |
1536 | ⭐⭐⭐ | $0.02/M tokens | 通用场景，性价比高 |
|---|---|---|---|---|
text-embedding-3-large |
3072 | ⭐⭐⭐ | $0.13/M tokens | 需要高精度的场景 |
embed-v4 (Cohere) |
1024 | ⭐⭐⭐⭐ | $0.1/M tokens | 多语言场景 |
BGE-M3 (BAAI) |
1024 | ⭐⭐⭐⭐⭐ | 免费(自部署) | 中文场景首选 |
bge-large-zh-v1.5 |
1024 | ⭐⭐⭐⭐⭐ | 免费(自部署) | 纯中文场景 |
Jina-embeddings-v3 |
1024 | ⭐⭐⭐⭐ | 免费额度 | 多语言、长文本 |
`python
# OpenAI Embedding示例
from openai import OpenAI
client = OpenAI()
def get_embedding(text: str, model: str = “text-embedding-3-small”) -> list[float]:
response = client.embeddings.create(
input=text,
model=model
)
return response.data[0].embedding
# BGE-M3 本地部署示例
from FlagEmbedding import BGEM3FlagModel
model = BGEM3FlagModel(‘BAAI/bge-m3’, use_fp16=True)
embeddings = model.encode(
[“这是第一段文本”, “这是第二段文本”],
batch_size=12,
max_length=8192,
)[‘dense_vecs’]
`
4.4 向量数据库对比
向量数据库是存储和检索Embedding向量的专用数据库。2026年的主流选择如下：
类比：向量数据库就像一个按”语义距离”组织的图书馆。传统图书馆按分类号排列，你需要知道分类号才能找到书；语义图书馆按”意思的远近”排列，你只需要描述你在找什么，它就能帮你找到最相关的书。
| 数据库 | 类型 | 分布式 | 特点 | 推荐场景 |
|---|
|——–|——|——–|——|———|
| Milvus | 专用 | ✅ | 高性能、云原生、GPU加速 | 大规模生产环境 |
|---|---|---|---|---|
| Qdrant | 专用 | ✅ | Rust实现、过滤能力强 | 需要复杂过滤的场景 |
| Weaviate | 专用 | ✅ | GraphQL接口、多模态 | 需要丰富API的场景 |
| Chroma | 专用 | ❌ | 轻量级、开发友好 | 原型开发、小规模 |
| Pinecone | 云服务 | ✅ | 全托管、零运维 | 不想自建基础设施 |
| pgvector | 扩展 | ✅ | 基于PostgreSQL | 已有PG基础设施 |
`python
# ChromaDB 快速上手示例
import chromadb
# 创建客户端
client = chromadb.PersistentClient(path=”./chroma_db”)
# 创建集合
collection = client.get_or_create_collection(
name=”knowledge_base”,
metadata={“hnsw:space”: “cosine”} # 使用余弦相似度
)
# 添加文档
collection.add(
documents=[“Harness Engineering是一种AI工程方法论”,
“Context层负责管理LLM的上下文信息”],
ids=[“doc1”, “doc2”],
metadatas=[{“source”: “article-01”}, {“source”: “article-04”}]
)
# 查询
results = collection.query(
query_texts=[“什么是Harness Engineering？”],
n_results=3,
where={“source”: {“$eq”: “article-01”}} # 元数据过滤
)
`
4.5 检索策略优化
单纯的向量相似度搜索往往不够。实际工程中需要多种检索策略的组合。
4.5.1 混合搜索（Hybrid Search）
结合关键词搜索和语义搜索的优势：
`python
def hybrid_search(query: str, alpha: float = 0.7) -> list[dict]:
“””混合搜索：语义搜索 + 关键词搜索”””
# 语义搜索结果
semantic_results = vector_db.search(
embedding=get_embedding(query),
top_k=20
)
# 关键词搜索结果（BM25）
keyword_results = bm25_search(query, top_k=20)
# 融合评分 (Reciprocal Rank Fusion)
combined = {}
for rank, doc in enumerate(semantic_results):
combined[doc.id] = combined.get(doc.id, 0) + alpha / (60 + rank)
for rank, doc in enumerate(keyword_results):
combined[doc.id] = combined.get(doc.id, 0) + (1-alpha) / (60 + rank)
# 按融合分数排序
return sorted(combined.items(), key=lambda x: x[1], reverse=True)[:10]
`
4.5.2 重排序（Reranking）
在初筛后用更精细的模型重新排序：
`python
from cohere import Client
co = Client(“your-api-key”)
def rerank(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
“””使用Cohere Reranker重排序”””
results = co.rerank(
query=query,
documents=documents,
top_n=top_n,
model=”rerank-v3.5″
)
return [
{“text”: documents[r.index], “score”: r.relevance_score}
for r in results.results
]
`
4.5.3 查询改写（Query Rewriting）
用LLM改写用户问题，提高检索质量：
`python
def rewrite_query(original_query: str) -> list[str]:
“””查询改写：生成多个检索查询”””
response = llm.chat(f”””
给定以下用户问题，生成3个不同角度的检索查询，用于向量数据库搜索。
只返回查询，每行一个。
用户问题：{original_query}
“””)
return response.strip().split(“n”)
# 生成多个查询 → 分别检索 → 合并去重 → 重排序
queries = rewrite_query(“Agent的记忆机制怎么设计？”)
all_results = []
for q in queries:
all_results.extend(vector_db.search(get_embedding(q), top_k=5))
final_results = deduplicate_and_rerank(all_results)
`
五、长上下文管理
5.1 128K/1M上下文的实际使用
随着Gemini 1.5 Pro支持1M tokens、Claude支持200K tokens，一个自然的问题是：既然窗口这么大了，还需要RAG吗？
答案是：需要，但用法变了。
类比：就像你有了一张更大的桌子，但桌子再大也不等于图书馆。大桌子的好处是你可以一次性摊开更多的文件做对比分析，但你仍然需要一个高效的系统来决定哪些文件应该上桌。
长上下文的实际成本：
`python
# 计算一次API调用的成本
def estimate_cost(input_tokens: int, output_tokens: int, model: str) -> float:
pricing = {
“gpt-4o”: {“input”: 2.50, “output”: 10.00}, # per 1M tokens
“claude-sonnet-4”: {“input”: 3.00, “output”: 15.00},
“gemini-1.5-pro”: {“input”: 1.25, “output”: 5.00},
}
p = pricing[model]
return (input_tokens * p[“input”] + output_tokens * p[“output”]) / 1_000_000
# 对比：RAG vs 长上下文
rag_cost = estimate_cost(4_000, 500, “gpt-4o”) # ~$0.015
long_ctx_cost = estimate_cost(120_000, 500, “gpt-4o”) # ~$0.305
print(f”RAG成本: ${rag_cost:.4f}”)
print(f”长上下文成本: ${long_ctx_cost:.4f}”)
print(f”长上下文贵 {long_ctx_cost/rag_cost:.1f} 倍”)
# 输出：长上下文贵 ~20倍
`
5.2 上下文压缩技术
当信息量超出窗口限制，或者想要降低成本时，上下文压缩是必要的。
类比：上下文压缩就像做会议纪要——你不会把两小时的会议录音一字不差地记下来，而是提取关键信息、决策和待办事项。
`python
def compress_context(context: str, query: str, max_tokens: int = 4000) -> str:
“””使用LLM压缩上下文，保留与查询相关的关键信息”””
response = llm.chat(f”””
请从以下上下文中提取与问题最相关的关键信息。
输出应该精简但完整，不超过{max_tokens} tokens。
问题：{query}
上下文：
{context}
请提取关键信息（保留原文中的重要细节和数据）：
“””)
return response
`
压缩策略对比：
| 策略 | 压缩率 | 信息保留率 | 适用场景 |
|---|
|——|——–|———–|———|
| LLM摘要 | 70-80% | ⭐⭐⭐⭐ | 通用文档压缩 |
|---|---|---|---|
| 关键句提取 | 50-70% | ⭐⭐⭐ | 长文档快速压缩 |
| 基于查询的压缩 | 80-90% | ⭐⭐⭐⭐⭐ | 明确查询的场景 |
| 结构化提取 | 60-80% | ⭐⭐⭐⭐ | 表格、JSON等结构化数据 |
5.3 滑动窗口策略
对于长对话，使用滑动窗口管理对话历史：
`python
class SlidingWindowContext:
def __init__(self, max_tokens: int = 8000, preserve_recent: int = 5):
self.messages = []
self.max_tokens = max_tokens
self.preserve_recent = preserve_recent
def add_message(self, role: str, content: str):
self.messages.append({“role”: role, “content”: content})
self._trim()
def _trim(self):
“””保留最近N条消息 + 尽可能多的历史消息”””
total_tokens = sum(count_tokens(m[“content”]) for m in self.messages)
if total_tokens <= self.max_tokens:
return
# 保留最近的消息
recent = self.messages[-self.preserve_recent:]
older = self.messages[:-self.preserve_recent]
# 从旧到新保留，直到超出限制
preserved = []
tokens = sum(count_tokens(m[“content”]) for m in recent)
for msg in reversed(older):
msg_tokens = count_tokens(msg[“content”])
if tokens + msg_tokens > self.max_tokens – 500: # 留500 token缓冲
break
preserved.insert(0, msg)
tokens += msg_tokens
# 添加摘要标记
if len(preserved) < len(older):
summary = summarize_messages(older[:len(older)-len(preserved)])
self.messages = [{“role”: “system”, “content”: f”[历史摘要] {summary}”}] + preserved + recent
else:
self.messages = preserved + recent
`
六、动态上下文注入
6.1 基于任务类型的上下文选择
不同的任务需要不同的上下文。动态上下文注入的核心是根据任务类型，智能选择和组装上下文。
类比：这就像一个专业的助手在给老板准备会议材料。财务会议准备财务报表，技术会议准备技术方案，客户会议准备客户资料——不同的场景，不同的材料。
`python
from enum import Enum
from typing import Optional
class TaskType(Enum):
QA = “question_answering” # 知识问答
SUMMARIZE = “summarization” # 文档总结
CODE = “code_generation” # 代码生成
ANALYSIS = “data_analysis” # 数据分析
CREATIVE = “creative_writing” # 创意写作
TRANSLATE = “translation” # 翻译
class ContextBuilder:
“””动态上下文组装器”””
# 不同任务类型的上下文模板
TEMPLATES = {
TaskType.QA: {
“system_prompt”: “你是一个知识渊博的助手，请基于提供的参考资料回答问题。”,
“retrieval_top_k”: 5,
“include_history”: True,
“max_context_tokens”: 6000,
},
TaskType.CODE: {
“system_prompt”: “你是一个专业的程序员，请根据需求和参考代码编写代码。”,
“retrieval_top_k”: 3,
“include_history”: False,
“max_context_tokens”: 8000,
},
TaskType.SUMMARIZE: {
“system_prompt”: “请总结以下内容的要点。”,
“retrieval_top_k”: 0, # 不需要检索，直接用原文
“include_history”: False,
“max_context_tokens”: 16000,
},
}
def build_context(
self,
query: str,
task_type: TaskType,
history: Optional[list] = None,
extra_docs: Optional[list] = None
) -> dict:
template = self.TEMPLATES.get(task_type, self.TEMPLATES[TaskType.QA])
context_parts = []
# 1. 系统指令
context_parts.append((“[SYSTEM]”, template[“system_prompt”]))
# 2. 检索相关内容
if template[“retrieval_top_k”] > 0:
docs = self.retriever.search(query, top_k=template[“retrieval_top_k”])
context_parts.append((“[RETRIEVED]”, “n—n”.join(docs)))
# 3. 对话历史
if template[“include_history”] and history:
history_text = self._format_history(history[-10:]) # 最近10轮
context_parts.append((“[HISTORY]”, history_text))
# 4. 额外文档
if extra_docs:
context_parts.append((“[DOCUMENTS]”, “n—n”.join(extra_docs)))
return self._assemble(context_parts, template[“max_context_tokens”])
`
6.2 上下文优先级排序
当上下文空间有限时，需要对候选内容按优先级排序：
`python
class PriorityRanker:
“””上下文优先级排序器”””
# 优先级权重
WEIGHTS = {
“relevance”: 0.40, # 与查询的相关性
“recency”: 0.20, # 信息的新鲜度
“authority”: 0.15, # 信息来源的权威性
“diversity”: 0.15, # 信息的多样性
“completeness”: 0.10, # 信息的完整性
}
def rank(self, candidates: list[dict], query: str) -> list[dict]:
scored = []
for doc in candidates:
score = (
self.WEIGHTS[“relevance”] * self._relevance_score(doc, query) +
self.WEIGHTS[“recency”] * self._recency_score(doc) +
self.WEIGHTS[“authority”] * self._authority_score(doc) +
self.WEIGHTS[“diversity”] * self._diversity_score(doc, scored) +
self.WEIGHTS[“completeness”] * self._completeness_score(doc)
)
scored.append((score, doc))
scored.sort(reverse=True, key=lambda x: x[0])
return [doc for _, doc in scored]
`
6.3 上下文缓存优化
频繁检索相同的文档会浪费资源。缓存是提高效率的关键：
`python
import hashlib
import time
from functools import lru_cache
class ContextCache:
“””上下文缓存管理器”””
def __init__(self, ttl: int = 3600):
self.cache = {}
self.ttl = ttl # 缓存过期时间（秒）
def _key(self, query: str, params: dict) -> str:
content = f”{query}:{sorted(params.items())}”
return hashlib.md5(content.encode()).hexdigest()
def get(self, query: str, params: dict) -> Optional[list]:
key = self._key(query, params)
if key in self.cache:
entry = self.cache[key]
if time.time() – entry[“time”] < self.ttl:
return entry[“data”]
else:
del self.cache[key] # 过期删除
return None
def set(self, query: str, params: dict, data: list):
key = self._key(query, params)
self.cache[key] = {
“data”: data,
“time”: time.time()
}
`
OpenAI的Prompt Caching：2026年，OpenAI等提供商已经支持Prompt级别的自动缓存。如果连续请求的前缀相同（比如相同的System Prompt），会自动命中缓存，减少延迟和成本：
`python
# 连续请求相同的System Prompt时，自动享受缓存折扣
# 第二次起，前缀部分的成本降低50-90%
response = client.chat.completions.create(
model=”gpt-4o”,
messages=[
{“role”: “system”, “content”: LONG_SYSTEM_PROMPT}, # 这部分会被缓存
{“role”: “user”, “content”: user_query} # 这部分每次不同
]
)
`
七、实战：构建一个文档问答系统
现在，让我们把前面学到的所有知识整合起来，构建一个完整的文档问答系统。
7.1 系统架构
`
┌──────────────────────────────────────────────────────────┐
│ 文档问答系统架构 │
│ │
│ ┌──────────┐ ┌──────────────┐ ┌───────────────┐ │
│ │ 文档导入 │───▶│ 分块+向量化 │───▶│ 向量数据库 │ │
│ │ PDF/TXT │ │ Embedding │ │ ChromaDB │ │
│ └──────────┘ └──────────────┘ └───────┬───────┘ │
│ │ │
│ ┌──────────┐ ┌──────────────┐ ┌───────▼───────┐ │
│ │ 用户查询 │───▶│ 查询处理 │───▶│ 混合检索 │ │
│ │ Web/API │ │ 改写+Embed │ │ 语义+关键词 │ │
│ └──────────┘ └──────────────┘ └───────┬───────┘ │
│ │ │
│ ┌──────────┐ ┌──────────────┐ ┌───────▼───────┐ │
│ │ 回答输出 │◀───│ LLM推理 │◀───│ 重排序+组装 │ │
│ │ Web/API │ │ GPT-4o │ │ Context Build │ │
│ └──────────┘ └──────────────┘ └───────────────┘ │
└──────────────────────────────────────────────────────────┘
`
7.2 完整代码实现
`python
“””
文档问答系统 – 完整实现
支持：PDF/TXT/MD文档导入、RAG检索、混合搜索、重排序
“””
import os
from dataclasses import dataclass, field
from typing import Optional
from pathlib import Path
import chromadb
from openai import OpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
# ==================== 数据结构 ====================
@dataclass
class Document:
“””文档数据结构”””
content: str
metadata: dict = field(default_factory=dict)
embedding: Optional[list[float]] = None
@dataclass
class SearchResult:
“””检索结果”””
content: str
score: float
metadata: dict
# ==================== 文档处理 ====================
class DocumentProcessor:
“””文档处理器：加载、分块、向量化”””
def __init__(self, chunk_size: int = 512, chunk_overlap: int = 64):
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=[“nn”, “n”, “。”, “.”, “！”, “？”, “；”, ” “],
)
self.client = OpenAI()
def load_document(self, file_path: str) -> str:
“””加载文档”””
path = Path(file_path)
if path.suffix == ‘.pdf’:
import PyPDF2
with open(path, ‘rb’) as f:
reader = PyPDF2.PdfReader(f)
return “n”.join(page.extract_text() for page in reader.pages)
else:
return path.read_text(encoding=’utf-8′)
def chunk_document(self, text: str, metadata: dict = None) -> list[Document]:
“””分块处理”””
chunks = self.splitter.split_text(text)
return [
Document(
content=chunk,
metadata={**(metadata or {}), “chunk_index”: i}
)
for i, chunk in enumerate(chunks)
]
def get_embedding(self, text: str) -> list[float]:
“””获取文本的Embedding向量”””
response = self.client.embeddings.create(
input=text,
model=”text-embedding-3-small”
)
return response.data[0].embedding
def process_file(self, file_path: str) -> list[Document]:
“””处理单个文件：加载→分块→向量化”””
text = self.load_document(file_path)
metadata = {“source”: file_path}
chunks = self.chunk_document(text, metadata)
# 批量获取Embedding（效率更高）
texts = [c.content for c in chunks]
embeddings = self._batch_embed(texts)
for doc, emb in zip(chunks, embeddings):
doc.embedding = emb
return chunks
def _batch_embed(self, texts: list[str], batch_size: int = 100) -> list[list[float]]:
“””批量Embedding”””
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
response = self.client.embeddings.create(
input=batch,
model=”text-embedding-3-small”
)
all_embeddings.extend([d.embedding for d in response.data])
return all_embeddings
# ==================== 向量存储 ====================
class VectorStore:
“””向量存储：基于ChromaDB”””
def __init__(self, collection_name: str = “knowledge_base”, persist_dir: str = “./chroma_db”):
self.client = chromadb.PersistentClient(path=persist_dir)
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={“hnsw:space”: “cosine”}
)
def add_documents(self, documents: list[Document]):
“””添加文档到向量库”””
self.collection.add(
ids=[f”doc_{i}” for i in range(len(documents))],
documents=[d.content for d in documents],
embeddings=[d.embedding for d in documents],
metadatas=[d.metadata for d in documents],
)
def search(self, query_embedding: list[float], top_k: int = 5) -> list[SearchResult]:
“””向量搜索”””
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=[“documents”, “metadatas”, “distances”]
)
return [
SearchResult(
content=doc,
score=1 – dist, # 余弦距离转相似度
metadata=meta
)
for doc, meta, dist in zip(
results[“documents”][0],
results[“metadatas”][0],
results[“distances”][0]
)
]
# ==================== RAG引擎 ====================
class RAGEngine:
“””RAG问答引擎”””
def __init__(self, vector_store: VectorStore):
self.store = vector_store
self.client = OpenAI()
self.processor = DocumentProcessor()
def ingest(self, file_paths: list[str]):
“””导入文档”””
for path in file_paths:
print(f”处理文档: {path}”)
docs = self.processor.process_file(path)
self.store.add_documents(docs)
print(f” → 分块数: {len(docs)}”)
def query(self, question: str, top_k: int = 5) -> dict:
“””问答查询”””
# 1. 查询Embedding
query_embedding = self.processor.get_embedding(question)
# 2. 向量检索
results = self.store.search(query_embedding, top_k=top_k)
# 3. 组装上下文
context_parts = []
for i, r in enumerate(results, 1):
source = r.metadata.get(“source”, “未知来源”)
context_parts.append(f”[参考{i}] (来源: {source})n{r.content}”)
context = “nn—nn”.join(context_parts)
# 4. LLM生成回答
response = self.client.chat.completions.create(
model=”gpt-4o”,
messages=[
{
“role”: “system”,
“content”: “””你是一个专业的知识助手。请基于提供的参考资料回答用户问题。
规则：
- 只基于参考资料回答，不要编造信息
- 如果参考资料中没有相关信息，请明确说明
- 回答要准确、简洁、有条理
- 引用参考资料时标注来源”””
},
{
“role”: “user”,
“content”: f”””参考资料：
{context}
用户问题：{question}
请基于以上参考资料回答问题。”””
}
],
temperature=0.1, # 低温度保证回答一致性
)
return {
“answer”: response.choices[0].message.content,
“sources”: [
{“content”: r.content, “score”: r.score, “metadata”: r.metadata}
for r in results
],
“usage”: {
“input_tokens”: response.usage.prompt_tokens,
“output_tokens”: response.usage.completion_tokens,
}
}
# ==================== 使用示例 ====================
if __name__ == “__main__”:
# 初始化
store = VectorStore()
engine = RAGEngine(store)
# 导入文档
engine.ingest([
“./docs/harness-engineering-01.pdf”,
“./docs/harness-engineering-04.md”,
])
# 查询
result = engine.query(“什么是Harness Engineering？它有哪几层架构？”)
print(“=” * 60)
print(“回答：”)
print(result[“answer”])
print(“n参考来源：”)
for src in result[“sources”]:
print(f” – [{src[‘score’]:.3f}] {src[‘metadata’].get(‘source’, ‘未知’)}”)
`
7.3 运行效果示例
`bash
$ python rag_system.py
处理文档: ./docs/harness-engineering-01.pdf
→ 分块数: 42
处理文档: ./docs/harness-engineering-04.md
→ 分块数: 38
============================================================
回答：
Harness Engineering是一种2026年兴起的AI工程方法论，专注于通过
系统化的”约束”和”引导”来释放大语言模型的潜力。
它包含六层架构：
- Prompt层（提示词）- 指令设计与输出格式
- Context层（上下文）- 信息管理与RAG
- Tool层（工具）- API、MCP与外部能力
- Memory层（记忆）- 短期/长期记忆与知识库
- Eval层（评估）- 测试、验证与质量保证
- Observability层（可观测性）- 监控、日志与追踪
参考来源：
- [0.923] ./docs/harness-engineering-01.pdf
- [0.887] ./docs/harness-engineering-04.md
- [0.856] ./docs/harness-engineering-01.pdf
`
八、常见陷阱与解决方案
在实际工程中，Context层设计有许多容易踩的坑。这里总结最常见的三大陷阱：
8.1 陷阱一：上下文溢出（Context Overflow）
问题：注入的上下文超出模型窗口限制，导致截断或错误。
`
┌──────────────────────────────────────┐
│ 系统指令 2,000 tokens │
│ 对话历史 15,000 tokens │
│ 检索文档 80,000 tokens │
│ 用户问题 200 tokens │
├──────────────────────────────────────┤
│ 总计 97,200 tokens │
│ 模型限制 128,000 tokens │
│ 安全余量 30,800 tokens ✓ │
└──────────────────────────────────────┘
`
解决方案：
`python
def safe_context_assembly(parts: list[str], max_tokens: int = 100000) -> str:
“””安全的上下文组装，确保不超出限制”””
total = 0
assembled = []
for part in parts:
part_tokens = count_tokens(part)
if total + part_tokens > max_tokens:
# 截断当前部分以适应限制
remaining = max_tokens – total
if remaining > 100: # 至少保留100 tokens才有意义
truncated = truncate_to_tokens(part, remaining)
assembled.append(truncated + “n[…内容已截断]”)
break
assembled.append(part)
total += part_tokens
return “nn”.join(assembled)
`
8.2 陷阱二：信息稀释（Information Dilution）
问题：注入了太多低相关信息，淹没了真正有用的信息。
类比：这就像你在图书馆找一本关于”量子力学”的书，但管理员给了你一整书架的书——其中只有一两本是关于量子力学的，其他都是关于古典文学的。信息量很大，但有效信息被稀释了。
解决方案：
`python
def filter_by_relevance(results: list[SearchResult], threshold: float = 0.7) -> list[SearchResult]:
“””过滤低相关性结果”””
filtered = [r for r in results if r.score >= threshold]
# 如果过滤后结果太少，降低阈值
if len(filtered) < 2:
filtered = sorted(results, key=lambda x: x.score, reverse=True)[:3]
return filtered
def deduplicate_results(results: list[SearchResult], similarity_threshold: float = 0.95) -> list[SearchResult]:
“””去重：移除内容高度相似的结果”””
unique = []
for r in results:
is_dup = False
for u in unique:
if cosine_similarity(r.content, u.content) > similarity_threshold:
is_dup = True
break
if not is_dup:
unique.append(r)
return unique
`
8.3 陷阱三：幻觉（Hallucination）
问题：LLM在上下文信息不足时，倾向于”编造”看似合理但错误的回答。
解决方案：
`python
# 方案1：在System Prompt中明确约束
ANTI_HALLUCINATION_PROMPT = “””
你是一个严谨的知识助手。请严格遵守以下规则：
- 只基于【参考资料】中的信息回答问题
- 如果参考资料中没有相关信息，回答”根据已有资料，我无法回答这个问题”
- 不要编造、推测或补充参考资料中没有的信息
- 回答时引用具体的参考资料编号，格式：[参考X]
- 如果信息可能不完整或有歧义，请明确说明
“””
# 方案2：添加置信度评估
def query_with_confidence(question: str, context: str) -> dict:
response = llm.chat(f”””
基于以下参考资料回答问题，并评估你的置信度。
参考资料：{context}
问题：{question}
请以JSON格式回答：
{{
“answer”: “你的回答”,
“confidence”: 0.0-1.0,
“sources_used”: [“参考1”, “参考2”],
“gaps”: “信息不足之处（如有）”
}}
“””)
return json.loads(response)
`
陷阱速查表
| 陷阱 | 表现 | 解决方案 | 预防措施 |
|---|
|——|——|———|———|
| 上下文溢出 | API报错、回答截断 | Token计数+截断 | 设定安全上限 |
|---|---|---|---|
| 信息稀释 | 回答笼统、不准确 | 相关性过滤+重排序 | 提高检索精度 |
| 幻觉 | 编造不存在的信息 | 约束Prompt+置信度评估 | 提供充足上下文 |
| 检索偏差 | 检索到错误文档 | 混合搜索+查询改写 | 优化Embedding |
| 重复内容 | 同一信息多次出现 | 去重算法 | 分块策略优化 |
| 延迟过高 | 响应时间>5秒 | 缓存+异步处理 | 架构优化 |
九、总结
9.1 核心要点回顾
`
┌─────────────────────────────────────────────────────────┐
│ Context层设计要点 │
│ │
│ 1. Context是Agent的工作记忆，决定回答质量的上限 │
│ 2. 三大策略：RAG + 长上下文 + 动态注入，通常混合使用 │
│ 3. RAG四要素：分块策略 × Embedding模型 × 向量数据库 × │
│ 检索策略 │
│ 4. 长上下文不等于万能，成本和注意力衰减是主要限制 │
│ 5. 动态注入根据任务类型智能组装最优上下文 │
│ 6. 避免三大陷阱：溢出、稀释、幻觉 │
└─────────────────────────────────────────────────────────┘
`
9.2 Context层设计决策树
`
你的场景是什么？
│
├─ 需要查询大量文档/知识库？
│ ├─ 文档量 < 100页 → 直接塞入长上下文
│ └─ 文档量 > 100页 → RAG方案
│ ├─ 需要精确匹配 → 混合搜索（语义+关键词）
│ └─ 需要语义理解 → 语义搜索 + 重排序
│
├─ 需要长文档分析？
│ ├─ 文档 < 模型窗口 → 直接使用
│ └─ 文档 > 模型窗口 → 分段处理 + 摘要链
│
└─ 需要多轮复杂对话？
├─ 轮次 < 20轮 → 完整历史
└─ 轮次 > 20轮 → 滑动窗口 + 摘要压缩
`
9.3 下篇预告
在下一篇《AI PRO·Harness Day 5 Tool层设计：Memory层设计》中，我们将深入探讨：
- 短期记忆 vs 长期记忆：Agent如何记住和遗忘？
- 记忆的存储与检索：如何高效管理Agent的记忆库？
- 记忆的衰减与强化：如何模拟人类的记忆机制？
- 多Agent记忆共享：多个Agent之间如何共享知识？
- 实战：构建一个具有长期记忆的个人助理
Context层解决的是”当前对话需要什么知识”，而Memory层要解决的是”Agent如何积累和运用长期经验”。这两层紧密配合，共同构成了Agent的”认知基础设施”。
敬请期待！
系列文章
– 第四篇：Context层设计 ← 你在这里
*本文约10,000字，阅读时间约25分钟。如有疑问或建议，欢迎在评论区交流。*
发表回复