第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