新增: F-15上下文压缩基础(compress_prompt四段式+compress_via_llm+context辅助方法)

This commit is contained in:
2026-06-17 00:50:47 +08:00
parent 76d823ded0
commit 63bff8bf96
4 changed files with 507 additions and 17 deletions

View File

@@ -106,7 +106,7 @@ impl ContextConfig {
/// 消息在逻辑上的分组标签,用于淘汰时保持原子性
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MessageGroup {
pub enum MessageGroup {
/// 普通 User / Assistant 文本消息(可独立淘汰)
Standalone,
/// Assistant 带 tool_calls是三元组的头
@@ -116,10 +116,14 @@ enum MessageGroup {
}
/// 带有 token 缓存和分组信息的消息条目
struct TrackedMessage {
message: ChatMessage,
token_count: u32,
group: MessageGroup,
///
/// 字段 `pub`供阶段2 IPC 经 `ContextManager::messages_mut()` 拿到可变切片后,
/// 直接改 `message.status` / 读 `token_count` 做 token 重算(Mutex 单线程访问,
/// 同 crate 内安全)。结构体本身也 `pub`(返回类型对外可见)。
pub struct TrackedMessage {
pub message: ChatMessage,
pub token_count: u32,
pub group: MessageGroup,
}
fn classify_group(msg: &ChatMessage) -> MessageGroup {
@@ -150,6 +154,9 @@ pub struct ContextManager {
history_tokens: u32,
config: ContextConfig,
estimator: TokenEstimator,
/// 压缩重入标志F-15 §4.3true 表示一次 LLM 压缩正在进行中。
/// agentic loop 顶部检测,防同一轮内多次触发压缩互相覆盖。纯内存态,不落库。
is_compressing: bool,
}
/// 保护区大小:最后 N 条消息永不淘汰(≈ 最近 2 个完整用户轮次)
@@ -166,6 +173,7 @@ impl ContextManager {
history_tokens: 0,
config,
estimator: TokenEstimator::default(),
is_compressing: false,
}
}
@@ -194,6 +202,7 @@ impl ContextManager {
pub fn clear(&mut self) {
self.messages.clear();
self.history_tokens = 0;
self.is_compressing = false;
}
/// 消息数量
@@ -522,14 +531,49 @@ impl ContextManager {
self.messages.iter().map(|t| &t.message)
}
// ── 内部方法 ──
// ── F-15 上下文管理增强辅助方法阶段1 基础,零行为变化)──
//
// 这些方法供阶段2 IPCai_chat_compress_context/ 阶段3 agentic loop
// 自动压缩调用。本阶段只暴露方法 + 单测,不接 IPC/前端/agentic.rs。
/// 构建淘汰单元列表
/// 配置(只读视图,供 agentic.rs 计算压缩触发阈值 `config().budget_limit()`
pub fn config(&self) -> &ContextConfig {
&self.config
}
/// 可变消息切片供阶段2 标记 status="compressed"/"archived_segment" + 调整 token
///
/// 每个单元是连续消息范围 [start, end),保证:
/// - 工具调用三元组ToolCallHead + ToolResultTail* + 紧随的文本 Assistant在同一单元
/// - 保护区内的消息不纳入任何单元
fn build_eviction_units(&self, protect_start: usize) -> Vec<EvictionUnit> {
/// 调用方约定:仅改 `message.status` / `message.content`,不增删条目(增删走
/// [`push`] / [`insert_at`]),否则 `history_tokens` 会与实际脱钩。
pub fn messages_mut(&mut self) -> &mut [TrackedMessage] {
&mut self.messages
}
/// 在给定位置插入一条消息(其余向后移),并把它计入 token 预算(active 才计)。
///
/// 供阶段2 在压缩点插入摘要 system 消息。`index` 越界则 panic(对齐 Vec::insert 语义,
/// 调用方负责算合法 index,如 `compress_end` 已由 `compress_old_messages` 校验)。
pub fn insert_at(&mut self, index: usize, message: ChatMessage) {
let tokens = self.estimator.estimate_message(&message);
let group = classify_group(&message);
if message.is_active() {
self.history_tokens += tokens;
}
self.messages.insert(index, TrackedMessage {
message,
token_count: tokens,
group,
});
}
/// 按淘汰单元分组消息范围(三元组原子性),供压缩定位/分段标记复用同一分组逻辑。
///
/// 返回每个单元的右开区间 end + token 总和,保证:
/// - 工具调用三元组(Head + Tail* + 紧随的 Standalone Assistant)在同一单元
/// - 保护区 `[protect_start, len)` 内的消息不纳入任何单元
///
/// 公开供阶段2 会话分段(`archived_segment` 按组原子标记)与压缩定位共用。
pub fn build_eviction_units(&self, protect_start: usize) -> Vec<EvictionUnit> {
let mut units = Vec::new();
let mut i = 0usize;
@@ -537,14 +581,12 @@ impl ContextManager {
let mut token_sum = 0u32;
if self.messages[i].group == MessageGroup::ToolCallHead {
// 收集完整三元组Head + 后续所有 ToolResultTail + 紧随的文本 Assistant
token_sum += self.messages[i].token_count;
i += 1;
while i < protect_start && self.messages[i].group == MessageGroup::ToolResultTail {
token_sum += self.messages[i].token_count;
i += 1;
}
// 紧随的 Standalone Assistant工具调用的最终文本回复
if i < protect_start
&& self.messages[i].group == MessageGroup::Standalone
&& matches!(self.messages[i].message.role, MessageRole::Assistant)
@@ -553,7 +595,6 @@ impl ContextManager {
i += 1;
}
} else {
// Standalone / ToolResultTail理论上孤立 Tail 不该出现,按单条处理)
token_sum += self.messages[i].token_count;
i += 1;
}
@@ -563,12 +604,64 @@ impl ContextManager {
units
}
/// 保护区外是否存在可压缩消息(供 agentic loop 顶部触发判断)。
///
/// "可压缩"= status 为 None/active 的消息(已 compressed/archived_segment/truncated
/// 不参与二次压缩,幂等)。`protect_start` 为保护区起点(如 `len - PROTECT_COUNT`)。
pub fn has_compressible_messages(&self, protect_start: usize) -> bool {
let end = protect_start.min(self.messages.len());
self.messages[..end]
.iter()
.any(|t| t.message.is_active())
}
/// 把保护区 `[0, compress_end)` 范围内的 active 消息标记为 `status="compressed"`,
/// 同步从 `history_tokens` 扣除其 token,返回被压缩消息的克隆(供阶段2 喂 LLM 摘要)。
///
/// **幂等**:已 compressed(或任何 !active)的消息跳过,不会被二次压缩;`history_tokens`
/// 也只扣首次标记的 token。返回的 Vec 仅含**本次新标记**的消息(已 compressed 的不返)。
///
/// **单向不可逆**:压缩后 DB 原始消息保留,但 LLM 上下文里被 is_active 白名单隔离
/// (sanitize step0 过滤)。`compress_end` 越界自动 clamp 到 `messages.len()`。
///
/// 返回空 Vec 表示本批次无可压缩消息(全部已 compressed 或范围空),调用方据此跳过 LLM 调用。
pub fn compress_old_messages(&mut self, compress_end: usize) -> Vec<ChatMessage> {
let end = compress_end.min(self.messages.len());
let mut newly_compressed = Vec::new();
for t in self.messages[..end].iter_mut() {
if t.message.is_active() {
newly_compressed.push(t.message.clone());
t.message.status = Some("compressed".to_string());
self.history_tokens = self.history_tokens.saturating_sub(t.token_count);
}
}
newly_compressed
}
/// 压缩重入标志(读)。true 表示一次 LLM 压缩正在进行中,触发方应跳过本轮压缩。
pub fn is_compressing(&self) -> bool {
self.is_compressing
}
/// 压缩重入标志(写)。`true`=开始压缩(进入 agentic loop 顶部前置置位),
/// `false`=压缩结束(无论成功或降级)。调用方必须成对调用,防止永久卡死。
pub fn set_compressing(&mut self, v: bool) {
self.is_compressing = v;
}
// ── 内部方法 ──
// (原 build_eviction_units / classify_group 等纯函数已在上方公开或文件级定义;
// ContextManager 的私有辅助如需新增放在这里。)
}
/// 淘汰单元:连续消息范围 [..end) + token 总和
struct EvictionUnit {
end: usize,
token_sum: u32,
///
/// `pub` 供 `build_eviction_units` 的返回类型对外可见(阶段2/3 调用方读 `end` / `token_sum`)。
pub struct EvictionUnit {
pub end: usize,
pub token_sum: u32,
}
#[cfg(test)]
@@ -944,4 +1037,171 @@ mod tests {
msgs.len()
);
}
// ── F-15 阶段1 辅助方法单测 ──
#[test]
fn compress_old_messages_marks_compressed_and_returns_refs() {
// F-15 §4.2/§4.3compress_old_messages 把 [0, end) 内 active 消息标 compressed,
// 同步扣 history_tokens,返回它们的克隆供 LLM 摘要。持久化全量保留。
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("旧消息1"));
mgr.push(ChatMessage::assistant("旧回复1"));
mgr.push(ChatMessage::user("新消息2"));
let tokens_before = mgr.history_tokens();
assert!(tokens_before > 0);
let compressed = mgr.compress_old_messages(2);
assert_eq!(compressed.len(), 2, "应压缩前 2 条 active");
assert_eq!(compressed[0].content, "旧消息1");
assert_eq!(compressed[1].content, "旧回复1");
// status 已改 compressed
assert_eq!(mgr.messages_mut()[0].message.status.as_deref(), Some("compressed"));
assert_eq!(mgr.messages_mut()[1].message.status.as_deref(), Some("compressed"));
// 保护区外(本例 index 2)仍 active
assert!(mgr.messages_mut()[2].message.is_active(), "保护区外消息不应被动");
// 持久化全量不变
assert_eq!(mgr.all_messages_clone().len(), 3, "compress 不应删消息(单向,全量保留)");
// token 已扣(剩第 3 条的)
let only_third_tokens = TokenEstimator::default().estimate_message(&ChatMessage::user("新消息2"));
assert_eq!(mgr.history_tokens(), only_third_tokens, "history_tokens 应扣除前两条");
}
#[test]
fn compress_old_messages_is_idempotent() {
// 幂等:已 compressed 不二次压缩,二次调用返回空 Vec 且 history_tokens 不再变。
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("a"));
mgr.push(ChatMessage::user("b"));
let first = mgr.compress_old_messages(2);
assert_eq!(first.len(), 2);
let tokens_after_first = mgr.history_tokens();
let second = mgr.compress_old_messages(2);
assert!(second.is_empty(), "二次压缩应返回空(已 compressed 不重压)");
assert_eq!(
mgr.history_tokens(),
tokens_after_first,
"二次压缩 history_tokens 不应再变(幂等)"
);
}
#[test]
fn compress_old_messages_clamps_oversized_end() {
// compress_end 越界自动 clamp 到 len,不 panic。
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("唯一"));
let compressed = mgr.compress_old_messages(999);
assert_eq!(compressed.len(), 1, "越界 end 应 clamp 到 len(1)");
assert_eq!(mgr.history_tokens(), 0, "全量压缩后 history_tokens 归零");
}
#[test]
fn compress_old_messages_skips_already_inactive() {
// 范围内含 truncated(已 !active)的消息:跳过,不返,不重复扣 token。
let mut mgr = ContextManager::new(cfg(100_000));
let mut truncated = ChatMessage::user("被截断");
truncated.status = Some("truncated".to_string());
mgr.push(truncated);
mgr.push(ChatMessage::user("active 一条"));
let tokens_before = mgr.history_tokens();
// truncated 已不计 token(见 push_token_only_active),所以 tokens_before 只含 active 一条
let compressed = mgr.compress_old_messages(2);
assert_eq!(compressed.len(), 1, "只压缩 active 那条,truncated 跳过");
assert_eq!(mgr.history_tokens(), 0);
assert_eq!(
mgr.history_tokens(),
tokens_before.saturating_sub(tokens_before),
"幂等扣除:truncated 本就没计 token,active 扣光"
);
// truncated 状态不被改成 compressed(保留原 truncated,语义不混淆)
assert_eq!(
mgr.messages_mut()[0].message.status.as_deref(),
Some("truncated"),
"已 truncated 不应被改写为 compressed"
);
}
#[test]
fn has_compressible_messages_respects_protect_zone() {
let mut mgr = ContextManager::new(cfg(100_000));
for i in 0..8 {
mgr.push(ChatMessage::user(&format!("消息 {}", i)));
}
// protect_start=6 → [0,6) 内有 active → true
assert!(mgr.has_compressible_messages(6));
// protect_start=0 → 空范围 → false
assert!(!mgr.has_compressible_messages(0));
// 全部压缩后 → false
mgr.compress_old_messages(6);
assert!(!mgr.has_compressible_messages(6), "全 compressed 后不应有可压缩消息");
}
#[test]
fn is_compressing_flag_round_trip() {
// 标志位读写 round-trip;clear() 复位。
let mut mgr = ContextManager::new(cfg(100_000));
assert!(!mgr.is_compressing(), "默认 false");
mgr.set_compressing(true);
assert!(mgr.is_compressing(), "set true 后应读到 true");
mgr.set_compressing(false);
assert!(!mgr.is_compressing(), "set false 后复位");
// clear 复位
mgr.set_compressing(true);
mgr.clear();
assert!(!mgr.is_compressing(), "clear() 应复位 is_compressing");
}
#[test]
fn insert_at_adds_to_budget_when_active() {
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("a"));
let tokens_before = mgr.history_tokens();
// 插入 active system 消息 → 计入 token
mgr.insert_at(0, ChatMessage::system("## 摘要"));
assert!(mgr.history_tokens() > tokens_before, "active 消息应计入 token");
assert_eq!(mgr.len(), 2);
assert_eq!(mgr.messages_mut()[0].message.content, "## 摘要");
// 插入 !active 消息 → 不计入 token
let tokens_before_inactive = mgr.history_tokens();
let mut inactive = ChatMessage::user("x");
inactive.status = Some("truncated".to_string());
mgr.insert_at(0, inactive);
assert_eq!(
mgr.history_tokens(),
tokens_before_inactive,
"!active 消息插入不应计 token"
);
}
#[test]
fn build_eviction_units_keeps_triplet_atomic_public() {
// 公开的 build_eviction_units:三元组(Head + Tail + Standalone Assistant)应落同一单元。
let mut mgr = ContextManager::new(cfg(100_000));
mgr.push(ChatMessage::user("前置"));
mgr.push(ChatMessage::assistant_with_tools(
"",
vec![ToolCall::new("c1", "fn", "{}")],
));
mgr.push(ChatMessage::tool_result("c1", "结果"));
mgr.push(ChatMessage::assistant("完成"));
mgr.push(ChatMessage::user("后置"));
// protect_start=5(全部纳入)
let units = mgr.build_eviction_units(5);
// 第一个单元是前置 Standalone(end=1);第二个单元应包含三元组三件套 + 后置应分开
// 确认三元组的 Head+Tail+Assistant 在同一单元(end 跳过 3)
let unit2 = units.iter().find(|u| u.end >= 4).expect("应有跨三元组的单元");
assert!(
unit2.end >= 4,
"三元组三件套应在同一淘汰单元, end={}",
unit2.end
);
}
}