lxy
6f5c6b7532
- 块状裁剪: 水位目标+轮边界吸附+病态回退,贪心路径保留为开关回退 - 存储态 System 锚点出请求时视图转 user,治前缀缓存整体折叠失配 - BP3 断点机制+缓存命中率观测日志,调参有据 - 压缩摘要可召回: JSONL 档案落盘+句柄行注入,take 语义一次性消费 - 字符估算修正: CJK 分档计权+改图地板,估算对象对齐实际消耗 - namespace 引用跨重启内容哈希懒回填
1758 lines
85 KiB
Rust
1758 lines
85 KiB
Rust
//! Anthropic 兼容 Provider — 通过 /v1/messages 端点实现
|
||
//!
|
||
//! 覆盖: Claude 官方 / GLM 订阅端点 (open.bigmodel.cn/api/anthropic) / 任意 Messages API 网关
|
||
//! 支持: 同步调用 + SSE 流式 + Tool Use
|
||
//!
|
||
//! 与 OpenAI 协议的关键差异由本模块内部完成转换，对外仍暴露统一的 LlmProvider trait，
|
||
//! 上层 (Agentic Loop / AiNode) 无需感知协议。
|
||
//!
|
||
//! 协议数据结构（请求/响应 struct）与 SSE 事件解析纯函数已抽至 `anthropic_helpers`，
|
||
//! 本模块仅保留 Provider struct + impl（HTTP 调用），Rust impl 块不可跨文件故作此切分。
|
||
|
||
use async_trait::async_trait;
|
||
use futures::StreamExt;
|
||
use reqwest::Client;
|
||
use std::time::Duration;
|
||
use tracing::{debug, error, warn};
|
||
|
||
use crate::provider::{
|
||
tool_call_id_or_fallback, CompletionRequest, CompletionResponse, LlmProvider, MessageRole,
|
||
StreamResult, TokenUsage, ToolCall,
|
||
};
|
||
// ChatMessage 仅单测构造 CompletionRequest 用，避免非 test 构建的 unused import 警告。
|
||
#[cfg(test)]
|
||
use crate::provider::ChatMessage;
|
||
use crate::retry::{
|
||
retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable,
|
||
};
|
||
|
||
// 协议数据结构 + SSE 纯解析（apply_anthropic_event / AnthropicRequest 等）抽至 anthropic_helpers，
|
||
// 此处 use 引入以保持本模块内引用路径不变（零行为变更搬迁）。
|
||
use crate::anthropic_helpers::{
|
||
apply_anthropic_event, AnthropicRequest, AnthropicResponse, AnthropicToolDef, SystemBlock,
|
||
ANTHROPIC_VERSION, DEFAULT_MAX_TOKENS,
|
||
};
|
||
|
||
// ============================================================
|
||
// Anthropic prompt caching 开关（进程级全局，受 AppState settings 同步）
|
||
// ============================================================
|
||
|
||
/// Anthropic prompt caching 总开关（默认关）。
|
||
///
|
||
/// 该分支同时服务 Claude 官方 / GLM 订阅端点 (open.bigmodel.cn/api/anthropic) / 任意 Messages
|
||
/// API 网关——**cache_control 字段可能被非官方网关拒收**，故必须默认关、用户对支持的端点显式开启。
|
||
///
|
||
/// 开关开：`convert_request` 把 system 包成 blocks 数组并附加 `cache_control:{type:ephemeral}`
|
||
/// （配合上层把易变段挪出 system，让 system 稳定段真正命中缓存），
|
||
/// 并在历史段稳定边界补消息级断点（使 system 之后的对话历史也可命中）。
|
||
/// 开关关：system 保持纯字符串、消息不打断点（现状行为，零风险兼容网关）。
|
||
///
|
||
/// 进程级全局而非 per-provider：上层(AppState settings)按「用户对当前启用的 anthropic 端点
|
||
/// 是否支持缓存」一次性开启，所有 anthropic 请求一致生效。默认关。
|
||
static ANTHROPIC_CACHE_ENABLED: std::sync::atomic::AtomicBool =
|
||
std::sync::atomic::AtomicBool::new(false);
|
||
|
||
/// 设置 anthropic prompt caching 开关（上层读取 AppState settings 后同步）。
|
||
pub fn set_anthropic_cache_enabled(enabled: bool) {
|
||
ANTHROPIC_CACHE_ENABLED.store(enabled, std::sync::atomic::Ordering::SeqCst);
|
||
}
|
||
|
||
/// 读取 anthropic prompt caching 开关（`convert_request` 组装 system 形态用）。
|
||
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 实现
|
||
// ============================================================
|
||
|
||
/// Anthropic 兼容 LLM Provider
|
||
pub struct AnthropicCompatProvider {
|
||
client: Client,
|
||
api_key: String,
|
||
base_url: String,
|
||
default_model: String,
|
||
}
|
||
|
||
impl AnthropicCompatProvider {
|
||
/// 创建 Provider
|
||
///
|
||
/// - `base_url`: 如 `https://api.anthropic.com`、`https://open.bigmodel.cn/api/anthropic`
|
||
/// - `api_key`: API 密钥（Anthropic 用 x-api-key 头，非 Bearer）
|
||
/// - `default_model`: 默认模型名称
|
||
pub fn new(base_url: impl Into<String>, api_key: impl Into<String>, default_model: impl Into<String>) -> Self {
|
||
// reqwest Client 构建抽 crate::build_provider_client（与 OpenAI 共用 DRY）。
|
||
let client = crate::build_provider_client();
|
||
Self {
|
||
client,
|
||
api_key: api_key.into(),
|
||
base_url: base_url.into(),
|
||
default_model: default_model.into(),
|
||
}
|
||
}
|
||
|
||
/// 构建 messages 端点 URL
|
||
///
|
||
/// - 已含 `/v1/messages` → 直接用
|
||
/// - 以 `/v1` 结尾 → 补 `/messages`
|
||
/// - 否则（如 `.../api/anthropic`、`api.anthropic.com`）→ 补 `/v1/messages`
|
||
fn messages_url(&self) -> String {
|
||
let base = self.base_url.trim_end_matches('/');
|
||
if base.ends_with("/v1/messages") {
|
||
return base.to_string();
|
||
}
|
||
if base.ends_with("/v1") {
|
||
return format!("{}/messages", base);
|
||
}
|
||
format!("{}/v1/messages", base)
|
||
}
|
||
|
||
/// 将统一 CompletionRequest 转换为 Anthropic 请求体
|
||
///
|
||
/// 转换要点：
|
||
/// - system 消息从 messages 抽离到顶层 system 字段
|
||
/// - assistant 带 tool_calls → content 数组含 text + tool_use 块
|
||
/// - tool_result（role=Tool）→ user 消息含 tool_result 块；连续多个合并为一条 user
|
||
/// - tool_definitions 的 parameters → input_schema
|
||
fn convert_request(&self, req: CompletionRequest) -> AnthropicRequest {
|
||
let model = if req.model.is_empty() {
|
||
self.default_model.clone()
|
||
} else {
|
||
req.model
|
||
};
|
||
|
||
// 抽离 system 消息。
|
||
// 开关开 → blocks 数组 + cache_control(Anthropic prompt caching,让 system 稳定段在
|
||
// 支持缓存的端点上命中);开关关 → 纯字符串(兼容非官方网关,零风险)。
|
||
let system: Option<SystemBlock> = {
|
||
let sys: Vec<String> = req
|
||
.messages
|
||
.iter()
|
||
.filter(|m| matches!(m.role, MessageRole::System))
|
||
.map(|m| m.content.clone())
|
||
.collect();
|
||
if sys.is_empty() {
|
||
None
|
||
} else if anthropic_cache_enabled() {
|
||
Some(SystemBlock::Cached(vec![serde_json::json!({
|
||
"type": "text",
|
||
"text": sys.join("\n\n"),
|
||
"cache_control": { "type": "ephemeral" },
|
||
})]))
|
||
} else {
|
||
Some(SystemBlock::Plain(sys.join("\n\n")))
|
||
}
|
||
};
|
||
|
||
// 构建非 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 (src_idx, m) in req.messages.iter().enumerate() {
|
||
match m.role {
|
||
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
|
||
// 报 'ClaudeContentBlockToolResult' object has no attribute 'id'，
|
||
// 致会话卡死持续 500），跳过该块并告警。
|
||
match &m.tool_call_id {
|
||
Some(tid) if !tid.is_empty() => {
|
||
pending_tool_results.push(serde_json::json!({
|
||
"type": "tool_result",
|
||
"tool_use_id": tid,
|
||
"content": m.content,
|
||
}));
|
||
}
|
||
_ => {
|
||
warn!(
|
||
content_preview = %m.content.chars().take(80).collect::<String>(),
|
||
"Anthropic tool_result 缺少 tool_use_id，已跳过该块（发 null 会触发 GLM 500）"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
MessageRole::User => {
|
||
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}}。
|
||
// Anthropic 协议要求 image 必须内嵌 base64（不接受 URL 直传）。
|
||
// 现状：前端只产 base64 模式图片片，url 模式当前不可达。
|
||
// 未来若加 url 图片输入，必须在 commands 层补 url→base64 预拉
|
||
//（provider 不发额外 HTTP），否则下方兜底会发空 data 致 Anthropic 400。
|
||
// 纯文本消息（无图）保持原字符串简写，与现有端点零回归。
|
||
if m.has_image() {
|
||
let blocks: Vec<serde_json::Value> = m
|
||
.flattened_parts()
|
||
.into_iter()
|
||
.map(Self::content_part_to_block)
|
||
.collect();
|
||
messages.push(serde_json::json!({ "role": "user", "content": blocks }));
|
||
} 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, &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 }));
|
||
}
|
||
if let Some(calls) = &m.tool_calls {
|
||
for tc in calls {
|
||
// arguments 非法 JSON(流式中断残留 / ToolCall::new 默认空串)
|
||
// → 空 object 兜底。Anthropic/GLM 要求 tool_use.input 必为 object,
|
||
// null 直触发 1214「messages 参数非法」。
|
||
let input: serde_json::Value = serde_json::from_str(&tc.function.arguments)
|
||
.unwrap_or_else(|_| serde_json::json!({}));
|
||
content.push(serde_json::json!({
|
||
"type": "tool_use",
|
||
"id": tc.id,
|
||
"name": tc.function.name,
|
||
"input": input,
|
||
}));
|
||
}
|
||
}
|
||
if content.is_empty() {
|
||
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, &mut msgs_after_anchor);
|
||
|
||
// 合并相邻 user 块。Anthropic 协议要求 user/assistant 严格交替,连续 user
|
||
// 触发 GLM 1214。场景:drainQueue 续发(前一轮以 tool_result 结尾 + 新 user)→ flush 把
|
||
// tool_result 转成 user 后紧跟 push 新 user → 连续两 user。合并成一条 user 含
|
||
// [tool_result..., text] blocks(Anthropic 允许一条 user 多 blocks),打破恶性循环。
|
||
Self::merge_consecutive_users(&mut messages);
|
||
|
||
// 保证首条为 user(Anthropic 协议硬性要求 messages[0].role == "user")。
|
||
// 上游绕过 ContextManager::sanitize_messages 的调用方(标题生成 / 知识注入 / 工作流 AI
|
||
// 节点等直接构造 CompletionRequest 的路径)可能传入首条 assistant 的序列——会话恢复、
|
||
// 续发或历史片段截取时,真正的首条 user 已被裁剪/压缩掉,直接发触发 precheck
|
||
// "首条 role=assistant 非法"。
|
||
// 根本解用"补"而非"砍":开头补一条 user 占位,保留全部上下文(assistant 的 tool_use
|
||
// 与其后 user 的 tool_result 配对完整),precheck 必过。砍会丢工具调用历史,且多轮
|
||
// [asst(tu),user(tr),asst(tu),user(tr),...] 逐对砍到空。占位是异常路径轻量噪声
|
||
// (正常会话首条本就是 user),远优于丢弃上下文。
|
||
Self::ensure_leading_user(&mut messages);
|
||
|
||
// 消息级缓存断点(开关开):BP2 打在历史段稳定边界,使 system 之后的对话历史
|
||
// 也可命中(否则只缓存 system,长会话历史段全价重算);BP3(缓存友好开关,默认开)
|
||
// 加打摘要锚点链之后第一条,BP2 因裁剪边界/压缩失效时 [system+摘要链] 小前缀仍命中。
|
||
if anthropic_cache_enabled() {
|
||
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| {
|
||
defs.into_iter()
|
||
.map(|d| AnthropicToolDef {
|
||
name: d.function.name,
|
||
description: Some(d.function.description).filter(|s| !s.is_empty()),
|
||
// input_schema 非 object(未来误用)→ 兜底 {"type":"object"},
|
||
// 防 Anthropic 拒非法 tool schema(当前全走 object_schema 恒 object,纯防御)。
|
||
input_schema: if d.function.parameters.is_object() {
|
||
d.function.parameters
|
||
} else {
|
||
serde_json::json!({"type": "object"})
|
||
},
|
||
})
|
||
.collect()
|
||
});
|
||
|
||
// 模型不支持 temperature(如商汤 sensenova 仅接受 1.0)→ 省略该字段,避免上游 HTTP 400。
|
||
// 判定走 model_probe(命中 presets 表;未命中默认支持,保持原样透传零回归)。
|
||
let temperature = if req.temperature.is_some()
|
||
&& !crate::model_probe::probe(&model).temperature_supported
|
||
{
|
||
tracing::debug!(model = %model, "模型不支持 temperature,已省略该参数");
|
||
None
|
||
} else {
|
||
req.temperature
|
||
};
|
||
|
||
AnthropicRequest {
|
||
model,
|
||
messages,
|
||
max_tokens: req.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS),
|
||
system,
|
||
temperature,
|
||
stream: req.stream,
|
||
tools,
|
||
tool_choice: req.tool_choice,
|
||
}
|
||
}
|
||
|
||
/// 单个 ContentPart → Anthropic content block（text/image）。
|
||
/// - Text 片 → {type:text, text}
|
||
/// - Image 片 → {type:image, source:{type:base64, media_type, data}}
|
||
/// base64 内嵌；本地 file:// 引用(#958)先展开为 base64，http(s) url 模式
|
||
/// 当前不可达，兜底发空 data + warn（保留原行为）。
|
||
/// match 取得 p 所有权后直接 move media_type/base64，避免大 base64 clone。
|
||
fn content_part_to_block(p: crate::provider::ContentPart) -> serde_json::Value {
|
||
let p = crate::image_ref::inline_local_image_part(p);
|
||
match p {
|
||
crate::provider::ContentPart::Text { text } => serde_json::json!({
|
||
"type": "text",
|
||
"text": text,
|
||
}),
|
||
crate::provider::ContentPart::Image { url, base64, media_type, alt: _ } => {
|
||
let mt = media_type.unwrap_or_else(|| "image/png".into());
|
||
let data = base64.unwrap_or_else(|| {
|
||
// 完整性兜底：当前 url 模式不可达（前端只产 base64 图片片）。
|
||
// 若未来接入 url 图片输入而 commands 层未补 url→base64 预拉，
|
||
// 此处会发空 data 致 Anthropic 400，warn 留痕但不阻塞（避免静默吞数据）。
|
||
if url.is_some() {
|
||
warn!(
|
||
url = ?url,
|
||
"Anthropic user 消息含 Image(url) 但 base64 缺失，将发空 data（commands 层未补 url→base64 预拉）"
|
||
);
|
||
}
|
||
String::new()
|
||
});
|
||
serde_json::json!({
|
||
"type": "image",
|
||
"source": {
|
||
"type": "base64",
|
||
"media_type": mt,
|
||
"data": data,
|
||
}
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 将累积的 tool_result 块作为一条 user 消息 flush 进消息列表
|
||
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;
|
||
}
|
||
}
|
||
|
||
/// 消息级缓存断点:BP2 打在「历史段稳定边界」的末个 content block 上,BP3(可选)
|
||
/// 加打「摘要锚点链之后第一条」——`after_anchor` 为锚点链后的消息条数
|
||
/// (Some(0)=锚点收尾无后继,不打;None=无锚点,只打 BP2)。
|
||
///
|
||
/// 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 + 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 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)) => {
|
||
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 续发,
|
||
/// 前一轮以 tool_result 结尾)。合并成一条 user 含 [tool_result..., text] blocks,合法。
|
||
/// content 形态:string(简写)或 array(blocks),统一规范化为 array 后拼接。
|
||
fn merge_consecutive_users(messages: &mut Vec<serde_json::Value>) {
|
||
let is_user = |m: &serde_json::Value| m.get("role").and_then(|r| r.as_str()) == Some("user");
|
||
let mut i = 0;
|
||
while i + 1 < messages.len() {
|
||
if is_user(&messages[i]) && is_user(&messages[i + 1]) {
|
||
let next_content = messages[i + 1].get("content").cloned().unwrap_or(serde_json::Value::Null);
|
||
let cur = &mut messages[i];
|
||
// 当前 user content 规范化为 array
|
||
let mut blocks: Vec<serde_json::Value> = match cur.get("content").cloned() {
|
||
Some(serde_json::Value::String(s)) => vec![serde_json::json!({ "type": "text", "text": s })],
|
||
Some(serde_json::Value::Array(a)) => a,
|
||
_ => vec![],
|
||
};
|
||
// 并入下一个 user content
|
||
match next_content {
|
||
serde_json::Value::String(s) => blocks.push(serde_json::json!({ "type": "text", "text": s })),
|
||
serde_json::Value::Array(a) => blocks.extend(a),
|
||
_ => {}
|
||
}
|
||
cur["content"] = serde_json::Value::Array(blocks);
|
||
messages.remove(i + 1);
|
||
// 不增 i:继续合并 i 与新 i+1(可能多个连续 user)
|
||
} else {
|
||
i += 1;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 保证 messages 首条为 user(Anthropic 协议硬性要求 messages[0].role=="user")。
|
||
///
|
||
/// 上游绕过 `ContextManager::sanitize_messages` 的调用方(标题生成 / 知识注入 / 工作流 AI
|
||
/// 节点等直接构造 CompletionRequest 的路径)可能传入首条 assistant 的序列——会话恢复、续发
|
||
/// 或从历史片段截取时,真正的首条 user 已被裁剪/压缩掉。直接发触发 precheck "首条
|
||
/// role=assistant 非法"。
|
||
///
|
||
/// **用"补"而非"砍"**:在开头插一条 user 占位消息。
|
||
/// - 砍掉首条 assistant 会丢失有效上下文(其 tool_use 与后续 user 的 tool_result 是完整
|
||
/// 配对),且多轮 [asst(tu),user(tr),asst(tu),user(tr),...] 会被逐对砍到空;
|
||
/// - 补占位则全部上下文保留(占位 user 紧贴原首条 assistant,不破坏 user/assistant 交替),
|
||
/// tool_use/tool_result 配对完整不动,**无 orphan 产生**(故砍策略那套 orphan 清理在此不需要),
|
||
/// 占位 content 非空过 precheck 的"user content 空"校验。
|
||
///
|
||
/// 仅异常路径触发(正常会话首条本就是 user),占位文案是轻量噪声,远优于丢弃工具调用历史。
|
||
fn ensure_leading_user(messages: &mut Vec<serde_json::Value>) {
|
||
let first_role = messages
|
||
.first()
|
||
.and_then(|m| m.get("role").and_then(|r| r.as_str()))
|
||
.unwrap_or("");
|
||
if first_role == "user" {
|
||
// 首条 user 但 content 为空串(会话压缩/续发残留,典型 text(0B)):原样发出会被
|
||
// precheck "user content 空" 拦下而整请求失败。与上方补占位同思路——就地填
|
||
// 占位文案(空串无信息可丢),上游实测接受非空占位,precheck 亦放行。
|
||
let empty_str = messages[0]
|
||
.get("content")
|
||
.and_then(|c| c.as_str())
|
||
.map(|s| s.is_empty())
|
||
.unwrap_or(false);
|
||
if empty_str {
|
||
warn!(
|
||
msg_count = messages.len(),
|
||
"ensure_leading_user: 首条 user content 为空串,填占位(防 precheck 拦截)"
|
||
);
|
||
messages[0]["content"] =
|
||
serde_json::json!("(continued from previous context)");
|
||
}
|
||
return;
|
||
}
|
||
// 空 Vec(异常会话经 sanitize 清空)或首条非 user → 补 user 占位:
|
||
// Anthropic 协议要求 messages 至少一条且首条 user,补占位让降级会话能继续(不丢这条兜底)。
|
||
warn!(
|
||
first_role,
|
||
msg_count = messages.len(),
|
||
"ensure_leading_user: 首条非 user(含空),补 user 占位(防 Anthropic 首条 assistant/空 messages 非法)"
|
||
);
|
||
messages.insert(0, serde_json::json!({
|
||
"role": "user",
|
||
"content": "(continued from previous context)",
|
||
}));
|
||
}
|
||
|
||
/// 生成 messages 诊断摘要(每条 role + content 形态 + tool 标记),不含敏感数据。
|
||
/// 1214 类错误时随 bail 文案直达前端 raw,定位哪条/字段非法。
|
||
fn summarize_messages(messages: &[serde_json::Value]) -> String {
|
||
let lines: Vec<String> = messages
|
||
.iter()
|
||
.enumerate()
|
||
.map(|(i, m)| {
|
||
let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("?");
|
||
let desc = match m.get("content") {
|
||
Some(serde_json::Value::String(s)) => format!("text({}B)", s.len()),
|
||
Some(serde_json::Value::Array(blocks)) => {
|
||
let parts: Vec<String> = blocks
|
||
.iter()
|
||
.map(|b| {
|
||
let ty = b.get("type").and_then(|t| t.as_str()).unwrap_or("?");
|
||
match ty {
|
||
"text" => format!(
|
||
"text({}B)",
|
||
b.get("text").and_then(|t| t.as_str()).map(|s| s.len()).unwrap_or(0)
|
||
),
|
||
"tool_use" => format!(
|
||
"tool_use[id={},input_obj={}]",
|
||
b.get("id").and_then(|t| t.as_str()).unwrap_or(""),
|
||
b.get("input").map(|v| v.is_object()).unwrap_or(false)
|
||
),
|
||
"tool_result" => format!(
|
||
"tool_result[tid={}]",
|
||
b.get("tool_use_id").and_then(|t| t.as_str()).unwrap_or("")
|
||
),
|
||
"image" => "image".to_string(),
|
||
_ => ty.to_string(),
|
||
}
|
||
})
|
||
.collect();
|
||
format!("[{}]", parts.join(","))
|
||
}
|
||
_ => "?".to_string(),
|
||
};
|
||
format!("#{}:{} {}", i, role, desc)
|
||
})
|
||
.collect();
|
||
format!("{} msgs: {}", lines.len(), lines.join(" | "))
|
||
}
|
||
|
||
/// 协议预检——扫 messages 发现确定非法形态,命中返回原因(仅诊断不修复)。
|
||
/// 覆盖:首条非 user / 连续同 role / tool_use input 非 object / 空 content / orphan tool_result
|
||
/// (tool_use_id 无前置 tool_use,常见于裁剪/过滤后 assistant 被删但 tool_result 留)。
|
||
fn precheck_messages(messages: &[serde_json::Value]) -> Result<(), String> {
|
||
if messages.is_empty() {
|
||
return Err("messages 为空".into());
|
||
}
|
||
let first_role = messages[0].get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
if first_role != "user" {
|
||
return Err(format!("首条 role={} 非法(须 user)", first_role));
|
||
}
|
||
for w in messages.windows(2) {
|
||
let r0 = w[0].get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
let r1 = w[1].get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
if r0 == r1 && (r0 == "user" || r0 == "assistant") {
|
||
return Err(format!("连续同 role={}", r0));
|
||
}
|
||
}
|
||
let mut tool_use_ids: Vec<&str> = Vec::new();
|
||
for (i, m) in messages.iter().enumerate() {
|
||
let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
match m.get("content") {
|
||
Some(serde_json::Value::String(s)) if s.is_empty() && role == "user" => {
|
||
return Err(format!("#{} user content 空", i));
|
||
}
|
||
Some(serde_json::Value::Array(blocks)) => {
|
||
if blocks.is_empty() && role == "user" {
|
||
return Err(format!("#{} user content 空数组", i));
|
||
}
|
||
for b in blocks {
|
||
Self::check_block(b, i, &mut tool_use_ids)?;
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// 校验单个 content block（precheck_messages 内部用）。
|
||
/// - tool_use: 收集 id,校验 input 为 object
|
||
/// - tool_result: 校验 tool_use_id 有前置 tool_use（非 orphan）
|
||
fn check_block<'a>(
|
||
b: &'a serde_json::Value,
|
||
idx: usize,
|
||
tool_use_ids: &mut Vec<&'a str>,
|
||
) -> Result<(), String> {
|
||
match b.get("type").and_then(|t| t.as_str()).unwrap_or("") {
|
||
"tool_use" => {
|
||
let id = b.get("id").and_then(|t| t.as_str()).unwrap_or("");
|
||
tool_use_ids.push(id);
|
||
if !b.get("input").map(|v| v.is_object()).unwrap_or(false) {
|
||
return Err(format!("#{} tool_use input 非 object", idx));
|
||
}
|
||
}
|
||
"tool_result" => {
|
||
let tid = b.get("tool_use_id").and_then(|t| t.as_str()).unwrap_or("");
|
||
if !tid.is_empty() && !tool_use_ids.contains(&tid) {
|
||
return Err(format!(
|
||
"#{} orphan tool_result(tid={} 无前置 tool_use)",
|
||
idx, tid
|
||
));
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// 统一鉴权头：x-api-key + anthropic-version
|
||
fn auth_headers(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
||
rb.header("x-api-key", &self.api_key)
|
||
.header("anthropic-version", ANTHROPIC_VERSION)
|
||
.header("Content-Type", "application/json")
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl LlmProvider for AnthropicCompatProvider {
|
||
async fn complete(&self, request: CompletionRequest) -> anyhow::Result<CompletionResponse> {
|
||
let mut req = request;
|
||
req.stream = false;
|
||
let body = self.convert_request(req);
|
||
|
||
// 协议预检——命中非法 bail 含 messages 摘要,把 GLM 模糊 1214 转明确诊断
|
||
if let Err(reason) = Self::precheck_messages(&body.messages) {
|
||
let summary = Self::summarize_messages(&body.messages);
|
||
warn!(%reason, %summary, "Anthropic messages 协议预检失败");
|
||
anyhow::bail!("messages 预检失败: {} | 摘要: {}", reason, summary);
|
||
}
|
||
|
||
debug!(model = %body.model, "Anthropic 同步调用");
|
||
|
||
// 指数退避重试: 包裹 send + 状态码判定。
|
||
// 同时补单请求 timeout(Anthropic 同步路径无 timeout 会 hang),此处加 60s,与 OpenAI 路径对齐。
|
||
let label = format!("Anthropic[{}]", body.model);
|
||
retry_with_backoff(&label, move |_| {
|
||
let client = self.client.clone();
|
||
let url = self.messages_url();
|
||
let api_key = self.api_key.clone();
|
||
let body = body.clone();
|
||
async move {
|
||
// send: 补 60s 单请求 timeout(此前缺失，FR-R4 遗漏)
|
||
let rb = client
|
||
.post(url)
|
||
.header("x-api-key", &api_key)
|
||
.header("anthropic-version", ANTHROPIC_VERSION)
|
||
.header("Content-Type", "application/json")
|
||
.timeout(Duration::from_secs(60))
|
||
.version(reqwest::Version::HTTP_11)
|
||
.json(&body);
|
||
let resp = match rb.send().await {
|
||
Ok(r) => r,
|
||
Err(e) => {
|
||
// 记 reqwest 错误源因链(is_*/source)。原仅 Display
|
||
// "error sending request for url" 无法定位 reset/TLS/超时/body 真因。
|
||
tracing::error!(
|
||
is_timeout = e.is_timeout(),
|
||
is_connect = e.is_connect(),
|
||
is_body = e.is_body(),
|
||
is_request = e.is_request(),
|
||
url = ?e.url(),
|
||
source = ?std::error::Error::source(&e),
|
||
"Anthropic 同步 send 失败"
|
||
);
|
||
// is_decode 补可重试:body 中途断连在 reqwest 归 decode 类,complete 重发
|
||
// 幂等安全(受 3 次/30s 预算约束,真正格式错重试同败,预算内止损)。
|
||
if is_reqwest_error_retryable(&e) || e.is_decode() {
|
||
return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e));
|
||
}
|
||
return AttemptOutcome::Fatal(format!(
|
||
"请求失败(不可重试): {} [timeout={} connect={} body={} src={:?}]",
|
||
e,
|
||
e.is_timeout(),
|
||
e.is_connect(),
|
||
e.is_body(),
|
||
std::error::Error::source(&e)
|
||
));
|
||
}
|
||
};
|
||
if !resp.status().is_success() {
|
||
let status = resp.status().as_u16();
|
||
let text = resp.text().await.unwrap_or_default();
|
||
let msg = format!("Anthropic API 错误 {}: {}", status, text);
|
||
if is_status_retryable(status) {
|
||
warn!(%status, "Anthropic 同步调用可重试状态码");
|
||
return AttemptOutcome::Retryable(msg);
|
||
}
|
||
error!(%status, %text, "Anthropic API 调用失败(不可重试)");
|
||
return AttemptOutcome::Fatal(msg);
|
||
}
|
||
// body 解析:text 先取再 serde 分层判定:读取中断(text Err,reqwest 归
|
||
// is_decode 类,complete 重发幂等安全)可重试;JSON 结构错属 Fatal(响应已
|
||
// 完整送达,重试同败)。Response::json 把 serde 错同包成 is_decode,先 text
|
||
// 后 from_str 才能区分二者,避免确定性结构错被重试白烧预算。
|
||
let text = match resp.text().await {
|
||
Ok(t) => t,
|
||
Err(e) => {
|
||
if is_reqwest_error_retryable(&e) || e.is_decode() {
|
||
return AttemptOutcome::Retryable(format!("响应读取失败(可重试): {}", e));
|
||
}
|
||
return AttemptOutcome::Fatal(format!("响应读取失败(不可重试): {}", e));
|
||
}
|
||
};
|
||
let resp: AnthropicResponse = match serde_json::from_str(&text) {
|
||
Ok(r) => r,
|
||
Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)),
|
||
};
|
||
// content 块中拼接 text，收集 tool_use，累积 thinking
|
||
let mut text = String::new();
|
||
let mut reasoning = String::new();
|
||
let mut tool_calls: Vec<ToolCall> = Vec::new();
|
||
// CR-空 id：按 tool_use 块在数组中的顺序计数（仅 tool_use 递增），用于 fallback index。
|
||
// 用独立计数器而非 for enumerate，避免 text/unknown 块占用 index 致 fallback 编号跳号。
|
||
let mut tool_use_idx: usize = 0;
|
||
for block in resp.content {
|
||
match block.block_type.as_str() {
|
||
"text" => {
|
||
if let Some(t) = block.text {
|
||
text.push_str(&t);
|
||
}
|
||
}
|
||
"thinking" => {
|
||
// thinking 块正文回填 reasoning_content 通道（与 DeepSeek 同路入库展示）。
|
||
// 仅入库与展示，绝不回传 anthropic 历史（丢弃 signature 的伪造回传必 400）。
|
||
if let Some(t) = block.thinking {
|
||
reasoning.push_str(&t);
|
||
}
|
||
}
|
||
// redacted_thinking 为端点侧密文块，不可解析也不可伪造回传，合规丢弃不告警
|
||
"redacted_thinking" => {}
|
||
"tool_use" => {
|
||
// CR-空 id：原逻辑空 id 直接 continue 跳过整个块（丢工具调用）。
|
||
// 改为兜底：id 非空原样，空 → `gen_anthropic_{idx}` fallback（DRY 共用
|
||
// tool_call_id_or_fallback）。Anthropic 一般非空，此为兼容缺陷兜底。
|
||
// 不再 warn+continue（continue 会丢工具调用致 LLM 拿不到结果）。
|
||
let raw_id = block.id.unwrap_or_default();
|
||
let id = tool_call_id_or_fallback(&raw_id, tool_use_idx, "gen_anthropic");
|
||
if raw_id.is_empty() {
|
||
warn!(
|
||
fallback_id = %id,
|
||
name = ?block.name,
|
||
"Anthropic tool_use 块 id 为空，已生成 fallback id（原 continue 跳过会丢工具调用）"
|
||
);
|
||
}
|
||
tool_use_idx += 1;
|
||
let name = block.name.unwrap_or_default();
|
||
let args = block
|
||
.input
|
||
.map(|v| serde_json::to_string(&v).unwrap_or_default())
|
||
.unwrap_or_default();
|
||
tool_calls.push(ToolCall::new(id, name, args));
|
||
}
|
||
other => warn!(block_type = other, "Anthropic 响应含未知 content 块类型，已忽略"),
|
||
}
|
||
}
|
||
let usage = TokenUsage {
|
||
prompt_tokens: resp.usage.input_tokens,
|
||
completion_tokens: resp.usage.output_tokens,
|
||
total_tokens: resp.usage.input_tokens + resp.usage.output_tokens,
|
||
// anthropic cache 字段:creation(写入)= miss 全价语义,read(命中)= hit 低价
|
||
prompt_cache_hit_tokens: resp.usage.cache_read_input_tokens,
|
||
prompt_cache_miss_tokens: resp.usage.cache_creation_input_tokens,
|
||
reasoning_tokens: 0,
|
||
};
|
||
log_cache_usage(&usage);
|
||
AttemptOutcome::Ok(CompletionResponse {
|
||
text,
|
||
model: resp.model,
|
||
usage,
|
||
tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) },
|
||
// 无 thinking 块时保持 None（语义与 DeepSeek 路径一致，空串不落库）
|
||
reasoning_content: if reasoning.is_empty() { None } else { Some(reasoning) },
|
||
})
|
||
}
|
||
})
|
||
.await
|
||
}
|
||
|
||
async fn stream(&self, request: CompletionRequest) -> anyhow::Result<StreamResult> {
|
||
let mut req = request;
|
||
req.stream = true;
|
||
let body = self.convert_request(req);
|
||
|
||
// 协议预检——命中非法 bail 含 messages 摘要,把 GLM 模糊 1214 转明确诊断
|
||
if let Err(reason) = Self::precheck_messages(&body.messages) {
|
||
let summary = Self::summarize_messages(&body.messages);
|
||
warn!(%reason, %summary, "Anthropic messages 协议预检失败");
|
||
anyhow::bail!("messages 预检失败: {} | 摘要: {}", reason, summary);
|
||
}
|
||
|
||
debug!(model = %body.model, "Anthropic 流式调用");
|
||
|
||
// send 阶段需 timeout 防 hang(实测 GLM 偶发建连后长时间不返回)。
|
||
// 注意:不能用 reqwest 的 .timeout()——它是整个请求(含 body 读取)的总超时,
|
||
// 流式长生成任务会被误砍。改用 tokio::time::timeout
|
||
// 包裹 send().await,只管建连+首响应头,不管后续 body 读取(后续由 stream_llm idle timeout 兜底)。
|
||
// 60s 选型:正常 send(建连+收 200 headers)<5s,60s 足够宽容。
|
||
// send 阶段弱网退避: connect 类失败(隧道断网/基站切换/DNS 抖动)请求未发出, 重试零副作用。
|
||
// 网络感知分速(net_probe 基准面): 本机断网进长等待按自适应间隔探测恢复(隧道级分钟级波动),
|
||
// 单点故障指数短退避 1s→8s; 10 分钟窗口耗尽才放弃。timeout/decode/HTTP 状态错不重试。
|
||
// 与 openai_compat 同构。
|
||
let resp = {
|
||
let start = std::time::Instant::now();
|
||
let mut offline_logged = false;
|
||
let mut offline_polls = 0usize;
|
||
let mut attempt = 0usize;
|
||
loop {
|
||
attempt += 1;
|
||
let send_future = self
|
||
.auth_headers(self.client.post(self.messages_url()))
|
||
.json(&body)
|
||
.version(reqwest::Version::HTTP_11)
|
||
.send();
|
||
let outcome = tokio::time::timeout(Duration::from_secs(60), send_future).await;
|
||
match outcome {
|
||
Ok(Ok(r)) => break r,
|
||
Ok(Err(e)) => {
|
||
// 谓词结构化分类(对齐 openai 路径,共用 crate::send_error_class)。原 bail 串
|
||
// [timeout={} connect={}=...] 的字面词让一切错误(含 builder 错)侥幸可重试。
|
||
// 类词改由谓词决定置串首;谓词转储缩写为 t/c/b/r 防误触分类词表,source 链
|
||
// 保留真实传输词(reset/broken pipe 等)作串兜底。
|
||
let preds = crate::SendErrPreds::of(&e);
|
||
let retryable = preds.is_connect || (preds.is_request && !preds.is_builder);
|
||
if retryable && start.elapsed() < crate::net_probe::SEND_RETRY_WINDOW {
|
||
// 本机断网: 等基准面恢复(自适应间隔), 不消耗退避档位
|
||
if !crate::net_probe::baseline_alive().await {
|
||
offline_polls += 1;
|
||
let wait = crate::net_probe::offline_probe_wait(offline_polls);
|
||
if !offline_logged {
|
||
warn!(
|
||
url = %self.messages_url(),
|
||
window_secs = crate::net_probe::SEND_RETRY_WINDOW.as_secs(),
|
||
"本机网络不可达(疑似隧道/电梯弱网), 按自适应间隔探测恢复, 恢复后立即重试"
|
||
);
|
||
offline_logged = true;
|
||
}
|
||
tokio::time::sleep(wait).await;
|
||
continue;
|
||
}
|
||
offline_polls = 0;
|
||
offline_logged = false;
|
||
let delay = Duration::from_millis(
|
||
1000u64.saturating_mul(1 << (attempt - 1).min(3)).min(8000),
|
||
);
|
||
warn!(
|
||
url = %self.messages_url(),
|
||
attempt,
|
||
delay_ms = delay.as_millis() as u64,
|
||
error = %e,
|
||
"流式 send 连接失败, 退避重试"
|
||
);
|
||
tokio::time::sleep(delay).await;
|
||
continue;
|
||
}
|
||
let class = crate::send_error_class(preds);
|
||
tracing::error!(
|
||
is_timeout = e.is_timeout(),
|
||
is_connect = e.is_connect(),
|
||
is_body = e.is_body(),
|
||
is_request = e.is_request(),
|
||
class = class,
|
||
url = %self.messages_url(),
|
||
source = ?std::error::Error::source(&e),
|
||
"Anthropic 流式 send 失败"
|
||
);
|
||
anyhow::bail!(
|
||
"send 失败({}): {} [t={} c={} b={} r={} src={:?}]",
|
||
class,
|
||
e,
|
||
e.is_timeout(),
|
||
e.is_connect(),
|
||
e.is_body(),
|
||
e.is_request(),
|
||
std::error::Error::source(&e)
|
||
);
|
||
}
|
||
Err(_elapsed) => {
|
||
// send 阶段超时(60s 未返回 HTTP 响应头):GLM 端点可能不可达或极慢
|
||
tracing::error!(
|
||
url = %self.messages_url(),
|
||
"Anthropic 流式 send 超时(60s 未返回响应头)"
|
||
);
|
||
anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应，可能服务不可达或被防火墙拦截)");
|
||
}
|
||
}
|
||
}
|
||
};
|
||
|
||
if !resp.status().is_success() {
|
||
let status = resp.status();
|
||
let text = resp.text().await.unwrap_or_default();
|
||
error!(%status, %text, "Anthropic 流式 API 调用失败");
|
||
anyhow::bail!("Anthropic 流式 API 错误 {}: {}", status, text);
|
||
}
|
||
|
||
// 原生 SSE 解析器替代 eventsource-stream(同 openai_compat)。
|
||
let mut usage_accum: Option<TokenUsage> = None;
|
||
// MidStream error(如 GLM 1214 messages 非法)时附 messages 摘要定位哪条非法。
|
||
let messages_summary = Self::summarize_messages(&body.messages);
|
||
|
||
let sse = crate::sse_parser::SseStream::new(resp.bytes_stream());
|
||
let stream = sse.flat_map(move |result: Result<Vec<String>, String>| {
|
||
let mut chunks: Vec<anyhow::Result<crate::provider::StreamChunk>> = Vec::new();
|
||
match result {
|
||
Ok(events) => {
|
||
for data in events {
|
||
let mut chunk = apply_anthropic_event(&data, &mut usage_accum);
|
||
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));
|
||
}
|
||
}
|
||
Err(e) => {
|
||
let ctx = format!("Anthropic SSE 错误: {}", e);
|
||
error!("{}", ctx);
|
||
chunks.push(Err(anyhow::anyhow!("{}", ctx)));
|
||
}
|
||
}
|
||
futures::stream::iter(chunks)
|
||
});
|
||
|
||
Ok(Box::pin(stream))
|
||
}
|
||
|
||
fn name(&self) -> &str {
|
||
"anthropic-compat"
|
||
}
|
||
|
||
fn endpoint(&self) -> String {
|
||
self.messages_url()
|
||
}
|
||
}
|
||
|
||
// ============================================================
|
||
// 单测（不发真实 HTTP，喂构造的 SSE data 字符串序列）
|
||
// ============================================================
|
||
|
||
#[cfg(test)]
|
||
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!(
|
||
r#"{{"type":"message_start","message":{{"usage":{{"input_tokens":{},"output_tokens":0}}}}}}"#,
|
||
input_tokens
|
||
)
|
||
}
|
||
|
||
/// 辅助：构造 message_delta 事件（output_tokens 为累计值）
|
||
fn message_delta(output_tokens: u32) -> String {
|
||
format!(
|
||
r#"{{"type":"message_delta","delta":{{"stop_reason":"end_turn"}},"usage":{{"output_tokens":{}}}}}"#,
|
||
output_tokens
|
||
)
|
||
}
|
||
|
||
/// 辅助：构造文本增量 content_block_delta
|
||
fn text_delta(text: &str) -> String {
|
||
format!(
|
||
r#"{{"type":"content_block_delta","index":0,"delta":{{"type":"text_delta","text":"{}"}}}}"#,
|
||
text
|
||
)
|
||
}
|
||
|
||
/// 辅助：构造 message_stop 事件
|
||
fn message_stop() -> &'static str {
|
||
r#"{"type":"message_stop"}"#
|
||
}
|
||
|
||
/// 完整流：message_start 初始化 input + 多次 message_delta 累计覆盖 output + message_stop 带出
|
||
#[test]
|
||
fn anthropic_full_stream_accumulates_usage() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
|
||
// 1) message_start：input=42，output=0
|
||
// （6bafdcd 起 usage 同帧挂载：message_start 即带初始快照，中途断连也能拿到真实 usage）
|
||
let c = apply_anthropic_event(&message_start(42), &mut acc);
|
||
assert!(!c.finished);
|
||
let u0 = c.usage.expect("message_start 应同帧带出初始 usage 快照");
|
||
assert_eq!(u0.prompt_tokens, 42);
|
||
assert_eq!(u0.completion_tokens, 0);
|
||
assert_eq!(u0.total_tokens, 42);
|
||
let a = acc.as_ref().expect("message_start 应初始化累加器");
|
||
assert_eq!(a.prompt_tokens, 42);
|
||
assert_eq!(a.completion_tokens, 0);
|
||
assert_eq!(a.total_tokens, 42);
|
||
|
||
// 2) 文本增量不影响 usage
|
||
let c = apply_anthropic_event(&text_delta("Hello"), &mut acc);
|
||
assert_eq!(c.delta, "Hello");
|
||
assert!(!c.finished);
|
||
assert_eq!(acc.as_ref().unwrap().completion_tokens, 0, "文本增量不应改 output");
|
||
|
||
// 3) message_delta：output_tokens=10（累计值，覆盖）
|
||
let c = apply_anthropic_event(&message_delta(10), &mut acc);
|
||
assert!(!c.finished);
|
||
let a = acc.as_ref().unwrap();
|
||
assert_eq!(a.prompt_tokens, 42, "input 保持");
|
||
assert_eq!(a.completion_tokens, 10, "output 被覆盖为累计值");
|
||
assert_eq!(a.total_tokens, 52, "total 重算 = input+output");
|
||
|
||
// 4) 再次 message_delta：output_tokens=30（更大累计值，再覆盖）
|
||
let _ = apply_anthropic_event(&message_delta(30), &mut acc);
|
||
let a = acc.as_ref().unwrap();
|
||
assert_eq!(a.completion_tokens, 30, "后续累计值覆盖前值");
|
||
assert_eq!(a.total_tokens, 72);
|
||
|
||
// 5) message_stop：带出累积 usage，finished=true，累加器清空
|
||
let c = apply_anthropic_event(message_stop(), &mut acc);
|
||
assert!(c.finished);
|
||
let u = c.usage.expect("message_stop 应带出累积 usage");
|
||
assert_eq!(u.prompt_tokens, 42);
|
||
assert_eq!(u.completion_tokens, 30);
|
||
assert_eq!(u.total_tokens, 72);
|
||
assert!(acc.is_none(), "take() 后累加器应清空");
|
||
}
|
||
|
||
/// message_delta 在没有 message_start 时也能补全累加器（get_or_insert 兜底）
|
||
#[test]
|
||
fn anthropic_message_delta_without_start_uses_default_input() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let _ = apply_anthropic_event(&message_delta(15), &mut acc);
|
||
let a = acc.as_ref().unwrap();
|
||
assert_eq!(a.prompt_tokens, 0, "无 message_start 时 input 兜底为 0");
|
||
assert_eq!(a.completion_tokens, 15);
|
||
assert_eq!(a.total_tokens, 15);
|
||
}
|
||
|
||
/// stop_reason=max_tokens(message_delta 帧携带)必须置截断标志;end_turn 不置。
|
||
/// 终态 finished 在后续 message_stop 帧,消费方跨帧 OR 累积标志。
|
||
#[test]
|
||
fn anthropic_max_tokens_stop_reason_sets_truncation_flag() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let data = r#"{"type":"message_delta","delta":{"stop_reason":"max_tokens"},"usage":{"output_tokens":9}}"#;
|
||
let c = apply_anthropic_event(data, &mut acc);
|
||
assert!(!c.finished, "message_delta 非终态帧");
|
||
assert!(c.truncated_by_max_tokens, "max_tokens 必须置截断标志");
|
||
|
||
let mut acc2: Option<TokenUsage> = None;
|
||
let c2 = apply_anthropic_event(&message_delta(5), &mut acc2);
|
||
assert!(!c2.truncated_by_max_tokens, "end_turn 不置截断标志");
|
||
}
|
||
|
||
/// message_delta 的 output_tokens 必须是累计覆盖（非累加）：连续两个 delta 5 和 8，结果应是 8 不是 13
|
||
#[test]
|
||
fn anthropic_message_delta_output_is_cumulative_not_incremental() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
apply_anthropic_event(&message_start(100), &mut acc);
|
||
apply_anthropic_event(&message_delta(5), &mut acc);
|
||
apply_anthropic_event(&message_delta(8), &mut acc);
|
||
let c = apply_anthropic_event(message_stop(), &mut acc);
|
||
let u = c.usage.unwrap();
|
||
assert_eq!(u.completion_tokens, 8, "output_tokens 是累计值，覆盖而非累加");
|
||
assert_eq!(u.total_tokens, 108);
|
||
}
|
||
|
||
/// 无 usage 字段的流：message_stop 时 usage 为 None
|
||
#[test]
|
||
fn anthropic_message_stop_without_any_usage() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let _ = apply_anthropic_event(&text_delta("hi"), &mut acc);
|
||
assert!(acc.is_none(), "文本增量不初始化累加器");
|
||
let c = apply_anthropic_event(message_stop(), &mut acc);
|
||
assert!(c.finished);
|
||
assert!(c.usage.is_none(), "无 usage 时 message_stop usage 为 None");
|
||
}
|
||
|
||
/// content_block_start (tool_use) 带 id+name
|
||
#[test]
|
||
fn anthropic_content_block_start_tool_use() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let data = r#"{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"tool_1","name":"get_weather"}}"#;
|
||
let c = apply_anthropic_event(data, &mut acc);
|
||
assert!(acc.is_none(), "content_block_start 不动 usage");
|
||
let tcs = c.tool_calls.expect("应有 tool_calls");
|
||
assert_eq!(tcs.len(), 1);
|
||
assert_eq!(tcs[0].index, 1);
|
||
assert_eq!(tcs[0].id.as_deref(), Some("tool_1"));
|
||
assert_eq!(tcs[0].function_name.as_deref(), Some("get_weather"));
|
||
assert!(tcs[0].function_arguments.is_none());
|
||
assert!(!c.finished);
|
||
}
|
||
|
||
/// content_block_delta (input_json_delta) → 工具入参增量
|
||
#[test]
|
||
fn anthropic_content_block_delta_input_json() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let data = r#"{"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"q\":"}}"#;
|
||
let c = apply_anthropic_event(data, &mut acc);
|
||
let tcs = c.tool_calls.expect("应有 tool_calls 增量");
|
||
assert_eq!(tcs[0].index, 2);
|
||
assert_eq!(tcs[0].function_arguments.as_deref(), Some("{\"q\":"));
|
||
assert!(tcs[0].id.is_none());
|
||
assert_eq!(c.delta, "");
|
||
assert!(!c.finished);
|
||
}
|
||
|
||
/// thinking_delta → reasoning_content 增量（不混入正文 delta）；signature_delta 安静忽略。
|
||
/// 推理正文走 reasoning_content 通道与 DeepSeek 同路入库展示。
|
||
#[test]
|
||
fn anthropic_thinking_delta_accumulates_reasoning() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let think1 = r#"{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"先查"}}"#;
|
||
let c1 = apply_anthropic_event(think1, &mut acc);
|
||
assert_eq!(c1.reasoning_content.as_deref(), Some("先查"), "thinking_delta 应产出 reasoning_content");
|
||
assert_eq!(c1.delta, "", "思考增量不得混入正文 delta");
|
||
assert!(!c1.finished);
|
||
let think2 = r#"{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"用户意图"}}"#;
|
||
let c2 = apply_anthropic_event(think2, &mut acc);
|
||
assert_eq!(c2.reasoning_content.as_deref(), Some("用户意图"));
|
||
// signature_delta：thinking 块签名字段，本侧无处消费，安静跳过（空 chunk 不告警）
|
||
let sig = r#"{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-abc"}}"#;
|
||
let c3 = apply_anthropic_event(sig, &mut acc);
|
||
assert!(c3.reasoning_content.is_none());
|
||
assert!(c3.delta.is_empty());
|
||
assert!(!c3.finished);
|
||
// 思考块之后正文照常
|
||
let c4 = apply_anthropic_event(&text_delta("答案"), &mut acc);
|
||
assert_eq!(c4.delta, "答案");
|
||
assert!(c4.reasoning_content.is_none());
|
||
}
|
||
|
||
/// error 事件 → error=Some + finished=false避免残缺响应被当正常完成入库，
|
||
/// 由 stream_llm 识别 error 非空发 AiError + 丢弃残缺，与 OpenAI 路径 Err 一致）
|
||
#[test]
|
||
fn anthropic_error_event_yields_error_not_finished() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
apply_anthropic_event(&message_start(10), &mut acc);
|
||
let c = apply_anthropic_event(r#"{"type":"error","error":{"message":"overloaded"}}"#, &mut acc);
|
||
assert!(!c.finished, "error 不走 finished 完成路径，否则残缺响应会被当正常完成");
|
||
assert_eq!(c.error.as_deref(), Some("overloaded"), "error 事件应携带错误消息");
|
||
assert!(c.usage.is_none(), "error 不带出 usage");
|
||
assert!(c.delta.is_empty(), "error 不带文本增量");
|
||
assert!(acc.is_some(), "error 不应清空已累积的 usage（与原实现一致）");
|
||
}
|
||
|
||
/// error 事件无 error.message 字段时兜底为 "stream error"
|
||
#[test]
|
||
fn anthropic_error_event_missing_message_falls_back() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let c = apply_anthropic_event(r#"{"type":"error"}"#, &mut acc);
|
||
assert_eq!(c.error.as_deref(), Some("stream error"));
|
||
assert!(!c.finished);
|
||
}
|
||
|
||
/// ping / content_block_stop 等事件 → 空且不 finished
|
||
#[test]
|
||
fn anthropic_ping_and_block_stop_yield_empty_chunk() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let c = apply_anthropic_event(r#"{"type":"ping"}"#, &mut acc);
|
||
assert!(!c.finished);
|
||
assert_eq!(c.delta, "");
|
||
assert!(acc.is_none());
|
||
let c = apply_anthropic_event(r#"{"type":"content_block_stop","index":0}"#, &mut acc);
|
||
assert!(!c.finished);
|
||
assert_eq!(c.delta, "");
|
||
}
|
||
|
||
/// 非法 JSON → 空 chunk，不 panic
|
||
#[test]
|
||
fn anthropic_malformed_json_yields_empty_chunk() {
|
||
let mut acc: Option<TokenUsage> = None;
|
||
let c = apply_anthropic_event("not json", &mut acc);
|
||
assert!(!c.finished);
|
||
assert_eq!(c.delta, "");
|
||
assert!(acc.is_none());
|
||
}
|
||
|
||
// ---------- 多模态 convert_request ----------
|
||
|
||
/// 含图 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(),
|
||
messages: vec![ChatMessage::user_parts(
|
||
"看图",
|
||
vec![crate::provider::ContentPart::image_base64("image/png", "iVBOR")],
|
||
)],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
// user 消息 content 应为数组形态
|
||
let user_msg = body
|
||
.messages
|
||
.iter()
|
||
.find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user"))
|
||
.expect("应有 user 消息");
|
||
let content = user_msg.get("content").and_then(|c| c.as_array()).expect("user content 应为数组");
|
||
// 顺序：content "看图" → text 块，image 片 → image 块
|
||
assert_eq!(content.len(), 2);
|
||
assert_eq!(content[0]["type"], "text");
|
||
assert_eq!(content[0]["text"], "看图");
|
||
assert_eq!(content[1]["type"], "image");
|
||
assert_eq!(content[1]["source"]["type"], "base64");
|
||
assert_eq!(content[1]["source"]["media_type"], "image/png");
|
||
assert_eq!(content[1]["source"]["data"], "iVBOR");
|
||
}
|
||
|
||
/// 本地 file:// 引用(#958)在转换层展开为 base64 source;缺失降级 text 块不炸请求
|
||
#[test]
|
||
fn anthropic_convert_expands_local_file_ref() {
|
||
use base64::Engine as _;
|
||
let _convert = convert_lock();
|
||
let dir = std::env::temp_dir().join("df-ai-anthropic-ref-test");
|
||
std::fs::create_dir_all(&dir).unwrap();
|
||
let path = dir.join("ref.png");
|
||
std::fs::write(&path, b"\x89PNG ok").unwrap();
|
||
|
||
let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet");
|
||
let req = CompletionRequest {
|
||
model: "claude-3-5-sonnet".into(),
|
||
messages: vec![ChatMessage::user_parts(
|
||
"看图",
|
||
vec![crate::provider::ContentPart::Image {
|
||
url: Some(format!("file://{}", path.display())),
|
||
base64: None,
|
||
media_type: Some("image/png".into()),
|
||
alt: None,
|
||
}],
|
||
)],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
let user_msg = body
|
||
.messages
|
||
.iter()
|
||
.find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user"))
|
||
.expect("应有 user 消息");
|
||
let content = user_msg.get("content").and_then(|c| c.as_array()).expect("user content 应为数组");
|
||
assert_eq!(content[1]["source"]["data"], base64::engine::general_purpose::STANDARD.encode(b"\x89PNG ok"));
|
||
std::fs::remove_file(&path).ok();
|
||
}
|
||
|
||
/// 纯文本 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(),
|
||
messages: vec![ChatMessage::user("hello")],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
let user_msg = body
|
||
.messages
|
||
.iter()
|
||
.find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user"))
|
||
.expect("应有 user 消息");
|
||
// 纯文本 → 字符串简写（非数组）
|
||
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, None);
|
||
// 前一条 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, None);
|
||
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, None);
|
||
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, 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) 消息
|
||
fn msg_assistant_with_tool_use(text: &str, tool_id: &str, tool_name: &str) -> ChatMessage {
|
||
ChatMessage::assistant_with_tools(
|
||
text,
|
||
vec![ToolCall::new(tool_id, tool_name, "{}")],
|
||
)
|
||
}
|
||
|
||
/// 精确复现线上场景——多轮 [asst(tool_use), tool_result] 链,首条 assistant。
|
||
/// 补一条 user 占位后:首条 user、tool_use/tool_result 配对完整保留、precheck 通过。
|
||
/// (原"砍"策略会把每对三元组砍掉,多轮砍到空,丢失全部工具调用历史——"补"策略零丢失。)
|
||
#[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(),
|
||
messages: vec![
|
||
msg_assistant_with_tool_use("我来帮你", "call_f892", "read_file"),
|
||
ChatMessage::tool_result("call_f892", "file content"),
|
||
msg_assistant_with_tool_use("继续", "call_003a", "write_file"),
|
||
ChatMessage::tool_result("call_003a", "done"),
|
||
],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
|
||
// 首条必须是 user(补的占位)
|
||
let first_role = body.messages[0].get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
assert_eq!(first_role, "user", "首条应为 user(补占位)");
|
||
|
||
// 上下文零丢失:占位 user + asst(tu) + user(tr) + asst(tu) + user(tr) = 5 条
|
||
assert_eq!(
|
||
body.messages.len(), 5,
|
||
"应保留全部上下文(占位 + 原 4 条),实际 {} 条", body.messages.len()
|
||
);
|
||
|
||
assert!(
|
||
AnthropicCompatProvider::precheck_messages(&body.messages).is_ok(),
|
||
"precheck 应通过,实际 messages: {}",
|
||
AnthropicCompatProvider::summarize_messages(&body.messages)
|
||
);
|
||
}
|
||
|
||
/// 首条 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(),
|
||
messages: vec![
|
||
ChatMessage::assistant("你好"),
|
||
ChatMessage::user("请帮我"),
|
||
],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
assert_eq!(body.messages.len(), 3, "占位 + 原 2 条");
|
||
assert_eq!(
|
||
body.messages[0].get("role").and_then(|r| r.as_str()),
|
||
Some("user"),
|
||
);
|
||
assert!(AnthropicCompatProvider::precheck_messages(&body.messages).is_ok());
|
||
}
|
||
|
||
/// 正常序列(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(),
|
||
messages: vec![
|
||
ChatMessage::user("hello"),
|
||
ChatMessage::assistant("hi there"),
|
||
],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
assert_eq!(body.messages.len(), 2, "正常序列不应补占位");
|
||
assert_eq!(
|
||
body.messages[0].get("role").and_then(|r| r.as_str()),
|
||
Some("user"),
|
||
);
|
||
assert!(AnthropicCompatProvider::precheck_messages(&body.messages).is_ok());
|
||
}
|
||
|
||
/// 线上 3 轮工具调用场景(6 条 [asst(tu),tool_result]×3,首条 assistant)。
|
||
/// 补一个 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(),
|
||
messages: vec![
|
||
msg_assistant_with_tool_use("a1", "call_1", "read_file"),
|
||
ChatMessage::tool_result("call_1", "r1"),
|
||
msg_assistant_with_tool_use("a2", "call_2", "write_file"),
|
||
ChatMessage::tool_result("call_2", "r2"),
|
||
msg_assistant_with_tool_use("a3", "call_3", "list_directory"),
|
||
ChatMessage::tool_result("call_3", "r3"),
|
||
],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
let first_role = body.messages[0].get("role").and_then(|r| r.as_str()).unwrap_or("");
|
||
assert_eq!(first_role, "user", "首条应为 user(补占位)");
|
||
// 占位 + 3×[asst(tu),user(tr)] = 7 条,全部保留
|
||
assert_eq!(
|
||
body.messages.len(), 7,
|
||
"3 轮链应全保留(占位 + 原 6 条),实际 {} 条", body.messages.len()
|
||
);
|
||
assert!(
|
||
AnthropicCompatProvider::precheck_messages(&body.messages).is_ok(),
|
||
"3 轮链补占位后 precheck 应通过,实际: {}",
|
||
AnthropicCompatProvider::summarize_messages(&body.messages)
|
||
);
|
||
}
|
||
|
||
/// 空 messages(异常会话经 sanitize 清空)→ convert 补 1 条 user 占位,
|
||
/// 避免发空 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(),
|
||
messages: vec![],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
let body = provider.convert_request(req);
|
||
assert_eq!(body.messages.len(), 1, "空 messages 应补 1 条 user 占位");
|
||
assert_eq!(
|
||
body.messages[0].get("role").and_then(|r| r.as_str()),
|
||
Some("user"),
|
||
);
|
||
assert!(
|
||
AnthropicCompatProvider::precheck_messages(&body.messages).is_ok(),
|
||
"补占位后 precheck 应通过"
|
||
);
|
||
}
|
||
|
||
// ---------- prompt caching:开关驱动 system 形态 ----------
|
||
|
||
/// 开关控制 system 形态:开 → blocks 数组含 cache_control:ephemeral;关 → 纯字符串。
|
||
/// (两态共用进程级全局开关,拆两个用例并行跑会互相污染——单测内顺序切换断言,末尾复位。)
|
||
#[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");
|
||
let mk = || CompletionRequest {
|
||
model: "claude-3-5-sonnet".into(),
|
||
messages: vec![
|
||
ChatMessage::system("a"),
|
||
ChatMessage::system("b"),
|
||
ChatMessage::user("hi"),
|
||
],
|
||
temperature: None,
|
||
max_tokens: None,
|
||
stream: false,
|
||
tools: None,
|
||
tool_choice: None,
|
||
reasoning_content: None,
|
||
};
|
||
|
||
// 开:数组形态 + cache_control(Anthropic prompt caching)
|
||
set_anthropic_cache_enabled(true);
|
||
let body = provider.convert_request(mk());
|
||
match body.system.expect("开关开时应有 system") {
|
||
SystemBlock::Cached(blocks) => {
|
||
assert_eq!(blocks.len(), 1, "开关开应包成单 blocks 数组");
|
||
assert_eq!(blocks[0]["type"], "text");
|
||
assert_eq!(blocks[0]["text"], "a\n\nb", "多条 system 应以 \\n\\n join 进单个 text 块");
|
||
assert_eq!(blocks[0]["cache_control"]["type"], "ephemeral");
|
||
}
|
||
SystemBlock::Plain(s) => panic!("开关开应为数组形态,实际 Plain: {}", s),
|
||
}
|
||
|
||
// 关:纯字符串(兼容非官方网关,零行为变更)
|
||
set_anthropic_cache_enabled(false);
|
||
let body = provider.convert_request(mk());
|
||
match body.system {
|
||
Some(SystemBlock::Plain(s)) => assert_eq!(s, "a\n\nb", "开关关多条 system 应以 \\n\\n join 纯字符串"),
|
||
other => panic!("开关关应为纯字符串,实际: {:?}", other),
|
||
}
|
||
}
|
||
}
|