AI 智能体上下文层：架构 + 代码
每个构建过 AI 智能体的工程师都撞过同一堵墙。你的原型在五个测试用例上跑得漂亮，在 demo 里惊艳了干系人，可一旦遇到一个带着真实历史、跨越多个对话轮次、有真实问题的真实用户，它瞬间就散了架。
模型没坏。坏的是上下文。
2025 年中，业界开始围绕一个术语达成共识：上下文工程（context engineering）——即设计一套系统，在正确的时间、动态地把正确的信息、工具与知识组装进 AI 智能体的工作记忆。Gartner 宣布它是提示词工程的继任者。Shopify 的 CEO 称它是使用 AI 最重要的技能。每一个在生产环境部署智能体的团队，都心照不宣地认同：这早已是最难的部分。
这篇文章，是我希望自己刚开始构建那些需要跨会话、跨用户、跨工具去记忆、检索、推理和行动的智能体时，就能拥有的指南。我们会深入架构，然后亲手把它构建出来——用的是真实、可运行的 Python 代码，基于 Groq 极速推理 API，你今天就能改造用于生产。
1、什么是上下文层？
上下文层是介于用户和你的 LLM 之间的子系统，负责组装模型生成有用回复所需的一切。它回答的是一个看似简单的问题：此刻模型应该看到什么？
那个"一切"通常包括：
- 对话历史——本会话中到目前为止说过的话
- 检索到的知识——文档、FAQ、从外部来源拉取的数据
- 记忆笔记——跨交互积累下来的持久事实
- 工具定义与结果——智能体能做什么、已经做了什么
- 系统指令——智能体的角色设定、护栏与行为规则
没有设计良好的上下文层，你基本就是递给 LLM 一张白纸，却指望它写出一部小说。而有了它，你递给它的则是一份针对它即将处理的确切情境、精心组织的简报包。
上下文层不是提示词。提示词是上下文层的输出——也就是最终组装好、发送给模型的载荷。上下文层是决定"什么进载荷、以什么顺序、按什么优先级"的整个系统。
2、为什么上下文工程比模型选择更重要
经验丰富的智能体构建者会很快学到一条反直觉的真理：从一个上下文优秀的中端模型，切换到一个上下文糟糕的旗舰模型，通常会让结果更糟。一个带着错误信息的更聪明模型，会信心十足地给出错误答案。而一个上下文精确地正确的较小模型，往往能命中要害。
这正是我们在实现中使用 Groq 加 Llama 3.3 70B 的原因——它快、能力强，并且印证了这一点。上下文架构的重要性，胜过把钱砸在最大的模型上。
上下文工程之所以重要，是因为：
1. Token 预算是真实约束。 即便有了很大的上下文窗口，盲目把一切塞进去也会让你的智能体变慢、变贵，而且矛盾地变得更不准确。当上下文充满噪声时，模型在"大海捞针"类任务上的表现会退化。你需要像做手术一样精准。
2. 智能体的可靠性来自上下文，而非聪明。 当智能体失败时——漏掉用户偏好、重复已经回答过的问题、忘记某个工具结果——这几乎总是一个上下文组装问题，而不是推理问题。
3. 多步工作流会放大上下文错误。 在智能体式工作流中，模型调用工具、解读结果、并在循环中决定下一步；第 2 步的一个小上下文错误，到第 5 步就会变成灾难性失败。
4. 成本随上下文浪费而放大。 你发给模型的每一个无关 token 都在消耗金钱和延迟。在规模化下，粗糙的上下文工程能让你的推理账单膨胀 10 倍。
3、核心架构：编译器流水线
这里有一个能把干净的上下文层和一团乱麻区分开的关键架构洞察：把上下文组装当作一组可组合的编译器阶段（compiler stage）流水线来处理。
每个阶段转换同一个 ContextPayload 对象——加载静态内容、压缩对话历史、注入检索到的文档，最终组装成 API 就绪的消息数组。各个阶段相互独立、可测试、可替换。
这与"构建一个巨大提示词字符串"的整体式思路有着本质不同的哲学。每个编译器阶段只承担单一职责，由流水线按顺序编排它们。要新增一个上下文来源——比如用户画像注入器或护栏检查器——意味着写一个类、插入流水线。其它什么都不用动。
我们来构建它。
3.1 数据结构：ContextItem 与 ContextPayload
每个上下文层都需要清晰、有类型的数据结构。我们的核心是两个：ContextItem（对话中的单条消息）和 ContextPayload（流经流水线的完整信息包）。
import os
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Any
from groq import Groq
import tiktoken
class ContentRole(Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL = "tool"
@dataclass
class ContextItem:
role: ContentRole
content: str
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ContextPayload:
system_prompt: str
tool_definitions: List[Dict[str, Any]] = field(default_factory=list)
conversation_history: List[ContextItem] = field(default_factory=list)
tool_outputs: Dict[str, str] = field(default_factory=dict)
retrieved_chunks: List[str] = field(default_factory=list)
memory_notes: List[str] = field(default_factory=list)
max_tokens: int = 32000
system_reserve_pct: float = 0.15
history_reserve_pct: float = 0.50
retrieval_reserve_pct: float = 0.25
overflow_strategy: str = "summarize"
几个值得点出的设计决策：
ContextPayload 自带预算。max_tokens、system_reserve_pct、history_reserve_pct 和 retrieval_reserve_pct 这几个字段，意味着每个编译器阶段都能做出有预算意识的决策，而无需外部配置。载荷就是阶段之间的契约。
Token 预算是按百分比的，而非固定值。 这让系统具备适应性。从 32K 上下文模型切换到 8K 模型？改一个数字——max_tokens——每个阶段就自动调整自己的分配。系统提示占 15%，对话历史占 50%，检索占 25%，剩下 10% 给工具定义和开销留了余地。
overflow_strategy 字段是面向未来的。现在我们用 "summarize"——当上下文溢出时，通过一次 LLM 调用压缩较早的对话轮次。但这个字段让实现替代方案变得轻而易举，比如 "truncate"（丢弃最旧的消息）或 "prioritize"（打分并只保留最相关的轮次）。
3.2 Token 工具：计量真正要紧的东西
你无法对无法计量的东西做预算。准确的 token 计数，是流水线的每一个决策的基础。
class TokenHelper:
@staticmethod
def count_tokens(text: str, model: str = "gpt-4o") -> int:
try:
enc = tiktoken.encoding_for_model(model)
except Exception:
enc = tiktoken.get_encoding("cl100k_base")
return len(enc.encode(text))
这是一个体量很小、重要性却极大的类。每个编译器阶段都用它来回答同一个问题："这段内容能放进我分配到的预算吗？"
我们这里用了 tiktoken，它为 OpenAI 兼容的 tokenizer 提供精确的 token 计数。cl100k_base 回退覆盖了那些没有 tiktoken 原生编码的模型（比如 Llama）——对于预算计算，这是一个足够接近的近似。在生产中，你可以把它换成目标模型的精确 tokenizer，但 cl100k_base 带来 5–10% 的方差，很少是压垮你系统的那根稻草。
关键洞察：在生产中永远不要用 len(text.split()) 来估算 token。对于代码、非英文文本和技术内容，词数和 token 数差异巨大。一个满是标点和短键的 JSON blob，其 token 数可能是你仅按词数猜测的两倍。
3.3 编译器流水线模式
这是架构的脊梁。我们流水线中的每个阶段都实现一个统一接口——compile(payload) → payload——这让系统可组合、可测试。
class ContextCompiler:
"""Base class for all pipeline stages."""
def compile(self, payload: ContextPayload):
raise NotImplementedError
class Pipeline:
"""Runs a sequence of compilers over a shared payload."""
def __init__(self, compilers: List[ContextCompiler]):
self.compilers = [c for c in compilers if c is not None]
def run(self, payload: ContextPayload):
for compiler in self.compilers:
payload = compiler.compile(payload)
return payload
这就是把 Unix 管道哲学应用到了上下文工程上。每个编译器读取载荷、转换它、再传给下游。Pipeline 类只是负责迭代。
为什么这比整体式组装器更好？三个原因：
可测试性。 你可以单独对每一阶段做单元测试。给 ConversationCompactor 喂一个含 100 条消息的载荷，验证它输出时处于预算之内——无需牵涉检索、工具或 LLM。
可组合性。 需要加一个在触达模型前抹除 PII 的护栏阶段？写一个类，插入流水线。需要为某些查询类型跳过检索？条件性地排除 RetrievalInjector。系统的其余部分浑然不知，也无需关心。
可调试性。 当你的智能体给出糟糕回复时，你可以检查每个阶段之后的载荷，精确定位上下文究竟在哪一步出了错。是系统提示太长？压缩器把某个关键细节摘要掉了？还是检索注入了无关片段？每个阶段都是一个检查点。
3.4 静态内容加载器
最简单的一个阶段——它把系统提示和工具定义加载到载荷上。但即便简单的阶段，显式化也有好处：
class StaticContentLoader(ContextCompiler):
def __init__(self, system_prompt: str, tools: List[dict]):
self.system_prompt = system_prompt
self.tools = tools
def compile(self, payload: ContextPayload):
payload.system_prompt = self.system_prompt
payload.tool_definitions = self.tools
return payload
你可能会想：何必把两个赋值包进一个类？因为它确立了模式。每一个上下文来源——无论是一段静态字符串，还是一个复杂的检索流水线——都通过同一个接口进入载荷。当你日后需要给系统提示做版本管理、做 A/B 测试，或从配置服务加载它们，你只改这一个类。流水线不动。
3.5 对话压缩器：聪明的记忆管理
这里开始有趣了。对话压缩器处理的是上下文管理中最难的一个问题：在契合 token 预算的同时，让对话历史保持信息量。
策略是：保留最近的消息原文（它们包含模型需要的即时上下文），并用一次快速、便宜的 LLM 调用把较早的消息压缩成一个精简块。
class ConversationCompactor(ContextCompiler):
def __init__(
self,
groq_client: Groq,
preserve_recent: int = 6,
summary_model: str = "llama3-8b-8192",
):
self.client = groq_client
self.preserve_recent = preserve_recent
self.summary_model = summary_model
def compile(self, payload: ContextPayload):
history = payload.conversation_history
if not history:
return payload
# Calculate the token budget for conversation history
history_budget = int(payload.max_tokens * payload.history_reserve_pct)
current_tokens = sum(
TokenHelper.count_tokens(item.content) for item in history
)
# If we're within budget, pass through unchanged
if current_tokens <= history_budget:
return payload
# Split into recent (keep verbatim) and older (summarize)
recent = history[-self.preserve_recent :]
older = history[: -self.preserve_recent]
if not older:
return payload
# Summarize the older messages via a fast LLM call
summary = self._summarize(older)
summary_item = ContextItem(
role=ContentRole.SYSTEM,
content=f"[PRIOR CONVERSATION SUMMARY]\n{summary}",
metadata={"compacted": True},
)
payload.conversation_history = [summary_item] + recent
return payload
def _summarize(self, items: List[ContextItem]) -> str:
transcript = "\n".join(
f"[{item.role.value}] {item.content}" for item in items
)
response = self.client.chat.completions.create(
model=self.summary_model,
messages=[
{
"role": "system",
"content": (
"Summarize this conversation while preserving key facts, "
"decisions, goals, and unresolved questions."
),
},
{"role": "user", "content": transcript},
],
temperature=0.0,
)
return response.choices[0].message.content.strip()
这里有几个架构选择值得解释：
摘要模型刻意用小而便宜的。 我们用 llama3-8b-8192 做摘要——而不是主力的 70B 智能体模型。摘要不需要旗舰级推理，它需要的是快且忠实。在 Groq 的基础设施上，一个 8B 模型在毫秒间返回。这让压缩成本可以忽略不计。
preserve_recent 默认为 6 条消息（3 个用户—助手轮次）。这个数字来自一个实际观察：模型几乎总是需要最近 2–3 次交换来维持对话连贯性。较早的上下文重要的是讨论过什么，而非怎么措辞的——而这恰恰是一个摘要能捕捉到的。
压缩由预算触发，而非由消息数触发。 一段有 20 条短消息的对话可能仍在预算内，完全跳过压缩。一段有 8 条很长消息的对话可能早早触发它。预算才是真相来源，而不是某个任意阈值。
摘要以 SYSTEM 角色的消息注入。这是刻意的——它向模型发出信号：这是背景上下文，而不是需要它回应的对话轮次。[PRIOR CONVERSATION SUMMARY] 标签让这一点显式化。
这就是你的短期记忆 vs. 长期记忆的切分在起作用。最近的消息是短期记忆——高保真、原文、昂贵。摘要则是被压缩的长期记忆——有损、便宜，但保留了本质信号。
3.6 检索注入器：你的 RAG 层
检索注入器让你的智能体得以访问对话之外的知识。这就是 RAG（检索增强生成）进入流水线的地方——而要把它做对，意味着在每一步都尊重 token 预算。
class RetrievalInjector(ContextCompiler):
def __init__(self, retriever, top_k: int = 5):
self.retriever = retriever
self.top_k = top_k
def compile(self, payload: ContextPayload):
if not self.retriever:
return payload
# Extract the latest user query as the search key
query = self._last_user_query(payload.conversation_history)
if not query:
return payload
# Calculate retrieval budget
retrieval_budget = int(payload.max_tokens * payload.retrieval_reserve_pct)
# Search and rank
chunks = self.retriever.search(query)[: self.top_k]
# Greedily select chunks that fit within budget
selected = []
used_tokens = 0
for chunk in chunks:
tokens = TokenHelper.count_tokens(chunk)
if used_tokens + tokens <= retrieval_budget:
selected.append(chunk)
used_tokens += tokens
else:
break
payload.retrieved_chunks = selected
return payload
@staticmethod
def _last_user_query(history):
for item in reversed(history):
if item.role == ContentRole.USER:
return item.content
return None
设计遵循三条原则：
查询提取是自动的。 注入器向后扫描对话，找到最近的用户消息并把它作为搜索查询。无需外部接线——流水线是自包含的。
预算执行是贪心的。 片段按排名顺序加入，直到预算耗尽。这保证了最相关的片段总能进来，较不相关的则被优雅地丢弃。没有硬性失败，没有句子中间被截断的片段。
检索器是可插拔的接口。 RetrievalInjector 不在乎你的检索器用的是向量嵌入、BM25 关键词搜索，还是混合方案。它只调用 .search(query)，并期望得到一个排好序的字符串列表。这种分离至关重要——你可以从一个简单的关键词检索器起步，再升级到完整的向量流水线，而无需触碰上下文层。
下面是一个用于开发和测试的简单检索器：
class SimpleRetriever:
"""Keyword-based retriever for development. Swap for a vector
store (Qdrant, Pinecone, pgvector, ChromaDB) in production."""
def __init__(self, documents: List[str]):
self.documents = documents
def search(self, query: str):
query_terms = set(query.lower().split())
scored = []
for doc in self.documents:
score = sum(1 for term in query_terms if term in doc.lower())
scored.append((score, doc))
scored.sort(reverse=True, key=lambda x: x[0])
return [doc for score, doc in scored if score > 0]
在生产中，你会把它替换成类似这样的东西：
# Production retriever example (pseudocode)
class VectorRetriever:
def __init__(self, collection, embedding_model):
self.collection = collection # Qdrant / Pinecone / pgvector
self.embedding_model = embedding_model # sentence-transformers
def search(self, query: str, top_k: int = 10) -> List[str]:
query_embedding = self.embedding_model.encode(query)
results = self.collection.search(
query_vector=query_embedding,
limit=top_k,
)
return [hit.payload["text"] for hit in results]
上下文层不变。变的只有检索器。
3.7 最终组装器：构建 API 载荷
最终组装器是流水线中的最后一个阶段。它把前面各阶段产出的一切——系统提示、检索到的片段、记忆笔记、对话历史——编译成 LLM API 期望的精确 messages 数组。
class FinalAssembler(ContextCompiler):
def compile(self, payload: ContextPayload):
messages = []
# Build the system message: base prompt + retrieved context + memory
system_text = payload.system_prompt
if payload.retrieved_chunks:
system_text += (
"\n\n[RETRIEVED CONTEXT]\n"
+ "\n---\n".join(payload.retrieved_chunks)
+ "\n[END RETRIEVED CONTEXT]"
)
if payload.memory_notes:
system_text += (
"\n\n[MEMORY NOTES]\n"
+ "\n".join(f"- {m}" for m in payload.memory_notes)
)
messages.append({"role": "system", "content": system_text})
# Append conversation history
for item in payload.conversation_history:
messages.append(
{
"role": item.role.value,
"content": item.content,
}
)
# Final safety net: hard-truncate if still over budget
total_tokens = TokenHelper.count_tokens(str(messages))
while total_tokens > payload.max_tokens and len(messages) > 2:
messages.pop(1) # remove oldest non-system message
total_tokens = TokenHelper.count_tokens(str(messages))
return messages
注意这里针对上下文溢出的分层防御：
第 1 层（上游）： 对话压缩器和检索注入器已经各自执行了它们的预算。
第 2 层（组装）： 检索到的片段和记忆笔记被折进系统消息，保持消息数低、结构对 API 干净。
第 3 层（安全网）： 底部的 while 循环是一个硬性兜底。如果尽管上游做了所有预算控制，总量仍然超过 max_tokens，它就一条条丢弃最旧的非系统消息。这极少触发——但一旦触发，你的智能体是优雅降级，而不是因 token 超限错误而崩溃。
检索到的上下文和记忆笔记被注入到系统消息内部，包裹在清晰的定界符中（[RETRIEVED CONTEXT] / [END RETRIEVED CONTEXT] 以及 [MEMORY NOTES]）。这给模型提供了无歧义的信号：哪些是接地知识、哪些是对话上下文——这个小小的细节，能切实地改善回复质量。
3.8 上下文层：编排一切
现在我们把流水线接入一个完整的 ContextLayer 类，它编译上下文并运行智能体循环：
class ContextLayer:
def __init__(
self,
system_prompt: str,
tools: List[dict],
retriever=None,
groq_api_key: Optional[str] = None,
model: str = "llama3-70b-8192",
max_tokens: int = 32000,
):
self.model = model
self.client = Groq(
api_key=groq_api_key or os.getenv("ABCD")
)
# Build the compiler pipeline
compilers = [
StaticContentLoader(system_prompt, tools),
ConversationCompactor(
groq_client=self.client,
preserve_recent=6,
),
]
if retriever:
compilers.append(RetrievalInjector(retriever=retriever))
compilers.append(FinalAssembler())
self.pipeline = Pipeline(compilers)
self.max_tokens = max_tokens
def compile(self, conversation_history: List[ContextItem]):
"""Run the full pipeline and return API-ready messages."""
payload = ContextPayload(
system_prompt="",
conversation_history=conversation_history,
max_tokens=self.max_tokens,
)
return self.pipeline.run(payload)
def run_agent(
self,
user_message: str,
conversation_history: Optional[List[ContextItem]] = None,
):
"""Process a user message through the full context pipeline
and return the agent's response + updated history."""
history = conversation_history or []
# Add the new user message
history.append(
ContextItem(role=ContentRole.USER, content=user_message)
)
# Compile context through the pipeline
messages = self.compile(history)
# Call Groq
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.2,
)
assistant_text = response.choices[0].message.content.strip()
# Append the assistant's response to history
history.append(
ContextItem(role=ContentRole.ASSISTANT, content=assistant_text)
)
return assistant_text, history
这是整个系统的公开 API。调用者只与两个方法交互：
compile()——当你想在不发起 LLM 调用的情况下检查组装好的上下文时使用。对于调试、测试和日志 invaluable。
run_agent()——完整的循环。接收一条用户消息，组装上下文，调用模型，返回回复和更新后的历史。
ContextLayer 的构造函数就是流水线组合发生的地方。注意检索器是如何被条件性包含的——如果你不传，流水线就直接跳过那个阶段。系统适应它的配置，而无需把 if/else 分支散落在代码各处。
3.9 整合起来：完整可运行示例
下面是整个系统端到端运行：
if __name__ == "__main__":
# A small knowledge base for RAG
docs = [
"Context engineering is the systematic design of information "
"flow into LLMs, including memory, retrieval, tools, and "
"conversation state.",
"Prompt engineering focuses mainly on crafting prompts, while "
"context engineering manages the full system context architecture.",
"RAG systems enhance model performance by injecting external "
"retrieved knowledge into prompts.",
]
retriever = SimpleRetriever(docs)
# Initialize the context layer
layer = ContextLayer(
system_prompt=(
"You are a helpful research assistant. Use retrieved context "
"when available; otherwise answer clearly from your own "
"knowledge."
),
tools=[],
retriever=retriever,
model="llama-3.3-70b-versatile",
)
history = []
# Turn 1
response, history = layer.run_agent(
"What is context engineering?",
history,
)
print("Agent:", response)
# Turn 2 — the agent maintains conversational continuity
response, history = layer.run_agent(
"How is it different from prompt engineering?",
history,
)
print("Agent:", response)
当你运行它时，下面是第二轮在底层发生的事情：
- StaticContentLoader 在载荷上设置系统提示和工具定义。
- ConversationCompactor 检查 4 条消息的历史（2 用户 + 2 助手）是否落在 50% 的历史预算（约 16K token）内。落在了，所以原样通过。
- RetrievalInjector 提取最近的用户查询（"How is it different from prompt engineering?"），搜索文档集合，找到比较提示词工程与上下文工程的那个片段。它落在 25% 的检索预算内，所以被加入
retrieved_chunks。 - FinalAssembler 把系统提示 + 检索片段编译成一条系统消息，追加对话历史，检查总 token 数，返回 API 就绪的消息数组。
- ContextLayer.run_agent 把它发给 Groq 的 Llama 3.3 70B，拿到回复，追加进历史，返回。
整个流水线在 LLM 调用开始之前，就在毫秒间执行完毕。在 Groq 的基础设施上，这条流程的端到端延迟通常不到 2 秒。
4、扩展系统：记忆笔记及更多
ContextPayload 已经支持 memory_notes——跨会话存活的持久事实。下面是你如何接入它们：
# Extend the ContextLayer with persistent memory
class ContextLayerWithMemory(ContextLayer):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._memory_notes: List[str] = []
def add_memory(self, note: str):
"""Store a persistent memory note."""
self._memory_notes.append(note)
def compile(self, conversation_history: List[ContextItem]):
payload = ContextPayload(
system_prompt="",
conversation_history=conversation_history,
max_tokens=self.max_tokens,
memory_notes=self._memory_notes,
)
return self.pipeline.run(payload)
现在你可以这样用：
layer = ContextLayerWithMemory(
system_prompt="You are a helpful research assistant.",
tools=[],
retriever=retriever,
model="llama-3.3-70b-versatile",
)
layer.add_memory("User prefers concise, bullet-point answers.")
layer.add_memory("User is building a customer support agent for e-commerce.")
response, history = layer.run_agent("How should I structure my retrieval?", history)
# The model now sees those memory notes and tailors its answer accordingly
在生产中，你会用一个以用户 ID 为键的数据库来支撑它，在每个会话开始时加载相关的记忆笔记。FinalAssembler 已经处理了格式化——它把它们注入到系统消息的 [MEMORY NOTES] 头下。
同样的模式可以扩展到任何新的上下文来源：用户画像、护栏、工具结果、外部 API 数据。写一个 ContextCompiler 子类，插入流水线，完事。
5、性能与扩展考量
上下文层会很快撞上真实的扩展问题。下面这些才是要紧的：
用小模型做压缩。 ConversationCompactor 用的是 Llama 3 8B——而非主力的 70B 模型——来做摘要。这是刻意的。摘要不需要旗舰级推理，它需要的是速度和忠实。在 Groq 上，8B 模型在 200 毫秒内返回。把昂贵模型留给需要它的任务。
Token 计数足够快。 tiktoken 在单 CPU 核上以约 4M token/秒的速度编码。对于一个 32K token 的上下文，那就是约 8 毫秒。不是免费，但相比 LLM 调用可以忽略。如果这对你的热路径来说还是太慢，可以按消息缓存 token 数，只在内容变化时重算。
预算百分比胜过固定分配。 当你切换模型（比如从 32K 上下文模型到 128K 上下文模型），固定 token 分配需要到处手动重新校准。百分比会自动缩放。无论总预算多少，系统提示始终 15%、历史 50%、检索 25%。
实现优雅降级。 FinalAssembler 的 while 循环是安全网——但你也应该处理检索器故障。如果你的向量数据库慢或挂了，RetrievalInjector 应该超时并返回一个空片段列表，而不是让流水线崩溃。智能体带着缩减的上下文继续运行，总好过没有回复。
激进缓存。 系统提示和工具定义在轮次之间不变。检索结果在会话内对相同查询通常可以缓存。只有对话历史保证每个轮次都变。
6、常见错误与规避方法
在构建和审查了几十个智能体系统之后，下面是我最常看到的错误：
错误 1：把上下文窗口当作无限的。 即便很大的上下文窗口也要花钱、也增加延迟。更糟的是，检索精度在极长上下文中会退化。这就是为什么我们的系统按百分比给 token 做预算——为约束而设计，而非为充裕。
错误 2：摘要过于激进。 每条消息之后都压缩会摧毁细微差别。我们的压缩器只在预算被超出时才触发，并且保留最近 6 条消息原文。摘要是有损的——把它当作一个泄压阀，而不是默认值。
错误 3：忽视系统提示的 token 成本。 一个 5,000 token 的系统提示，在每次调用中都吃掉 32K 预算的 15%。我们的架构让这一点可见——system_reserve_pct 字段迫使你有意识地给它分配空间。保持系统提示精简，把详细指令移到检索文档里，只在相关时才取用。
错误 4：不用真实的对话长度测试。 你的上下文层应该用 50+ 条消息的对话测试，而不是三轮的 demo。围绕压缩、检索和 token 限制的边界情况，只有在真实规模下才会浮现。构建一个测试工具，模拟一段 30 轮对话，并在每一步检查编译好的载荷。
错误 5：在整体式函数里混搭关注点。 这正是流水线模式存在的原因。当你把检索、记忆、摘要和组装塞进一个 200 行的函数，你就失去了单独测试、缓存和扩展各组件的能力。每个阶段都应该是可独立部署、可独立测试的。
错误 6：忘记检索质量。 最好的上下文架构也修不好糟糕的检索。如果你的片段太大、太小，或在句子中间被切断，模型得到的是噪声上下文，产出的是噪声答案。独立测试你的检索流水线——在责怪 LLM 之前，先量一下 recall@5 和 precision@5。
7、结束语
模型会持续变好。上下文窗口会持续变大。但那个根本性的挑战——决定模型此刻、为了这个具体任务、需要看到什么——不会消失。如果说有什么变化，那就是它正成为"能用的智能体系统"和"只在 demo 里惊艳的智能体系统"之间的首要区分因素。
那些把上下文当作一等工程要务的团队——有显式预算、有类型化载荷、有可组合的流水线阶段、有生产级检索——将交付出感觉智能、响应迅速、可靠可信的智能体。而那些把一切倒进一个巨大提示词、指望模型自己理顺的团队，会一直纳闷：为什么他们的智能体又慢、又贵、又不一致。
我们在这里构建的流水线模式，给了你一个可以向任意方向扩展的基础：
- 加一个
UserProfileInjector阶段，从数据库加载用户偏好，并把它加入载荷的记忆笔记。 - 加一个
GuardrailStage，在组装好的上下文触达模型之前，扫描其中是否有 PII 或策略违规。 - 加一个
ToolResultCompactor，在冗长的工具输出消耗你的历史预算之前，先把它摘要。 - 用向量库替换
SimpleRetriever——流水线不变，变的只是检索器的实现。
每一次扩展都是一个类、流水线构造函数里的一行。这就是可组合架构的力量。
上下文层，是工程与智能交汇之处。把它构建好。
原文链接: How to Design a Context Layer for Your AI Agent: Architecture + Code
汇智网翻译整理，转载请标明出处