优化: 缓存命中字段解析与提示词前缀稳定化

- openai 兼容口径补解析嵌套 cached_tokens
- cache 显示区分无数据与真零,估算轮不渲染
- 轨迹展开折叠空过程记录行降噪
- token 统计行恒完整可见不截断
- openai_compat 纳入 system 稳定易变拆分
- 系统警示与主题标记改尾部注入不位移前缀
- 工具结果视图压缩按入库时刻冻结
- 会话内工具集冻结仅增不减
- anthropic 缓存补消息级断点,设置页加开关
- 补 intent 测试对 Agent domain 覆盖
This commit is contained in:
lxy
2026-10-01 19:35:21 +08:00
parent 2187ffe173
commit 423acfdd00
14 changed files with 499 additions and 99 deletions
+120 -2
View File
@@ -43,8 +43,9 @@ use crate::anthropic_helpers::{
/// API 网关——**cache_control 字段可能被非官方网关拒收**,故必须默认关、用户对支持的端点显式开启。
///
/// 开关开:`convert_request` 把 system 包成 blocks 数组并附加 `cache_control:{type:ephemeral}`
/// (配合上层把易变段挪出 system,让 system 稳定段真正命中缓存)。
/// 开关关:system 保持纯字符串(现状行为,零风险兼容网关)。
/// (配合上层把易变段挪出 system,让 system 稳定段真正命中缓存),
/// 并在历史段稳定边界补消息级断点(使 system 之后的对话历史也可命中)。
/// 开关关:system 保持纯字符串、消息不打断点(现状行为,零风险兼容网关)。
///
/// 进程级全局而非 per-provider:上层(AppState settings)按「用户对当前启用的 anthropic 端点
/// 是否支持缓存」一次性开启,所有 anthropic 请求一致生效。默认关。
@@ -239,6 +240,12 @@ impl AnthropicCompatProvider {
// (正常会话首条本就是 user),远优于丢弃上下文。
Self::ensure_leading_user(&mut messages);
// 消息级缓存断点(开关开):除 system 断点外在历史段尾部再打一个 cache_control,
// 使 system 之后的对话历史也可命中(否则只缓存 system,长会话历史段全价重算)。
if anthropic_cache_enabled() {
Self::apply_message_cache_breakpoint(&mut messages);
}
let tools = req.tools.map(|defs| {
defs.into_iter()
.map(|d| AnthropicToolDef {
@@ -316,6 +323,45 @@ impl AnthropicCompatProvider {
messages.push(serde_json::json!({ "role": "user", "content": blocks }));
}
/// 消息级缓存断点:在「历史段稳定边界」消息的末个 content block 上打 cache_control。
///
/// 断点落点取**末条 user 消息之前的相邻前一条**(末条非 user 收尾的异常序列则取末条):
/// 末条 user 是每轮请求的漂移边界——缓存拆分模式下它是请求末尾追加的易变尾段
/// (日期/清单/augmentation,每次重组),非拆分模式下是本轮新问题,二者都不属于
/// 只增历史;打在它身上断点位置随请求漂移,永远无法命中。标其前一条(assistant /
/// tool_result 归并的 user)则断点始终落在只增历史的末尾,历史只增不改位:
/// 上一轮写入的缓存前缀是本轮请求前缀的逐字节前缀 → 命中后增量续写,滚动前进。
///
/// content 形态:string(简写)→ 升级为单 text block 数组再标;array(blocks)→ 标末块。
/// 断点预算:Anthropic 单请求最多 4 个 cache_control,system 1 个 + 消息级 1 个,余量充足。
fn apply_message_cache_breakpoint(messages: &mut [serde_json::Value]) {
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 cache_control = serde_json::json!({ "type": "ephemeral" });
match msg.get("content").cloned() {
Some(serde_json::Value::Array(mut blocks)) => {
if let Some(last_block) = blocks.last_mut() {
last_block["cache_control"] = cache_control;
}
msg["content"] = serde_json::Value::Array(blocks);
}
Some(serde_json::Value::String(text)) => {
msg["content"] = serde_json::json!([{
"type": "text",
"text": text,
"cache_control": cache_control,
}]);
}
_ => {}
}
}
/// 合并相邻 user 消息为一条(content 拼成 blocks 数组)。
/// Anthropic 协议要求 user/assistant 严格交替,连续 user 触发 1214。
/// 触发场景:flush_tool_results 把 tool_result 转 user 后紧跟新 user(drainQueue 续发,
@@ -752,6 +798,14 @@ mod tests {
use super::*;
// apply_anthropic_event / TokenUsage 经 super::*(含 anthropic_helpers::apply_anthropic_event 的 use)可见。
/// convert_request 读取进程级缓存开关(全局 AtomicBool)。开关切换用例的「开」窗口内,
/// 并行运行的普通转换用例会带上缓存形态(system blocks / 消息级断点)致断言互踩。
/// 所有走 convert_request 的用例统一先取本锁串行化,消除进程级状态的用例间污染。
static CONVERT_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn convert_lock() -> std::sync::MutexGuard<'static, ()> {
CONVERT_LOCK.lock().unwrap_or_else(|p| p.into_inner())
}
/// 辅助:构造 message_start 事件
fn message_start(input_tokens: u32) -> String {
format!(
@@ -945,6 +999,7 @@ mod tests {
/// 含图 user 消息 → content blocks(text + image source.base64)
#[test]
fn anthropic_convert_multimodal_user_blocks() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -980,6 +1035,7 @@ mod tests {
/// 纯文本 user 消息 → content 仍是字符串简写(无图不数组化)
#[test]
fn anthropic_convert_text_only_user_remains_string() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1001,6 +1057,62 @@ mod tests {
assert_eq!(user_msg.get("content").and_then(|c| c.as_str()), Some("hello"));
}
// ---------- 消息级缓存断点(apply_message_cache_breakpoint,纯函数直测) ----------
// 注:不开全局 ANTHROPIC_CACHE_ENABLED(convert_request 层开关默认关,已有断言不受扰),
// 直接测纯函数落点,避免进程级开关与并行测试互踩。
/// 末条 user(漂移边界:易变尾段/本轮新问题)不打断点,标其前一条稳定历史消息。
#[test]
fn cache_breakpoint_marks_message_before_trailing_user() {
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "q1" }),
serde_json::json!({ "role": "assistant", "content": [{"type": "text", "text": "a1"}] }),
serde_json::json!({ "role": "user", "content": "q2" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
// 前一条 assistant 的末块带 cache_control
let a1 = &messages[1];
let blocks = a1["content"].as_array().unwrap();
assert_eq!(blocks.last().unwrap()["cache_control"]["type"], "ephemeral");
// 末条 user 保持字符串简写,未被打点
assert_eq!(messages[2]["content"].as_str(), Some("q2"));
}
/// string content 升级为单 text block 数组并附带 cache_control。
#[test]
fn cache_breakpoint_upgrades_string_content_to_blocks() {
let mut messages = vec![
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);
let blocks = messages[1]["content"].as_array().expect("string 应升级为 blocks 数组");
assert_eq!(blocks.len(), 1);
assert_eq!(blocks[0]["type"], "text");
assert_eq!(blocks[0]["text"], "a1");
assert_eq!(blocks[0]["cache_control"]["type"], "ephemeral");
}
/// 单条消息(无前一条可标)→ 标自身,不 panic。
#[test]
fn cache_breakpoint_single_message_marks_itself() {
let mut messages = vec![
serde_json::json!({ "role": "user", "content": "only" }),
];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
let blocks = messages[0]["content"].as_array().expect("单条应升级为 blocks");
assert_eq!(blocks[0]["cache_control"]["type"], "ephemeral");
}
/// 空 messages 不 panic。
#[test]
fn cache_breakpoint_empty_messages_noop() {
let mut messages: Vec<serde_json::Value> = vec![];
AnthropicCompatProvider::apply_message_cache_breakpoint(&mut messages);
assert!(messages.is_empty());
}
// ---------- ensure_leading_user(首条 assistant → 补 user 占位,保留上下文)----------
/// 辅助:构造 assistant(tool_use) 消息
@@ -1016,6 +1128,7 @@ mod tests {
/// (原"砍"策略会把每对三元组砍掉,多轮砍到空,丢失全部工具调用历史——"补"策略零丢失。)
#[test]
fn anthropic_ensure_leading_user_tool_use_chain_preserves_context() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1054,6 +1167,7 @@ mod tests {
/// 首条 assistant 无 tool_use → 补占位,首条 user,原上下文保留。
#[test]
fn anthropic_ensure_leading_user_plain_assistant() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1080,6 +1194,7 @@ mod tests {
/// 正常序列(user 开头)不补占位——零回归验证。
#[test]
fn anthropic_ensure_leading_user_normal_sequence_unchanged() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1107,6 +1222,7 @@ mod tests {
/// 补一个 user 占位后全部保留,验证多轮链不丢数据、precheck 通过(原"砍"策略此场景砍到空)。
#[test]
fn anthropic_ensure_leading_user_three_round_chain() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1144,6 +1260,7 @@ mod tests {
/// 避免发空 messages 触发 precheck "messages 为空"(降级让会话能继续)。
#[test]
fn anthropic_ensure_leading_user_empty_messages_gets_placeholder() {
let _convert = convert_lock();
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
let req = CompletionRequest {
model: "claude-3-5-sonnet".into(),
@@ -1173,6 +1290,7 @@ mod tests {
/// (两态共用进程级全局开关,拆两个用例并行跑会互相污染——单测内顺序切换断言,末尾复位。)
#[test]
fn anthropic_cache_switch_controls_system_shape() {
let _convert = convert_lock();
set_anthropic_cache_enabled(false); // 兜底复位,防污染其他并行用例
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
+3 -1
View File
@@ -1273,9 +1273,10 @@ mod tests {
];
for intent in all_intents {
let subset = tool_subset_for(&intent);
// 全 registry 工具名(五 domain 并集:Data + DataReadOnly + File + Exec + Http)
// 全 registry 工具名(六 domain 并集:Data + DataReadOnly + File + Exec + Http + Agent)
// 注:DataReadOnly 工具名是 Data 的子集,chain 它仅为语义显式(并集去重无副作用),
// 防 Code subset 里 list_projects 等 DataReadOnly 工具被判"不在 registry"。
// Agent(spawn_agent)是编排委派基本盘成员,子集合法出现,漏 chain 会误报。
let registry: std::collections::HashSet<&str> = ToolDomain::Data
.tools()
.iter()
@@ -1283,6 +1284,7 @@ mod tests {
.chain(ToolDomain::File.tools().iter())
.chain(ToolDomain::Exec.tools().iter())
.chain(ToolDomain::Http.tools().iter())
.chain(ToolDomain::Agent.tools().iter())
.copied()
.collect();
for name in &subset {
+115 -9
View File
@@ -97,25 +97,78 @@ pub(crate) struct OpenAiFunctionResp {
pub arguments: String,
}
/// OpenAI 标准嵌套缓存口径形态(usage.prompt_tokens_details):
/// GLM/Qwen/OpenAI 官方在嵌套结构报缓存命中,DeepSeek 用扁平 prompt_cache_hit_tokens。
/// cached_tokens 用 Option 容错给空 details 对象的网关。
#[derive(Debug, Default, Deserialize)]
pub(crate) struct PromptTokensDetails {
#[serde(default)]
pub cached_tokens: Option<u64>,
}
/// usage 反序列化中间载体:与 OpenAiUsage 字段同构,经 `From` 做缓存口径归一后产出
/// OpenAiUsage(见 impl From)。流式 SSE 与非流式 JSON 两条解析路径因此都自动获得
/// 嵌套 cached_tokens 回填,消费方一律读归一后的扁平字段。
#[derive(Debug, Deserialize)]
pub(crate) struct OpenAiUsageRaw {
pub prompt_tokens: u32,
pub completion_tokens: u32,
pub total_tokens: u32,
/// DeepSeek 扩展:缓存命中 token(低价,deepseek-chat/reasoner)。无此字段 → default 0。
#[serde(default)]
pub prompt_cache_hit_tokens: u32,
/// DeepSeek 扩展:未命中 token(全价真实输入,prompt_cache_miss_tokens)。无此字段 → default 0。
#[serde(default)]
pub prompt_cache_miss_tokens: u32,
/// DeepSeek-reasoner / OpenAI o1 扩展:思考 token(隐藏输出,reasoning_tokens)。
/// 非 reasoning 模型无此字段 → default 0。
#[serde(default)]
pub reasoning_tokens: u32,
/// OpenAI 标准嵌套缓存口径(GLM/Qwen/OpenAI 官方)。
#[serde(default)]
pub prompt_tokens_details: Option<PromptTokensDetails>,
}
/// OpenAI usage(缓存口径已归一)
///
/// 归一发生在反序列化时(`#[serde(from)]`):扁平 hit 有值优先(DeepSeek 自报 hit+miss);
/// 扁平 hit 未报而嵌套 cached_tokens > 0 时回填 hit,miss 推断为 prompt - hit(嵌套口径
/// 缺全价分项);两种口径都没有则保持 0。消费方(apply_openai_sse / openai_compat 非流式)
/// 直接读扁平字段,无需感知 provider 差异。
#[derive(Debug, Deserialize)]
#[serde(from = "OpenAiUsageRaw")]
pub(crate) struct OpenAiUsage {
pub prompt_tokens: u32,
pub completion_tokens: u32,
pub total_tokens: u32,
/// DeepSeek 扩展:缓存命中 token(低价,deepseek-chat/reasoner prompt_cache_hit_tokens)。
/// OpenAI 官方(o1 等)无此字段 → serde default 0。其他 OpenAI 兼容网关若支持 cache 也用此名。
#[serde(default)]
/// 缓存命中 token(低价;扁平优先,缺失时回填自嵌套 cached_tokens)
pub prompt_cache_hit_tokens: u32,
/// DeepSeek 扩展:未命中 token(全价真实输入,prompt_cache_miss_tokens)。
/// OpenAI 官方无此字段 → serde default 0。
#[serde(default)]
/// 缓存未命中 token(全价真实输入;嵌套口径推断时 = prompt - hit)
pub prompt_cache_miss_tokens: u32,
/// DeepSeek-reasoner / OpenAI o1 扩展:思考 token(隐藏输出,reasoning_tokens)。
/// 非 reasoning 模型无此字段 → serde default 0。
#[serde(default)]
/// 思考 token(隐藏输出)
pub reasoning_tokens: u32,
}
impl From<OpenAiUsageRaw> for OpenAiUsage {
fn from(raw: OpenAiUsageRaw) -> Self {
let (hit, miss) = match raw.prompt_tokens_details.and_then(|d| d.cached_tokens) {
Some(nested) if nested > 0 && raw.prompt_cache_hit_tokens == 0 => {
let hit = u32::try_from(nested).unwrap_or(u32::MAX);
(hit, raw.prompt_tokens.saturating_sub(hit))
}
_ => (raw.prompt_cache_hit_tokens, raw.prompt_cache_miss_tokens),
};
Self {
prompt_tokens: raw.prompt_tokens,
completion_tokens: raw.completion_tokens,
total_tokens: raw.total_tokens,
prompt_cache_hit_tokens: hit,
prompt_cache_miss_tokens: miss,
reasoning_tokens: raw.reasoning_tokens,
}
}
}
/// SSE 流式响应 chunk
#[derive(Debug, Deserialize)]
pub(crate) struct OpenAiStreamChunk {
@@ -287,3 +340,56 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// DeepSeek 扁平口径:hit/miss 各取其值,不参与嵌套回填
#[test]
fn usage_deepseek_flat_cache() {
let u: OpenAiUsage = serde_json::from_str(
r#"{"prompt_tokens":100,"completion_tokens":10,"total_tokens":110,
"prompt_cache_hit_tokens":64,"prompt_cache_miss_tokens":36}"#,
)
.unwrap();
assert_eq!(u.prompt_tokens, 100);
assert_eq!(u.prompt_cache_hit_tokens, 64);
assert_eq!(u.prompt_cache_miss_tokens, 36);
}
/// OpenAI/GLM 嵌套口径:扁平未报,hit 回填自 cached_tokens,miss 推断为 prompt - hit
#[test]
fn usage_nested_cached_tokens_backfill() {
let u: OpenAiUsage = serde_json::from_str(
r#"{"prompt_tokens":100,"completion_tokens":10,"total_tokens":110,
"prompt_tokens_details":{"cached_tokens":64}}"#,
)
.unwrap();
assert_eq!(u.prompt_cache_hit_tokens, 64);
assert_eq!(u.prompt_cache_miss_tokens, 36);
}
/// 两种口径都没有:hit/miss 保持 0(前端不渲染 cache 段)
#[test]
fn usage_no_cache_fields() {
let u: OpenAiUsage = serde_json::from_str(
r#"{"prompt_tokens":100,"completion_tokens":10,"total_tokens":110}"#,
)
.unwrap();
assert_eq!(u.prompt_cache_hit_tokens, 0);
assert_eq!(u.prompt_cache_miss_tokens, 0);
}
/// 端到端:usage-only chunk 经 apply_openai_sse 累加路径带出归一后的缓存分项
#[test]
fn sse_usage_chunk_nested_cache_reaches_token_usage() {
let data = r#"{"choices":[],"usage":{"prompt_tokens":100,"completion_tokens":10,
"total_tokens":110,"prompt_tokens_details":{"cached_tokens":64}}}"#;
let mut accum = None;
let chunk = apply_openai_sse(data, &mut accum);
let usage = chunk.usage.expect("usage chunk 应携带 usage");
assert_eq!(usage.prompt_cache_hit_tokens, 64);
assert_eq!(usage.prompt_cache_miss_tokens, 36);
}
}