发布日期:2026-08-21
作者:小玉米 (Littlecorn AI)
标签:#AIAgent #上下文工程 #长程记忆 #LLM #生产级实践
随着 AI Agent 承担的任务愈发复杂和长期化,单一的 LLM 上下文窗口(即使达到 128K 甚至 1M Token)也难以承载完整的对话历史、工具调用记录和领域知识。上下文工程(Context Engineering) 正在成为继 Prompt Engineering 之后的新一代核心能力。本文深入探讨了 AI Agent 在长程任务中的上下文管理策略,包括分层记忆架构、动态上下文压缩、语义检索增强以及跨会话状态持久化的生产级工程实践。
在传统的 AI Agent 开发中,开发者往往将完整的对话历史、工具调用日志和系统提示词全部塞入 LLM 的上下文窗口。然而这种方式在长程任务中会遭遇三大核心瓶颈:
┌─────────────────────────────────────────────────────┐ │ AI Agent 长程记忆分层架构 │ ├─────────────────────────────────────────────────────┤ │ L0: Working Memory (当前推理窗口) ~8K-32K │ │ └─ 即时对话上下文 + 活跃工具结果 │ │ L1: Short-term Memory (滚动摘要) ~32K-128K │ │ └─ 近期会话压缩摘要 + 待执行任务队列 │ │ L2: Long-term Memory (持久向量库) 无上限 │ │ └─ 语义向量索引 + 知识图谱 + 技能库 │ │ L3: Episodic Memory (事件日志) 无上限 │ │ └─ 历史行动轨迹 + 决策回放 + 反馈记录 │ └─────────────────────────────────────────────────────┘
工作记忆是 Agent 当前推理窗口中可直接引用的信息子集。关键设计原则:
critical / normal / low 优先级标签,在窗口溢出时自动清除低优先级内容。每隔 N 轮对话(如 10 轮),自动将早期对话通过摘要模型压缩为结构化摘要:
import json
from datetime import datetime
from typing import List, Dict
class RollingMemorySummarizer:
def __init__(self, summarization_interval: int = 10):
self.interval = summarization_interval
self.conversation_buffer: List[Dict] = []
self.summaries: List[Dict] = []
def add_message(self, role: str, content: str):
self.conversation_buffer.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def should_summarize(self) -> bool:
return len(self.conversation_buffer) >= self.interval
async def summarize(self, llm_client) -> Dict:
"""使用轻量级 LLM 对缓冲区对话进行结构化摘要"""
messages_text = "\n".join(
f"[{m['role']}] {m['content']}" for m in self.conversation_buffer
)
prompt = f"""请将以下对话历史压缩为结构化摘要,提取关键决策、工具调用结果和待办事项:
{messages_text}
输出 JSON 格式:
{{"summary": "简短摘要", "key_decisions": [...], "tool_results": [...], "pending_tasks": [...]}}"""
response = await llm_client.complete(prompt, model="lightweight")
summary = json.loads(response)
self.summaries.append({
"segment_id": len(self.summaries),
"timestamp": datetime.now().isoformat(),
**summary
})
self.conversation_buffer.clear()
return summary
def get_context_block(self) -> str:
"""生成注入 L0 工作记忆的摘要块"""
if not self.summaries:
return ""
recent = self.summaries[-3:] # 仅保留最近3段摘要
blocks = []
for s in recent:
blocks.append(f"[摘要 #{s['segment_id']}] {s['summary']}")
return "\n".join(blocks)
长期记忆基于向量数据库(如 ChromaDB、Qdrant)实现语义检索:
除了滚动摘要,更细粒度的上下文压缩技术正在成为 Agent 上下文工程的关键环节:
| 策略 | 适用场景 | 压缩率 | 信息损失风险 |
|---|---|---|---|
| 全量摘要 | 长对话历史归档 | ~90% | 中 |
| 选择性保留 | 保留关键决策节点 | ~60% | 低 |
| 指针引用 | 替换原文为摘要指针 | ~75% | 低 |
| 语义去重 | 合并重复相似内容 | ~40% | 极低 |
| 渐进降级 | 从详述变为概述 | ~80% | 中 |
class ContextSelector:
"""根据任务相关性动态选择上下文条目"""
PRIORITY_WEIGHTS = {
"system_prompt": 1.0,
"user_instruction": 0.95,
"tool_result_critical": 0.85,
"assistant_decision": 0.80,
"tool_result_normal": 0.50,
"casual_conversation": 0.20,
}
def __init__(self, max_tokens: int = 32000):
self.max_tokens = max_tokens
def select(self, context_entries: List[Dict], current_task: str) -> List[Dict]:
scored = []
for entry in context_entries:
priority = self.PRIORITY_WEIGHTS.get(entry.get("type", "normal"), 0.3)
relevance = self._compute_relevance(entry["content"], current_task)
score = priority * 0.6 + relevance * 0.4
scored.append((score, entry))
scored.sort(key=lambda x: x[0], reverse=True)
selected, token_count = [], 0
for score, entry in scored:
entry_tokens = len(entry["content"]) // 4 # 粗略估算
if token_count + entry_tokens > self.max_tokens:
continue
selected.append(entry)
token_count += entry_tokens
return selected
def _compute_relevance(self, content: str, task: str) -> float:
"""简化版语义相关性评分"""
content_words = set(content.lower().split())
task_words = set(task.lower().split())
if not task_words:
return 0.0
overlap = len(content_words & task_words)
return min(overlap / len(task_words), 1.0)
对于多日运行的长程 Agent,会话之间的状态桥接至关重要:
import os
import json
from pathlib import Path
class AgentCheckpoint:
def __init__(self, workspace: str = ".agent_state"):
self.workspace = Path(workspace)
self.workspace.mkdir(parents=True, exist_ok=True)
def save(self, agent_state: Dict, session_id: str):
filepath = self.workspace / f"checkpoint_{session_id}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(agent_state, f, ensure_ascii=False, indent=2)
# 仅保留最近 5 个检查点
checkpoints = sorted(self.workspace.glob("checkpoint_*.json"), key=os.path.getmtime)
for old in checkpoints[:-5]:
old.unlink()
def load_latest(self) -> Dict | None:
checkpoints = sorted(self.workspace.glob("checkpoint_*.json"), key=os.path.getmtime)
if not checkpoints:
return None
with open(checkpoints[-1], "r", encoding="utf-8") as f:
return json.load(f)
上下文工程正在从"把什么放进 Prompt"演进为"如何管理 Agent 的完整记忆生命周期"。在 Littlecorn-ai 的生产环境实践中,分层记忆架构 + 动态压缩 + 跨会话持久化的组合策略,使 Agent 在持续运行数周的长程任务中仍能保持 95% 以上的关键决策召回率,同时将 LLM 推理成本降低约 60%。
未来,随着模型原生支持更长上下文窗口(如 Gemini 的 2M Token),上下文工程的核心将从"容量管理"转向"质量筛选"——在近乎无限的窗口中精准定位最关键的信息片段,这将是一个更加激动人心的工程挑战。