lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化: 上下文压缩缓存友好化与召回句柄

Browse Source 
- 块状裁剪: 水位目标+轮边界吸附+病态回退,贪心路径保留为开关回退
- 存储态 System 锚点出请求时视图转 user,治前缀缓存整体折叠失配
- BP3 断点机制+缓存命中率观测日志,调参有据
- 压缩摘要可召回: JSONL 档案落盘+句柄行注入,take 语义一次性消费
- 字符估算修正: CJK 分档计权+改图地板,估算对象对齐实际消耗
- namespace 引用跨重启内容哈希懒回填
This commit is contained in:
lxy
2026-10-11 22:27:08 +08:00
parent c7d0ed939c
commit 6f5c6b7532
6 changed files with 1006 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-ai/src/anthropic_compat.rs
+253 -18
View File
@@ -62,6 +62,46 @@ pub fn anthropic_cache_enabled() -> bool {
ANTHROPIC_CACHE_ENABLED.load(std::sync::atomic::Ordering::SeqCst)
}
/// BP3 摘要链尾断点开关（缓存友好上下文总开关在本模块的投影，默认开）。
///
/// 本模块不读 src-tauri KV（跨 crate），上层读 settings `df-ai-cache-friendly-context`
/// 后经 [`set_bp3_enabled`] 同步（同 [`set_anthropic_cache_enabled`] 先例）。
/// 关时消息级只打 BP2，断点布局与历史逐字节一致。BP3 仍内嵌于 anthropic 缓存总开关
/// 之下（它发的也是 cache_control），两层正交。
static BP3_ENABLED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
/// 设置 BP3 摘要链尾断点开关（上层读 settings 后同步）。
pub fn set_bp3_enabled(enabled: bool) {
BP3_ENABLED.store(enabled, std::sync::atomic::Ordering::SeqCst);
}
/// 读取 BP3 开关（`convert_request` 组装消息级断点用）。
pub fn bp3_enabled() -> bool {
BP3_ENABLED.load(std::sync::atomic::Ordering::SeqCst)
}
/// 缓存命中率观测日志（usage 就绪点每轮一条，供后续水位调参取数）。
///
/// anthropic 语义：input_tokens 为未缓存输入，cache_read/creation 独立计数，
/// 故命中率分母 = read + creation + input；三者全零（端点未回 usage）跳过不报。
fn log_cache_usage(usage: &TokenUsage) {
let total_input = usage.prompt_cache_hit_tokens as u64
+ usage.prompt_cache_miss_tokens as u64
+ usage.prompt_tokens as u64;
if total_input == 0 {
return;
}
let hit_rate = usage.prompt_cache_hit_tokens as f64 / total_input as f64;
let hit_pct = format!("{:.1}%", hit_rate * 100.0);
tracing::info!(
input_tokens = usage.prompt_tokens,
cache_read_input_tokens = usage.prompt_cache_hit_tokens,
cache_creation_input_tokens = usage.prompt_cache_miss_tokens,
hit_rate = %hit_pct,
"Anthropic prompt cache 命中率观测"
);
}
// ============================================================
// Provider 实现
// ============================================================
@@ -147,10 +187,21 @@ impl AnthropicCompatProvider {
// 构建非 system 消息（保留顺序，合并连续 tool_result）
let mut messages: Vec<serde_json::Value> = Vec::new();
let mut pending_tool_results: Vec<serde_json::Value> = Vec::new();
// 摘要锚点 = index>0 的 System（存储态：压缩摘要/话题标记/催办；首条是组装层
// system_prompt，归 BP1 顶层 system 非锚点）。记录「末锚点之后的消息条数」而非
// 下标：取用时按 len 反推目标，此前的 merge/补首条等数组增删天然自校正。
// 锚点以存储态 System 识别——view 层 System→user 转换落地后锚点以 user 形态
// 到达，此处识别不到，BP3 自然旁路（不加断点），无副作用。
let mut msgs_after_anchor: Option<usize> = None;
for m in req.messages.iter() {
for (src_idx, m) in req.messages.iter().enumerate() {
match m.role {
MessageRole::System => continue,
MessageRole::System => {
if src_idx > 0 {
msgs_after_anchor = Some(0);
}
continue;
}
MessageRole::Tool => {
// 累积 tool_result 块，遇到非 Tool 消息时 flush。
// tool_use_id 为 None 时绝不能发 null（GLM anthropic 端点会 500
@@ -173,7 +224,7 @@ impl AnthropicCompatProvider {
}
}
MessageRole::User => {
Self::flush_tool_results(&mut messages, &mut pending_tool_results);
Self::flush_tool_results(&mut messages, &mut pending_tool_results, &mut msgs_after_anchor);
// 多模态 user 消息 → content blocks 数组（text/image）。
// 含图时把 content + parts 拍平成 blocks：Text 片 → {type:text}，
// Image 片 → {type:image, source:{type:base64, media_type, data}}。
@@ -192,9 +243,12 @@ impl AnthropicCompatProvider {
} else {
messages.push(serde_json::json!({ "role": "user", "content": m.content }));
}
if let Some(k) = &mut msgs_after_anchor {
*k += 1;
}
}
MessageRole::Assistant => {
Self::flush_tool_results(&mut messages, &mut pending_tool_results);
Self::flush_tool_results(&mut messages, &mut pending_tool_results, &mut msgs_after_anchor);
let mut content: Vec<serde_json::Value> = Vec::new();
if !m.content.is_empty() {
content.push(serde_json::json!({ "type": "text", "text": m.content }));
@@ -218,10 +272,13 @@ impl AnthropicCompatProvider {
content.push(serde_json::json!({ "type": "text", "text": "" }));
}
messages.push(serde_json::json!({ "role": "assistant", "content": content }));
if let Some(k) = &mut msgs_after_anchor {
*k += 1;
}
}
}
}
Self::flush_tool_results(&mut messages, &mut pending_tool_results);
Self::flush_tool_results(&mut messages, &mut pending_tool_results, &mut msgs_after_anchor);
// 合并相邻 user 块。Anthropic 协议要求 user/assistant 严格交替,连续 user
// 触发 GLM 1214。场景:drainQueue 续发(前一轮以 tool_result 结尾 + 新 user)→ flush 把
@@ -240,10 +297,12 @@ impl AnthropicCompatProvider {
// (正常会话首条本就是 user),远优于丢弃上下文。
Self::ensure_leading_user(&mut messages);
// 消息级缓存断点(开关开):除 system 断点外在历史段尾部再打一个 cache_control,
// 使 system 之后的对话历史也可命中(否则只缓存 system,长会话历史段全价重算)。
// 消息级缓存断点(开关开):BP2 打在历史段稳定边界,使 system 之后的对话历史
// 也可命中(否则只缓存 system,长会话历史段全价重算);BP3(缓存友好开关,默认开)
// 加打摘要锚点链之后第一条,BP2 因裁剪边界/压缩失效时 [system+摘要链] 小前缀仍命中。
if anthropic_cache_enabled() {
Self::apply_message_cache_breakpoint(&mut messages);
let after_anchor = if bp3_enabled() { msgs_after_anchor } else { None };
Self::apply_message_cache_breakpoint(&mut messages, after_anchor);
}
let tools = req.tools.map(|defs| {
@@ -328,34 +387,61 @@ impl AnthropicCompatProvider {
fn flush_tool_results(
messages: &mut Vec<serde_json::Value>,
pending: &mut Vec<serde_json::Value>,
msgs_after_anchor: &mut Option<usize>,
) {
if pending.is_empty() {
return;
}
let blocks: Vec<serde_json::Value> = pending.drain(..).collect();
messages.push(serde_json::json!({ "role": "user", "content": blocks }));
if let Some(k) = msgs_after_anchor {
*k += 1;
}
}
/// 消息级缓存断点:在「历史段稳定边界」消息的末个 content block 上打 cache_control。
/// 消息级缓存断点:BP2 打在「历史段稳定边界」的末个 content block 上,BP3(可选)
/// 加打「摘要锚点链之后第一条」——`after_anchor` 为锚点链后的消息条数
/// (Some(0)=锚点收尾无后继,不打;None=无锚点,只打 BP2)。
///
/// 断点落点取**末条 user 消息之前的相邻前一条**(末条非 user 收尾的异常序列则取末条):
/// BP2 落点取**末条 user 消息之前的相邻前一条**(末条非 user 收尾的异常序列则取末条):
/// 末条 user 是每轮请求的漂移边界——缓存拆分模式下它是请求末尾追加的易变尾段
/// (日期/清单/augmentation,每次重组),非拆分模式下是本轮新问题,二者都不属于
/// 只增历史;打在它身上断点位置随请求漂移,永远无法命中。标其前一条(assistant /
/// tool_result 归并的 user)则断点始终落在只增历史的末尾,历史只增不改位:
/// 上一轮写入的缓存前缀是本轮请求前缀的逐字节前缀 → 命中后增量续写,滚动前进。
///
/// BP3 服务于 BP2 频繁失效的场景:BP2 随裁剪边界移动/压缩事件/新近度豁免改写而
/// 迁位,前缀反复全价重算;BP3 前缀只含 [system+摘要链+1 条],失效源仅压缩事件,
/// 为该小前缀保底命中。与 BP2 同条(短序列)时跳过,BP2 优先。
///
/// content 形态:string(简写)→ 升级为单 text block 数组再标;array(blocks)→ 标末块。
/// 断点预算:Anthropic 单请求最多 4 个 cache_control,system 1 个 + 消息级 1 个,余量充足。
fn apply_message_cache_breakpoint(messages: &mut [serde_json::Value]) {
/// 断点预算:Anthropic 单请求最多 4 个 cache_control——system 1 + BP2 1 + BP3 1 = 3,
/// 余 1 备用。
fn apply_message_cache_breakpoint(
messages: &mut [serde_json::Value],
after_anchor: Option<usize>,
) {
let n = messages.len();
if n == 0 {
return;
}
let last_is_user =
messages[n - 1].get("role").and_then(|r| r.as_str()) == Some("user");
let idx = if last_is_user && n >= 2 { n - 2 } else { n - 1 };
let msg = &mut messages[idx];
let bp2_idx = if last_is_user && n >= 2 { n - 2 } else { n - 1 };
if let Some(k) = after_anchor {
if k > 0 && k <= n {
let bp3_idx = n - k;
if bp3_idx != bp2_idx {
Self::mark_cache_control(&mut messages[bp3_idx]);
}
}
}
Self::mark_cache_control(&mut messages[bp2_idx]);
}
/// 单条消息打 cache_control:string content 升级为单 text block 数组再标;
/// array(blocks) 标末块;其余形态(异常)不动。
fn mark_cache_control(msg: &mut serde_json::Value) {
let cache_control = serde_json::json!({ "type": "ephemeral" });
match msg.get("content").cloned() {
Some(serde_json::Value::Array(mut blocks)) => {
@@ -729,6 +815,7 @@ impl LlmProvider for AnthropicCompatProvider {
prompt_cache_miss_tokens: resp.usage.cache_creation_input_tokens,
reasoning_tokens: 0,
};
log_cache_usage(&usage);
AttemptOutcome::Ok(CompletionResponse {
text,
model: resp.model,
@@ -874,6 +961,12 @@ impl LlmProvider for AnthropicCompatProvider {
if let Some(err) = chunk.error.as_mut() {
*err = format!("{} | messages 摘要: {}", err, messages_summary);
}
// 流式收尾(message_stop 终态帧)即 usage 就绪点,输出命中率观测日志
if chunk.finished {
if let Some(u) = &chunk.usage {
log_cache_usage(u);
}
}
chunks.push(Ok(chunk));
}
}
@@ -1258,7 +1351,7 @@ mod tests {
serde_json::json!({ "role": "assistant", "content": [{"type": "text", "text": "a1"}] }),
serde_json::json!({ "role": "user", "content": "q2" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, None);
// 前一条 assistant 的末块带 cache_control
let a1 = &messages[1];
let blocks = a1["content"].as_array().unwrap();
@@ -1275,7 +1368,7 @@ mod tests {
serde_json::json!({ "role": "assistant", "content": "a1" }),
serde_json::json!({ "role": "user", "content": "q2" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, None);
let blocks = messages[1]["content"].as_array().expect("string 应升级为 blocks 数组");
assert_eq!(blocks.len(), 1);
assert_eq!(blocks[0]["type"], "text");
@@ -1289,7 +1382,7 @@ mod tests {
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "only" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, None);
let blocks = messages[0]["content"].as_array().expect("单条应升级为 blocks");
assert_eq!(blocks[0]["cache_control"]["type"], "ephemeral");
}
@@ -1298,10 +1391,152 @@ mod tests {
#[test]
fn cache_breakpoint_empty_messages_noop() {
let mut messages: Vec<serde_json::Value> = vec![];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, None);
assert!(messages.is_empty());
}
/// BP3 落「锚点链后首条」(按锚点后条数反推下标),与 BP2 各占一条互不覆盖。
#[test]
fn cache_breakpoint_bp3_marks_first_message_after_anchor_chain() {
// 模拟 [锚点链][q0][a0][q1][a1][q2] 转换后的数组:锚点后 5 条
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "q0" }),
serde_json::json!({ "role": "assistant", "content": "a0" }),
serde_json::json!({ "role": "user", "content": "q1" }),
serde_json::json!({ "role": "assistant", "content": "a1" }),
serde_json::json!({ "role": "user", "content": "q2" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, Some(5));
// BP3 → idx0(q0);BP2(末条 user,取 n-2)→ idx3(a1)
for idx in [0usize, 3] {
let blocks = messages[idx]["content"].as_array().unwrap();
assert_eq!(
blocks.last().unwrap()["cache_control"]["type"],
"ephemeral",
"idx{} 应带断点",
idx
);
}
assert_eq!(messages[4]["content"].as_str(), Some("q2"), "末条 user 不打点");
}
/// BP3 与 BP2 同条(短序列)→ 跳过 BP3,BP2 优先,断点预算不重复占。
#[test]
fn cache_breakpoint_bp3_skipped_when_colliding_with_bp2() {
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "q0" }),
serde_json::json!({ "role": "assistant", "content": "a0" }),
serde_json::json!({ "role": "user", "content": "q1" }),
];
// 锚点后 2 条 → BP3 目标 idx1,与 BP2(n-2=1)同条 → 跳过
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, Some(2));
let blocks = messages[1]["content"].as_array().unwrap();
assert_eq!(blocks.last().unwrap()["cache_control"]["type"], "ephemeral");
assert!(messages[0]["content"].as_str().is_some(), "q0 应保持字符串简写");
}
/// 锚点收尾无后继(after_anchor=0)→ 不打 BP3,BP2 照常落点。
#[test]
fn cache_breakpoint_bp3_zero_after_anchor_noop() {
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "q0" }),
serde_json::json!({ "role": "user", "content": "q1" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages, Some(0));
// BP2(末条 user → n-2=0)照常标 q0
let blocks = messages[0]["content"].as_array().expect("BP2 应照常落点");
assert_eq!(blocks.last().unwrap()["cache_control"]["type"], "ephemeral");
}
// ---------- BP3 开关门(convert 级,进程级开关用例须持 convert_lock) ----------
/// 构造 BP3 开关用例的请求:首条组装层 system_prompt + (可选)锚点链 + 历史段。
/// 转换后数组恒为 [q0, a0, q1](锚点折叠进顶层 system)。
fn bp3_req(with_anchor: bool) -> CompletionRequest {
let mut messages = vec![
ChatMessage::system("sys-prompt"),
ChatMessage::user("q0"),
ChatMessage::assistant("a0"),
ChatMessage::user("q1"),
];
if with_anchor {
messages.insert(1, ChatMessage::system("## 压缩摘要"));
messages.insert(2, ChatMessage::system("## 话题标记"));
}
CompletionRequest {
model: "claude-3-5-sonnet".into(),
messages,
temperature: None,
max_tokens: None,
stream: false,
tools: None,
tool_choice: None,
reasoning_content: None,
}
}
/// BP3 开 + 有锚点 → 断点落摘要链后第一条(q0),BP2 照常落历史尾(a0)。
#[test]
fn bp3_enabled_with_anchor_marks_first_message_after_chain() {
let _convert = convert_lock();
set_anthropic_cache_enabled(true);
set_bp3_enabled(true);
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let body = provider.convert_request(bp3_req(true));
assert_eq!(body.messages.len(), 3);
// BP3:锚点链后首条 q0(原字符串升级为 blocks + cache_control)
let q0_blocks = body.messages[0]["content"].as_array().expect("BP3 应标在锚点链后首条 q0");
assert_eq!(q0_blocks[0]["text"], "q0");
assert_eq!(q0_blocks[0]["cache_control"]["type"], "ephemeral");
// BP2 照常:末条 user 前相邻一条 a0
let a0_blocks = body.messages[1]["content"].as_array().unwrap();
assert_eq!(a0_blocks.last().unwrap()["cache_control"]["type"], "ephemeral");
// 末条 user 保持字符串简写
assert_eq!(body.messages[2]["content"].as_str(), Some("q1"));
set_anthropic_cache_enabled(false);
set_bp3_enabled(true); // 复位到进程默认(默认开),防污染其他并行用例
}
/// BP3 关 → 断点布局与历史一致(仅 BP2,锚点链后首条不打点)。
#[test]
fn bp3_disabled_keeps_legacy_breakpoint_layout() {
let _convert = convert_lock();
set_anthropic_cache_enabled(true);
set_bp3_enabled(false);
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let body = provider.convert_request(bp3_req(true));
assert_eq!(body.messages.len(), 3);
assert_eq!(
body.messages[0]["content"].as_str(),
Some("q0"),
"BP3 关,q0 保持字符串简写"
);
let a0_blocks = body.messages[1]["content"].as_array().unwrap();
assert_eq!(a0_blocks.last().unwrap()["cache_control"]["type"], "ephemeral", "BP2 照常");
set_anthropic_cache_enabled(false);
set_bp3_enabled(true);
}
/// 无锚点(仅组装层首条 system)→ 不加 BP3,仅 BP2。
#[test]
fn bp3_no_anchor_no_extra_breakpoint() {
let _convert = convert_lock();
set_anthropic_cache_enabled(true);
set_bp3_enabled(true);
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let body = provider.convert_request(bp3_req(false));
assert_eq!(body.messages.len(), 3);
assert_eq!(
body.messages[0]["content"].as_str(),
Some("q0"),
"无锚点应不加 BP3"
);
let a0_blocks = body.messages[1]["content"].as_array().unwrap();
assert_eq!(a0_blocks.last().unwrap()["cache_control"]["type"], "ephemeral", "BP2 照常");
set_anthropic_cache_enabled(false);
set_bp3_enabled(true);
}
// ---------- ensure_leading_user（首条 assistant → 补 user 占位,保留上下文）----------
/// 辅助:构造 assistant(tool_use) 消息