第6章 Agent的记忆系统
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch05-memory.html 所属：第二篇-Agent的大脑
让 Agent 拥有短期的专注和长期的经验 📝 先看一个最小记忆例子
如果用户第一轮说"我喜欢简洁风格",第二轮说"帮我写周报",没有记忆的 Agent 会重新追问风格;有记忆的 Agent 会直接按简洁风格输出。
所以你可以把本章先简单理解成一句话:记忆不是为了存很多内容,而是为了让 Agent 在下一轮做出更一致、更省心的决定。
6.1 没有记忆的 Agent 会怎样?
想象一个失忆的助手:你上午告诉他你喜欢简洁的报告,下午他又问你"你需要什么风格?";你刚给了他一份资料,下一秒他就忘了。
这就是没有记忆的 Agent。它只能处理单轮、即时的请求,无法积累经验、无法保持一致性、无法处理需要跨轮引用的复杂任务。 💡 记忆的本质
记忆让 Agent 从"活在当下"变成"有过去、有经验"。它影响 Agent 的连贯性(同一对话内的一致性)、个性化(记住用户偏好)和学习进化(从历史中优化)。
6.2 三层记忆架构
人类有感觉记忆、短期记忆、长期记忆三层结构。Agent 也有类似的设计: | 类型 | 类比 | 实现 | 容量 | 持久性 | | --- | --- | --- | --- | --- | | 短期记忆 | 工作台上的文件 | LLM 上下文窗口 | 4K~200K token | 对话期间 | | 工作记忆 | 便签纸 | Scratchpad / 变量 | 任务相关 | 任务期间 | | 长期记忆 | 图书馆/档案室 | 向量数据库 | 理论无限 | 永久 | ## 6.3 短期记忆的管理策略
短期记忆 = LLM 上下文窗口。对话越长,上下文越大。但窗口有上限,必须管理:
四种管理策略
1
滑窗截断(Sliding Window)
保留最近 N 轮对话,丢弃最旧的消息。简单粗暴,但可能丢失关键早期信息。
← 1 / 4 → 重播
6.4 长期记忆:向量数据库
长期记忆让 Agent 能回忆起几天前甚至几个月前的交互。核心技术是向量数据库 + 语义检索。
主流向量数据库对比 | 数据库 | 类型 | 特点 | 适用场景 | | --- | --- | --- | --- | | Chroma | 嵌入式 | 轻量、Python原生、零配置 | 原型开发、小规模 | | Pinecone | 云服务 | 全托管、高性能、易扩展 | 生产环境、大规模 | | Weaviate | 自部署 | 支持混合搜索、GraphQL | 需要复杂查询 | | Milvus | 分布式 | 高可用、支持十亿级向量 | 超大规模企业 | | pgvector | PostgreSQL扩展 | SQL生态、事务支持 | 已有PG基础设施 | ## 6.5 上下文压缩实战
在长任务中,Agent 的上下文窗口会不断膨胀--每次工具调用返回结果、每次推理步骤、每次对话轮次都在累积 token。这种现象称为上下文膨胀(Context Bloat),也叫上下文腐烂(Context Rot)。 ⚠️ Context Rot 的典型症状
- 行为漂移:Agent 早期遵循的规则(如"用中文回答"),在 50 轮对话后可能被遗忘
- 注意力稀释:上下文中有太多无关历史,LLM 的注意力被分散,输出质量下降
- 成本飙升:每次调用都要处理全部历史 token,费用随对话长度线性增长
- 窗口溢出:超过模型最大上下文长度后,被迫截断或报错
五种压缩策略详解
1
截断(Truncation)
最简单的方案:超过窗口上限时,直接丢弃最早的消息。实现成本为零,但会导致行为不一致--Agent 可能忘记用户在第一轮设定的约束条件。
← 1 / 5 → 重播
压缩前后 Token 对比
# ===== 模拟一次 30 轮对话的上下文膨胀 =====
# 原始对话(无压缩)
raw_tokens_per_turn = 150 # 每轮平均 150 token
turns = 30
total_raw = raw_tokens_per_turn * turns + 500 # +500 为 System Prompt
print(f"原始上下文: {total_raw} tokens") # 输出: 5000 tokens
# ----- 策略 1: 截断 -----
# 保留最近 10 轮,丢弃其余
truncated = raw_tokens_per_turn * 10 + 500
print(f"截断后: {truncated} tokens") # 输出: 2000 tokens
# 损失: 20 轮对话完全丢失,包含用户初始约束条件
# ----- 策略 2: 摘要 -----
# 旧 20 轮压缩为一段摘要(约 200 token)
summarized = 200 + raw_tokens_per_turn * 10 + 500
print(f"摘要后: {summarized} tokens") # 输出: 2200 tokens
# 损失: 细节丢失,但保留了关键决策和偏好
# ----- 策略 3: 滑动窗口 -----
# 等同于截断策略(按时间维度)
sliding = truncated
print(f"滑动窗口后: {sliding} tokens") # 输出: 2000 tokens
# ----- 策略 4: 选择性保留 -----
# 从 30 轮中标记 8 轮为"重要",其余丢弃
selected = raw_tokens_per_turn * 8 + 500 + 150 # +150 为类型标签开销
print(f"选择性保留后: {selected} tokens") # 输出: 1850 tokens
# 信息密度最高,但可能遗漏看似不重要实则关键的上下文
# ----- 策略 5: 语义保护型压缩 -----
# System Prompt 全量 + 6 个关键行动-结果对 + 12 轮推理压缩为 300 token
semantic = 500 + raw_tokens_per_turn * 6 + 300 + 100 # +100 为因果链标注
print(f"语义保护型后: {semantic} tokens") # 输出: 1800 tokens
# 最低压缩率 + 最高信息完整性
# ===== 效果总结 =====
print("\n--- 压缩效果对比 ---")
strategies = {
"原始": total_raw,
"截断": truncated,
"摘要": summarized,
"滑动窗口": sliding,
"选择性保留": selected,
"语义保护型": semantic
}
for name, tokens in strategies.items():
ratio = tokens / total_raw
print(f"{name:10s}: {tokens} tokens (压缩率 {ratio:.1%})")
# 输出:
# 原始 : 5000 tokens (压缩率 100.0%)
# 截断 : 2000 tokens (压缩率 40.0%)
# 摘要 : 2200 tokens (压缩率 44.0%)
# 滑动窗口 : 2000 tokens (压缩率 40.0%)
# 选择性保留: 1850 tokens (压缩率 37.0%)
# 语义保护型: 1800 tokens (压缩率 36.0%)
// ===== 模拟一次 30 轮对话的上下文膨胀 =====
// 原始对话(无压缩)
const rawTokensPerTurn = 150; // 每轮平均 150 token
const turns = 30;
const totalRaw = rawTokensPerTurn * turns + 500; // +500 为 System Prompt
console.log(`原始上下文: ${totalRaw} tokens`); // 输出: 5000 tokens
// ----- 策略 1: 截断 -----
const truncated = rawTokensPerTurn * 10 + 500;
console.log(`截断后: ${truncated} tokens`); // 输出: 2000 tokens
// ----- 策略 2: 摘要 -----
const summarized = 200 + rawTokensPerTurn * 10 + 500;
console.log(`摘要后: ${summarized} tokens`); // 输出: 2200 tokens
// ----- 策略 3: 滑动窗口 -----
const sliding = truncated;
console.log(`滑动窗口后: ${sliding} tokens`); // 输出: 2000 tokens
// ----- 策略 4: 选择性保留 -----
const selected = rawTokensPerTurn * 8 + 500 + 150;
console.log(`选择性保留后: ${selected} tokens`); // 输出: 1850 tokens
// ----- 策略 5: 语义保护型压缩 -----
const semantic = 500 + rawTokensPerTurn * 6 + 300 + 100;
console.log(`语义保护型后: ${semantic} tokens`); // 输出: 1800 tokens
// ===== 效果总结 =====
console.log("\n--- 压缩效果对比 ---");
const strategies: Record = {
"原始": totalRaw,
"截断": truncated,
"摘要": summarized,
"滑动窗口": sliding,
"选择性保留": selected,
"语义保护型": semantic
};
for (const [name, tokens] of Object.entries(strategies)) {
const ratio = tokens / totalRaw;
console.log(`${name.padEnd(10)}: ${tokens} tokens (压缩率 ${(ratio * 100).toFixed(1)}%)`);
}
// 输出:
// 原始 : 5000 tokens (压缩率 100.0%)
// 截断 : 2000 tokens (压缩率 40.0%)
// 摘要 : 2200 tokens (压缩率 44.0%)
// 滑动窗口 : 2000 tokens (压缩率 40.0%)
// 选择性保留: 1850 tokens (压缩率 37.0%)
// 语义保护型: 1800 tokens (压缩率 36.0%)
package main
import "fmt"
func main() {
// ===== 模拟一次 30 轮对话的上下文膨胀 =====
// 原始对话(无压缩)
rawTokensPerTurn := 150 // 每轮平均 150 token
turns := 30
totalRaw := rawTokensPerTurn*turns + 500 // +500 为 System Prompt
fmt.Printf("原始上下文: %d tokens\n", totalRaw) // 输出: 5000 tokens
// ----- 策略 1: 截断 -----
truncated := rawTokensPerTurn*10 + 500
fmt.Printf("截断后: %d tokens\n", truncated) // 输出: 2000 tokens
// ----- 策略 2: 摘要 -----
summarized := 200 + rawTokensPerTurn*10 + 500
fmt.Printf("摘要后: %d tokens\n", summarized) // 输出: 2200 tokens
// ----- 策略 3: 滑动窗口 -----
sliding := truncated
fmt.Printf("滑动窗口后: %d tokens\n", sliding) // 输出: 2000 tokens
// ----- 策略 4: 选择性保留 -----
selected := rawTokensPerTurn*8 + 500 + 150
fmt.Printf("选择性保留后: %d tokens\n", selected) // 输出: 1850 tokens
// ----- 策略 5: 语义保护型压缩 -----
semantic := 500 + rawTokensPerTurn*6 + 300 + 100
fmt.Printf("语义保护型后: %d tokens\n", semantic) // 输出: 1800 tokens
// ===== 效果总结 =====
fmt.Println("\n--- 压缩效果对比 ---")
strategies := []struct {
name string
tokens int
}{
{"原始", totalRaw},
{"截断", truncated},
{"摘要", summarized},
{"滑动窗口", sliding},
{"选择性保留", selected},
{"语义保护型", semantic},
}
for _, s := range strategies {
ratio := float64(s.tokens) / float64(totalRaw)
fmt.Printf("%-10s: %d tokens (压缩率 %.1f%%)\n", s.name, s.tokens, ratio*100)
}
// 输出:
// 原始 : 5000 tokens (压缩率 100.0%)
// 截断 : 2000 tokens (压缩率 40.0%)
// 摘要 : 2200 tokens (压缩率 44.0%)
// 滑动窗口 : 2000 tokens (压缩率 40.0%)
// 选择性保留: 1850 tokens (压缩率 37.0%)
// 语义保护型: 1800 tokens (压缩率 36.0%)
}
import java.util.LinkedHashMap;
import java.util.Map;
public class TokenCompressionComparison {
public static void main(String[] args) {
// ===== 模拟一次 30 轮对话的上下文膨胀 =====
// 原始对话(无压缩)
int rawTokensPerTurn = 150; // 每轮平均 150 token
int turns = 30;
int totalRaw = rawTokensPerTurn * turns + 500; // +500 为 System Prompt
System.out.printf("原始上下文: %d tokens%n", totalRaw); // 输出: 5000 tokens
// ----- 策略 1: 截断 -----
int truncated = rawTokensPerTurn * 10 + 500;
System.out.printf("截断后: %d tokens%n", truncated); // 输出: 2000 tokens
// ----- 策略 2: 摘要 -----
int summarized = 200 + rawTokensPerTurn * 10 + 500;
System.out.printf("摘要后: %d tokens%n", summarized); // 输出: 2200 tokens
// ----- 策略 3: 滑动窗口 -----
int sliding = truncated;
System.out.printf("滑动窗口后: %d tokens%n", sliding); // 输出: 2000 tokens
// ----- 策略 4: 选择性保留 -----
int selected = rawTokensPerTurn * 8 + 500 + 150;
System.out.printf("选择性保留后: %d tokens%n", selected); // 输出: 1850 tokens
// ----- 策略 5: 语义保护型压缩 -----
int semantic = 500 + rawTokensPerTurn * 6 + 300 + 100;
System.out.printf("语义保护型后: %d tokens%n", semantic); // 输出: 1800 tokens
// ===== 效果总结 =====
System.out.println("\n--- 压缩效果对比 ---");
Map strategies = new LinkedHashMap<>();
strategies.put("原始", totalRaw);
strategies.put("截断", truncated);
strategies.put("摘要", summarized);
strategies.put("滑动窗口", sliding);
strategies.put("选择性保留", selected);
strategies.put("语义保护型", semantic);
for (Map.Entry entry : strategies.entrySet()) {
double ratio = (double) entry.getValue() / totalRaw;
System.out.printf("%-10s: %d tokens (压缩率 %.1f%%)%n",
entry.getKey(), entry.getValue(), ratio * 100);
}
// 输出:
// 原始 : 5000 tokens (压缩率 100.0%)
// 截断 : 2000 tokens (压缩率 40.0%)
// 摘要 : 2200 tokens (压缩率 44.0%)
// 滑动窗口 : 2000 tokens (压缩率 40.0%)
// 选择性保留: 1850 tokens (压缩率 37.0%)
// 语义保护型: 1800 tokens (压缩率 36.0%)
}
}
Solon AI 框架的 SummarizationInterceptor
Solon AI 是一个国产的 Java AI 框架,它提供了 SummarizationInterceptor 作为上下文压缩的内置方案。其核心思路是:当对话轮次超过阈值时,自动触发摘要压缩。
// Solon AI 框架 - 上下文压缩拦截器配置
@Configuration
public class AiConfig {
@Bean
public ChatService chatService() {
return new ChatService.builder()
.model(openAiModel) // LLM 模型
.interceptor(summarizationInterceptor()) // 添加压缩拦截器
.build();
}
@Bean
public SummarizationInterceptor summarizationInterceptor() {
return new SummarizationInterceptor()
.threshold(10) // 超过 10 轮对话时触发压缩
.summaryModel(openAiModel) // 用哪个模型做摘要(可以用更便宜的模型)
.preserveSystemPrompt(true) // 保护 System Prompt 不被压缩
.preserveLastN(3) // 保留最近 3 轮原始对话
.summaryPrompt("请将以下对话历史压缩为简洁摘要," +
"保留所有关键决策、用户偏好和事实信息。" +
"丢弃闲聊和重复内容。");
}
}
// ===== 工作原理 =====
// 1. 用户每轮对话后,Interceptor 检查当前对话轮次
// 2. 轮次 threshold:
// - System Prompt 全量保留(preserveSystemPrompt=true)
// - 最近 3 轮完整保留(preserveLastN=3)
// - 其余轮次调用 summaryModel 生成摘要
// - 摘要替代原始对话,拼接到上下文中
// ===== 压缩前后对比 =====
// 压缩前: [System] + [15轮原始对话] = ~3000 tokens
// 压缩后: [System] + [摘要~200tokens] + [最近3轮] = ~1150 tokens
// 压缩率: 38.3%,且核心指令和近期上下文无损保留
// // Solon AI 框架 - 上下文压缩拦截器配置
// @Configuration
// public class AiConfig {
// @Bean
// public ChatService chatService() {
return new ChatService.builder();
// .model(openAiModel) // LLM 模型
// .interceptor(summarizationInterceptor()) // 添加压缩拦截器
// .build();
// }
// @Bean
// public SummarizationInterceptor summarizationInterceptor() {
return new SummarizationInterceptor();
// .threshold(10) // 超过 10 轮对话时触发压缩
// .summaryModel(openAiModel) // 用哪个模型做摘要(可以用更便宜的模型)
// .preserveSystemPrompt(true) // 保护 System Prompt 不被压缩
// .preserveLastN(3) // 保留最近 3 轮原始对话
// .summaryPrompt("请将以下对话历史压缩为简洁摘要," +
// "保留所有关键决策、用户偏好和事实信息。" +
// "丢弃闲聊和重复内容。");
// }
// }
// // ===== 工作原理 =====
// // 1. 用户每轮对话后,Interceptor 检查当前对话轮次
// // 2. 轮次 threshold:
// // - System Prompt 全量保留(preserveSystemPrompt=true)
// // - 最近 3 轮完整保留(preserveLastN=3)
// // - 其余轮次调用 summaryModel 生成摘要
// // - 摘要替代原始对话,拼接到上下文中
// // ===== 压缩前后对比 =====
// // 压缩前: [System] + [15轮原始对话] = ~3000 tokens
// // 压缩后: [System] + [摘要~200tokens] + [最近3轮] = ~1150 tokens
// // 压缩率: 38.3%,且核心指令和近期上下文无损保留
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: // Solon AI 框架 - 上下文压缩拦截器配置
// Python: public class AiConfig {
// Python: public ChatService chatService() {
return new ChatService.builder()
// Python: .model(openAiModel) // LLM 模型
// Python: .interceptor(summarizationInterceptor()) // 添加压缩拦截器
// Python: .build();
// Python: }
// Python: public SummarizationInterceptor summarizationInterceptor() {
return new SummarizationInterceptor()
// Python: .threshold(10) // 超过 10 轮对话时触发压缩
// Python: .summaryModel(openAiModel) // 用哪个模型做摘要(可以用更便宜的模型)
// Python: .preserveSystemPrompt(true) // 保护 System Prompt 不被压缩
// Python: .preserveLastN(3) // 保留最近 3 轮原始对话
// Python: .summaryPrompt("请将以下对话历史压缩为简洁摘要," +
// Python: "保留所有关键决策、用户偏好和事实信息。" +
// Python: "丢弃闲聊和重复内容。");
// Python: }
// Python: }
// Python: // ===== 工作原理 =====
// Python: // 1. 用户每轮对话后,Interceptor 检查当前对话轮次
// Python: // 2. 轮次 threshold:
// Python: // - System Prompt 全量保留(preserveSystemPrompt=true)
// Python: // - 最近 3 轮完整保留(preserveLastN=3)
// Python: // - 其余轮次调用 summaryModel 生成摘要
// Python: // - 摘要替代原始对话,拼接到上下文中
// Python: // ===== 压缩前后对比 =====
// Python: // 压缩前: [System] + [15轮原始对话] = ~3000 tokens
// Python: // 压缩后: [System] + [摘要~200tokens] + [最近3轮] = ~1150 tokens
// Python: // 压缩率: 38.3%,且核心指令和近期上下文无损保留
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: // Solon AI 框架 - 上下文压缩拦截器配置
// Python: public class AiConfig {
// Python: public ChatService chatService() {
return new ChatService.builder();
// Python: .model(openAiModel) // LLM 模型
// Python: .interceptor(summarizationInterceptor()) // 添加压缩拦截器
// Python: .build();
// Python: }
// Python: public SummarizationInterceptor summarizationInterceptor() {
return new SummarizationInterceptor();
// Python: .threshold(10) // 超过 10 轮对话时触发压缩
// Python: .summaryModel(openAiModel) // 用哪个模型做摘要(可以用更便宜的模型)
// Python: .preserveSystemPrompt(true) // 保护 System Prompt 不被压缩
// Python: .preserveLastN(3) // 保留最近 3 轮原始对话
// Python: .summaryPrompt("请将以下对话历史压缩为简洁摘要," +
// Python: "保留所有关键决策、用户偏好和事实信息。" +
// Python: "丢弃闲聊和重复内容。");
// Python: }
// Python: }
// Python: // ===== 工作原理 =====
// Python: // 1. 用户每轮对话后,Interceptor 检查当前对话轮次
// Python: // 2. 轮次 threshold:
// Python: // - System Prompt 全量保留(preserveSystemPrompt=true)
// Python: // - 最近 3 轮完整保留(preserveLastN=3)
// Python: // - 其余轮次调用 summaryModel 生成摘要
// Python: // - 摘要替代原始对话,拼接到上下文中
// Python: // ===== 压缩前后对比 =====
// Python: // 压缩前: [System] + [15轮原始对话] = ~3000 tokens
// Python: // 压缩后: [System] + [摘要~200tokens] + [最近3轮] = ~1150 tokens
// Python: // 压缩率: 38.3%,且核心指令和近期上下文无损保留
}
压缩不是删减,而是蒸馏。好的压缩策略就像化学蒸馏--把混合物中的精华提取出来,丢弃的是杂质而非有效成分。
6.6 记忆架构实战
选择合适的向量数据库和 Embedding 模型是构建记忆系统的关键决策。本节提供完整的选型对比和架构实现。
向量数据库深度对比 | 数据库 | 部署方式 | 最大向量数 | 混合搜索 | 过滤查询 | 事务支持 | 成本 | 适用场景 | | --- | --- | --- | --- | --- | --- | --- | --- | | Chroma | 嵌入式/内存 | ~百万级 | ❌ | 基础 | ❌ | 免费 | 原型/小规模 | | Pinecone | 全托管云服务 | 亿级 | ✅(稀疏+稠密) | 丰富 | ❌ | 按量计费 | 生产/大规模 | | Weaviate | 自部署/Docker | 千万级 | ✅(BM25+向量) | GraphQL | ❌ | 免费/云付费 | 复杂查询场景 | | Milvus | 分布式集群 | 十亿级 | ✅ | 丰富 | ❌ | 免费/运维成本高 | 超大规模企业 | | pgvector | PG扩展/Docker | 百万级 | ✅(全文+向量) | SQL全能力 | ✅ | 免费/低成本 | 已有PG基础设施 | 🎯 选型决策树
- 快速原型验证 → Chroma(零配置,pip install 即用)
- 不想运维 → Pinecone(全托管,按量付费,注册即用)
- 需要混合搜索(关键词+语义) → Weaviate(BM25+向量双路召回)
- 已有 PostgreSQL 且数据量不大 → pgvector(不用引入新组件,SQL+向量一体化)
- 十亿级向量 + 高可用 → Milvus(分布式架构,专门为超大规模设计)
完整记忆系统架构设计
短期/长期/工作记忆协同代码实现
import chromadb
from openai import OpenAI
class MemorySystem:
"""三层记忆协同系统"""
def __init__(self, embedding_model="text-embedding-3-small"):
self.client = OpenAI()
self.chroma = chromadb.PersistentClient(path="./memory_db")
self.collection = self.chroma.get_or_create_collection(
name="agent_memory",
metadata={"hnsw:space": "cosine"}
)
self.short_term = [] # 短期记忆:对话历史列表
self.working = {} # 工作记忆:当前任务的中间变量
self.compress_threshold = 15 # 超过 15 轮触发压缩
# ===== 短期记忆管理 =====
def add_to_short_term(self, role, content):
"""添加一条消息到短期记忆"""
self.short_term.append({"role": role, "content": content})
# 检查是否需要压缩
if len(self.short_term) > self.compress_threshold:
self._compress_short_term()
def _compress_short_term(self):
"""压缩短期记忆:摘要 + 转存长期记忆"""
# 保留最近 3 轮和 System Prompt
system_msgs = [m for m in self.short_term if m["role"] == "system"]
recent_msgs = self.short_term[-6:] # 最近 3 轮(每轮2条)
# 其余消息交给 LLM 生成摘要
old_msgs = self.short_term[len(system_msgs):-6]
if old_msgs:
summary = self._summarize(old_msgs)
# 摘要存入长期记忆(向量数据库)
self._save_to_long_term(
text=summary,
metadata={"type": "summary", "turn_range":
f"turns {len(system_msgs)}-{len(self.short_term)-6}"}
)
# 用摘要替代原始对话
self.short_term = system_msgs + [
{"role": "system", "content": f"[历史摘要]: {summary}"}
] + recent_msgs
def _summarize(self, messages):
"""用 LLM 对旧对话生成摘要"""
text = "\n".join([f"{m['role']}: {m['content']}" for m in messages])
response = self.client.chat.completions.create(
model="gpt-4o-mini", # 用便宜模型做摘要
messages=[
{"role": "system", "content":
"将以下对话压缩为简洁摘要。保留所有关键决策、"
"用户偏好、事实信息和约束条件。丢弃闲聊和重复内容。"},
{"role": "user", "content": text}
],
max_tokens=300
)
return response.choices[0].message.content
# ===== 长期记忆管理 =====
def _save_to_long_term(self, text, metadata=None):
"""将内容存入长期记忆(向量数据库)"""
vector = self._get_embedding(text)
doc_id = f"mem_{len(self.short_term)}_{hash(text) % 10000}"
self.collection.upsert(
ids=[doc_id],
embeddings=[vector],
documents=[text],
metadatas=[metadata or {}]
)
def retrieve_from_long_term(self, query, top_k=5):
"""从长期记忆中检索相关内容"""
results = self.collection.query(
query_texts=[query],
n_results=top_k,
include=["documents", "metadatas", "distances"]
)
memories = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
if dist this.compressThreshold) {
this.compressShortTerm();
}
}
private async compressShortTerm(): Promise {
const systemMsgs = this.shortTerm.filter(m => m.role === 'system');
const recentMsgs = this.shortTerm.slice(-6);
const oldMsgs = this.shortTerm.slice(systemMsgs.length, -6);
if (oldMsgs.length > 0) {
const summary = await this.summarize(oldMsgs);
this.saveToLongTerm(summary, {
type: 'summary',
turnRange: `turns ${systemMsgs.length}-${this.shortTerm.length - 6}`
});
this.shortTerm = [
...systemMsgs,
{ role: 'system', content: `[历史摘要]: ${summary}` },
...recentMsgs
];
}
}
private async summarize(messages: Message[]): Promise {
const text = messages.map(m => `${m.role}: ${m.content}`).join('\n');
const response = await this.client.chat.completions.create({
model: 'gpt-4o-mini',
messages: [
{ role: 'system', content:
'将以下对话压缩为简洁摘要。保留所有关键决策、用户偏好、事实信息和约束条件。丢弃闲聊和重复内容。' },
{ role: 'user', content: text }
],
max_tokens: 300
});
return response.choices[0].message.content || '';
}
// ===== 长期记忆管理 =====
async saveToLongTerm(text: string, metadata?: Record): Promise {
const vector = await this.getEmbedding(text);
const docId = `mem_${this.shortTerm.length}_${text.length % 10000}`;
this.collection.push({ id: docId, embedding: vector, document: text, metadata: metadata || {} });
}
async retrieveFromLongTerm(query: string, topK = 5): Promise {
const queryVector = await this.getEmbedding(query);
// Simplified: in production use proper vector similarity search
const results = this.collection.slice(0, topK);
return results.filter(r => true).map(r => ({
text: r.document,
metadata: r.metadata,
similarity: 0.85
}));
}
// ===== 工作记忆管理 =====
updateWorking(key: string, value: any): void { this.working[key] = value; }
getWorking(key: string): any { return this.working[key]; }
clearWorking(): void { this.working = {}; }
// ===== Embedding =====
private async getEmbedding(text: string): Promise {
const response = await this.client.embeddings.create({
model: 'text-embedding-3-small',
input: text
});
return response.data[0].embedding;
}
// ===== 完整响应流程 =====
async respond(userInput: string): Promise {
const longMemories = await this.retrieveFromLongTerm(userInput, 3);
const memoryContext = longMemories.map(m => m.text).join('\n');
const workingContext = Object.entries(this.working)
.map(([k, v]) => `${k}: ${v}`).join('\n');
const messages: Message[] = [
...this.shortTerm,
{ role: 'system', content:
`[相关历史记忆]\n${memoryContext}\n\n[当前任务状态]\n${workingContext}` },
{ role: 'user', content: userInput }
];
const response = await this.client.chat.completions.create({
model: 'gpt-4o',
messages: messages as any
});
const answer = response.choices[0].message.content || '';
this.addToShortTerm('user', userInput);
this.addToShortTerm('assistant', answer);
await this.saveToLongTerm(
`用户: ${userInput}\n助手: ${answer}`,
{ type: 'interaction' }
);
return answer;
}
}
// ===== 使用示例 =====
const memory = new MemorySystem();
const answer = await memory.respond("帮我分析一下最近的销售数据趋势");
console.log(answer);
package main
import (
"context"
"fmt"
"log"
openai "github.com/sashabaranov/go-openai"
)
// MemorySystem 三层记忆协同系统
type MemorySystem struct {
client *openai.Client
collection []MemoryEntry // 简化的向量存储
shortTerm []Message
working map[string]string
compressThreshold int
}
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
type MemoryEntry struct {
ID string
Embedding []float32
Document string
Metadata map[string]string
}
type Memory struct {
Text string
Metadata map[string]string
Similarity float64
}
func NewMemorySystem() *MemorySystem {
return &MemorySystem{
client: openai.NewClient(),
collection: []MemoryEntry{},
shortTerm: []Message{},
working: make(map[string]string),
compressThreshold: 15,
}
}
// ===== 短期记忆管理 =====
func (m *MemorySystem) AddToShortTerm(role, content string) {
m.shortTerm = append(m.shortTerm, Message{Role: role, Content: content})
if len(m.shortTerm) > m.compressThreshold {
m.compressShortTerm()
}
}
func (m *MemorySystem) compressShortTerm() {
var systemMsgs, recentMsgs, oldMsgs []Message
for _, msg := range m.shortTerm {
if msg.Role == "system" {
systemMsgs = append(systemMsgs, msg)
}
}
if len(m.shortTerm) >= 6 {
recentMsgs = m.shortTerm[len(m.shortTerm)-6:]
oldMsgs = m.shortTerm[len(systemMsgs) : len(m.shortTerm)-6]
}
if len(oldMsgs) > 0 {
summary := m.summarize(oldMsgs)
m.saveToLongTerm(summary, map[string]string{
"type": "summary",
"turn_range": fmt.Sprintf("turns %d-%d", len(systemMsgs), len(m.shortTerm)-6),
})
m.shortTerm = append(systemMsgs,
Message{Role: "system", Content: fmt.Sprintf("[历史摘要]: %s", summary)})
m.shortTerm = append(m.shortTerm, recentMsgs...)
}
}
func (m *MemorySystem) summarize(messages []Message) string {
var text string
for _, msg := range messages {
text += fmt.Sprintf("%s: %s\n", msg.Role, msg.Content)
}
resp, err := m.client.CreateChatCompletion(context.Background(),
openai.ChatCompletionRequest{
Model: openai.GPT4oMini,
Messages: []openai.ChatCompletionMessage{
{Role: "system", Content: "将以下对话压缩为简洁摘要。保留所有关键决策、用户偏好、事实信息和约束条件。丢弃闲聊和重复内容。"},
{Role: "user", Content: text},
},
MaxTokens: 300,
})
if err != nil {
log.Printf("summarize error: %v", err)
return ""
}
return resp.Choices[0].Message.Content
}
// ===== 长期记忆管理 =====
func (m *MemorySystem) saveToLongTerm(text string, metadata map[string]string) {
vector := m.getEmbedding(text)
docID := fmt.Sprintf("mem_%d_%d", len(m.shortTerm), len(text)%10000)
m.collection = append(m.collection, MemoryEntry{
ID: docID, Embedding: vector, Document: text, Metadata: metadata,
})
}
func (m *MemorySystem) retrieveFromLongTerm(query string, topK int) []Memory {
_ = m.getEmbedding(query) // 简化:实际应做向量相似度搜索
var memories []Memory
for i, entry := range m.collection {
if i >= topK {
break
}
memories = append(memories, Memory{
Text: entry.Document, Metadata: entry.Metadata, Similarity: 0.85,
})
}
return memories
}
// ===== 工作记忆管理 =====
func (m *MemorySystem) updateWorking(key, value string) { m.working[key] = value }
func (m *MemorySystem) getWorking(key string) string { return m.working[key] }
func (m *MemorySystem) clearWorking() { m.working = make(map[string]string) }
// ===== Embedding =====
func (m *MemorySystem) getEmbedding(text string) []float32 {
resp, err := m.client.CreateEmbeddings(context.Background(),
openai.EmbeddingRequest{
Model: openai.SmallEmbedding3,
Input: []string{text},
})
if err != nil {
log.Printf("embedding error: %v", err)
return nil
}
return resp.Data[0].Embedding
}
// ===== 完整响应流程 =====
func (m *MemorySystem) respond(userInput string) string {
longMemories := m.retrieveFromLongTerm(userInput, 3)
var memoryContext string
for _, mem := range longMemories {
memoryContext += mem.Text + "\n"
}
var workingContext string
for k, v := range m.working {
workingContext += fmt.Sprintf("%s: %s\n", k, v)
}
messages := append(m.shortTerm,
Message{Role: "system", Content: fmt.Sprintf("[相关历史记忆]\n%s\n\n[当前任务状态]\n%s", memoryContext, workingContext)},
Message{Role: "user", Content: userInput},
)
resp, err := m.client.CreateChatCompletion(context.Background(),
openai.ChatCompletionRequest{
Model: openai.GPT4o,
Messages: toOpenAIMessages(messages),
})
if err != nil {
log.Printf("respond error: %v", err)
return ""
}
answer := resp.Choices[0].Message.Content
m.AddToShortTerm("user", userInput)
m.AddToShortTerm("assistant", answer)
m.saveToLongTerm(fmt.Sprintf("用户: %s\n助手: %s", userInput, answer),
map[string]string{"type": "interaction"})
return answer
}
func toOpenAIMessages(msgs []Message) []openai.ChatCompletionMessage {
result := make([]openai.ChatCompletionMessage, len(msgs))
for i, m := range msgs {
result[i] = openai.ChatCompletionMessage{Role: m.Role, Content: m.Content}
}
return result
}
// ===== 使用示例 =====
func main() {
memory := NewMemorySystem()
answer := memory.respond("帮我分析一下最近的销售数据趋势")
fmt.Println(answer)
}
import com.openai.client.OpenAIClient;
import com.openai.models.*;
import java.util.*;
public class MemorySystem {
private OpenAIClient client;
private List collection = new ArrayList<>();
private List shortTerm = new ArrayList<>();
private Map working = new HashMap<>();
private int compressThreshold = 15;
static class Message {
String role;
String content;
Message(String role, String content) { this.role = role; this.content = content; }
}
static class MemoryEntry {
String id;
String document;
Map metadata;
}
static class Memory {
String text;
Map metadata;
double similarity;
}
public MemorySystem() {
this.client = new OpenAIClient();
}
// ===== 短期记忆管理 =====
public void addToShortTerm(String role, String content) {
shortTerm.add(new Message(role, content));
if (shortTerm.size() > compressThreshold) {
compressShortTerm();
}
}
private void compressShortTerm() {
List systemMsgs = new ArrayList<>();
for (Message m : shortTerm) {
if ("system".equals(m.role)) systemMsgs.add(m);
}
List recentMsgs = shortTerm.subList(
Math.max(shortTerm.size() - 6, 0), shortTerm.size());
List oldMsgs = shortTerm.subList(
systemMsgs.size(), Math.max(shortTerm.size() - 6, systemMsgs.size()));
if (!oldMsgs.isEmpty()) {
String summary = summarize(oldMsgs);
saveToLongTerm(summary, Map.of("type", "summary"));
shortTerm.clear();
shortTerm.addAll(systemMsgs);
shortTerm.add(new Message("system", "[历史摘要]: " + summary));
shortTerm.addAll(recentMsgs);
}
}
private String summarize(List messages) {
StringBuilder text = new StringBuilder();
for (Message m : messages) {
text.append(m.role).append(": ").append(m.content).append("\n");
}
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o-mini")
.addMessage(ChatCompletionMessageParam.builder()
.role(ChatCompletionMessageParam.Role.SYSTEM)
.content("将以下对话压缩为简洁摘要。保留所有关键决策、用户偏好、事实信息和约束条件。")
.build())
.addMessage(ChatCompletionMessageParam.builder()
.role(ChatCompletionMessageParam.Role.USER)
.content(text.toString())
.build())
.maxCompletionTokens(300)
.build();
ChatCompletion resp = client.chat().completions().create(params);
return resp.choices().get(0).message().content().orElse("");
}
// ===== 长期记忆管理 =====
public void saveToLongTerm(String text, Map metadata) {
MemoryEntry entry = new MemoryEntry();
entry.id = "mem_" + shortTerm.size() + "_" + (text.hashCode() % 10000);
entry.document = text;
entry.metadata = metadata != null ? metadata : new HashMap<>();
collection.add(entry);
}
public List retrieveFromLongTerm(String query, int topK) {
List memories = new ArrayList<>();
for (int i = 0; i (shortTerm);
messages.add(new Message("system",
"[相关历史记忆]\n" + memoryContext + "\n[当前任务状态]\n" + workingContext));
messages.add(new Message("user", userInput));
// 调用 LLM 生成响应
String answer = callLLM(messages);
addToShortTerm("user", userInput);
addToShortTerm("assistant", answer);
saveToLongTerm("用户: " + userInput + "\n助手: " + answer,
Map.of("type", "interaction"));
return answer;
}
private String callLLM(List messages) {
// Simplified - build params and call OpenAI
return "Response from LLM";
}
// ===== 使用示例 =====
public static void main(String[] args) {
MemorySystem memory = new MemorySystem();
String answer = memory.respond("帮我分析一下最近的销售数据趋势");
System.out.println(answer);
}
}
Embedding 模型选择 | 模型 | 提供商 | 维度 | 质量 | 成本 | 延迟 | 适用场景 | | --- | --- | --- | --- | --- | --- | --- | | text-embedding-3-large | OpenAI | 3072 | ⭐⭐⭐⭐⭐ | $0.13/1M | ~100ms | 高���度检索 | | text-embedding-3-small | OpenAI | 1536 | ⭐⭐⭐⭐ | $0.02/1M | ~50ms | 通用场景 | | bge-large-zh-v1.5 | BAAI(本地) | 1024 | ⭐⭐⭐⭐ | 免费 | ~20ms | 中文场景最佳 | | all-MiniLM-L6-v2 | SentenceTransformers | 384 | ⭐⭐⭐ | 免费 | ~5ms | 快速原型/轻量 | 💡 Embedding 选型建议
- 中文为主 → BGE 系列(BAAI 北京智源),中文语义理解显著优于 OpenAI 模型
- 成本敏感 → 本地模型(bge / MiniLM),零 API 费用,但需要 GPU 或 CPU 推理
- 追求最高质量 → OpenAI text-embedding-3-large,MTEB 排行榜领先
- 平衡成本与质量 → OpenAI text-embedding-3-small,性价比最优
6.7 反思记忆与情景记忆
前面介绍的三层记忆架构解决了存储和检索问题,但记忆系统还有更高阶的能力:从经验中学习(反思记忆)和记录完整事件(情景记忆)。
🪞 反思记忆(Reflective Memory)
从执行结果中提炼经验教训。不是存储原始对话,而是存储"我学到了什么"。
例:任务失败后反思 → "查询天气 API 时需要加超时处理,否则会卡死整个流程"
📖 情景记忆(Episodic Memory)
记录具体事件的完整上下文:谁、什么、何时、何地、为什么、结果如何。
例:记录完整事件 → "2024-03-15,用户请求分析报告,用了 Python 分析工具,耗时 3 分钟,输出 500 字摘要,用户满意"
与三层记忆的关系
反思记忆和情景记忆都存储在长期记忆(向量数据库)中,但它们的元数据标签不同,检索时可以按类型过滤: | 记忆类型 | 存储内容 | 元数据标签 | 检索场景 | | --- | --- | --- | --- | | 短期记忆 | 原始对话轮次 | 无(直接拼接) | 当前对话上下文 | | 反思记忆 | 提炼的经验教训 | {type: "reflection", task: "xxx"} | 遇到相似任务时参考经验 | | 情景记忆 | 完整事件的5W1H | {type: "episodic", timestamp: "xxx"} | 需要历史上下文时参考事件 | | 工作记忆 | 当前任务中间变量 | 无(Scratchpad) | 多步推理过程中的状态 | ### Reflexion 框架的反思循环
Reflexion 是一个经典的反思型 Agent 框架,其核心创新在于引入反思循环:Agent 执行任务后,如果结果不理想,会进行自我反思,将反思结果存入记忆,下次执行时参考这些经验。
class ReflexionAgent:
"""基于 Reflexion 框架的反思型 Agent"""
def __init__(self, llm, memory_system):
self.llm = llm
self.memory = memory_system
self.max_retries = 3 # 最大反思重试次数
def execute_task(self, task_description):
"""执行任务,失败时触发反思循环"""
for attempt in range(self.max_retries):
# 1. 检索相关反思经验
past_reflections = self.memory.retrieve_from_long_term(
query=task_description,
top_k=3
)
reflection_context = "\n".join([
f"[过往经验 #{i+1}]: {r['text']}"
for i, r in enumerate(past_reflections)
if r.get('metadata', {}).get('type') == 'reflection'
])
# 2. 构建带反思经验的 prompt
prompt = f"""
任务: {task_description}
{f'过往反思经验:\n{reflection_context}' if reflection_context else '这是首次尝试此任务。'}
请执行任务,注意避免过去犯过的错误。
"""
# 3. 执行 ReAct 循环
result = self._react_loop(prompt)
# 4. 评估结果
evaluation = self._evaluate(task_description, result)
if evaluation["success"]:
# 成功 → 记录为情景记忆
self.memory._save_to_long_term(
text=f"任务成功: {task_description}\n方法: {result['approach']}\n结果: {result['output']}",
metadata={"type": "episodic", "outcome": "success"}
)
return result
# 5. 失败 → 触发反思
reflection = self._reflect(task_description, result, evaluation)
# 6. 将反思存入记忆
self.memory._save_to_long_term(
text=reflection,
metadata={
"type": "reflection",
"task": task_description,
"attempt": attempt + 1,
"outcome": "failure"
}
)
print(f"第 {attempt+1} 次尝试失败,反思已存入记忆")
return {"output": "任务失败,已达最大重试次数", "success": False}
def _reflect(self, task, result, evaluation):
"""对失败结果进行反思"""
reflect_prompt = f"""
任务: {task}
我的行动: {result['actions_taken']}
得到的结果: {result['output']}
评估反馈: {evaluation['feedback']}
请反思以下问题:
1. 我哪里做错了?具体是哪个步骤出了问题?
2. 为什么会犯这个错误?根本原因是什么?
3. 下次应该怎么避免?给出具体的改进策略。
用简洁的要点总结你的反思。
"""
response = self.llm.generate(reflect_prompt)
return response
def _evaluate(self, task, result):
"""用 LLM 评估任务结果"""
eval_prompt = f"""
任务: {task}
结果: {result['output']}
这个结果是否成功完成了任务?请判断:
- 成功: 返回 JSON {"success": true}
- 失败: 返回 JSON {"success": false, "feedback": "具体问题描述"}
"""
response = self.llm.generate(eval_prompt)
# 解析评估结果...
return {"success": False, "feedback": "结果不完整"} # 简化示例
class ReflexionAgent {
private llm: { generate: (prompt: string) => Promise };
private memory: MemorySystem;
private maxRetries = 3;
constructor(llm: any, memorySystem: MemorySystem) {
this.llm = llm;
this.memory = memorySystem;
}
async executeTask(taskDescription: string): Promise {
for (let attempt = 0; attempt r.metadata?.type === 'reflection')
.map((r, i) => `[过往经验 #${i + 1}]: ${r.text}`)
.join('\n');
// 2. 构建带反思经验的 prompt
const prompt = `任务: ${taskDescription}\n\n` +
(reflectionContext ? `过往反思经验:\n${reflectionContext}` : '这是首次尝试此任务。') +
'\n\n请执行任务,注意避免过去犯过的错误。';
// 3. 执行 ReAct 循环
const result = await this.reactLoop(prompt);
// 4. 评估结果
const evaluation = await this.evaluate(taskDescription, result);
if (evaluation.success) {
await this.memory.saveToLongTerm(
`任务成功: ${taskDescription}\n方法: ${result.approach}\n结果: ${result.output}`,
{ type: 'episodic', outcome: 'success' }
);
return result;
}
// 5. 失败 → 触发反思
const reflection = await this.reflect(taskDescription, result, evaluation);
// 6. 将反思存入记忆
await this.memory.saveToLongTerm(reflection, {
type: 'reflection', task: taskDescription, attempt: attempt + 1, outcome: 'failure'
});
console.log(`第 ${attempt + 1} 次尝试失败,反思已存入记忆`);
}
return { output: '任务失败,已达最大重试次数', success: false };
}
private async reflect(task: string, result: any, evaluation: any): Promise {
const prompt = `任务: ${task}\n我的行动: ${result.actions_taken}\n` +
`得到的结果: ${result.output}\n评估反馈: ${evaluation.feedback}\n\n` +
'请反思:\n1. 哪里做错了?\n2. 为什么会犯这个错误?\n3. 下次怎么避免?';
return await this.llm.generate(prompt);
}
private async evaluate(task: string, result: any): Promise {
const prompt = `任务: ${task}\n结果: ${result.output}\n这个结果是否成功完成任务?`;
const response = await this.llm.generate(prompt);
return { success: false, feedback: '结果不完整' };
}
private async reactLoop(prompt: string): Promise {
const response = await this.llm.generate(prompt);
return { output: response, approach: 'ReAct', actions_taken: 'execute' };
}
}
package main
import "fmt"
// ReflexionAgent 基于Reflexion框架的反思型Agent
type ReflexionAgent struct {
llm LLMInterface
memory *MemorySystem
maxRetries int
}
type LLMInterface interface {
Generate(prompt string) string
}
type Evaluation struct {
Success bool
Feedback string
}
type TaskResult struct {
Output string
Approach string
ActionsTaken string
Success bool
}
func NewReflexionAgent(llm LLMInterface, memory *MemorySystem) *ReflexionAgent {
return &ReflexionAgent{llm: llm, memory: memory, maxRetries: 3}
}
func (a *ReflexionAgent) executeTask(taskDescription string) *TaskResult {
for attempt := 0; attempt ←
1 / 5
→
重播
### 记忆治理策略代码实现
import time from datetime import datetime, timedelta
class MemoryGovernor: """记忆生命周期治理引擎"""
def init(self, memory_system): self.memory = memory_system self.config = { "max_short_term_turns": 15, # 短期记忆最大轮次 "compress_threshold": 12, # 触发压缩的轮次 "preserve_last_n": 3, # 压缩时保留最近 N 轮 "preserve_system_prompt": True, # 保护 System Prompt "eviction_age_days": 90, # 超过 90 天的低重要性记忆淘汰 "eviction_max_count": 10000, # 总记忆条数上限 "retrieval_similarity_threshold": 0.7, # 检索相似度阈值 "retrieval_top_k": 5, # 检索返回 Top-K "jit_mode": True # JIT 上下文模式 } self.access_stats = {} # 记忆访问统计(用于淘汰决策)
===== 写入阶段 =====
def write(self, content, metadata=None): """写入记忆,自动评估重要性""" importance = self._assess_importance(content)
enriched_meta = { **(metadata or {}), "importance": importance, "created_at": datetime.now().isoformat(), "access_count": 0, "last_accessed": None }
self.memory._save_to_long_term(content, enriched_meta) self.memory.add_to_short_term( enriched_meta.get("role", "user"), content )
def _assess_importance(self, content): """用 LLM 快速评估内容重要性(1-5分)"""
简化版:基于规则的启发式评估
score = 1 if any(kw in content for kw in ["决策", "偏好", "规则", "约束"]): score += 2 if any(kw in content for kw in ["失败", "错误", "反思"]): score += 2 if len(content) > 200: # 长内容通常更重要 score += 1 return min(score, 5)
===== 索引阶段 =====
def index(self): """确保记忆索引完整且高效"""
Chroma/Milvus 等向量数据库自动维护向量索引
额外维护:时间索引、类型索引、访问频率索引
pass # 由向量数据库自动处理
===== 检索阶段(JIT 模式) =====
def retrieve_jit(self, current_step_description): """Just-in-Time 检索:只取当前步骤需要的信息""" if not self.config["jit_mode"]:
非 JIT 模式:一次性加载所有相关记忆
return self.memory.retrieve_from_long_term( current_step_description, top_k=self.config["retrieval_top_k"] )
JIT 模式:分步检索,按需注入
results = self.memory.retrieve_from_long_term( current_step_description, top_k=3 # JIT 模式下减少初始检索量 )
更新访问统计(用于淘汰决策)
for r in results:
doc_id = r.get("metadata", {}).get("doc_id", "")
if doc_id:
self.access_stats[doc_id] =
self.access_stats.get(doc_id, 0) + 1
相似度过滤
filtered = [ r for r in results if r["similarity"] >= self.config["retrieval_similarity_threshold"] ]
return filtered
===== 压缩阶段 =====
def compress_if_needed(self): """检查并执行上下文压缩""" if len(self.memory.short_term) > self.config["compress_threshold"]: self._semantic_preserving_compress()
def _semantic_preserving_compress(self): """语义保护型压缩""" short = self.memory.short_term
分层提取
system_msgs = [m for m in short if m["role"] == "system"] recent = short[-(self.config["preserve_last_n"] * 2):] middle = short[len(system_msgs):-(self.config["preserve_last_n"] * 2)]
识别行动-结果对(工具调用 + 返回值)
action_pairs = [] i = 0 while i content.includes(kw))) score += 2; if (['失败', '错误', '反思'].some(kw => content.includes(kw))) score += 2; if (content.length > 200) score += 1; return Math.min(score, 5); }
// ===== 检索阶段(JIT模式) ===== retrieveJit(currentStepDescription: string): any[] { if (!this.config.jitMode) { return this.memory.retrieveFromLongTerm(currentStepDescription, this.config.retrievalTopK); } const results = this.memory.retrieveFromLongTerm(currentStepDescription, 3); for (const r of results) { const docId = r.metadata?.docId || ''; if (docId) this.accessStats[docId] = (this.accessStats[docId] || 0) + 1; } return results.filter(r => r.similarity >= this.config.retrievalSimilarityThreshold); }
// ===== 压缩阶段 ===== compressIfNeeded(): void { if (this.memory.shortTerm.length > this.config.compressThreshold) { this.semanticPreservingCompress(); } }
private semanticPreservingCompress(): void { // 简化版:保留system和最近N轮,其余摘要 const short = this.memory.shortTerm; const systemMsgs = short.filter(m => m.role === 'system'); const recent = short.slice(-(this.config.preserveLastN * 2)); // 中间内容做摘要... this.memory.shortTerm = [...systemMsgs, ...recent]; }
// ===== 淘汰阶段 ===== evict(): void { const cutoff = new Date(Date.now() - this.config.evictionAgeDays * 86400000); // 时效淘汰 + 容量淘汰 + 质量淘汰 const leastAccessed = Object.entries(this.accessStats) .sort((a, b) => a[1] - b[1]).slice(0, 100); // 删除最少访问的记忆... } }
// ===== 使用示例 ===== const governor = new MemoryGovernor(new MemorySystem()); governor.write("用户偏好简洁的中文回答", { type: 'preference' }); const context = governor.retrieveJit("分析销售数据趋势"); governor.compressIfNeeded(); governor.evict();
package main
import ( "fmt" "time" )
// MemoryGovernor 记忆生命周期治理引擎 type MemoryGovernor struct { memory *MemorySystem config GovernorConfig accessStats map[string]int }
type GovernorConfig struct { MaxShortTermTurns int CompressThreshold int PreserveLastN int PreserveSystemPrompt bool EvictionAgeDays int EvictionMaxCount int RetrievalSimilarityThreshold float64 RetrievalTopK int JITMode bool }
func NewMemoryGovernor(mem *MemorySystem) *MemoryGovernor { return &MemoryGovernor{ memory: mem, config: GovernorConfig{ MaxShortTermTurns: 15, CompressThreshold: 12, PreserveLastN: 3, PreserveSystemPrompt: true, EvictionAgeDays: 90, EvictionMaxCount: 10000, RetrievalSimilarityThreshold: 0.7, RetrievalTopK: 5, JITMode: true, }, accessStats: make(map[string]int), } }
// ===== 写入阶段 ===== func (g *MemoryGovernor) write(content string, metadata map[string]string) { importance := g.assessImportance(content) enriched := map[string]string{ "importance": fmt.Sprintf("%d", importance), "created_at": time.Now().Format(time.RFC3339), } for k, v := range metadata { enriched[k] = v } g.memory.saveToLongTerm(content, enriched) role := "user" if r, ok := metadata["role"]; ok { role = r } g.memory.addToShortTerm(role, content) }
func (g *MemoryGovernor) assessImportance(content string) int { score := 1 keywords := []string{"决策", "偏好", "规则", "约束"} for _, kw := range keywords { if contains(content, kw) { score += 2; break } } failKeywords := []string{"失败", "错误", "反思"} for _, kw := range failKeywords { if contains(content, kw) { score += 2; break } } if len(content) > 200 { score += 1 } if score > 5 { score = 5 } return score }
func contains(s, sub string) bool { return len(s) >= len(sub) && (s == sub || len(sub) == 0 || findSubstring(s, sub)) } func findSubstring(s, sub string) bool { for i := 0; i = g.config.RetrievalSimilarityThreshold { filtered = append(filtered, r) } } return filtered }
// ===== 压缩阶段 ===== func (g *MemoryGovernor) compressIfNeeded() { if len(g.memory.shortTerm) > g.config.CompressThreshold { g.semanticPreservingCompress() } }
func (g MemoryGovernor) semanticPreservingCompress() { short := g.memory.shortTerm var systemMsgs, recent []Message for _, m := range short { if m.Role == "system" { systemMsgs = append(systemMsgs, m) } } if len(short) >= g.config.PreserveLastN2 { recent = short[len(short)-g.config.PreserveLastN*2:] } g.memory.shortTerm = append(systemMsgs, recent...) }
// ===== 淘汰阶段 ===== func (g *MemoryGovernor) evict() { cutoff := time.Now().AddDate(0, 0, -g.config.EvictionAgeDays) _ = cutoff // 时效淘汰 // 容量淘汰 + 质量淘汰 }
// ===== 使用示例 ===== func main() { governor := NewMemoryGovernor(NewMemorySystem()) governor.write("用户偏好简洁的中文回答", map[string]string{"type": "preference"}) _ = governor.retrieveJIT("分析销售数据趋势") governor.compressIfNeeded() governor.evict() }
import java.util.*;
public class MemoryGovernor { private MemorySystem memory; private Map config = new HashMap<>(); private Map accessStats = new HashMap<>();
public MemoryGovernor(MemorySystem memorySystem) { this.memory = memorySystem; config.put("maxShortTermTurns", 15); config.put("compressThreshold", 12); config.put("preserveLastN", 3); config.put("evictionAgeDays", 90); config.put("evictionMaxCount", 10000); config.put("retrievalSimilarityThreshold", 0.7); config.put("retrievalTopK", 5); config.put("jitMode", true); }
// ===== 写入阶段 ===== public void write(String content, Map metadata) { int importance = assessImportance(content); Map enriched = new HashMap<>(metadata != null ? metadata : Map.of()); enriched.put("importance", String.valueOf(importance)); enriched.put("createdAt", java.time.Instant.now().toString()); memory.saveToLongTerm(content, enriched); String role = enriched.getOrDefault("role", "user"); memory.addToShortTerm(role, content); }
private int assessImportance(String content) { int score = 1; if (content.contains("决策") || content.contains("偏好") || content.contains("规则")) score += 2; if (content.contains("失败") || content.contains("错误") || content.contains("反思")) score += 2; if (content.length() > 200) score += 1; return Math.min(score, 5); }
// ===== 检索阶段(JIT模式) ===== @SuppressWarnings("unchecked") public List retrieveJit(String stepDesc) { boolean jitMode = (Boolean) config.get("jitMode"); int topK = (Integer) config.get("retrievalTopK"); if (!jitMode) return memory.retrieveFromLongTerm(stepDesc, topK); List results = memory.retrieveFromLongTerm(stepDesc, 3); for (Memory r : results) { String docId = r.metadata.getOrDefault("docId", ""); if (!docId.isEmpty()) accessStats.merge(docId, 1, Integer::sum); } double threshold = (Double) config.get("retrievalSimilarityThreshold"); List filtered = new ArrayList<>(); for (Memory r : results) { if (r.similarity >= threshold) filtered.add(r); } return filtered; }
// ===== 压缩阶段 ===== public void compressIfNeeded() { int threshold = (Integer) config.get("compressThreshold"); if (memory.shortTermSize() > threshold) { semanticPreservingCompress(); } }
private void semanticPreservingCompress() { // 保留system和最近N轮,其余摘要压缩 int preserveN = (Integer) config.get("preserveLastN"); memory.compressShortTerm(preserveN); }
// ===== 淘汰阶段 ===== public void evict() { int maxCount = (Integer) config.get("evictionMaxCount"); // 时效淘汰 + 容量淘汰 + 质量淘汰 if (accessStats.size() > maxCount) { // 删除最少访问的 } }
public static void main(String[] args) { MemoryGovernor governor = new MemoryGovernor(new MemorySystem()); governor.write("用户偏好简洁的中文回答", Map.of("type", "preference")); List context = governor.retrieveJit("分析销售数据趋势"); governor.compressIfNeeded(); governor.evict(); } }
**✅ 记忆治理的黄金法则**
- **写入要慷慨**:宁可多存,后续靠检索和淘汰来筛选
- **检索要精准**:JIT 原则,只在需要时取需要的量
- **压缩要保守**:保护核心指令和行动-结果对,只压缩中间推理
- **淘汰要渐进**:先降优先级再删除,给"冷记忆"一个回暖的机会
- **指标要量化**:追踪命中率、压缩率、检索延迟,用数据驱动治理策略
## 6.9 长期记忆的静态与动态分类
长期记忆不是铁板一块。仔细观察就会发现,有些记忆**几乎不变**,有些记忆**持续更新**。把长期记忆分为**静态长期记忆**和**动态长期记忆**,是面试中的高频考点,也是工程落地时的关键设计决策。
### 为什么要做这个分类?
如果所有长期记忆都用同一种策略管理,会出现两类问题:
**❌ 不分类的后果**
- **存储浪费**:用户偏好这种不变的信息,如果存入向量数据库每次做语义检索,检索开销大且没有必要--它永远都该被注入
- **信息过时**:历史对话摘要这种高频更新的信息,如果像用户偏好一样"写入就不管了",很快就会变得过时和冗余
- **召回混乱**:不同性质的记忆混在一起检索,模型很难区分"这是永远有效的规则"还是"这是上次的经验"
所以,分类的核心目的是**让不同性质的记忆用不同的管理策略**,从而提升存储效率、召回精准度和上下文利用率。
### 静态长期记忆:几乎不变的核心知识
静态长期记忆的特点是**变化频率极低**,写入后几乎不需要更新。它像一本"个人手册",定义了 Agent 与用户交互的基本规则。
**📖 静态长期记忆的典型内容**
- **用户偏好**:"用户偏好简洁的中文回答"、"用户不喜欢表格输出"--这些偏好一旦确定,几乎不变
- **系统规则**:"不要生成代码中的硬编码密码"、"回答必须附带引用来源"--规则是确定性的约束
- **领域知识**:"公司 API 文档的 Base URL 是 https://api.example.com/v2"--事实性知识,除非系统变更才更新
### 动态长期记忆:持续更新的经验积累
动态长期记忆的特点是**变化频率高**,需要持续更新、压缩和淘汰。它像一本"工作日记",记录了 Agent 的成长轨迹。
**📝 动态长期记忆的典型内容**
- **历史对话摘要**:之前多轮对话的压缩摘要--每次新对话都可能产生新的摘要,旧的摘要需要定期合并
- **任务执行经验**:"上次用 pandas 读 CSV 时遇到了编码问题,解决方案是指定 encoding='utf-8'"--经验会不断积累
- **用户行为模式**:"用户通常在下午 3 点后提问"、"用户习惯先问背景再问具体方案"--模式需要持续观察和修正
### 分类的意义:不同策略适配不同性质
分类带来三方面的差异化策略:
- **不同的存储策略**:静态记忆存本地配置文件(YAML/JSON),读写简单、确定性高;动态记忆存向量数据库,支持语义检索和大规模管理
- **不同的召回策略**:静态记忆直接注入 System Prompt,每次对话都有,不需要检索;动态记忆按相关性从向量库检索,只在需要时注入
- **不同的压缩策略**:静态记忆本身已经是精炼的表达(一条偏好就是一句话),不需要压缩;动态记忆需要定期压缩摘要、淘汰过时内容
### 对比表格 | 维度 | 静态长期记忆 | 动态长期记忆 | | --- | --- | --- | | 典型内容 | 用户偏好、系统规则、领域知识 | 对话摘要、执行经验、行为模式 | | 变化频率 | 极低,几乎不变 | 高,持续更新 | | 存储方式 | 本地配置文件(YAML/JSON) | 向量数据库(Chroma/Milvus) | | 召回方式 | 直接注入 System Prompt | 按相关性语义检索 | | 压缩策略 | 不压缩(本身已精炼) | 定期压缩摘要 + 淘汰过时内容 | | 更新触发 | 用户明确修改偏好时 | 每轮有价值交互后 | **✅ 面试要点**
面试中回答"长期记忆的管理策略"时,**先分类再谈策略**。不要一上来就说"存向量数据库",而要先区分:"静态记忆存配置文件、直接注入 Prompt;动态记忆存向量库、按需检索、定期压缩"。这体现了你对记忆系统不同性质的深刻理解。
关联面试题 → Q10
## 6.10 记忆存储触发链路
一个常见误解是:**"每轮对话都触发长期记忆写入"**。如果真的每轮都写,长期记忆库会变成完整对话日志的备份--噪音爆炸,检索失效。
记忆写入不是"录一切",而是"记值得记的"。就像人类不会记住每顿饭吃了什么,但会记住第一次吃到惊艳味道的那顿。
### 不是每轮都触发!有筛选机制
记忆写入有明确的**触发条件**,只有满足条件的信息才进入长期记忆:
**🎯 记忆写入触发条件**
- **用户表达了新偏好** → 写入**静态长期记忆**(如"我不喜欢表格输出")
- **完成了一个有价值的任务** → 写入**动态长期记忆**(执行经验,如"CSV 编码问题的解决方案")
- **发现了新的上下文信息** → 写入**动态长期记忆**(知识片段,如"公司的 API 版本升级到 v2 了")
- **简单闲聊** → **不触发长期记忆写入**(如"你好"、"今天天气不错")
### 判断标准:什么信息"值得记住"?
判断一条信息是否值得写入长期记忆,核心标准是两个问题:
❓ 问题一:是否有长期参考价值?
如果这条信息**在未来的对话中还可能被用到**,就值得记住。用户的偏好、问题的解决方案、关键的事实信息--这些都可能在下次交互中再次需要。
反例:"今天中午吃什么"只在当天有价值,不值得写入长期记忆。
❓ 问题二:是否包含决策/偏好/知识点?
如果这条信息包含**明确的决策、稳定的偏好或可复用的知识**,就值得记住。
反例:"嗯嗯好的"这种确认性回复不包含任何决策或知识,不值得写入。
### 链路设计:对话→信息提取→价值评估→分类→写入
记忆写入是一个**五步链路**,每一步都有明确的职责:
### 记忆写入触发引擎代码示例
class MemoryWriteTrigger: """记忆写入触发引擎:判断什么信息值得写入长期记忆"""
def init(self, llm_client, static_store, dynamic_store): self.llm = llm_client self.static_store = static_store # 本地配置文件存储 self.dynamic_store = dynamic_store # 向量数据库存储
===== 第一步:信息提取 =====
def extract_key_info(self, conversation_turn): """从对话中提取关键信息片段""" prompt = f"""从以下对话中提取关键信息。只提取可能具有长期参考价值的内容:
对话内容: {conversation_turn}
提取格式(JSON):
- type: preference / decision / knowledge / experience / none
- content: 提取的关键内容(精炼为一句话)
- confidence: 置信度(0-1)
如果对话中没有值得长期记住的信息,返回 type=none."""
result = self.llm.generate(prompt) return self._parse_extraction(result)
===== 第二步:价值评估 =====
def evaluate_value(self, extracted_info): """评估信息是否值得写入长期记忆""" if extracted_info["type"] == "none": return False, "无关键信息"
规则+LLM 双重评估
规则层:简单闲聊模式直接跳过
trivial_patterns = ["你好", "谢谢", "好的", "嗯嗯", "再见"] if any(p in extracted_info["content"] for p in trivial_patterns): return False, "简单闲聊,无长期价值"
LLM 层:评估长期参考价值
prompt = f"""评估以下信息是否具有长期参考价值(未来对话中可能再次需要):
信息:{extracted_info["content"]} 类型:{extracted_info["type"]}
回答 YES 或 NO,并简要说明理由."""
evaluation = self.llm.generate(prompt) is_valuable = "YES" in evaluation.upper() reason = evaluation.strip()
return is_valuable, reason
===== 第三步:分类 =====
def classify_memory(self, extracted_info): """判断信息属于静态记忆还是动态记忆""" type_mapping = { "preference": "static", # 用户偏好 → 静态记忆 "decision": "static", # 决策规则 → 静态记忆 "knowledge": "dynamic", # 知识片段 → 动态记忆 "experience": "dynamic", # 执行经验 → 动态记忆 } return type_mapping.get(extracted_info["type"], "dynamic")
===== 第四步:写入 =====
def write_memory(self, extracted_info, memory_type): """写入对应类型的记忆存储""" if memory_type == "static":
静态记忆:写入本地配置文件
self.static_store.save( key=extracted_info["type"], value=extracted_info["content"], metadata={"updated_at": datetime.now().isoformat()} ) else:
动态记忆:Embedding → 向量数据库
vector = self.llm.embed(extracted_info["content"]) self.dynamic_store.save( vector=vector, content=extracted_info["content"], metadata={ "type": extracted_info["type"], "created_at": datetime.now().isoformat(), "confidence": extracted_info.get("confidence", 0.8) } )
===== 完整链路 =====
def process_turn(self, conversation_turn): """处理一轮对话的完整记忆写入链路"""
Step 1: 信息提取
extracted = self.extract_key_info(conversation_turn)
Step 2: 价值评估
is_valuable, reason = self.evaluate_value(extracted) if not is_valuable: return {"action": "skip", "reason": reason}
Step 3: 分类
memory_type = self.classify_memory(extracted)
Step 4: 写入
self.write_memory(extracted, memory_type)
return { "action": "write", "type": memory_type, "content": extracted["content"], "reason": reason }
===== 使用示例 =====
trigger = MemoryWriteTrigger(llm, static_config, vector_db)
场景 1:用户表达偏好 → 写入静态记忆
result = trigger.process_turn("以后回答都用简洁的中文,不要用英文")
→ {"action": "write", "type": "static", "content": "用户偏好简洁中文回答"}
场景 2:任务经验 → 写入动态记忆
result = trigger.process_turn("用 pandas 读 CSV 遇到编码问题,指定 encoding='utf-8' 解决了")
→ {"action": "write", "type": "dynamic", "content": "CSV编码问题解决方案..."}
场景 3:简单闲聊 → 不触发写入
result = trigger.process_turn("你好,今天天气不错")
→ {"action": "skip", "reason": "简单闲聊,无长期价值"}
class MemoryWriteTrigger { private llm: any; private memory: MemorySystem; private static readonly TRIGGER_KEYWORDS = ['偏好', '喜欢', '讨厌', '规则', '记住', '别再']; private static readonly VALUE_KEYWORDS = ['决策', '偏好', '规则', '约束', '失败', '错误', '经验'];
constructor(llm: any, memory: MemorySystem) { this.llm = llm; this.memory = memory; }
// 判断是否触发记忆写入 shouldWrite(userInput: string, agentResponse: string): boolean { // 1. 用户明确要求记住 if (MemoryWriteTrigger.TRIGGER_KEYWORDS.some(kw => userInput.includes(kw))) return true; // 2. 包含决策/偏好/知识点 const combined = userInput + agentResponse; if (MemoryWriteTrigger.VALUE_KEYWORDS.some(kw => combined.includes(kw))) return true; // 3. 任务完成(简化判断) if (agentResponse.includes('完成') || agentResponse.includes('成功')) return true; return false; }
// 提取值得记住的信息
async extract(userInput: string, agentResponse: string): Promise {
const prompt = 从以下对话中提取值得长期记住的信息,返回JSON数组:\n +
用户: ${userInput}\n助手: ${agentResponse}\n +
提取规则:1.用户偏好 2.关键决策 3.事实信息 4.经验教训;
const result = await this.llm.generate(prompt);
try { return JSON.parse(result); } catch { return []; }
}
// 完整写入流程 async process(userInput: string, agentResponse: string): Promise { if (!this.shouldWrite(userInput, agentResponse)) return; const memories = await this.extract(userInput, agentResponse); for (const mem of memories) { const type = this.classify(mem.content); if (type === 'static') { this.memory.saveToLongTerm(mem.content, { type: 'preference', storage: 'config' }); } else { this.memory.saveToLongTerm(mem.content, { type: 'experience', storage: 'vector' }); } } }
private classify(content: string): 'static' | 'dynamic' { if (['偏好', '规则', '约束'].some(kw => content.includes(kw))) return 'static'; return 'dynamic'; } }
package main
import ( "fmt" "strings" )
// MemoryWriteTrigger 记忆写入触发引擎 type MemoryWriteTrigger struct { llm LLMInterface memory *MemorySystem }
var triggerKeywords = []string{"偏好", "喜欢", "讨厌", "规则", "记住", "别再"} var valueKeywords = []string{"决策", "偏好", "规则", "约束", "失败", "错误", "经验"}
func NewMemoryWriteTrigger(llm LLMInterface, mem *MemorySystem) *MemoryWriteTrigger { return &MemoryWriteTrigger{llm: llm, memory: mem} }
// 判断是否触发记忆写入 func (t *MemoryWriteTrigger) shouldWrite(userInput, agentResponse string) bool { for _, kw := range triggerKeywords { if strings.Contains(userInput, kw) { return true } } combined := userInput + agentResponse for _, kw := range valueKeywords { if strings.Contains(combined, kw) { return true } } if strings.Contains(agentResponse, "完成") || strings.Contains(agentResponse, "成功") { return true } return false }
// 完整写入流程 func (t *MemoryWriteTrigger) process(userInput, agentResponse string) { if !t.shouldWrite(userInput, agentResponse) { return } // 提取值得记住的信息 prompt := fmt.Sprintf("从以下对话中提取值得长期记住的信息:\n用户: %s\n助手: %s", userInput, agentResponse) result := t.llm.Generate(prompt) // 分类并写入 memType := t.classify(result) metadata := map[string]string{"type": memType} t.memory.saveToLongTerm(result, metadata) }
func (t *MemoryWriteTrigger) classify(content string) string { for _, kw := range []string{"偏好", "规则", "约束"} { if strings.Contains(content, kw) { return "static" } } return "dynamic" }
import java.util.*;
public class MemoryWriteTrigger { private LLMInterface llm; private MemorySystem memory; private static final String[] TRIGGER_KEYWORDS = {"偏好", "喜欢", "讨厌", "规则", "记住", "别再"}; private static final String[] VALUE_KEYWORDS = {"决策", "偏好", "规则", "约束", "失败", "错误", "经验"};
public MemoryWriteTrigger(LLMInterface llm, MemorySystem memory) { this.llm = llm; this.memory = memory; }
public boolean shouldWrite(String userInput, String agentResponse) { for (String kw : TRIGGER_KEYWORDS) { if (userInput.contains(kw)) return true; } String combined = userInput + agentResponse; for (String kw : VALUE_KEYWORDS) { if (combined.contains(kw)) return true; } if (agentResponse.contains("完成") || agentResponse.contains("成功")) return true; return false; }
public void process(String userInput, String agentResponse) { if (!shouldWrite(userInput, agentResponse)) return; String prompt = "从以下对话中提取值得长期记住的信息:\n用户: " + userInput + "\n助手: " + agentResponse; String result = llm.generate(prompt); String type = classify(result); memory.saveToLongTerm(result, Map.of("type", type)); }
private String classify(String content) { String[] staticKws = {"偏好", "规则", "约束"}; for (String kw : staticKws) { if (content.contains(kw)) return "static"; } return "dynamic"; } }
**✅ 面试要点**
面试中回答"记忆写入机制"时,**重点强调筛选而非全量**。链路是:对话→信息提取→价值评估→分类→写入。不是每轮都触发,只有"有长期参考价值"的信息才写入。简单闲聊不触发,偏好/规则写入静态记忆,经验/知识写入动态记忆。
关联面试题 → Q11
## 6.11 记忆召回决策与上下文污染防控
记忆召回的核心问题不是"能不能找到",而是**"该不该注入"**。召回太多无关记忆,比没有记忆更危险--这就是**上下文污染**。
上下文污染就像往一杯清水里倒泥浆--信息越多,水质越差。精准的召回不是"把所有相关的都找出来",而是"只把最必要的放进去"。
### 召回决策机制:如何判断哪些记忆需要召回?
大模型判断哪些长期记忆需要召回,依赖三个核心维度的综合评分:
**🎯 意图匹配**
当前对话意图与记忆标签的**语义相似度**。
如用户问"怎么处理 CSV 编码问题",与记忆标签"CSV编码"的相似度高,应召回。
⏰ 时间衰减
**越近期的记忆权重越高**。
时间衰减函数:weight × e^(-λ × Δt)
λ 是衰减系数,Δt 是距今时间。3天前的经验比3个月前的更有参考价值。
**📊 频率加权**
**多次被召回验证有效的记忆权重更高**。
如果一条经验被召回了5次且每次都有帮助,说明它是高价值的。从未被召回的记忆权重低。
综合评分公式:
recall_score = intent_similarity × w1 + time_decay_weight × w2 + frequency_weight × w3
其中: intent_similarity = cosine_similarity(query_embedding, memory_embedding) time_decay_weight = e^(-λ × days_since_creation) frequency_weight = log(1 + successful_recall_count) w1 + w2 + w3 = 1(权重归一化)
### 召回数量限制:不是越多越好
很多开发者认为"多召回一些总比漏掉好",但这恰恰是上下文污染的根源。
**❌ 召回过多的三大危害**
- **注意力稀释**:模型关注无关内容,忽略核心问题。就像人在嘈杂环境中听不清关键信息
- **指令冲突**:多条记忆互相矛盾。如一条说"偏好简洁回答",另一条说"上次用户要求详细分析"--模型该听谁的?
- **Token浪费**:无关记忆消耗宝贵的上下文窗口,挤占真正有用的信息空间
### 上下文污染防控策略
防控上下文污染需要四道防线:
1
数量上限
每次召回不超过 5 条记忆。宁可少召回一条有用信息,也不要多召回三条噪音。上限可以动态调整:简单问题 ≤3 条,复杂问题 ≤5 条。
←
1 / 4
→
重播
### 智能召回引擎代码示例
class SmartRecallEngine: """智能召回引擎:精准召回 + 上下文污染防控"""
def init(self, vector_db, llm_client, config=None): self.db = vector_db self.llm = llm_client self.config = config or { "max_recall_count": 5, # 召回数量上限 "similarity_threshold": 0.7, # 相关性阈值 "time_decay_lambda": 0.05, # 时间衰减系数 "weights": { # 评分权重 "intent": 0.5, "time": 0.3, "frequency": 0.2 }, "priority_order": ["core", "recent_experience", "supplementary"] } self.recall_stats = {} # 记忆召回统计
===== 第一步:多维度评分 =====
def compute_recall_scores(self, query_embedding, candidate_memories): """为每条候选记忆计算综合召回评分""" scores = []
for memory in candidate_memories:
意图匹配:语义相似度
intent_score = cosine_similarity( query_embedding, memory["embedding"] )
时间衰减:越近期权重越高
days_since = (datetime.now() - memory["created_at"]).days time_score = math.exp(-self.config["time_decay_lambda"] * days_since)
频率加权:多次验证有效的记忆权重更高
mem_id = memory["id"] freq_score = math.log(1 + self.recall_stats.get(mem_id, {}).get("success_count", 0))
综合评分
w = self.config["weights"] total_score = ( intent_score * w["intent"] + time_score * w["time"] + freq_score * w["frequency"] )
scores.append({ "memory": memory, "total_score": total_score, "intent_score": intent_score, "time_score": time_score, "freq_score": freq_score })
return scores
===== 第二步:污染防控过滤 =====
def pollution_defense(self, scored_memories): """四道防线过滤,防止上下文污染"""
防线 1:相关性阈值过滤
filtered = [ s for s in scored_memories if s["intent_score"] >= self.config["similarity_threshold"] ]
防线 2:数量上限截断
filtered.sort(key=lambda x: x["total_score"], reverse=True) filtered = filtered[:self.config["max_recall_count"]]
防线 3:冲突检测
filtered = self._resolve_conflicts(filtered)
防线 4:优先级排序
filtered = self._priority_sort(filtered)
return filtered
def _resolve_conflicts(self, scored_memories): """冲突检测:语义矛盾的记忆取最新的""" result = [] used_indices = set()
for i, m1 in enumerate(scored_memories): if i in used_indices: continue conflict_found = False
for j, m2 in enumerate(scored_memories): if j in used_indices or j == i: continue
检测语义矛盾(简化:同类别但内容相反)
if m1["memory"].get("category") == m2["memory"].get("category"):
同类记忆可能冲突,取时间更新的
if m2["memory"]["created_at"] > m1["memory"]["created_at"]: result.append(m2) used_indices.add(j) conflict_found = True break
if not conflict_found: result.append(m1) used_indices.add(i)
return result
def _priority_sort(self, scored_memories): """优先级排序:核心指令 > 最近经验 > 补充信息""" priority_map = {"core": 0, "recent_experience": 1, "supplementary": 2}
for m in scored_memories: cat = m["memory"].get("priority_category", "supplementary") m["priority_rank"] = priority_map.get(cat, 2)
return sorted(scored_memories, key=lambda x: x["priority_rank"])
===== 完整召回流程 =====
def recall(self, query, task_complexity="normal"): """智能召回完整流程"""
Step 1: 向量检索候选记忆
query_embedding = self.llm.embed(query) candidates = self.db.search(query_embedding, top_k=20)
Step 2: 多维度评分
scored = self.compute_recall_scores(query_embedding, candidates)
Step 3: 动态调整召回上限
max_count = self.config["max_recall_count"] if task_complexity == "simple": max_count = 3 # 简单问题召回更少 elif task_complexity == "complex": max_count = 5 # 复杂问题允许更多
self.config["max_recall_count"] = max_count
Step 4: 污染防控过滤
final_memories = self.pollution_defense(scored)
Step 5: 更新召回统计
for m in final_memories: mem_id = m["memory"]["id"] if mem_id not in self.recall_stats: self.recall_stats[mem_id] = {"success_count": 0, "total_count": 0} self.recall_stats[mem_id]["total_count"] += 1
return final_memories
===== 使用示例 =====
engine = SmartRecallEngine(chroma_db, llm)
简单问题:召回 ≤3 条
results = engine.recall("今天有什么日程?", task_complexity="simple")
→ 只返回最相关的 2-3 条近期日程记忆
复杂问题:召回 ≤5 条
results = engine.recall("分析上个季度的销售趋势并给出优化建议", task_complexity="complex")
→ 返回最多 5 条:1条核心指令 + 2条最近经验 + 2条补充知识
class SmartRecallEngine { private memory: MemorySystem; private llm: any; private maxRecall = 5; private similarityThreshold = 0.7;
constructor(memory: MemorySystem, llm: any) { this.memory = memory; this.llm = llm; }
// 三维度评分:意图匹配 × 时间衰减 × 频率加权 async recall(userInput: string, intent?: string): Promise { // 1. 向量检索候选记忆 const candidates = await this.memory.retrieveFromLongTerm( intent || userInput, this.maxRecall * 2 );
// 2. 三维度评分 const scored = candidates.map(mem => ({ ...mem, score: this.score(mem, intent || userInput), }));
// 3. 排序+截断(≤5条) scored.sort((a, b) => b.score - a.score); let recalled = scored.slice(0, this.maxRecall);
// 4. 冲突检测:矛盾记忆取最新 recalled = this.resolveConflicts(recalled);
// 5. 相关性阈值过滤 recalled = recalled.filter(r => r.similarity >= this.similarityThreshold);
return recalled; }
private score(mem: any, query: string): number { const intentScore = mem.similarity; const timeDecay = this.timeDecay(mem.metadata?.createdAt); const frequency = Math.log10((mem.metadata?.accessCount || 1) + 1); return intentScore * 0.6 + timeDecay * 0.3 + frequency * 0.1; }
private timeDecay(createdAt?: string): number { if (!createdAt) return 0.5; const days = (Date.now() - new Date(createdAt).getTime()) / 86400000; return Math.exp(-days / 30); // 30天衰减常数 }
private resolveConflicts(memories: any[]): any[] { // 简化:按内容去重,保留最新的 const seen = new Map(); for (const m of memories) { const key = m.text.substring(0, 50); const existing = seen.get(key); if (!existing || (m.metadata?.createdAt > existing.metadata?.createdAt)) { seen.set(key, m); } } return Array.from(seen.values()); } }
interface RecalledMemory { text: string; metadata: Record; similarity: number; score: number; }
package main
import ( "math" "sort" "time" )
// SmartRecallEngine 智能召回引擎 type SmartRecallEngine struct { memory *MemorySystem maxRecall int similarityThreshold float64 }
func NewSmartRecallEngine(mem *MemorySystem) *SmartRecallEngine { return &SmartRecallEngine{ memory: mem, maxRecall: 5, similarityThreshold: 0.7, } }
// 三维度评分召回 func (e SmartRecallEngine) recall(userInput string) []ScoredMemory { // 1. 向量检索候选 candidates := e.memory.retrieveFromLongTerm(userInput, e.maxRecall2)
// 2. 评分 var scored []ScoredMemory for _, mem := range candidates { scored = append(scored, ScoredMemory{ Memory: mem, Score: e.score(mem, userInput), }) }
// 3. 排序+截断 sort.Slice(scored, func(i, j int) bool { return scored[i].Score > scored[j].Score }) if len(scored) > e.maxRecall { scored = scored[:e.maxRecall] }
// 4. 相关性阈值过滤 var result []ScoredMemory for _, s := range scored { if s.Similarity >= e.similarityThreshold { result = append(result, s) } } return result }
func (e SmartRecallEngine) score(mem Memory, query string) float64 { intentScore := mem.Similarity timeDecay := e.timeDecay(mem.Metadata["created_at"]) frequency := 0.5 // 简化 return intentScore0.6 + timeDecay0.3 + frequency0.1 }
func (e *SmartRecallEngine) timeDecay(createdAt string) float64 { if createdAt == "" { return 0.5 } t, err := time.Parse(time.RFC3339, createdAt) if err != nil { return 0.5 } days := time.Since(t).Hours() / 24 return math.Exp(-days / 30) }
type ScoredMemory struct { Memory Score float64 }
import java.util.*; import java.util.stream.Collectors;
public class SmartRecallEngine { private MemorySystem memory; private int maxRecall = 5; private double similarityThreshold = 0.7;
public SmartRecallEngine(MemorySystem memory) { this.memory = memory; }
// 三维度评分召回 public List recall(String userInput) { // 1. 向量检索候选 List candidates = memory.retrieveFromLongTerm(userInput, maxRecall * 2);
// 2. 评分 List scored = candidates.stream() .map(mem -> new ScoredMemory(mem, score(mem, userInput))) .collect(Collectors.toList());
// 3. 排序+截断 scored.sort((a, b) -> Double.compare(b.score, a.score)); if (scored.size() > maxRecall) scored = scored.subList(0, maxRecall);
// 4. 相关性阈值过滤 return scored.stream() .filter(s -> s.memory.similarity >= similarityThreshold) .collect(Collectors.toList()); }
private double score(Memory mem, String query) { double intentScore = mem.similarity; double timeDecay = timeDecay(mem.metadata.get("created_at")); double frequency = Math.log10(1 + 1); return intentScore * 0.6 + timeDecay * 0.3 + frequency * 0.1; }
private double timeDecay(String createdAt) { if (createdAt == null) return 0.5; long days = (System.currentTimeMillis() - java.time.Instant.parse(createdAt).toEpochMilli()) / 86400000; return Math.exp(-days / 30.0); }
static class ScoredMemory { Memory memory; double score; ScoredMemory(Memory m, double s) { memory = m; score = s; } } }
**✅ 面试要点**
面试中回答"记忆召回策略"时,**先谈召回决策再谈污染防控**。召回决策靠三维度评分(意图匹配×时间衰减×频率加权),污染防控靠四道防线(数量上限≤5、相关性阈值≥0.7、冲突检测取最新、优先级排序核心>经验>补充)。不要只说"用向量搜索找相关记忆"--那只是第一步,真正的难点是防控上下文污染。
关联面试题 → Q12
## 6.11a Prompt 工程:静态框架与动态组装
面试追问:**"动态 Prompt 和静态 Prompt 有什么区别?Agent 每轮对话开始时,上下文是如何动态组装的?"**上一节讲了记忆的召回决策,这一节讲召回的结果怎么**组装进 Prompt**--以及哪些部分永远不动(静态),哪些每轮都变(动态)。
把 Prompt 想象成一栋建筑:**静态部分是地基和框架**(角色定义、安全规则、工具列表)--不会随对话变化;**动态部分是家具和装饰**(记忆片段、对话历史、当前任务)--每轮都换。地基不稳,房子会塌;家具不换,住着不舒服。
### 静态 Prompt:不随对话变化的"固定框架"
静态 Prompt 包含**在所有对话中都相同**的内容,是 Agent 的"宪法":
🔒 静态 Prompt 的四个组成部分
- **角色定义**:"你是一个专业的代码审查专家"--定义 Agent 身份
- **安全规则**:"不要泄露用户隐私、不要执行危险命令"--行为边界
- **输出格式约束**:"回答格式:风险等级→问题→修复建议"--控制输出
- **工具/Skill 描述**:Function Calling Schema + Skill 触发词列表--Agent 的"能力菜单"
**📌 静态 Prompt 的三个特点**
- **不变性**:同一个会话内,内容完全不变(除非用户明确修改偏好)
- **最高优先级**:System Prompt 具有最高权重,模型始终遵循
- **缓存友好**:因为不变,放在 Prompt 前缀处可100%命中 Prompt Cache
在 OpenClaw 等框架中,静态 Prompt 就是从 `SOUL.md`(角色)、`AGENTS.md`(规则)和 ``(Skill列表)读取的内容,注入为 System Prompt 的一部分。这部分**每次请求都一样**,所以是最容易命中缓存的部分。
### 动态 Prompt:每轮对话变化的"活的内容"
动态 Prompt 包含**根据当前对话上下文实时变化**的内容:
**🔄 动态 Prompt 的三个来源**
- **长期记忆召回**:根据当前意图检索的5条相关记忆
- **对话历史**:累积的多轮交互记录(每轮增加)
- **当前任务上下文**:用户刚输入的指令 + 工具返回结果
**⚡ 动态内容的变化频率**
- **慢变化**:Skill完整指令(只在触发时加载)、静态记忆偏好(很少更新)
- **中变化**:长期记忆召回片段(每几轮可能换一批)
- **快变化**:对话历史和工具结果(每轮必变)
**⚠️ 动态内容的缓存特征**
- **慢变化**:多轮后可能命中缓存
- **中变化**:难命中,但可控制注入量
- **快变化**:完全不命中缓存
### 每轮对话的完整 Prompt 组装流程
每轮对话开始时,Agent 需要把静态和动态内容**按顺序组装**成完整 Prompt。顺序很重要--因为缓存是前缀匹配,���定部分必须放最前面:
class PromptAssembler: """每轮对话的 Prompt 动态组装引擎"""
def init(self, system_prompt, tool_schemas, skill_triggers, memory_engine, chat_history_manager):
Layer 1: 静态层(整个会话不变)
self.static_prefix = f"{system_prompt}\n\n{tool_schemas}\n\n{skill_triggers}" self.memory_engine = memory_engine self.history_manager = chat_history_manager
def assemble(self, user_input, intent=None): """组装当前轮次的完整 Prompt""" messages = []
===== Layer 1: 静态框架 =====
System Prompt + 工具Schema + Skill触发词
每次相同 → 100%命中缓存
messages.append({"role": "system", "content": self.static_prefix})
===== Layer 2: 记忆注入 =====
根据当前意图召回长期记忆
只注入与当前任务相关的 ≤5 条记忆
if intent: recalled = self.memory_engine.recall( query=intent, top_k=5, threshold=0.7 ) if recalled: memory_block = "# Relevant Memory\n" + "\n".join(recalled) messages.append({"role": "system", "content": memory_block})
===== Layer 3: 对话历史 =====
如果历史过长,先压缩再注入
history = self.history_manager.get_history(max_tokens=50000) for msg in history: messages.append(msg)
===== Layer 4: 当前请求 =====
messages.append({"role": "user", "content": user_input})
return messages
使用示例
assembler = PromptAssembler( system_prompt="你是代码审查专家...", tool_schemas=json.dumps(tool_definitions), skill_triggers="...", memory_engine=vector_db, chat_history_manager=history_mgr )
每轮调用
messages = assembler.assemble("帮我审查这段代码", intent="代码审查")
→ [system(fixed), system(memory), ...history..., user(current)]
class PromptAssembler { private staticPrefix: string; private memoryEngine: any; private historyManager: any;
constructor(opts: {
systemPrompt: string;
toolSchemas: string;
skillTriggers: string;
memoryEngine: any;
chatHistoryManager: any;
}) {
// Layer 1: 静态层(整个会话不变)
this.staticPrefix = ${opts.systemPrompt}\n\n${opts.toolSchemas}\n\n${opts.skillTriggers};
this.memoryEngine = opts.memoryEngine;
this.historyManager = opts.chatHistoryManager;
}
async assemble(userInput: string, intent?: string): Promise { const messages: Message[] = [];
// ===== Layer 1: 静态框架 ===== messages.push({ role: 'system', content: this.staticPrefix });
// ===== Layer 2: 记忆注入 ===== if (intent) { const recalled = await this.memoryEngine.recall(intent, 5, 0.7); if (recalled.length > 0) { const memoryBlock = '# Relevant Memory\n' + recalled.join('\n'); messages.push({ role: 'system', content: memoryBlock }); } }
// ===== Layer 3: 对话历史 ===== const history = this.historyManager.getHistory(50000); messages.push(...history);
// ===== Layer 4: 当前请求 ===== messages.push({ role: 'user', content: userInput });
return messages; } }
interface Message { role: string; content: string; }
// 使用示例 const assembler = new PromptAssembler({ systemPrompt: '你是代码审查专家...', toolSchemas: JSON.stringify(toolDefinitions), skillTriggers: '...', memoryEngine: vectorDb, chatHistoryManager: historyMgr, });
const messages = await assembler.assemble('帮我审查这段代码', '代码审查');
package main
import "encoding/json"
// PromptAssembler 每轮对话的Prompt动态组装引擎 type PromptAssembler struct { staticPrefix string memoryEngine *MemorySystem historyManager *HistoryManager }
type ChatMessage struct {
Role string json:"role"
Content string json:"content"
}
func NewPromptAssembler(systemPrompt, toolSchemas, skillTriggers string, memEngine *MemorySystem, histMgr *HistoryManager) *PromptAssembler { return &PromptAssembler{ staticPrefix: systemPrompt + "\n\n" + toolSchemas + "\n\n" + skillTriggers, memoryEngine: memEngine, historyManager: histMgr, } }
func (a *PromptAssembler) assemble(userInput, intent string) []ChatMessage { var messages []ChatMessage
// ===== Layer 1: 静态框架 ===== messages = append(messages, ChatMessage{Role: "system", Content: a.staticPrefix})
// ===== Layer 2: 记忆注入 ===== if intent != "" { recalled := a.memoryEngine.retrieveFromLongTerm(intent, 5) if len(recalled) > 0 { memoryBlock := "# Relevant Memory\n" for _, r := range recalled { memoryBlock += r.Text + "\n" } messages = append(messages, ChatMessage{Role: "system", Content: memoryBlock}) } }
// ===== Layer 3: 对话历史 ===== history := a.historyManager.getHistory(50000) messages = append(messages, history...)
// ===== Layer 4: 当前请求 ===== messages = append(messages, ChatMessage{Role: "user", Content: userInput})
return messages }
// 使用示例 func main() { schemas, _ := json.Marshal(toolDefinitions) assembler := NewPromptAssembler( "你是代码审查专家...", string(schemas), "...", NewMemorySystem(), NewHistoryManager(), ) messages := assembler.assemble("帮我审查这段代码", "代码审查") _ = messages }
import java.util.*;
public class PromptAssembler { private String staticPrefix; private MemorySystem memoryEngine; private HistoryManager historyManager;
public PromptAssembler(String systemPrompt, String toolSchemas, String skillTriggers, MemorySystem memEngine, HistoryManager histMgr) { // Layer 1: 静态层(整个会话不变) this.staticPrefix = systemPrompt + "\n\n" + toolSchemas + "\n\n" + skillTriggers; this.memoryEngine = memEngine; this.historyManager = histMgr; }
public List assemble(String userInput, String intent) { List messages = new ArrayList<>();
// ===== Layer 1: 静态框架 ===== messages.add(new ChatMessage("system", staticPrefix));
// ===== Layer 2: 记忆注入 ===== if (intent != null && !intent.isEmpty()) { List recalled = memoryEngine.retrieveFromLongTerm(intent, 5); if (!recalled.isEmpty()) { StringBuilder memoryBlock = new StringBuilder("# Relevant Memory\n"); for (Memory r : recalled) { memoryBlock.append(r.text).append("\n"); } messages.add(new ChatMessage("system", memoryBlock.toString())); } }
// ===== Layer 3: 对话历史 ===== List history = historyManager.getHistory(50000); messages.addAll(history);
// ===== Layer 4: 当前请求 ===== messages.add(new ChatMessage("user", userInput));
return messages; }
static class ChatMessage { String role; String content; ChatMessage(String r, String c) { role = r; content = c; } } }
**✅ 面试要点**
回答"动态Prompt和静态Prompt的区别"时,**先定义再对比**:
**静态Prompt** = System Prompt(角色+规则+工具Schema+Skill触发词),整个会话不变,放Prompt最前面,100%命中缓存。
**动态Prompt** = 记忆召回片段 + Skill完整指令 + 对话历史 + 用户当前输入,每轮都变,放Prompt后面,按变化频率分层。
**组装策略**:固定→半固定→动态→当前请求,四层叠放。核心原则:**不变的放前面最大化缓存,变化的放后面最小化缓存失效**。
关联面试题 → Q11(动态vs静态Prompt)、Q12(缓存命中设计)
关联章节 → 1.1.6(上下文窗口Token经济学)、4.13(Token Budget)
## 6.11b 多轮对话状态爆炸与完整解决方案
面试追问:**"多轮对话Agent,怎么解决状态爆炸、上下文溢出?"**前面5.3~5.11分别讲了压缩、治理、召回防控,但面试需要的是**一个完整的系统级解决方案**--怎么把这些策略串起来?这一节给你一个全景框架。
状态爆炸不是某一个环节的问题,而是**整个系统的压力累积**。就像水管爆裂--不是某一个接头坏了,而是整条管路压力超标。解决方案不是"堵一个洞",而是"设计一套泄压系统"。
### 什么是"状态爆炸"?为什么多轮对话会导致这个问题?
单轮对话很简单:用户问→模型答→结束。但多轮对话Agent中,**状态会从线性增长变为指数增长**:
💥 三种状态爆炸模式
- **对话历史膨胀**:每轮新增 ~1000-3000 Token,10轮就是 10K-30K,50轮可能50K-150K--线性增长但速度很快
- **工具调用结果嵌套**:一个工具返回3K Token,模型基于结果再调2个工具,每个又返回2K--嵌套增长呈**树状展开**
- **推理路径分叉**:ToT/Reflexion模式中,每条路径各自累积上下文,多条路径并行时Token消耗是**倍数增长**
📉 状态爆炸的四大危害
- **上下文溢出**:总Token超过窗口128K,直接截断丢失关键信息
- **注意力稀释**:模型在100K上下文中"迷失",重要信息被淹没(Lost in the Middle)
- **成本爆炸**:每轮都发50K+上下文给API,20轮就是$1+的账单
- **行为漂移**:长上下文导致模型偏离原始指令(Context Rot)
### 完整解决方案:四层泄压体系
解决状态爆炸不是单一策略,而是**四层递进的泄压体系**--从预防到压缩到淘汰到兜底:
**🛡️ Layer 1: 预防--不让压力累积**
- **Token Budget**:设定全局上限(如20万Token),实时累加,超预算立即停止(详见4.13节)
- **JIT召回原则**:只检索当前步骤需要的记忆,不提前加载全量(详见6.8节)
- **输出长度约束**:要求模型"只输出关键结论"而非长篇大论,减少输出Token
- **工具结果精简**:工具返回后先做格式化过滤(只保留关键行),不原样塞入上下文
**📦 Layer 2: 压缩--在信息完整的前提下缩减体积**
- **语义保护型压缩**:锁定System Prompt+原始问题,保护行动-结果对完整性,只压缩中间推理(详见6.5节)
- **三层压缩触发**:主动压缩(80%阈值)→后台压缩(60%)→反应式压缩(413状态码触发紧急压缩)
- **增量压缩**:只压缩"上次压缩后新增"的部分,避免全量重复压缩(详见6.12节)
- **对话历史摘要**:旧对话压缩为摘要存入长期记忆,上下文中只保留摘要和最近3轮原文
🗑️ Layer 3: 淘汰--扔掉不再需要的
- **记忆淘汰策略**:时效淘汰(过期记忆清理)+容量淘汰(超上限淘汰最低价值)+质量淘汰(从未被召回的记忆降优先级)(详见6.8节)
- **召回数量硬限制**:每次召回≤5条,简单问题≤3条,宁可少召回也不多注入噪音
- **工具结果裁剪**:长输出只保留Top-K关键行,命令输出只保留最后20行
- **淘汰前归档**:淘汰的记忆做最终摘要存入"归档层",彻底删除前保留摘要
🚨 Layer 4: 兜底--当所有策略都不够时
- **���急压缩(413反应式)**:上下文超限导致API返回413状态码时,立即触发最激进压缩--只保留System Prompt + 最近1轮 + 任务摘要
- **降级回答**:压缩后仍然超限,告诉用户"当前对话过长,建议开启新会话"
- **自动会话切分**:超过阈值(如50轮)后自动将当前对话摘要存入长期记忆,开启新会话继续
- **人工介入**:预算超限+压缩失败+降级回答后,通知用户人工决策
class StateOverflowGuard: """多轮对话状态爆炸的完整解决方案:四层泄压体系"""
def init(self, token_budget=200000, context_window=128000): self.token_budget = token_budget # 全局Token预算 self.context_window = context_window # 上下文窗口大小 self.current_tokens = 0 # 当前累计Token self.compressor = SemanticCompressor() # 语义保护型压缩器 self.memory_governor = MemoryGovernor() # 记忆治理引擎
===== Layer 1: 预防 =====
def check_budget(self, new_tokens): """预算检查:预防状态累积""" self.current_tokens += new_tokens if self.current_tokens > self.token_budget: raise BudgetExceededError( f"Token预算耗尽: {self.current_tokens}/{self.token_budget}"+ "建议开启新会话或人工介入" )
JIT召回:只检索当前需要的记忆
输出约束:限制模型输出长度
工具精简:过滤工具返回结果
return True
===== Layer 2: 压缩 =====
def compress_if_needed(self, messages): """压缩检查:超80%阈值触发语义保护型压缩""" total = count_tokens(messages) threshold = self.context_window * 0.8 # 80%阈值
if total > threshold:
语义保护型压缩:
锁定层(System Prompt+原始问题) → 不压缩
压缩层(中间推理步骤) → 摘要化
保留层(最近3轮) → 原样保留
compressed = self.compressor.compress(messages) return compressed return messages
===== Layer 3: 淘汰 =====
def evict_if_needed(self, messages): """淘汰检查:压缩后仍超阈值时淘汰低价值内容""" total = count_tokens(messages) threshold = self.context_window * 0.6 # 60%阈值
if total > threshold:
淘汰策略:时效淘汰 + 容量淘汰 + 质量淘汰
召回限制:每次≤5条记忆
工具裁剪:只保留关键行
淘汰前归档:做最终摘要存入长期记忆
evicted = self.memory_governor.evict(messages) return evicted return messages
===== Layer 4: 兜底 =====
def emergency_handler(self, error): """紧急兜底:413状态码或预算超限""" if isinstance(error, ContextOverflowError):
紧急压缩:只保留System Prompt + 最近1轮 + 摘要
return self.compressor.emergency_compress() elif isinstance(error, BudgetExceededError):
降级回答 + 自动会话切分 + 人工介入
return { "action": "degrade", "message": "对话过长,建议开启新会话", "summary": self.memory_governor.archive_session() }
===== 完整流程 =====
def process_turn(self, messages, new_tokens): """每轮对话的状态管理完整流程""" try:
Layer 1: 预防(预算检查)
self.check_budget(new_tokens)
Layer 2: 压缩(超80%触发)
messages = self.compress_if_needed(messages)
Layer 3: 淘汰(超60%触发)
messages = self.evict_if_needed(messages)
return messages except (ContextOverflowError, BudgetExceededError) as e:
Layer 4: 兜底(紧急处理)
return self.emergency_handler(e)
class StateOverflowGuard { private tokenBudget: number; private contextWindow: number; private currentTokens = 0; private compressor: SemanticCompressor; private memoryGovernor: MemoryGovernor;
constructor(tokenBudget = 200000, contextWindow = 128000) { this.tokenBudget = tokenBudget; this.contextWindow = contextWindow; this.compressor = new SemanticCompressor(); this.memoryGovernor = new MemoryGovernor(); }
// ===== Layer 1: 预防 =====
checkBudget(newTokens: number): boolean {
this.currentTokens += newTokens;
if (this.currentTokens > this.tokenBudget) {
throw new BudgetExceededError(
Token预算耗尽: ${this.currentTokens}/${this.tokenBudget}建议开启新会话
);
}
return true;
}
// ===== Layer 2: 压缩 ===== compressIfNeeded(messages: Message[]): Message[] { const total = countTokens(messages); const threshold = this.contextWindow * 0.8; if (total > threshold) { return this.compressor.compress(messages); } return messages; }
// ===== Layer 3: 淘汰 ===== evictIfNeeded(messages: Message[]): Message[] { const total = countTokens(messages); const threshold = this.contextWindow * 0.6; if (total > threshold) { return this.memoryGovernor.evict(messages); } return messages; }
// ===== Layer 4: 兜底 ===== emergencyHandler(error: Error): any { if (error instanceof ContextOverflowError) { return this.compressor.emergencyCompress(); } else if (error instanceof BudgetExceededError) { return { action: 'degrade', message: '对话过长,建议开启新会话', summary: this.memoryGovernor.archiveSession(), }; } }
// ===== 完整流程 ===== processTurn(messages: Message[], newTokens: number): Message[] | any { try { this.checkBudget(newTokens); messages = this.compressIfNeeded(messages); messages = this.evictIfNeeded(messages); return messages; } catch (e) { return this.emergencyHandler(e as Error); } } }
function countTokens(messages: Message[]): number { return messages.reduce((sum, m) => sum + m.content.length / 4, 0); }
package main
import ( "fmt" )
// StateOverflowGuard 多轮对话状态爆炸的完整解决方案 type StateOverflowGuard struct { tokenBudget int contextWindow int currentTokens int compressor *SemanticCompressor memoryGovernor *MemoryGovernor }
func NewStateOverflowGuard(tokenBudget, contextWindow int) *StateOverflowGuard { return &StateOverflowGuard{ tokenBudget: tokenBudget, contextWindow: contextWindow, compressor: NewSemanticCompressor(), memoryGovernor: NewMemoryGovernor(NewMemorySystem()), } }
// ===== Layer 1: 预防 ===== func (g *StateOverflowGuard) checkBudget(newTokens int) error { g.currentTokens += newTokens if g.currentTokens > g.tokenBudget { return fmt.Errorf("Token预算耗尽: %d/%d建议开启新会话", g.currentTokens, g.tokenBudget) } return nil }
// ===== Layer 2: 压缩 ===== func (g *StateOverflowGuard) compressIfNeeded(messages []Message) []Message { total := countTokens(messages) threshold := int(float64(g.contextWindow) * 0.8) if total > threshold { return g.compressor.compress(messages) } return messages }
// ===== Layer 3: 淘汰 ===== func (g *StateOverflowGuard) evictIfNeeded(messages []Message) []Message { total := countTokens(messages) threshold := int(float64(g.contextWindow) * 0.6) if total > threshold { return g.memoryGovernor.evictMessages(messages) } return messages }
// ===== Layer 4: 兜底 ===== func (g *StateOverflowGuard) emergencyHandler(err error) interface{} { if isContextOverflow(err) { return g.compressor.emergencyCompress() } if isBudgetExceeded(err) { return map[string]interface{}{ "action": "degrade", "message": "对话过长,建议开启新会话", "summary": g.memoryGovernor.archiveSession(), } } return nil }
// ===== 完整流程 ===== func (g *StateOverflowGuard) processTurn(messages []Message, newTokens int) interface{} { err := g.checkBudget(newTokens) if err != nil { return g.emergencyHandler(err) } messages = g.compressIfNeeded(messages) messages = g.evictIfNeeded(messages) return messages }
func countTokens(messages []Message) int { total := 0 for _, m := range messages { total += len(m.Content) / 4 } return total }
func isContextOverflow(err error) bool { return false } func isBudgetExceeded(err error) bool { return false }
import java.util.*;
public class StateOverflowGuard { private int tokenBudget; private int contextWindow; private int currentTokens = 0; private SemanticCompressor compressor; private MemoryGovernor memoryGovernor;
public StateOverflowGuard(int tokenBudget, int contextWindow) { this.tokenBudget = tokenBudget; this.contextWindow = contextWindow; this.compressor = new SemanticCompressor(); this.memoryGovernor = new MemoryGovernor(new MemorySystem()); }
// ===== Layer 1: 预防 ===== public boolean checkBudget(int newTokens) throws BudgetExceededException { currentTokens += newTokens; if (currentTokens > tokenBudget) { throw new BudgetExceededException( "Token预算耗尽: " + currentTokens + "/" + tokenBudget + "建议开启新会话"); } return true; }
// ===== Layer 2: 压缩 ===== public List compressIfNeeded(List messages) { int total = countTokens(messages); int threshold = (int) (contextWindow * 0.8); if (total > threshold) { return compressor.compress(messages); } return messages; }
// ===== Layer 3: 淘汰 ===== public List evictIfNeeded(List messages) { int total = countTokens(messages); int threshold = (int) (contextWindow * 0.6); if (total > threshold) { return memoryGovernor.evictMessages(messages); } return messages; }
// ===== Layer 4: 兜底 ===== public Object emergencyHandler(Exception error) { if (error instanceof ContextOverflowException) { return compressor.emergencyCompress(); } else if (error instanceof BudgetExceededException) { Map result = new HashMap<>(); result.put("action", "degrade"); result.put("message", "对话过长,建议开启新会话"); result.put("summary", memoryGovernor.archiveSession()); return result; } return null; }
// ===== 完整流程 ===== public Object processTurn(List messages, int newTokens) { try { checkBudget(newTokens); messages = compressIfNeeded(messages); messages = evictIfNeeded(messages); return messages; } catch (Exception e) { return emergencyHandler(e); } }
private int countTokens(List messages) { int total = 0; for (Message m : messages) total += m.content.length() / 4; return total; } }
**✅ 面试要点:四层泄压体系答题模板**
面试回答"多轮对话状态爆炸"时,**不要只说"压缩上下文"**,要给系统级方案:
**1 先定义问题**--多轮对话中状态从线性增长变为树状展开(工具嵌套)和倍数增长(推理分叉),导致上下文溢出、注意力稀释、成本爆炸和行为漂移。
**2 再给方案框架**--四层泄压体系递进式处理:
Layer 1 预防:Token Budget全局预算 + JIT召回原则 + 输出长度约束 + 工具结果精简
Layer 2 压缩:语义保护型压缩(锁定System Prompt和行动-结果对)+ 三层触发机制
Layer 3 淘汰:记忆治理策略 + 召回硬限制≤5条 + 工具结果裁剪 + 淘汰前归档
Layer 4 兜底:紧急压缩(413)+ 降级回答 + 自动会话切分 + 人工介入
**3 最后强调**--不是某一层单独起作用,而是四层递进。预防失败→压缩兜住,压缩不够→淘汰兜住,淘汰也不够→紧急兜底。
关联面试题 → Q23(状态爆炸)
关联章节 → 4.13(Token Budget)、5.3(短期记忆管理)、5.5(上下文压缩)、5.8(记忆治理)、5.11(召回防控)、5.12(生产级压缩系统)
## 6.12 工程深度:生产级上下文压缩系统
4.5 节介绍了上下文压缩的概念,但生产级系统远比概念复杂:什么时候压缩?用什么策略压缩?压缩后如何避免频繁重复压缩?压缩失败怎么办?本节基于 WaLiCode 项目的 contextCompressor.ts 和 ChatContextManager.ts 真实实现,讲解生产级上下文压缩系统的六大设计。
### 6.12.1 轮次分割算法
压缩的第一步是理解对话的结构。对话不是扁平的消息列表,而是由多个"轮次"组成--每个轮次以 user message 开始,包含后续的所有 assistant/tool 消息。WaLiCode 的 splitIntoRounds 算法实现了这个分割:
interface ConversationRound { startIndex: number; endIndex: number; messages: ClaudeMessage[]; tokenCount: number; }
function splitIntoRounds(messages: ClaudeMessage[]): ConversationRound[] { const rounds: ConversationRound[] = []; let currentRound: ConversationRound | null = null;
for (let i = 0; i c.type === 'tool_result');
if (!isInlineToolResult && currentRound) { // 结束上一轮 rounds.push(currentRound); currentRound = null; }
if (!currentRound) { currentRound = { startIndex: i, endIndex: i, messages: [msg], tokenCount: estimateMessageTokens(msg), }; continue; } }
// 继续当前轮次 if (currentRound) { currentRound.endIndex = i; currentRound.messages.push(msg); currentRound.tokenCount += estimateMessageTokens(msg); } else { // 开头的 system/assistant 消息 currentRound = { startIndex: i, endIndex: i, messages: [msg], tokenCount: estimateMessageTokens(msg), }; } }
if (currentRound) rounds.push(currentRound); return rounds; }
关键细节:**inline tool_result 不开新轮次**。Anthropic 格式中,工具结果以 `{ role: 'user', content: [{ type: 'tool_result', ... }] }` 形式出现。如果把它当新轮次,会把一个完整的"AI 调工具→获取结果→继续推理"轮次错误地拆成两半。
### 6.12.2 三种压缩触发模式
什么时候该压缩?WaLiCode 设计了三种触发模式,形成三层防线: | 模式 | 触发条件 | 执行方式 | 阻塞对话 | 压缩力度 | | --- | --- | --- | --- | --- | | **主动压缩** | token 占比 > 80% | 同步执行 | 是(短暂) | 中等(保留近期 3-5 轮) | | **后台压缩** | token 占比 > 60% | 异步执行 | 否 | 中等 | | **反应式压缩** | API 返回 413 | 同步紧急执行 | 是 | 激进(只保留 1-2 轮) | ### 6.12.3 AI 摘要 vs 规则摘要
压缩的核心是生成摘要。WaLiCode 支持两种摘要方式:
**AI 摘要(generateAiSummary)**:用专门的压缩 Prompt 让 AI 生成 500 字以内的摘要,保留:用户意图、文件操作、错误信息与解决方案、关键决策、当前进度。质量高但消耗 token。
**规则摘要(generateRuleSummary)**:纯规则提取关键信息,不调 AI。质量低但零成本、零延迟。
// AI 摘要的 System Prompt const COMPRESSION_SYSTEM_PROMPT = `你是一个对话压缩器。将长对话压缩为 500 字以内的摘要。 必须保留:
- 用户的原始意图和需求
- 已执行的关键文件操作(创建/修改/删除)
- 遇到的错误和解决方案
- 做出的重要决策
- 当前进度和下一步计划 可以省略:
- 工具调用的详细参数
- AI 的推理过程
- 重复的失败尝试`;
async function generateAiSummary(messages: ClaudeMessage[]): Promise {
const conversationText = messages.map(msg => {
const content = typeof msg.content === 'string'
? msg.content : JSON.stringify(msg.content);
const label = msg.role === 'user' ? '👤 用户'
: msg.role === 'assistant' ? '🤖 助手'
: msg.role === 'tool' ? '🔧 工具' : '📋 系统';
// 截断过长内容(避免压缩本身消耗太多 token)
const truncated = content.length > 2000
? content.slice(0, 2000) + '\n... (已截断)' : content;
return ${label}:\n${truncated};
}).join('\n\n---\n\n');
try {
let fullResponse = '';
for await (const event of ai.stream([
{ role: 'user', content: COMPRESSION_SYSTEM_PROMPT },
{ role: 'user', content: 请压缩以下对话历史:\n\n${conversationText} },
], undefined, undefined, undefined, { disableTools: true })) {
if (event.type === 'text') fullResponse += event.content || '';
}
return fullResponse || generateRuleSummary(messages);
} catch (err) {
// 降级为规则摘要
return generateRuleSummary(messages);
}
}
// // AI 摘要的 System Prompt
// const COMPRESSION_SYSTEM_PROMPT = 你是一个对话压缩器。将长对话压缩为 500 字以内的摘要。 // 必须保留: // 1. 用户的原始意图和需求 // 2. 已执行的关键文件操作(创建/修改/删除) // 3. 遇到的错误和解决方案 // 4. 做出的重要决策 // 5. 当前进度和下一步计划 // 可以省略: // - 工具调用的详细参数 // - AI 的推理过程 // - 重复的失败尝试;
// async function generateAiSummary(messages: ClaudeMessage[]): Promise {
// const conversationText = messages.map(msg => {
// const content = typeof msg.content === 'string'
// ? msg.content : JSON.stringify(msg.content);
// const label = msg.role === 'user' ? '👤 用户'
// : msg.role === 'assistant' ? '🤖 助手'
// : msg.role === 'tool' ? '🔧 工具' : '📋 系统';
// // 截断过长内容(避免压缩本身消耗太多 token)
// const truncated = content.length > 2000
// ? content.slice(0, 2000) + '\n... (已截断)' : content;
return ${label}:\n${truncated};;
// }).join('\n\n---\n\n');
// try {
// let fullResponse = '';
// for await (const event of ai.stream([
// { role: 'user', content: COMPRESSION_SYSTEM_PROMPT },
// { role: 'user', content: 请压缩以下对话历史:\n\n${conversationText} },
// ], undefined, undefined, undefined, { disableTools: true })) {
// if (event.type === 'text') fullResponse += event.content || '';
// }
return fullResponse || generateRuleSummary(messages);;
// } catch (err) {
// // 降级为规则摘要
return generateRuleSummary(messages);;
// }
// }
package main
import ( "fmt" "os" "os/exec" "strings" )
// Python: // AI 摘要的 System Prompt
// Python: const COMPRESSION_SYSTEM_PROMPT = `你是一个对话压缩器。将长对话压缩为 500 字以内的摘要。
// Python: 必须保留:
// Python: 1. 用户的原始意图和需求
// Python: 2. 已执行的关键文件操作(创建/修改/删除)
// Python: 3. 遇到的错误和解决方案
// Python: 4. 做出的重要决策
// Python: 5. 当前进度和下一步计划
// Python: 可以省略:
// Python: - 工具调用的详细参数
// Python: - AI 的推理过程
// Python: - 重复的失败尝试`;
// Python: async function generateAiSummary(messages: ClaudeMessage[]): Promise {
// Python: const conversationText = messages.map(msg => {
// Python: const content = typeof msg.content === 'string'
// Python: ? msg.content : JSON.stringify(msg.content);
// Python: const label = msg.role === 'user' ? '👤 用户'
// Python: : msg.role === 'assistant' ? '🤖 助手'
// Python: : msg.role === 'tool' ? '🔧 工具' : '📋 系统';
// Python: // 截断过长内容(避免压缩本身消耗太多 token)
// Python: const truncated = content.length > 2000
// Python: ? content.slice(0, 2000) + '\n... (已截断)' : content;
return `${label}:\n${truncated}`;
// Python: }).join('\n\n---\n\n');
// Python: try {
// Python: let fullResponse = '';
// Python: for await (const event of ai.stream([
// Python: { role: 'user', content: COMPRESSION_SYSTEM_PROMPT },
// Python: { role: 'user', content: `请压缩以下对话历史:\n\n${conversationText}` },
// Python: ], undefined, undefined, undefined, { disableTools: true })) {
// Python: if (event.type === 'text') fullResponse += event.content || '';
// Python: }
return fullResponse || generateRuleSummary(messages);
// Python: } catch (err) {
// Python: // 降级为规则摘要
return generateRuleSummary(messages);
// Python: }
// Python: }
import java.util.; import java.util.concurrent.; import java.util.regex.; import java.io.;
// Python: // AI 摘要的 System Prompt
// Python: const COMPRESSION_SYSTEM_PROMPT = 你是一个对话压缩器。将长对话压缩为 500 字以内的摘要。 // Python: 必须保留: // Python: 1. 用户的原始意图和需求 // Python: 2. 已执行的关键文件操作(创建/修改/删除) // Python: 3. 遇到的错误和解决方案 // Python: 4. 做出的重要决策 // Python: 5. 当前进度和下一步计划 // Python: 可以省略: // Python: - 工具调用的详细参数 // Python: - AI 的推理过程 // Python: - 重复的失败尝试;
// Python: async function generateAiSummary(messages: ClaudeMessage[]): Promise {
// Python: const conversationText = messages.map(msg => {
// Python: const content = typeof msg.content === 'string'
// Python: ? msg.content : JSON.stringify(msg.content);
// Python: const label = msg.role === 'user' ? '👤 用户'
// Python: : msg.role === 'assistant' ? '🤖 助手'
// Python: : msg.role === 'tool' ? '🔧 工具' : '📋 系统';
// Python: // 截断过长内容(避免压缩本身消耗太多 token)
// Python: const truncated = content.length > 2000
// Python: ? content.slice(0, 2000) + '\n... (已截断)' : content;
return ${label}:\n${truncated};;
// Python: }).join('\n\n---\n\n');
// Python: try {
// Python: let fullResponse = '';
// Python: for await (const event of ai.stream([
// Python: { role: 'user', content: COMPRESSION_SYSTEM_PROMPT },
// Python: { role: 'user', content: 请压缩以下对话历史:\n\n${conversationText} },
// Python: ], undefined, undefined, undefined, { disableTools: true })) {
// Python: if (event.type === 'text') fullResponse += event.content || '';
// Python: }
return fullResponse || generateRuleSummary(messages);;
// Python: } catch (err) {
// Python: // 降级为规则摘要
return generateRuleSummary(messages);;
// Python: }
// Python: }
}
**降级策略**:AI 摘要失败(API 错误、超时、返回空)时自动降级为规则摘要。为什么不直接报错?因为压缩是基础设施--如果压缩本身都能失败,那对话就没法继续了。规则摘要虽然质量低,但至少保留了基本信息,让对话能继续。
### 6.12.4 增量压缩
问题:如果每次 token 超过 80% 就压缩,压缩后降到 50%,用户继续对话又涨到 80%,又要压缩......这会导致频繁压缩,每次压缩都消耗 token(AI 摘要本身也要花 token)。
WaLiCode 的解决方案:**增量压缩阈值**--只有当 token 再次上升到上次压缩水平的 80% 时才再次压缩。
class ContextCompressor { private lastCompressTokenLevel: number = 0; // 上次压缩时的 token 水平
shouldCompress(currentTokens: number, maxTokens: number): boolean { const ratio = currentTokens / maxTokens;
// 首次压缩:80% 阈值 if (this.lastCompressTokenLevel === 0) { return ratio >= 0.8; }
// 后续压缩:必须超过上次压缩水平的 80% // 例如:上次在 80000 token 时压缩,这次必须到 80000 * 0.8 = 64000 才再压缩 // 但由于压缩后降到了 ~50000,用户需要再聊 14000 token 才触发 const threshold = this.lastCompressTokenLevel * 0.8; return currentTokens >= threshold; }
async compress(messages: ClaudeMessage[], maxTokens: number) { const summary = await generateAiSummary(messages); this.lastCompressTokenLevel = countTokens(messages); return this.applySummary(messages, summary); } }
// class ContextCompressor { // private lastCompressTokenLevel: number = 0; // 上次压缩时的 token 水平
// shouldCompress(currentTokens: number, maxTokens: number): boolean { // const ratio = currentTokens / maxTokens;
// // 首次压缩:80% 阈值 // if (this.lastCompressTokenLevel === 0) { return ratio >= 0.8;; // }
// // 后续压缩:必须超过上次压缩水平的 80% // // 例如:上次在 80000 token 时压缩,这次必须到 80000 * 0.8 = 64000 才再压缩 // // 但由于压缩后降到了 ~50000,用户需要再聊 14000 token 才触发 // const threshold = this.lastCompressTokenLevel * 0.8; return currentTokens >= threshold;; // }
// async compress(messages: ClaudeMessage[], maxTokens: number) { // const summary = await generateAiSummary(messages); // this.lastCompressTokenLevel = countTokens(messages); return this.applySummary(messages, summary);; // } // }
package main
import ( "fmt" "os" "os/exec" "strings" )
// ContextCompressor - CLI Agent class type ContextCompressor struct { // Python: private lastCompressTokenLevel: number = 0; // 上次压缩时的 token 水平
// Python: shouldCompress(currentTokens: number, maxTokens: number): boolean {
// Python: const ratio = currentTokens / maxTokens;
// Python: // 首次压缩:80% 阈值
// Python: if (this.lastCompressTokenLevel === 0) {
return ratio >= 0.8;
// Python: }
// Python: // 后续压缩:必须超过上次压缩水平的 80%
// Python: // 例如:上次在 80000 token 时压缩,这次必须到 80000 * 0.8 = 64000 才再压缩
// Python: // 但由于压缩后降到了 ~50000,用户需要再聊 14000 token 才触发
// Python: const threshold = this.lastCompressTokenLevel * 0.8;
return currentTokens >= threshold;
// Python: }
// Python: async compress(messages: ClaudeMessage[], maxTokens: number) {
// Python: const summary = await generateAiSummary(messages);
// Python: this.lastCompressTokenLevel = countTokens(messages);
return this.applySummary(messages, summary);
// Python: }
// Python: }
}
import java.util.; import java.util.concurrent.; import java.util.regex.; import java.io.;
public class ContextCompressor { // Python: private lastCompressTokenLevel: number = 0; // 上次压缩时的 token 水平
// Python: shouldCompress(currentTokens: number, maxTokens: number): boolean { // Python: const ratio = currentTokens / maxTokens;
// Python: // 首次压缩:80% 阈值 // Python: if (this.lastCompressTokenLevel === 0) { return ratio >= 0.8;; // Python: }
// Python: // 后续压缩:必须超过上次压缩水平的 80% // Python: // 例如:上次在 80000 token 时压缩,这次必须到 80000 * 0.8 = 64000 才再压缩 // Python: // 但由于压缩后降到了 ~50000,用户需要再聊 14000 token 才触发 // Python: const threshold = this.lastCompressTokenLevel * 0.8; return currentTokens >= threshold;; // Python: }
// Python: async compress(messages: ClaudeMessage[], maxTokens: number) { // Python: const summary = await generateAiSummary(messages); // Python: this.lastCompressTokenLevel = countTokens(messages); return this.applySummary(messages, summary);; // Python: } // Python: } } }
### 6.12.5 Provider/Reducer 可插拔架构
WaLiCode 的上下文管理采用**Provider/Reducer 可插拔架构**,这是生产级系统与 Demo 级系统的核心区别:
**Provider 模式**:可插拔的上下文来源,每个 Provider 负责一种类型的上下文。
interface ContextProvider { name: string; provide(): MessageContext[]; }
// 文件上下文:当前打开的文件、最近修改的文件
class FileProvider implements ContextProvider {
name = 'FileProvider';
provide(): MessageContext[] {
return openFiles.map(f => ({
role: 'system',
content: 当前文件: ${f.path}\n${f.content.slice(0, 2000)},
priority: MessagePriority.High,
}));
}
}
// 任务上下文:当前任务的目标和进度
class TaskProvider implements ContextProvider {
name = 'TaskProvider';
provide(): MessageContext[] {
return [{
role: 'system',
content: 当前任务: ${currentTask},
priority: MessagePriority.Critical,
}];
}
}
// 里程碑上下文:关键决策记录
class MilestoneProvider implements ContextProvider {
name = 'MilestoneProvider';
provide(): MessageContext[] {
return milestones.map(m => ({
role: 'system',
content: [${m.type}] ${m.summary},
priority: MessagePriority.High,
}));
}
}
// interface ContextProvider { // name: string; // provide(): MessageContext[]; // }
// // 文件上下文:当前打开的文件、最近修改的文件
// class FileProvider implements ContextProvider {
const name = 'FileProvider';;
// provide(): MessageContext[] {
return openFiles.map(f => ({;
// role: 'system',
// content: 当前文件: ${f.path}\n${f.content.slice(0, 2000)},
// priority: MessagePriority.High,
// }));
// }
// }
// // 任务上下文:当前任务的目标和进度
// class TaskProvider implements ContextProvider {
const name = 'TaskProvider';;
// provide(): MessageContext[] {
return [{;
// role: 'system',
// content: 当前任务: ${currentTask},
// priority: MessagePriority.Critical,
// }];
// }
// }
// // 里程碑上下文:关键决策记录
// class MilestoneProvider implements ContextProvider {
const name = 'MilestoneProvider';;
// provide(): MessageContext[] {
return milestones.map(m => ({;
// role: 'system',
// content: [${m.type}] ${m.summary},
// priority: MessagePriority.High,
// }));
// }
// }
package main
import ( "fmt" "os" "os/exec" "strings" )
// Python: interface ContextProvider {
// Python: name: string;
// Python: provide(): MessageContext[];
// Python: }
// Python: // 文件上下文:当前打开的文件、最近修改的文件
// FileProvider - CLI Agent class
type FileProvider struct {
// Python: name = 'FileProvider';
// Python: provide(): MessageContext[] {
return openFiles.map(f => ({
// Python: role: 'system',
// Python: content: 当前文件: ${f.path}\n${f.content.slice(0, 2000)},
// Python: priority: MessagePriority.High,
// Python: }));
// Python: }
// Python: }
// Python: // 任务上下文:当前任务的目标和进度
// TaskProvider - CLI Agent class
type TaskProvider struct {
// Python: name = 'TaskProvider';
// Python: provide(): MessageContext[] {
return [{
// Python: role: 'system',
// Python: content: 当前任务: ${currentTask},
// Python: priority: MessagePriority.Critical,
// Python: }];
// Python: }
// Python: }
// Python: // 里程碑上下文:关键决策记录
// MilestoneProvider - CLI Agent class
type MilestoneProvider struct {
// Python: name = 'MilestoneProvider';
// Python: provide(): MessageContext[] {
return milestones.map(m => ({
// Python: role: 'system',
// Python: content: [${m.type}] ${m.summary},
// Python: priority: MessagePriority.High,
// Python: }));
// Python: }
// Python: }
}
import java.util.; import java.util.concurrent.; import java.util.regex.; import java.io.;
// Python: interface ContextProvider { // Python: name: string; // Python: provide(): MessageContext[]; // Python: }
// Python: // 文件上下文:当前打开的文件、最近修改的文件
public class FileProvider {
// Python: name = 'FileProvider';
// Python: provide(): MessageContext[] {
return openFiles.map(f => ({;
// Python: role: 'system',
// Python: content: 当前文件: ${f.path}\n${f.content.slice(0, 2000)},
// Python: priority: MessagePriority.High,
// Python: }));
// Python: }
// Python: }
// Python: // 任务上下文:当前任务的目标和进度
public class TaskProvider {
// Python: name = 'TaskProvider';
// Python: provide(): MessageContext[] {
return [{;
// Python: role: 'system',
// Python: content: 当前任务: ${currentTask},
// Python: priority: MessagePriority.Critical,
// Python: }];
// Python: }
// Python: }
// Python: // 里程碑上下文:关键决策记录
public class MilestoneProvider {
// Python: name = 'MilestoneProvider';
// Python: provide(): MessageContext[] {
return milestones.map(m => ({;
// Python: role: 'system',
// Python: content: [${m.type}] ${m.summary},
// Python: priority: MessagePriority.High,
// Python: }));
// Python: }
// Python: }
}
}
**Reducer 模式**:可插拔的裁剪策略,按不同策略裁剪消息。
interface MessageReducer { name: string; reduce(messages: MessageContext[], maxTokens: number): MessageContext[]; }
class PriorityReducer implements MessageReducer { name = 'PriorityReducer';
reduce(messages: MessageContext[], maxTokens: number): MessageContext[] { // 第一轮:Critical 和 High 无条件保留 // 第二轮:按优先级从高到低,在预算内逐条追加 // 第三轮:按时间排序输出 const result: MessageContext[] = []; let usedChars = 0; const maxChars = maxTokens * 4; // ~4 chars/token
// Critical + High 无条件保留 for (const msg of messages) { if (msg.priority === MessagePriority.Critical || msg.priority === MessagePriority.High) { result.push(msg); usedChars += msg.content.length; } }
// 按优先级填充剩余预算 for (const priority of [MessagePriority.Medium, MessagePriority.Low]) { for (const msg of messages) { if (msg.priority !== priority || result.includes(msg)) continue; if (usedChars + msg.content.length > maxChars) continue; result.push(msg); usedChars += msg.content.length; } }
// 按时间排序 return result.sort((a, b) => a.timestamp - b.timestamp); } }
// interface MessageReducer { // name: string; // reduce(messages: MessageContext[], maxTokens: number): MessageContext[]; // }
// class PriorityReducer implements MessageReducer { const name = 'PriorityReducer';;
// reduce(messages: MessageContext[], maxTokens: number): MessageContext[] { // // 第一轮:Critical 和 High 无条件保留 // // 第二轮:按优先级从高到低,在预算内逐条追加 // // 第三轮:按时间排序输出 // const result: MessageContext[] = []; // let usedChars = 0; // const maxChars = maxTokens * 4; // ~4 chars/token
// // Critical + High 无条件保留 // for (const msg of messages) { // if (msg.priority === MessagePriority.Critical // || msg.priority === MessagePriority.High) { // result.push(msg); // usedChars += msg.content.length; // } // }
// // 按优先级填充剩余预算 // for (const priority of [MessagePriority.Medium, MessagePriority.Low]) { // for (const msg of messages) { // if (msg.priority !== priority || result.includes(msg)) continue; // if (usedChars + msg.content.length > maxChars) continue; // result.push(msg); // usedChars += msg.content.length; // } // }
// // 按时间排序 return result.sort((a, b) => a.timestamp - b.timestamp);; // } // }
package main
import ( "fmt" "os" "os/exec" "strings" )
// Python: interface MessageReducer {
// Python: name: string;
// Python: reduce(messages: MessageContext[], maxTokens: number): MessageContext[];
// Python: }
// PriorityReducer - CLI Agent class type PriorityReducer struct { // Python: name = 'PriorityReducer';
// Python: reduce(messages: MessageContext[], maxTokens: number): MessageContext[] {
// Python: // 第一轮:Critical 和 High 无条件保留
// Python: // 第二轮:按优先级从高到低,在预算内逐条追加
// Python: // 第三轮:按时间排序输出
// Python: const result: MessageContext[] = [];
// Python: let usedChars = 0;
// Python: const maxChars = maxTokens * 4; // ~4 chars/token
// Python: // Critical + High 无条件保留
// Python: for (const msg of messages) {
// Python: if (msg.priority === MessagePriority.Critical
// Python: || msg.priority === MessagePriority.High) {
// Python: result.push(msg);
// Python: usedChars += msg.content.length;
// Python: }
// Python: }
// Python: // 按优先级填充剩余预算
// Python: for (const priority of [MessagePriority.Medium, MessagePriority.Low]) {
// Python: for (const msg of messages) {
// Python: if (msg.priority !== priority || result.includes(msg)) continue;
// Python: if (usedChars + msg.content.length > maxChars) continue;
// Python: result.push(msg);
// Python: usedChars += msg.content.length;
// Python: }
// Python: }
// Python: // 按时间排序
return result.sort((a, b) => a.timestamp - b.timestamp);
// Python: }
// Python: }
}
import java.util.; import java.util.concurrent.; import java.util.regex.; import java.io.;
// Python: interface MessageReducer { // Python: name: string; // Python: reduce(messages: MessageContext[], maxTokens: number): MessageContext[]; // Python: }
public class PriorityReducer { // Python: name = 'PriorityReducer';
// Python: reduce(messages: MessageContext[], maxTokens: number): MessageContext[] { // Python: // 第一轮:Critical 和 High 无条件保留 // Python: // 第二轮:按优先级从高到低,在预算内逐条追加 // Python: // 第三轮:按时间排序输出 // Python: const result: MessageContext[] = []; // Python: let usedChars = 0; // Python: const maxChars = maxTokens * 4; // ~4 chars/token
// Python: // Critical + High 无条件保留 // Python: for (const msg of messages) { // Python: if (msg.priority === MessagePriority.Critical // Python: || msg.priority === MessagePriority.High) { // Python: result.push(msg); // Python: usedChars += msg.content.length; // Python: } // Python: }
// Python: // 按优先级填充剩余预算 // Python: for (const priority of [MessagePriority.Medium, MessagePriority.Low]) { // Python: for (const msg of messages) { // Python: if (msg.priority !== priority || result.includes(msg)) continue; // Python: if (usedChars + msg.content.length > maxChars) continue; // Python: result.push(msg); // Python: usedChars += msg.content.length; // Python: } // Python: }
// Python: // 按时间排序 return result.sort((a, b) => a.timestamp - b.timestamp);; // Python: } // Python: } } }
消息优先级推断规则:
- **Critical**:错误消息、系统指令、当前任务
- **High**:文件操作、用户配置、里程碑事件
- **Medium**:工具调用结果、AI 推理过程
- **Low**:AI 长文本解释、闲聊内容
为什么用可插拔设计?因为不同场景需要不同的 Provider 和 Reducer 组合。编码场景需要 FileProvider + TaskProvider + PriorityReducer;聊天场景可能只需要 MemoryProvider + SlidingWindowReducer。写死在代码里意味着每次改需求都要改核心代码,可插拔意味着只需注册新组件。
### 6.12.6 双作用域记忆系统
WaLiCode 的记忆系统分为两个作用域: | 维度 | User Scope(用户级) | Project Scope(项目级) | | --- | --- | --- | | 作用范围 | 跨项目共享 | 仅当前项目 | | 存储内容 | 用户偏好、通用知识、历史决策 | 架构决策、模块依赖、编码规范 | | 容量限制 | LRU 淘汰,最多 20 条 | LRU 淘汰,最多 20 条 | | 示例 | "用户喜欢用 TypeScript" | "本项目用 pnpm,不要用 npm" | ```
class MemoryService {
private userMemory: Map = new Map();
private projectMemory: Map = new Map();
private readonly MAX_ENTRIES = 20;
set(scope: 'user' | 'project', key: string, value: string) {
const store = scope === 'user' ? this.userMemory : this.projectMemory;
// LRU 淘汰:超过 20 条时删除最久未访问的
if (store.size >= this.MAX_ENTRIES && !store.has(key)) {
const oldest = [...store.entries()]
.sort((a, b) => a[1].lastAccessed - b[1].lastAccessed)[0];
store.delete(oldest[0]);
}
store.set(key, { value, lastAccessed: Date.now() });
}
get(scope: 'user' | 'project', key: string): string | undefined {
const store = scope === 'user' ? this.userMemory : this.projectMemory;
const entry = store.get(key);
if (entry) entry.lastAccessed = Date.now(); // LRU 更新
return entry?.value;
}
// 生成记忆 Prompt 注入
toPrompt(scope: 'user' | 'project' | 'both'): string {
const entries: MemoryEntry[] = [];
if (scope === 'user' || scope === 'both') {
entries.push(...this.userMemory.values());
}
if (scope === 'project' || scope === 'both') {
entries.push(...this.projectMemory.values());
}
return entries.map(e => `- ${e.value}`).join('\n');
}
}
// class MemoryService {
// private userMemory: Map = new Map();
// private projectMemory: Map = new Map();
// private readonly MAX_ENTRIES = 20;
// set(scope: 'user' | 'project', key: string, value: string) {
// const store = scope === 'user' ? this.userMemory : this.projectMemory;
// // LRU 淘汰:超过 20 条时删除最久未访问的
// if (store.size >= this.MAX_ENTRIES && !store.has(key)) {
// const oldest = [...store.entries()]
// .sort((a, b) => a[1].lastAccessed - b[1].lastAccessed)[0];
// store.delete(oldest[0]);
// }
// store.set(key, { value, lastAccessed: Date.now() });
// }
// get(scope: 'user' | 'project', key: string): string | undefined {
// const store = scope === 'user' ? this.userMemory : this.projectMemory;
// const entry = store.get(key);
// if (entry) entry.lastAccessed = Date.now(); // LRU 更新
return entry?.value;;
// }
// // 生成记忆 Prompt 注入
// toPrompt(scope: 'user' | 'project' | 'both'): string {
// const entries: MemoryEntry[] = [];
// if (scope === 'user' || scope === 'both') {
// entries.push(...this.userMemory.values());
// }
// if (scope === 'project' || scope === 'both') {
// entries.push(...this.projectMemory.values());
// }
return entries.map(e => `- ${e.value}`).join('\n');;
// }
// }
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// MemoryService - CLI Agent class
type MemoryService struct {
// Python: private userMemory: Map = new Map();
// Python: private projectMemory: Map = new Map();
// Python: private readonly MAX_ENTRIES = 20;
// Python: set(scope: 'user' | 'project', key: string, value: string) {
// Python: const store = scope === 'user' ? this.userMemory : this.projectMemory;
// Python: // LRU 淘汰:超过 20 条时删除最久未访问的
// Python: if (store.size >= this.MAX_ENTRIES && !store.has(key)) {
// Python: const oldest = [...store.entries()]
// Python: .sort((a, b) => a[1].lastAccessed - b[1].lastAccessed)[0];
// Python: store.delete(oldest[0]);
// Python: }
// Python: store.set(key, { value, lastAccessed: Date.now() });
// Python: }
// Python: get(scope: 'user' | 'project', key: string): string | undefined {
// Python: const store = scope === 'user' ? this.userMemory : this.projectMemory;
// Python: const entry = store.get(key);
// Python: if (entry) entry.lastAccessed = Date.now(); // LRU 更新
return entry?.value;
// Python: }
// Python: // 生成记忆 Prompt 注入
// Python: toPrompt(scope: 'user' | 'project' | 'both'): string {
// Python: const entries: MemoryEntry[] = [];
// Python: if (scope === 'user' || scope === 'both') {
// Python: entries.push(...this.userMemory.values());
// Python: }
// Python: if (scope === 'project' || scope === 'both') {
// Python: entries.push(...this.projectMemory.values());
// Python: }
return entries.map(e => `- ${e.value}`).join('\n');
// Python: }
// Python: }
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
public class MemoryService {
// Python: private userMemory: Map = new Map();
// Python: private projectMemory: Map = new Map();
// Python: private readonly MAX_ENTRIES = 20;
// Python: set(scope: 'user' | 'project', key: string, value: string) {
// Python: const store = scope === 'user' ? this.userMemory : this.projectMemory;
// Python: // LRU 淘汰:超过 20 条时删除最久未访问的
// Python: if (store.size >= this.MAX_ENTRIES && !store.has(key)) {
// Python: const oldest = [...store.entries()]
// Python: .sort((a, b) => a[1].lastAccessed - b[1].lastAccessed)[0];
// Python: store.delete(oldest[0]);
// Python: }
// Python: store.set(key, { value, lastAccessed: Date.now() });
// Python: }
// Python: get(scope: 'user' | 'project', key: string): string | undefined {
// Python: const store = scope === 'user' ? this.userMemory : this.projectMemory;
// Python: const entry = store.get(key);
// Python: if (entry) entry.lastAccessed = Date.now(); // LRU 更新
return entry?.value;;
// Python: }
// Python: // 生成记忆 Prompt 注入
// Python: toPrompt(scope: 'user' | 'project' | 'both'): string {
// Python: const entries: MemoryEntry[] = [];
// Python: if (scope === 'user' || scope === 'both') {
// Python: entries.push(...this.userMemory.values());
// Python: }
// Python: if (scope === 'project' || scope === 'both') {
// Python: entries.push(...this.projectMemory.values());
// Python: }
return entries.map(e => `- ${e.value}`).join('\n');;
// Python: }
// Python: }
}
}
6.12.6 压缩可靠性保障
上下文压缩是 Agent 对话稳定性的最后一道防线,但如果压缩本身不可靠,反而会成为新的问题源。生产环境中必须解决三个可靠性挑战:压缩失败回退、语义丢失检测、压缩对对话连贯性的影响。
挑战 1：压缩失败回退
AI 摘要依赖 LLM 生成,而 LLM 可能超时、限流或返回乱码。WaLiCode 的做法是三层降级链: | 层级 | 触发条件 | 策略 | 代价 | | --- | --- | --- | --- | | L1 AI 摘要 | 正常流程 | 调 LLM 生成 500 字以内摘要 | 消耗 ~2000 token | | L2 规则摘要 | AI 超时/限流/JSON 解析失败 | 正则提取关键信息（用户意图/文件操作/错误信息/决策） | 质量较低,零 token | | L3 硬截断 | 规则摘要也失败（极端情况） | 只保留最近 2 轮对话 + system prompt | 丢失全部历史 | ``` class CompressionFallbackChain: """三层降级压缩链"""
def init(self, llm_client): self.llm = llm_client self.max_ai_attempts = 2 self.max_ai_tokens = 2000
def compress(self, messages: list, target_ratio: float = 0.3) -> dict: """压缩消息列表,返回 {success, strategy, compressed, error}"""
L1: 尝试 AI 摘要
for attempt in range(self.max_ai_attempts): try: result = self._ai_compress(messages, target_ratio) if result and self._validate(result): return {"success": True, "strategy": "ai", "compressed": result} except (TimeoutError, RateLimitError) as e: print(f"AI 摘要第{attempt+1}次失败: {e}") continue except Exception as e: print(f"AI 摘要异常: {e}") break
L2: 降级为规则摘要
try: result = self._rule_compress(messages) if result: return {"success": True, "strategy": "rule", "compressed": result} except Exception as e: print(f"规则摘要失败: {e}")
L3: 硬截断
return { "success": True, "strategy": "truncate", "compressed": messages[:4], # system + 最近2轮 "warning": "硬截断,历史已丢失" }
def _ai_compress(self, messages, ratio): """AI 摘要""" prompt = f"将以下对话压缩为500字以内的摘要,保留:用户意图/文件操作/错误与修复/关键决策/当前进度\n\n{messages}" return self.llm.chat(prompt, max_tokens=self.max_ai_tokens)
def _rule_compress(self, messages): """规则摘要:正则提取关键信息""" import re summary_parts = [] for msg in messages: content = msg.get("content", "")
提取用户意图
if msg["role"] == "user": summary_parts.append(f"[用户] {content[:100]}")
提取错误信息
elif "error" in content.lower() or "报错" in content: error_match = re.search(r'(error|报错|异常)[:\s]*([^\n]+)', content, re.I) if error_match: summary_parts.append(f"[错误] {error_match.group(2)[:80]}")
提取文件操作
file_match = re.search(r'(创建|修改|删除|读取)[文件了]*(.+?.[a-z]+)', content) if file_match: summary_parts.append(f"[文件] {file_match.group(0)[:80]}") return "\n".join(summary_parts) if summary_parts else None
def _validate(self, result): """校验摘要质量""" if not result or len(result) 2000: return False return True
interface CompressionResult { success: boolean; strategy: 'ai' | 'rule' | 'truncate'; compressed: Message[]; warning?: string; }
class CompressionFallbackChain { private llm: LLMClient; private maxAiAttempts = 2;
constructor(llm: LLMClient) { this.llm = llm; }
async compress(messages: Message[], targetRatio = 0.3): Promise {
// L1: AI 摘要
for (let i = 0; i {
const prompt = 将以下对话压缩为500字以内的摘要...\n\n${JSON.stringify(messages)};
const result = await this.llm.chat(prompt, { maxTokens: 2000 });
return [{ role: 'system', content: [历史摘要] ${result} }];
}
private ruleCompress(messages: Message[]): Message[] | null {
const parts: string[] = [];
for (const msg of messages) {
if (msg.role === 'user') parts.push([用户] ${msg.content.slice(0, 100)});
if (/error|报错|异常/i.test(msg.content)) {
const m = msg.content.match(/(error|报错|异常)[:\s]*([^\n]+)/i);
if (m) parts.push([错误] ${m[2].slice(0, 80)});
}
}
return parts.length ? [{ role: 'system', content: [历史摘要] ${parts.join('\n')} }] : null;
}
private validate(result: Message[]): boolean { const content = result[0]?.content ?? ''; return content.length >= 20 && content.length 2 { parts = append(parts, "[错误] "+truncate(m[2], 80)) } } } if len(parts) == 0 { return nil, errors.New("无关键信息") } return []Message{{Role: "system", Content: "[历史摘要] " + strings.Join(parts, "\n")}}, nil }
func (c *CompressionFallbackChain) validate(result []Message) bool { if len(result) == 0 { return false } content := result[0].Content return len(content) >= 20 && len(content) > compressed; String warning; }
class CompressionFallbackChain { private final LLMClient llm; private final int maxAiAttempts = 2;
public CompressionFallbackChain(LLMClient llm) { this.llm = llm; }
public CompressionResult compress(List> messages, double targetRatio) { // L1: AI 摘要 for (int i = 0; i > result = aiCompress(messages, targetRatio); if (result != null && validate(result)) { CompressionResult r = new CompressionResult(); r.success = true; r.strategy = "ai"; r.compressed = result; return r; } } catch (Exception e) { System.err.println("AI 摘要第" + (i + 1) + "次失败: " + e.getMessage()); } }
// L2: 规则摘要 try { List> result = ruleCompress(messages); if (result != null) { CompressionResult r = new CompressionResult(); r.success = true; r.strategy = "rule"; r.compressed = result; return r; } } catch (Exception e) { System.err.println("规则摘要失败: " + e.getMessage()); }
// L3: 硬截断 CompressionResult r = new CompressionResult(); r.success = true; r.strategy = "truncate"; int endIdx = Math.min(4, messages.size()); r.compressed = new ArrayList<>(messages.subList(0, endIdx)); r.warning = "硬截断,历史已丢失"; return r; }
private List> aiCompress(List> messages, double ratio) throws Exception { String prompt = "将以下对话压缩为500字以内的摘要...\n\n" + messages.toString(); String result = llm.chat(prompt, 2000); if (result == null || result.length() 2000) { return null; } return List.of(Map.of("role", "system", "content", "[历史摘要] " + result)); }
private List> ruleCompress(List> messages) { List parts = new ArrayList<>(); Pattern errorPattern = Pattern.compile("(?i)(error|报错|异常)[:\\s]*([^\\n]+)"); for (Map msg : messages) { String content = msg.getOrDefault("content", ""); if ("user".equals(msg.get("role"))) { parts.add("[用户] " + content.substring(0, Math.min(100, content.length()))); } Matcher m = errorPattern.matcher(content); if (m.find()) { parts.add("[错误] " + m.group(2).substring(0, Math.min(80, m.group(2).length()))); } } if (parts.isEmpty()) return null; return List.of(Map.of("role", "system", "content", "[历史摘要] " + String.join("\n", parts))); }
private boolean validate(List> result) { if (result == null || result.isEmpty()) return false; String content = result.get(0).getOrDefault("content", ""); return content.length() >= 20 && content.length() !constraintsAfter.includes(c));
if (lost.length > 0) {
// 将丢失的约束重新注入摘要
summary = [重要约束] ${lost.join('; ')}\n + summary;
}
**实现要点**:约束提取用正则匹配常见模式（"不要..."/"必须..."/"用..."/"目标..."）,不依赖 LLM。这是一个**低成本高收益**的保障措施——零额外 API 调用,只需几十行正则代码,就能避免最常见的压缩事故。
#### 挑战 3：压缩对对话连贯性的影响
压缩会改变上下文结构,可能导致 LLM 的回答出现**断层**——比如用户问"刚才那个文件改好了吗?",但压缩后历史中已经没有文件操作的记录了。三种应对策略: | 策略 | 做法 | 适用场景 | | --- | --- | --- | | **压缩标记注入** | 在压缩位置插入 `[已压缩 N 轮对话]` 标记,让 LLM 知道有历史被省略 | 大多数场景 | | **关键实体保留** | 压缩时强制保留最近提及的文件名、函数名、变量名,即使轮次被压缩 | 代码 Agent / IDE Agent | | **按需展开** | 用户追问历史细节时,从向量库检索被压缩的原始对话片段,临时注入上下文 | 长对话 > 50 轮 | **🔗 5.12 核心要点**
**轮次分割**:按 user message 分割对话为多个轮次,inline tool_result 不开新轮次。
**三层压缩触发**:主动(80%)→ 后台(60%)→ 反应式(413),层层递进。
**AI 摘要 vs 规则摘要**:AI 质量高但贵,规则零成本但糙,AI 失败降级为规则。
**增量压缩**:上次压缩水平的 80% 才再压缩,避免频繁压缩浪费 token。
**Provider/Reducer 架构**:可插拔设计,按场景组合不同的上下文来源和裁剪策略。
**双作用域记忆**:User Scope 跨项目,Project Scope 仅当前项目,LRU 淘汰。
## 6.13 主流记忆框架对比（2026更新）
前面几节讲了记忆系统的原理和工程实践。2024-2025 年,一批专门的**记忆框架**崛起,它们把记忆管理从 Agent 主逻辑中解耦出来,提供开箱即用的记忆层。面试中经常被问到:"你了解哪些记忆框架?它们有什么区别?"
### Mem0:三层记忆架构（Episodic + Semantic + Procedural）
Mem0 是 2024 年开源的记忆框架,核心理念是把记忆按**性质**分为三层,而非简单堆在向量库里:
**💡 Mem0 的三层记忆**
- **Episodic（情景记忆）**:存储具体的交互片段,如"用户上周问了X问题,Agent回答了Y"。按时间线组织,支持回溯。
- **Semantic（语义记忆）**:从交互中提取的结构化知识,如"用户偏好简洁风格""用户是Java开发者"。去重后存储,支持增量更新。
- **Procedural（过程记忆）**:从历史交互中总结的"怎么做"类知识,如"处理JSON解析错误时,先检查编码再检查格式"。类似人类的技能记忆。
Mem0 的工作流程:**对话输入 → LLM 提取记忆 → 分类（episodic/semantic/procedural）→ 去重/合并 → 存储**。检索时按当前意图从三层记忆分别召回,合并后注入上下文。面试要点:Mem0 的三层不是物理分区,而是**逻辑分类**,底层可以共用同一个向量库。
### Letta（原 MemGPT）:自编辑记忆 + 记忆分页
Letta 是 MemGPT 团队的商业化产品,核心创新是**把操作系统的虚拟内存概念引入 Agent 记忆**:
**💡 Letta 的两大核心机制**
- **自编辑记忆（Self-Editing Memory）**:Agent 可以主动修改自己的记忆。当发现旧记忆过时或矛盾时,LLM 自行调用 memory_edit 工具更新记忆内容,而非简单追加。类似人类"纠正记忆"的过程。
- **记忆分页（Memory Paging）**:把记忆分为**Core Memory**（常驻上下文,类似RAM）和**Archival Memory**（外部存储,类似磁盘）。Core Memory 满了时,自动把低优先级记忆"换出"到 Archival,需要时再"换入"。这和操作系统的页面置换机制几乎一模一样。
面试要点:Letta 的记忆分页解决的是**有限上下文窗口 vs 无限记忆需求**的矛盾。Core Memory 放最关键的用户信息和当前任务状态（通常 < 2000 token）,Archival Memory 放历史交互。LLM 通过 search_archival 和 edit_core_memory 两个工具自主管理记忆流转。
### Zep:时序知识图谱 + 自动摘要
Zep 走了一条完全不同的路线——用**知识图谱**组织记忆,并引入**时间维度**:
**💡 Zep 的两个核心设计**
- **时序知识图谱（Temporal Knowledge Graph）**:把对话中的实体、关系、事件抽取为三元组,并附带时间戳。当用户说"我上个月换了一台Mac"时,Zep 不会只存原始文本,而是建一条 (用户, 换设备, Mac, 2025-06) 的图谱边。后续查询时可以回答"用户什么时候换的电脑"这类时序问题。
- **自动摘要（Auto-Summarization）**:每轮对话后自动生成摘要,摘要本身也会被图谱化。旧摘要会随时间衰减,新摘要权重更高。这解决了"半年前的摘要和昨天的摘要权重一样"的问题。
Zep 的优势在于**结构化检索**。向量检索只能回答"什么和X相关",Zep 的图谱检索还能回答"X是什么时候发生的""X和Y是什么关系""X最近有变化吗"。代价是抽取质量依赖 NLP 能力,复杂场景下图谱构建成本较高。
### 三者对比 | 维度 | Mem0 | Letta (MemGPT) | Zep | | --- | --- | --- | --- | | 记忆组织 | 三层分类（episodic/semantic/procedural） | 两级分页（core/archival） | 时序知识图谱 | | 记忆更新 | LLM 提取 + 去重合并 | Agent 自编辑（工具调用） | 图谱增量 + 自动摘要 | | 检索方式 | 向量相似度（分层召回） | Core 直接注入 + Archival 语义搜索 | 图谱查询 + 向量检索 + 时序过滤 | | 核心优势 | 分类清晰,开箱即用 | Agent 自主管理记忆,最像OS | 支持时序查询,结构化最强 | | 适用场景 | 通用对话 Agent、客服 | 长期伴随型 Agent、个人助手 | 需要时序推理、实体关系追踪 | | 开源协议 | Apache 2.0 | Apache 2.0 | Apache 2.0 | **🔗 5.13 核心要点**
**Mem0**:三层记忆（episodic/semantic/procedural）按性质分类,LLM 自动提取+去重,适合通用场景。
**Letta**:OS 式记忆分页（core/archival）,Agent 自编辑记忆,最接近"Agent 自我管理"的理念。
**Zep**:时序知识图谱 + 自动摘要,支持"什么时候""什么关系"类时序查询,结构化最强。
**选型建议**:通用对话选 Mem0,长期伴随选 Letta,需要时序推理选 Zep。也可组合使用——Zep 做图谱存储 + Mem0 做分层检索。
**📋 八股总结 - 面试高频考点**
Q1: Agent 的记忆系统分为哪几层?各自的特点和实现方式?
**短期记忆**:= LLM 上下文窗口,存当前对话历史。容量 4K~200K token,速度最快,对话结束即消失。
**工作记忆**:= Scratchpad/变量,存当前任务的中间状态。任务结束清除,辅助多步推理。
**长期记忆**:= 向量数据库,存历史交互经验。理论无限容量,需语义检索,永久持久化。
三者关系:短期→工作(任务相关提取),短期→长期(压缩存储),长期→工作(检索回忆)。
Q2: 短期记忆的压缩策略有哪些?各有什么优缺点?
**1 滑窗截断**:保留最近 N 轮,丢弃旧的。简单但可能丢失关键信息。
**2 摘要压缩**:用 LLM 对旧对话生成摘要替代原文。保留关键信息但消耗额外 token。
**3 选择性保留**:用 LLM 判断消息重要性,只保留重要的。信息密度高但判断可能不准。
**4 分层压缩**:结合以上策略,近期完整保留、中期摘要、远期存入长期记忆。效果最好但实现最复杂。
Q3: 向量数据库在 Agent 记忆中的作用?检索流程是什么?
**作用**:作为长期记忆的存储引擎,支持语义相似度检索,让 Agent 能"回忆"与当前查询相关的历史交互。
**写入流程**:交互文本 → Embedding 模型 → 向量 → 存入向量数据库(附带元数据如时间、用户ID)。
**检索流程**:用户查询 → Embedding → 在向量库中做相似度搜索(如余弦相似度)→ 返回 Top-K 最相关记忆 → 注入 prompt。
Q4: 反思记忆、实体记忆、情景记忆分别是什么?
**反思记忆**:定期对历史交互进行反思,提取规律和偏好(如"用户偏好简洁回答"),存储提炼后的知识而非原始对话。
**实体记忆**:以实体为中心组织记忆,维护实体知识图谱(如"小明"的属性和关系),实现实体级别的快速回忆。
**情景记忆**:按时间线存储情节,带时间戳和上下文标签,适合"上次聊到哪了"这类时间敏感场景。
Q5: 为什么不用 SQL 数据库而用向量数据库存 Agent 记忆?
SQL 数据库做精确查询(WHERE name='小明'),但记忆检索是**语义相似度**查询--"和用户当前问题意思相近的历史交互"。这需要把文本转为向量,在向量空间中计算距离。
SQL 无法高效做语义相似度搜索(只能用 LIKE 做模糊匹配,效果差)。向量数据库专门优化了高维向量的相似度搜索,能在毫秒级从百万条记忆中找到最相关的几条。
实践中也可用 pgvector(PostgreSQL 扩展)兼顾两者。
Q6: 上下文膨胀(Context Rot)是什么?五种压缩策略的优先级如何排序?
**Context Rot**:长任务中上下文不断膨胀,导致行为漂移、注意力稀释、成本飙升和窗口溢出。
五种压缩策略优先级(从最优到最差):
**1 语义保护型**:锁定核心指令 + 保护行动-结果对完整性,只压缩中间推理。压缩率和信息完整性最优。
**2 选择性保留**:按消息类型分别设定保留策略,信息密度高。
**3 摘要压缩**:有损压缩,保留关键信息但可能丢失细节。
**4 滑动窗口**:按时间丢弃,不考虑重要性。
**5 截断**:最简单但最粗暴,可能导致行为不一致。
Q7: Context Engineering 和 Prompt Engineering 的区别是什么?
**Prompt Engineering**:关注单次请求的措辞优化,是静态的、一次性的。如:优化 System Prompt、设计 Few-shot 示例。
**Context Engineering**:关注整个上下文的动态编排,是持续性的、动态的。如:记忆的检索/压缩/淘汰策略、工具结果的格式化过滤、多源信息的优先级排序。
类比:Prompt Engineering 是写一首诗(关注措辞),Context Engineering 是管理一个图书馆(关注信息流治理)。
Q8: 记忆生命周期包含哪些阶段?各阶段的治理要点是什么?
五个阶段:
**1 写入**:记录交互 + 附带元数据,默认慷慨写入,后续靠淘汰筛选。
**2 索引**:向量索引 + 关键词索引 + 时间索引 + 类型索引,多路召回提高命中率。
**3 检索**:JIT 模式按需取信息,相似度阈值过滤 + 元数据过滤 + Top-K 截断。
**4 压缩**:语义保护型优先,锁定 System Prompt 和行动-结果对。
**5 淘汰**:时效淘汰 + 容量淘汰 + 质量淘汰,淘汰前可做最终摘要归档。
Q9: Reflexion 框架的反思循环是如何工作的?
Reflexion 的核心流程:
**1 接收任务** → 检索相关反思经验(如有)
**2 执行行动** → 用 ReAct 循环完成任务
**3 评估结果** → 判断是否成功
**4 成功** → 记录为情景记忆(完整事件)
**5 失败** → 触发反思(分析失败原因、提炼改进策略)
**6 存入反思记忆** → 下次重试时参考经验教训
关键创新:不是简单重试,而是每次失败都提炼经验存入记忆,使 Agent 从"重复犯错的傻瓜"进化为"从失败中学习的聪明人"。
Q10: 长期记忆为什么要分为静态和动态?各自的存储、召回、压缩策略有何不同?
不分类会导致三类问题:**存储浪费**(偏好存向量库每次检索没必要)、**信息过时**(摘要不更新会冗余)、**召回混乱**(规则和经验混在一起模型无法区分)。
**静态长期记忆**(偏好/规则/知识):变化频率低 → 存本地配置文件 → 直接注入 System Prompt → 不压缩(本身已精炼)。
**动态长期记忆**(摘要/经验/模式):变化频率高 → 存向量数据库 → 按相关性语义检索 → 定期压缩摘要+淘汰过时内容。
面试要点:先分类再谈策略,不要一上来就说"存向量数据库"。
Q11: 为什么不是每轮对话都触发长期记忆写入?什么信息"值得记住"?
**不是每轮都触发**,否则长期记忆库会变成完整对话日志的备份,噪音爆炸,检索失效。
记忆写入触发条件:用户表达新偏好→写入静态记忆;完成有价值任务→写入动态记忆(执行经验);发现新上下文信息→写入动态记忆(知识片段);简单闲聊→不触发写入。
判断"值得记住"的两个标准:**1 是否有长期参考价值**(未来对话中可能再次需要);**2 是否包含决策/偏好/知识点**(明确的、可复用的信息)。
链路:对话→信息提取→价值评估→分类→写入。重点强调筛选而非全量。
Q12: 召回太多长期记忆会导致什么问题?如何防控上下文污染?
**三大危害**:
**1 注意力稀释**:模型关注无关内容,忽略核心问题
**2 指令冲突**:多条记忆互相矛盾(如"偏好简洁"vs"要求详细")
**3 Token浪费**:无关记忆消耗宝贵上下文窗口
**四道防线**:
**1 数量上限**:每次召回 ≤5 条,简单问题 ≤3 条
**2 相关性阈值**:相似度 最近经验 > 补充信息
召回决策靠三维度评分:意图匹配×时间衰减×频率加权。面试要点:先谈召回决策再谈污染防控,不要只说"用向量搜索"。
Q13: 多轮对话 Agent 怎么解决状态爆炸 / 上下文溢出?生产级的三板斧方案是什么?
这是落地项目一定会遇到的痛点,面试官深挖项目必问。我主要靠三套方案组合解决:
**1 规范状态结构**:用 LangGraph 的 TypedDict 定义固定的状态模板,只留存核心业务变量,去除无效冗余信息,防止状态字段无限膨胀。
**2 智能截断**:不做简单粗暴的字符删减,而是按优先级保留上下文--**系统提示词 > 最近对话记录 > 当前核心任务目标**。低优先级的早期内容优先被截断。
**3 对话摘要**:把老旧的历史对话压缩成简短摘要,放在上下文头部。既保留完整语境,又大幅减少上下文占用。类似人类的"回忆"而非"逐字回放"。
面试要点：不要只说"用向量数据库",要展示"规范状态→智能截断→摘要压缩"三板斧组合思维，体现生产级落地经验而非纸上谈兵。
Q14: 动态 Prompt 和静态 Prompt 有什么区别？Agent 每轮对话开始时，上下文是如何动态组装的？
**静态Prompt** = System Prompt（角色定义+安全规则+输出格式约束） + 工具Schema + Skill触发词。整个会话内不变，放在Prompt最前面，100%命中缓存，是Agent的"宪法"。
**动态Prompt** = 记忆召回片段（按意图检索≤5条） + Skill完整指令（触发时加载） + 对话历史（逐轮增长） + 用户当前输入。每轮都变，放Prompt后面，按变化频率分层。
**组装策略**：四层叠放，固定→半固定→动态→当前请求：
Layer 1 静态框架：System Prompt + 工具Schema + Skill触发词（100%缓存命中）
Layer 2 记忆注入：长期记忆召回片段（同会话内部分命中缓存）
Layer 3 对话历史：多轮交互记录+工具返回结果（旧部分可缓存）
Layer 4 当前请求：用户输入（完全不命中缓存）
核心原则：**不变的放前面最大化缓存，变化的放后面最小化缓存失效**。面试要点：先定义静态和动态，再说组装顺序和缓存策略，不要只说"每次都重新生成Prompt"。
Q15: 多轮对话Agent的状态爆炸，除了三板斧还有什么更完整的系统级解决方案？
Q13的三板斧（规范状态+智能截断+对话摘要）是入门级回答。**完整生产级方案是四层递进的泄压体系**：
**Layer 1 预防**——不让压力累积：
① Token Budget全局预算控制（超预算立即停止）
② JIT召回原则（只检索当前步骤需要的记忆）
③ 输出长度约束（要求模型只输出关键结论）
④ 工具结果精简（只保留关键行而非原样塞入）
**Layer 2 压缩**——在信息完整前提下缩减体积：
① 语义保护型压缩（锁定System Prompt和行动-结果对，只压缩中间推理）
② 三层压缩触发（主动80%→后台60%→反应式413）
③ 增量压缩（只压缩上次压缩后新增部分）
**Layer 3 淘汰**——扔掉不再需要的：
① 记忆治理策略（时效+容量+质量三维度淘汰）
② 召回数量硬限制（≤5条）
③ 工具结果裁剪（只保留Top-K关键行）
④ 淘汰前归档（做最终摘要存入长期记忆）
**Layer 4 兜底**——当所有策略都不够时：
① 紧急压缩（413状态码触发最激进压缩）
② 降级回答（告诉用户建议开新会话）
③ 自动会话切分（超50轮自动摘要+开新会话）
④ 人工介入（预算超限+压缩失败后通知用户）
面试要点：不是某一层单独起作用，而是四层递进。预防失败→压缩兜住，压缩不够→淘汰兜住，淘汰也不够→紧急兜底。体现"从预防到兜底的全链路思维"。
Q16: Mem0 的三层记忆（Episodic / Semantic / Procedural）分别是什麽？如何协同工作？
**Episodic（情景记忆）**:存储具体交互片段,按时间线组织,支持回溯。如"用户上周问了X,Agent回答了Y"。
**Semantic（语义记忆）**:从交互中提取的结构化知识,去重后增量更新。如"用户偏好简洁风格""用户是Java开发者"。
**Procedural（过程记忆）**:从历史中总结的"怎么做"类技能知识。如"处理JSON解析错误时,先检查编码再检查格式"。
**协同流程**:对话输入 → LLM 提取记忆 → 分类（episodic/semantic/procedural）→ 去重/合并 → 存储。检索时按当前意图从三层分别召回,合并后注入上下文。三层是**逻辑分类**而非物理分区,底层可共用同一个向量库。
面试要点:先说三层分别是什么,再说提取→分类→去重→存储的写入流程,最后说分层召回的检索策略。不要把三层说成三个数据库。
Q17: Letta（MemGPT）如何实现记忆分页？Zep 的时序知识图谱有什么优势？
**Letta 记忆分页**:借鉴 OS 虚拟内存机制,把记忆分为 Core Memory（常驻上下文,类似 RAM,放最关键的用户信息和任务状态,通常 < 2000 token）和 Archival Memory（外部存储,类似磁盘,放历史交互）。Core 满了自动把低优先级记忆"换出"到 Archival,需要时通过 search_archival 工具"换入"。Agent 还能通过 edit_core_memory 工具**自编辑**记忆,纠正过时信息。
**Zep 时序知识图谱**:把对话中的实体、关系、事件抽取为带时间戳的三元组,如 (用户, 换设备, Mac, 2025-06)。优势:①支持时序查询（"用户什么时候换的电脑"）,向量检索做不到;②支持关系查询（"X和Y什么关系"）;③自动摘要随时间衰减,新摘要权重更高,解决旧摘要和新摘要同等权重的问题。代价是图谱构建质量依赖 NLP 抽取能力。
面试要点:Letta 重点说"OS 式分页 + 自编辑",Zep 重点说"时序 + 结构化"。对比时强调:Letta 解决的是有限窗口 vs 无限记忆的矛盾,Zep 解决的是向量检索无法回答时序和关系查询的矛盾。
### 🧠 L3 开放题(面试高频)
深度思考 · 无标准答案
开放题
为什么不是每轮对话都触发长期记忆写入?什么信息"值得记住"?请描述记忆写入的完整链路。
**参考思路:**
不是每轮都触发,否则记忆库变成对话日志备份,噪音爆炸检索失效。
**触发条件**:用户表达新偏好→静态记忆;完成有价值任务→动态记忆(经验);发现新上下文→动态记忆(知识);简单闲聊→不触发。
**判断标准**:1是否有长期参考价值(未来可能再次需要)2是否包含决策/偏好/知识点(明确的、可复用的信息)。
**完整链路**:对话→信息提取→价值评估→分类(静态/动态)→写入(配置文件/向量数据库)。
关键词:筛选机制、价值评估、分类写入、不是全量记录。
关联章节:4.10 | 面试题:Q11
开放题
如果召回太多长期记忆,会导致什么问题?如何防控上下文污染?请描述召回决策机制和四道防线。
**参考思路:**
**三大危害**:1注意力稀释(模型关注无关内容忽略核心问题)2指令冲突(多条记忆互相矛盾)3Token浪费(无关记忆消耗上下文窗口)。
**召回决策机制**(三维度评分):意图匹配(语义相似度)× 时间衰减(越近期权重越高)× 频率加权(多次验证有效的记忆权重更高)。综合评分决定是否召回。
**四道防线**:1数量上限(≤5条,简单≤3)2相关性阈值(相似度最近经验>补充信息)。
关键词:上下文污染、注意力稀释、指令冲突、四道防线、召回评分公式。
关联章节:4.11 | 面试题:Q12
开放题
动态Prompt和静态Prompt有什么区别？Agent每轮对话开始时，上下文是如何动态组装的？为什么固定前缀放最前面能最大化缓存命中率？
**参考思路:**
**静态Prompt** = System Prompt（角色+规则） + 工具Schema + Skill触发词，整个会话不变，是Agent的"宪法"。
**动态Prompt** = 记忆召回片段 + Skill完整指令 + 对话历史 + 用户当前输入，每轮都变。
**组装策略**：四层叠放，固定→半固定→动态→当前请求。
Layer 1 静态框架（100%缓存命中）
Layer 2 记忆注入（同会话内部分命中缓存）
Layer 3 对话历史（旧部分可缓存）
Layer 4 当前请求（完全不命中缓存）
**为什么固定前缀放最前面**：Prompt Caching是前缀缓存，只有Prompt开头的不变部分才能命中缓存。System Prompt+工具Schema放在最前面确保每轮前缀一致，100%命中缓存，节省50%+输入成本。合并固定层为一个system消息增加前缀连续性，更容易达到OpenAI缓存最低5个消息触发阈值。
关键词:静态框架、动态组装、四层叠放、缓存命中率、前缀缓存。
关联章节:5.11a | 面试题:Q14
开放题
多轮对话Agent的状态爆炸怎么解决？请描述四层泄压体系的完整框架，以及每层递进的关系。
**参考思路:**
**问题定义**：多轮对话中状态从线性增长变为树状展开（工具嵌套）和倍数增长（推理分叉），导致上下文溢出、注意力稀释、成本爆炸和行为漂移。
**四层泄压体系**：
**Layer 1 预防**——不让压力累积：Token Budget全局预算控制、JIT召回原则、输出长度约束、工具结果精简。
**Layer 2 压缩**——在信息完整前提下缩减体积：语义保护型压缩（锁定System Prompt和行动-结果对）、三层压缩触发（主动80%→后台60%→反应式413）、增量压缩。
**Layer 3 淘汰**——扔掉不再需要的：记忆治理策略（时效+容量+质量三维度淘汰）、召回数量硬限制≤5条、工具结果裁剪、淘汰前归档。
**Layer 4 兜底**——当所有策略都不够时：紧急压缩（413触发最激进压缩）、降级回答、自动会话切分、人工介入。
**递进关系**：预防失败→压缩兜住，压缩不够→淘汰兜住，淘汰也不够→紧急兜底。不是某一层单独起作用，而是四层递进式泄压。
关键词:状态爆炸、四层泄压、预算控制、语义保护型压缩、记忆治理、紧急兜底。
关联章节:5.11b | 面试题:Q15