重构: audit第三批cache + anthropic_compat拆分(strategy并行)
- audit 第三批: 新建 audit/cache.rs(151行 find_cached_high_risk_result/canonical_args_key/sort_object_keys/PENDING_APPROVAL_PLACEHOLDER), audit/mod.rs 586→451(-135); pub(crate) re-export, F-09 per_conv保留 - anthropic_compat 拆分: 新建 df-ai/anthropic_helpers.rs(223行 apply_anthropic_event + 5 struct + 2常量), anthropic_compat.rs 1036→842; Provider impl保留(impl块约束), 字段pub(crate) 主代兜底: cargo check --workspace 0 + test df-ai 119 + test devflow 98 strategy: 核心库自底向上, 单批1-2文件原子, impl块约束(Provider方法保留只抽纯函数/类型) git add指定(audit/* + df-ai/*, 不含其他会话)
This commit is contained in:
@@ -5,18 +5,20 @@
|
||||
//!
|
||||
//! 与 OpenAI 协议的关键差异由本模块内部完成转换，对外仍暴露统一的 LlmProvider trait，
|
||||
//! 上层 (Agentic Loop / AiNode) 无需感知协议。
|
||||
//!
|
||||
//! 协议数据结构（请求/响应 struct）与 SSE 事件解析纯函数已抽至 `anthropic_helpers`，
|
||||
//! 本模块仅保留 Provider struct + impl（HTTP 调用），Rust impl 块不可跨文件故作此切分。
|
||||
|
||||
use async_trait::async_trait;
|
||||
use eventsource_stream::Eventsource;
|
||||
use futures::StreamExt;
|
||||
use reqwest::Client;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::provider::{
|
||||
CompletionRequest, CompletionResponse, LlmProvider, MessageRole,
|
||||
StreamChunk, StreamResult, TokenUsage, ToolCall, ToolCallDelta,
|
||||
StreamResult, TokenUsage, ToolCall,
|
||||
};
|
||||
// ChatMessage 仅单测构造 CompletionRequest 用，避免非 test 构建的 unused import 警告。
|
||||
#[cfg(test)]
|
||||
@@ -25,204 +27,12 @@ use crate::retry::{
|
||||
retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable,
|
||||
};
|
||||
|
||||
// ============================================================
|
||||
// Anthropic API 请求/响应结构体
|
||||
// ============================================================
|
||||
|
||||
/// Anthropic 请求体
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
struct AnthropicRequest {
|
||||
model: String,
|
||||
messages: Vec<serde_json::Value>,
|
||||
max_tokens: u32,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
system: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
temperature: Option<f32>,
|
||||
stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
tools: Option<Vec<AnthropicToolDef>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
tool_choice: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters）
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
struct AnthropicToolDef {
|
||||
name: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
description: Option<String>,
|
||||
input_schema: serde_json::Value,
|
||||
}
|
||||
|
||||
/// Anthropic 同步响应
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct AnthropicResponse {
|
||||
// SW-260618-25: Anthropic 响应反序列化字段,保留以对齐响应结构(响应 id),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
id: String,
|
||||
model: String,
|
||||
content: Vec<AnthropicContentBlock>,
|
||||
// SW-260618-25: Anthropic 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
stop_reason: Option<String>,
|
||||
usage: AnthropicUsage,
|
||||
}
|
||||
|
||||
/// 响应 content 块（text 或 tool_use）
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct AnthropicContentBlock {
|
||||
#[serde(rename = "type")]
|
||||
block_type: String,
|
||||
#[serde(default)]
|
||||
text: Option<String>,
|
||||
/// tool_use 块字段
|
||||
id: Option<String>,
|
||||
name: Option<String>,
|
||||
input: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct AnthropicUsage {
|
||||
input_tokens: u32,
|
||||
output_tokens: u32,
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// SSE 解析纯函数（与 HTTP 解耦，便于单测）
|
||||
// ============================================================
|
||||
|
||||
/// 将一条 Anthropic Messages SSE 事件 data 解析为 StreamChunk，并按需更新 usage 累加器。
|
||||
///
|
||||
/// 按 `type` 字段分发：
|
||||
/// - `message_start` → 用 `message.usage.input_tokens` 初始化累加器（output 置 0）。
|
||||
/// - `message_delta` → **output_tokens 是累计值（非增量）**，直接覆盖 `completion_tokens` 并重算 `total`。
|
||||
/// - `content_block_delta` (text_delta/input_json_delta) → 文本/工具入参增量。
|
||||
/// - `content_block_start` (tool_use) → 工具块开始，带 id+name。
|
||||
/// - `message_stop` → 返回 `finished=true` 终态 chunk，`usage` 取自累加器（`take()`）。
|
||||
/// - `error` → 返回 `finished=true` 终态空 chunk。
|
||||
/// - 其它（content_block_stop / ping 等）→ 空 chunk。
|
||||
///
|
||||
/// 等价性：content_block / message_stop / error 等事件分支与原 stream() 闭包逐字一致；
|
||||
/// usage 透传（message_stop 终态 take() 带出、message_delta 的 output_tokens 按累计值覆盖）
|
||||
/// 为本次新增能力，对应 StreamChunk 新增的 usage 字段。
|
||||
pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUsage>) -> StreamChunk {
|
||||
// 解析 data 中的 JSON，按 type 字段决定如何转 StreamChunk
|
||||
let v: serde_json::Value = match serde_json::from_str(data) {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
};
|
||||
let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
|
||||
match ty {
|
||||
// 消息开始：取 input_tokens 初始化累积器（output 此时未知，置 0）
|
||||
"message_start" => {
|
||||
if let Some(inp) = v
|
||||
.get("message")
|
||||
.and_then(|m| m.get("usage"))
|
||||
.and_then(|u| u.get("input_tokens"))
|
||||
.and_then(|t| t.as_u64())
|
||||
{
|
||||
*usage_accum = Some(TokenUsage {
|
||||
prompt_tokens: inp as u32,
|
||||
completion_tokens: 0,
|
||||
total_tokens: inp as u32,
|
||||
});
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 消息增量：output_tokens 是累计值（非增量），直接覆盖 completion + 重算 total
|
||||
"message_delta" => {
|
||||
if let Some(out) = v.get("usage").and_then(|u| u.get("output_tokens")).and_then(|t| t.as_u64()) {
|
||||
let acc = usage_accum
|
||||
.get_or_insert(TokenUsage { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 });
|
||||
acc.completion_tokens = out as u32;
|
||||
acc.total_tokens = acc.prompt_tokens + acc.completion_tokens;
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 文本增量
|
||||
"content_block_delta" => {
|
||||
if let Some(delta) = v.get("delta") {
|
||||
if delta.get("type").and_then(|t| t.as_str()) == Some("text_delta") {
|
||||
let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("").to_string();
|
||||
return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None };
|
||||
}
|
||||
// 工具入参增量
|
||||
if delta.get("type").and_then(|t| t.as_str()) == Some("input_json_delta") {
|
||||
let partial = delta.get("partial_json").and_then(|t| t.as_str()).unwrap_or("").to_string();
|
||||
let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32;
|
||||
return StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: false,
|
||||
tool_calls: Some(vec![ToolCallDelta {
|
||||
index: idx,
|
||||
id: None,
|
||||
function_name: None,
|
||||
function_arguments: Some(partial),
|
||||
}]),
|
||||
usage: None,
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
};
|
||||
}
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 工具块开始：带 id + name
|
||||
"content_block_start" => {
|
||||
if let Some(cb) = v.get("content_block") {
|
||||
if cb.get("type").and_then(|t| t.as_str()) == Some("tool_use") {
|
||||
let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32;
|
||||
let name = cb.get("name").and_then(|t| t.as_str()).map(|s| s.to_string());
|
||||
// id 缺失时用占位 id 兜底：流式后续 input_json_delta 按 index 累加，
|
||||
// 中途无法整体跳过；占位 id 保证回传的 tool_use_id 非空，避免 GLM 500。
|
||||
let id = match cb.get("id").and_then(|t| t.as_str()).map(|s| s.to_string()) {
|
||||
Some(id) if !id.is_empty() => Some(id),
|
||||
_ => {
|
||||
let placeholder = format!("tool_missing_{}", idx);
|
||||
warn!(%placeholder, name = ?name, "Anthropic 流式 tool_use 块缺少 id，已填占位 id（原样回传会触发 GLM 500）");
|
||||
Some(placeholder)
|
||||
}
|
||||
};
|
||||
return StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: false,
|
||||
tool_calls: Some(vec![ToolCallDelta {
|
||||
index: idx,
|
||||
id,
|
||||
function_name: name,
|
||||
function_arguments: None,
|
||||
}]),
|
||||
usage: None,
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
};
|
||||
}
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 消息结束：带出累积 usage
|
||||
"message_stop" => StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: true,
|
||||
tool_calls: None,
|
||||
usage: usage_accum.take(),
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
},
|
||||
// 错误事件：流中途出错。不走 finished 完成路径（避免残缺响应被当正常完成入库），
|
||||
// 改由 stream_llm 识别 error 非空 → 发 AiError + 丢弃残缺（与 OpenAI 路径 Err 一致）。
|
||||
"error" => {
|
||||
let msg = v.get("error").and_then(|e| e.get("message")).and_then(|m| m.as_str()).unwrap_or("stream error").to_string();
|
||||
error!(%msg, "Anthropic 流式错误事件");
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None }
|
||||
}
|
||||
// content_block_stop / ping 等不产出 chunk
|
||||
_ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None },
|
||||
}
|
||||
}
|
||||
// 协议数据结构 + SSE 纯解析（apply_anthropic_event / AnthropicRequest 等）抽至 anthropic_helpers，
|
||||
// 此处 use 引入以保持本模块内引用路径不变（零行为变更搬迁）。
|
||||
use crate::anthropic_helpers::{
|
||||
apply_anthropic_event, AnthropicRequest, AnthropicResponse, AnthropicToolDef,
|
||||
ANTHROPIC_VERSION, DEFAULT_MAX_TOKENS,
|
||||
};
|
||||
|
||||
// ============================================================
|
||||
// Provider 实现
|
||||
@@ -236,11 +46,6 @@ pub struct AnthropicCompatProvider {
|
||||
default_model: String,
|
||||
}
|
||||
|
||||
/// Anthropic 流式协议版本头
|
||||
const ANTHROPIC_VERSION: &str = "2023-06-01";
|
||||
/// Anthropic max_tokens 必填，缺省时的兜底值
|
||||
const DEFAULT_MAX_TOKENS: u32 = 4096;
|
||||
|
||||
impl AnthropicCompatProvider {
|
||||
/// 创建 Provider
|
||||
///
|
||||
@@ -788,6 +593,7 @@ impl LlmProvider for AnthropicCompatProvider {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
// apply_anthropic_event / TokenUsage 经 super::*（含 anthropic_helpers::apply_anthropic_event 的 use）可见。
|
||||
|
||||
/// 辅助：构造 message_start 事件
|
||||
fn message_start(input_tokens: u32) -> String {
|
||||
|
||||
223
crates/df-ai/src/anthropic_helpers.rs
Normal file
223
crates/df-ai/src/anthropic_helpers.rs
Normal file
@@ -0,0 +1,223 @@
|
||||
//! Anthropic 协议适配 — 协议数据结构与 SSE 事件解析（纯函数）。
|
||||
//!
|
||||
//! 本模块从 `anthropic_compat.rs` 抽离，承载与 HTTP 无关的纯协议逻辑：
|
||||
//! - 请求/响应结构体（`AnthropicRequest` 等）
|
||||
//! - 协议常量（`ANTHROPIC_VERSION` / `DEFAULT_MAX_TOKENS`）
|
||||
//! - SSE 事件 → `StreamChunk` 转换纯函数（`apply_anthropic_event`）
|
||||
//!
|
||||
//! Provider struct + impl（含 HTTP 调用）仍留在 `anthropic_compat.rs`，
|
||||
//! Rust impl 块不可跨文件，故仅搬迁 impl 块外部的类型/常量/纯函数。
|
||||
//! 零行为变更（纯搬迁）。
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{error, warn};
|
||||
|
||||
use crate::provider::{StreamChunk, TokenUsage, ToolCallDelta};
|
||||
|
||||
// ============================================================
|
||||
// Anthropic API 请求/响应结构体
|
||||
// ============================================================
|
||||
|
||||
/// Anthropic 请求体
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub(crate) struct AnthropicRequest {
|
||||
pub model: String,
|
||||
pub messages: Vec<serde_json::Value>,
|
||||
pub max_tokens: u32,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub system: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub temperature: Option<f32>,
|
||||
pub stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub tools: Option<Vec<AnthropicToolDef>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub tool_choice: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters）
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub(crate) struct AnthropicToolDef {
|
||||
pub name: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub description: Option<String>,
|
||||
pub input_schema: serde_json::Value,
|
||||
}
|
||||
|
||||
/// Anthropic 同步响应
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct AnthropicResponse {
|
||||
// SW-260618-25: Anthropic 响应反序列化字段,保留以对齐响应结构(响应 id),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
id: String,
|
||||
pub model: String,
|
||||
pub content: Vec<AnthropicContentBlock>,
|
||||
// SW-260618-25: Anthropic 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
stop_reason: Option<String>,
|
||||
pub usage: AnthropicUsage,
|
||||
}
|
||||
|
||||
/// 响应 content 块（text 或 tool_use）
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct AnthropicContentBlock {
|
||||
#[serde(rename = "type")]
|
||||
pub block_type: String,
|
||||
#[serde(default)]
|
||||
pub text: Option<String>,
|
||||
/// tool_use 块字段
|
||||
pub id: Option<String>,
|
||||
pub name: Option<String>,
|
||||
pub input: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct AnthropicUsage {
|
||||
pub input_tokens: u32,
|
||||
pub output_tokens: u32,
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 协议常量
|
||||
// ============================================================
|
||||
|
||||
/// Anthropic 流式协议版本头
|
||||
pub(crate) const ANTHROPIC_VERSION: &str = "2023-06-01";
|
||||
/// Anthropic max_tokens 必填，缺省时的兜底值
|
||||
pub(crate) const DEFAULT_MAX_TOKENS: u32 = 4096;
|
||||
|
||||
// ============================================================
|
||||
// SSE 解析纯函数（与 HTTP 解耦，便于单测）
|
||||
// ============================================================
|
||||
|
||||
/// 将一条 Anthropic Messages SSE 事件 data 解析为 StreamChunk，并按需更新 usage 累加器。
|
||||
///
|
||||
/// 按 `type` 字段分发：
|
||||
/// - `message_start` → 用 `message.usage.input_tokens` 初始化累加器（output 置 0）。
|
||||
/// - `message_delta` → **output_tokens 是累计值（非增量）**，直接覆盖 `completion_tokens` 并重算 `total`。
|
||||
/// - `content_block_delta` (text_delta/input_json_delta) → 文本/工具入参增量。
|
||||
/// - `content_block_start` (tool_use) → 工具块开始，带 id+name。
|
||||
/// - `message_stop` → 返回 `finished=true` 终态 chunk，`usage` 取自累加器（`take()`）。
|
||||
/// - `error` → 返回 `finished=true` 终态空 chunk。
|
||||
/// - 其它（content_block_stop / ping 等）→ 空 chunk。
|
||||
///
|
||||
/// 等价性：content_block / message_stop / error 等事件分支与原 stream() 闭包逐字一致；
|
||||
/// usage 透传（message_stop 终态 take() 带出、message_delta 的 output_tokens 按累计值覆盖）
|
||||
/// 为本次新增能力，对应 StreamChunk 新增的 usage 字段。
|
||||
pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUsage>) -> StreamChunk {
|
||||
// 解析 data 中的 JSON，按 type 字段决定如何转 StreamChunk
|
||||
let v: serde_json::Value = match serde_json::from_str(data) {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
return StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
};
|
||||
let ty = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
|
||||
match ty {
|
||||
// 消息开始：取 input_tokens 初始化累积器（output 此时未知，置 0）
|
||||
"message_start" => {
|
||||
if let Some(inp) = v
|
||||
.get("message")
|
||||
.and_then(|m| m.get("usage"))
|
||||
.and_then(|u| u.get("input_tokens"))
|
||||
.and_then(|t| t.as_u64())
|
||||
{
|
||||
*usage_accum = Some(TokenUsage {
|
||||
prompt_tokens: inp as u32,
|
||||
completion_tokens: 0,
|
||||
total_tokens: inp as u32,
|
||||
});
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 消息增量：output_tokens 是累计值（非增量），直接覆盖 completion + 重算 total
|
||||
"message_delta" => {
|
||||
if let Some(out) = v.get("usage").and_then(|u| u.get("output_tokens")).and_then(|t| t.as_u64()) {
|
||||
let acc = usage_accum
|
||||
.get_or_insert(TokenUsage { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 });
|
||||
acc.completion_tokens = out as u32;
|
||||
acc.total_tokens = acc.prompt_tokens + acc.completion_tokens;
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 文本增量
|
||||
"content_block_delta" => {
|
||||
if let Some(delta) = v.get("delta") {
|
||||
if delta.get("type").and_then(|t| t.as_str()) == Some("text_delta") {
|
||||
let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("").to_string();
|
||||
return StreamChunk { delta: text, finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None };
|
||||
}
|
||||
// 工具入参增量
|
||||
if delta.get("type").and_then(|t| t.as_str()) == Some("input_json_delta") {
|
||||
let partial = delta.get("partial_json").and_then(|t| t.as_str()).unwrap_or("").to_string();
|
||||
let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32;
|
||||
return StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: false,
|
||||
tool_calls: Some(vec![ToolCallDelta {
|
||||
index: idx,
|
||||
id: None,
|
||||
function_name: None,
|
||||
function_arguments: Some(partial),
|
||||
}]),
|
||||
usage: None,
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
};
|
||||
}
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 工具块开始：带 id + name
|
||||
"content_block_start" => {
|
||||
if let Some(cb) = v.get("content_block") {
|
||||
if cb.get("type").and_then(|t| t.as_str()) == Some("tool_use") {
|
||||
let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as u32;
|
||||
let name = cb.get("name").and_then(|t| t.as_str()).map(|s| s.to_string());
|
||||
// id 缺失时用占位 id 兜底：流式后续 input_json_delta 按 index 累加，
|
||||
// 中途无法整体跳过；占位 id 保证回传的 tool_use_id 非空，避免 GLM 500。
|
||||
let id = match cb.get("id").and_then(|t| t.as_str()).map(|s| s.to_string()) {
|
||||
Some(id) if !id.is_empty() => Some(id),
|
||||
_ => {
|
||||
let placeholder = format!("tool_missing_{}", idx);
|
||||
warn!(%placeholder, name = ?name, "Anthropic 流式 tool_use 块缺少 id，已填占位 id（原样回传会触发 GLM 500）");
|
||||
Some(placeholder)
|
||||
}
|
||||
};
|
||||
return StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: false,
|
||||
tool_calls: Some(vec![ToolCallDelta {
|
||||
index: idx,
|
||||
id,
|
||||
function_name: name,
|
||||
function_arguments: None,
|
||||
}]),
|
||||
usage: None,
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
};
|
||||
}
|
||||
}
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
|
||||
}
|
||||
// 消息结束：带出累积 usage
|
||||
"message_stop" => StreamChunk {
|
||||
delta: String::new(),
|
||||
finished: true,
|
||||
tool_calls: None,
|
||||
usage: usage_accum.take(),
|
||||
error: None,
|
||||
reasoning_content: None,
|
||||
},
|
||||
// 错误事件：流中途出错。不走 finished 完成路径（避免残缺响应被当正常完成入库），
|
||||
// 改由 stream_llm 识别 error 非空 → 发 AiError + 丢弃残缺（与 OpenAI 路径 Err 一致）。
|
||||
"error" => {
|
||||
let msg = v.get("error").and_then(|e| e.get("message")).and_then(|m| m.as_str()).unwrap_or("stream error").to_string();
|
||||
error!(%msg, "Anthropic 流式错误事件");
|
||||
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: Some(msg), reasoning_content: None }
|
||||
}
|
||||
// content_block_stop / ping 等不产出 chunk
|
||||
_ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None },
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
pub mod ai_tools;
|
||||
pub mod anthropic_compat;
|
||||
pub mod anthropic_helpers;
|
||||
pub mod context;
|
||||
pub mod context_helpers;
|
||||
pub mod coordinator;
|
||||
|
||||
151
src-tauri/src/commands/ai/audit/cache.rs
Normal file
151
src-tauri/src/commands/ai/audit/cache.rs
Normal file
@@ -0,0 +1,151 @@
|
||||
//! F-260616-05 高危工具去重缓存(第三批 helper 抽离,行为零变更)。
|
||||
//!
|
||||
//! 从 audit/mod.rs 搬迁:`find_cached_high_risk_result` + `canonical_args_key`
|
||||
//! + `sort_object_keys` + `PENDING_APPROVAL_PLACEHOLDER`(去重与 process_tool_calls
|
||||
//! 共用占位常量)。三函数内部闭环(find_cached 调 canonical→sort),无外部 helper 依赖。
|
||||
|
||||
use df_ai::provider::ChatMessage;
|
||||
use df_storage::crud::AiToolExecutionRepo;
|
||||
|
||||
use super::super::AiSession;
|
||||
|
||||
/// Med/High 工具挂起审批时回填的占位 tool_result 内容。
|
||||
///
|
||||
/// 单一常量避免「写入处」(process_tool_calls) 与「去重排查处」
|
||||
/// (find_cached_high_risk_result) 各持一份字面量致耦合——若两者漂移,
|
||||
/// 去重会把 pending 占位误判为已落定结果命中缓存,污染 LLM 上下文。
|
||||
pub(crate) const PENDING_APPROVAL_PLACEHOLDER: &str = "需要用户审批，等待确认";
|
||||
|
||||
/// F-260616-05：高危工具去重（根治 run_command 超时→重试→重新审批循环）。
|
||||
///
|
||||
/// ⚠ 性能注记(CR-260618-11#5):本函数在 session lock 持有期间对每个 high risk 工具
|
||||
/// 串行查 AiToolExecutionRepo::find_by_tool_call_id,工具数多时锁持有线性增长。
|
||||
/// 批量预取(改签名传 tool_call_ids 批量查)属架构改暂未做,后续 High risk 工具增多时优先处理。
|
||||
///
|
||||
/// **根因链**（F-04 batch42 已做超时标注）：LLM 调 run_command 超时 → F-04 把超时标注成
|
||||
/// tool_result 回传 LLM → LLM（不可靠）仍重试同命令 → 新 tool_call_id（provider 每轮新 id）
|
||||
/// → `process_tool_calls` 重新 insert pending(audit.rs:418) → 用户被迫重新审批，循环。
|
||||
///
|
||||
/// **治本（F-05）**：即使 LLM 仍重试，同 (tool_name, args) 不重复审批。本函数扫描会话历史，
|
||||
/// 若发现**已落定**（executed/failed/rejected，非 pending）的同类同参高危调用，返回其
|
||||
/// tool_result 内容，调用方把缓存结果作为新 tool_call_id 的 tool_result 回传 LLM，跳过审批。
|
||||
///
|
||||
/// **安全边界**：
|
||||
/// - 仅 High risk（循环源头）；Low/Med 不去重（Low 直行无审批，Med 重试场景少且去重易误伤）。
|
||||
/// - 仅匹配**已落定**结果（pending 的不命中——pending 已有独立流程，不会触发循环；
|
||||
/// 且避免两个 pending 互相吞掉审批）。LLM 只有在收到 tool_result 后才会重试，
|
||||
/// 故循环必然是「上一条已落定 → 重试」形态，pending 不命中不影响治本。
|
||||
/// - args 走 JSON 规范化比较（键序无关），避免 LLM 两次生成键序不同误判为不同命令。
|
||||
/// - 返回 None 表示无缓存命中（走原审批流程）。
|
||||
///
|
||||
/// `session` 只读扫描 messages（不写），调用方据返回值决定是否跳过 insert pending。
|
||||
///
|
||||
/// F-260616-09 B 批2:经`conv_id` 索引 per_conv.messages(顶层 messages 批2 后是死字段)。
|
||||
/// conv_id 来源:process_tool_calls 入参 → 由 agentic.rs run_agentic_loop 入参透传。
|
||||
pub(crate) async fn find_cached_high_risk_result(
|
||||
session: &AiSession,
|
||||
conv_id: &str,
|
||||
audit_repo: &AiToolExecutionRepo,
|
||||
tool_name: &str,
|
||||
args: &serde_json::Value,
|
||||
) -> Option<(String, String)> {
|
||||
use df_ai::provider::MessageRole;
|
||||
|
||||
// 规范化新调用的 args 为可比字符串（排序键，键序无关）
|
||||
let new_args_key = canonical_args_key(args);
|
||||
|
||||
// F-260616-09 B 批2:读 per_conv.messages。process_tool_calls 调用前 loop 入口已桥接建立 per_conv,
|
||||
// 故 conv_read 必命中;防御性 None 时返 None(无缓存命中,走原审批流程)。
|
||||
let conv = session.conv_read(conv_id)?;
|
||||
// ContextManager::iter 返回 impl Iterator（非 DoubleEnded），collect 成 Vec 再反向遍历。
|
||||
// 单对话消息量小（百级），collect 开销可忽略。
|
||||
let msgs: Vec<&ChatMessage> = conv.messages.iter().collect();
|
||||
|
||||
// 1) 反向扫描 assistant tool_calls，找最近一条同名同参的 High 工具调用 → 拿到旧 tool_call_id
|
||||
// 反向：循环是「最近一次超时→重试」，命中通常是末尾附近，反向先停省全扫。
|
||||
let mut prev_tool_call_id: Option<String> = None;
|
||||
for msg in msgs.iter().rev() {
|
||||
if !matches!(msg.role, MessageRole::Assistant) {
|
||||
continue;
|
||||
}
|
||||
let Some(tcs) = msg.tool_calls.as_ref() else { continue };
|
||||
for tc in tcs {
|
||||
if tc.function.name != tool_name {
|
||||
continue;
|
||||
}
|
||||
// 旧调用的 args 是流式拼接的 JSON 字符串，解析失败跳过（不误判为命中）
|
||||
let Ok(old_args) = serde_json::from_str::<serde_json::Value>(&tc.function.arguments) else {
|
||||
continue;
|
||||
};
|
||||
if canonical_args_key(&old_args) == new_args_key {
|
||||
prev_tool_call_id = Some(tc.id.clone());
|
||||
break;
|
||||
}
|
||||
}
|
||||
if prev_tool_call_id.is_some() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// 2) 用旧 tool_call_id 找对应 tool_result。注意：审批拒绝/超时失败也属「已落定」，
|
||||
// 其 tool_result 内容同样回传（LLM 看到原反馈自行决定，不再逼用户二次审批）。
|
||||
// pending 占位（「需要用户审批，等待确认」）不命中——仍在审批中，走原流程。
|
||||
let old_id = prev_tool_call_id?;
|
||||
for msg in msgs.iter().rev() {
|
||||
if !matches!(msg.role, MessageRole::Tool) {
|
||||
continue;
|
||||
}
|
||||
if msg.tool_call_id.as_deref() != Some(old_id.as_str()) {
|
||||
continue;
|
||||
}
|
||||
// 命中旧 tool_result：排除 pending 占位（内容固定为 PENDING_APPROVAL_PLACEHOLDER）
|
||||
if msg.content == PENDING_APPROVAL_PLACEHOLDER {
|
||||
return None;
|
||||
}
|
||||
// SW-260618-16: 查审计表拿缓存来源真实 status(completed/rejected/failed),透传给
|
||||
// audit_tool_call 而非固定 completed(审计语义与结果内容一致,防"rejected/failed 结果
|
||||
// 记 completed"误导安全追溯)。审计记录缺失/查询失败 fallback completed(不阻塞去重,降级原行为)。
|
||||
let status = audit_repo
|
||||
.find_by_tool_call_id(old_id.as_str())
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|opt| opt.map(|rec| rec.status))
|
||||
.unwrap_or_else(|| "completed".to_string());
|
||||
return Some((msg.content.clone(), status));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// 把 JSON args 规范化为可比字符串：对象键按字典序排序后序列化，
|
||||
/// 键序不同的等价参数生成同一 key（防 LLM 两次生成键序不同误判为不同命令）。
|
||||
fn canonical_args_key(args: &serde_json::Value) -> String {
|
||||
let mut v = args.clone();
|
||||
sort_object_keys(&mut v);
|
||||
// 紧凑序列化（无空白），保证稳定可比
|
||||
serde_json::to_string(&v).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// 递归对 JSON 对象的键做字典序排序（就地），数组成员也递归排序。
|
||||
fn sort_object_keys(v: &mut serde_json::Value) {
|
||||
match v {
|
||||
serde_json::Value::Object(map) => {
|
||||
// BTreeMap 按键排序，重建 Object
|
||||
let mut entries: Vec<(String, serde_json::Value)> = map
|
||||
.iter()
|
||||
.map(|(k, val)| (k.clone(), val.clone()))
|
||||
.collect();
|
||||
entries.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
map.clear();
|
||||
for (k, mut val) in entries {
|
||||
sort_object_keys(&mut val);
|
||||
map.insert(k, val);
|
||||
}
|
||||
}
|
||||
serde_json::Value::Array(arr) => {
|
||||
for item in arr {
|
||||
sort_object_keys(item);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
@@ -18,13 +18,6 @@ use crate::commands::err_str;
|
||||
use super::{AiChatEvent, AiSession, PendingApproval, ToolCallDraft};
|
||||
use super::tool_registry::generate_diff;
|
||||
|
||||
/// Med/High 工具挂起审批时回填的占位 tool_result 内容。
|
||||
///
|
||||
/// 单一常量避免「写入处」(process_tool_calls) 与「去重排查处」
|
||||
/// (find_cached_high_risk_result) 各持一份字面量致耦合——若两者漂移,
|
||||
/// 去重会把 pending 占位误判为已落定结果命中缓存,污染 LLM 上下文。
|
||||
const PENDING_APPROVAL_PLACEHOLDER: &str = "需要用户审批，等待确认";
|
||||
|
||||
/// RiskLevel → 审计记录字符串（low/medium/high）
|
||||
pub(crate) fn risk_str(r: RiskLevel) -> &'static str {
|
||||
match r {
|
||||
@@ -177,139 +170,11 @@ pub(crate) fn emit_data_changed(app_handle: &AppHandle, tool_name: &str) {
|
||||
}
|
||||
}
|
||||
|
||||
/// F-260616-05：高危工具去重（根治 run_command 超时→重试→重新审批循环）。
|
||||
///
|
||||
/// ⚠ 性能注记(CR-260618-11#5):本函数在 session lock 持有期间对每个 high risk 工具
|
||||
/// 串行查 AiToolExecutionRepo::find_by_tool_call_id,工具数多时锁持有线性增长。
|
||||
/// 批量预取(改签名传 tool_call_ids 批量查)属架构改暂未做,后续 High risk 工具增多时优先处理。
|
||||
///
|
||||
/// **根因链**（F-04 batch42 已做超时标注）：LLM 调 run_command 超时 → F-04 把超时标注成
|
||||
/// tool_result 回传 LLM → LLM（不可靠）仍重试同命令 → 新 tool_call_id（provider 每轮新 id）
|
||||
/// → `process_tool_calls` 重新 insert pending(audit.rs:418) → 用户被迫重新审批，循环。
|
||||
///
|
||||
/// **治本（F-05）**：即使 LLM 仍重试，同 (tool_name, args) 不重复审批。本函数扫描会话历史，
|
||||
/// 若发现**已落定**（executed/failed/rejected，非 pending）的同类同参高危调用，返回其
|
||||
/// tool_result 内容，调用方把缓存结果作为新 tool_call_id 的 tool_result 回传 LLM，跳过审批。
|
||||
///
|
||||
/// **安全边界**：
|
||||
/// - 仅 High risk（循环源头）；Low/Med 不去重（Low 直行无审批，Med 重试场景少且去重易误伤）。
|
||||
/// - 仅匹配**已落定**结果（pending 的不命中——pending 已有独立流程，不会触发循环；
|
||||
/// 且避免两个 pending 互相吞掉审批）。LLM 只有在收到 tool_result 后才会重试，
|
||||
/// 故循环必然是「上一条已落定 → 重试」形态，pending 不命中不影响治本。
|
||||
/// - args 走 JSON 规范化比较（键序无关），避免 LLM 两次生成键序不同误判为不同命令。
|
||||
/// - 返回 None 表示无缓存命中（走原审批流程）。
|
||||
///
|
||||
/// `session` 只读扫描 messages（不写），调用方据返回值决定是否跳过 insert pending。
|
||||
///
|
||||
/// F-260616-09 B 批2:经 `conv_id` 索引 per_conv.messages(顶层 messages 批2 后是死字段)。
|
||||
/// conv_id 来源:process_tool_calls 入参 → 由 agentic.rs run_agentic_loop 入参透传。
|
||||
async fn find_cached_high_risk_result(
|
||||
session: &AiSession,
|
||||
conv_id: &str,
|
||||
audit_repo: &AiToolExecutionRepo,
|
||||
tool_name: &str,
|
||||
args: &serde_json::Value,
|
||||
) -> Option<(String, String)> {
|
||||
use df_ai::provider::MessageRole;
|
||||
|
||||
// 规范化新调用的 args 为可比字符串（排序键，键序无关）
|
||||
let new_args_key = canonical_args_key(args);
|
||||
|
||||
// F-260616-09 B 批2:读 per_conv.messages。process_tool_calls 调用前 loop 入口已桥接建立 per_conv,
|
||||
// 故 conv_read 必命中;防御性 None 时返 None(无缓存命中,走原审批流程)。
|
||||
let conv = session.conv_read(conv_id)?;
|
||||
// ContextManager::iter 返回 impl Iterator（非 DoubleEnded），collect 成 Vec 再反向遍历。
|
||||
// 单对话消息量小（百级），collect 开销可忽略。
|
||||
let msgs: Vec<&ChatMessage> = conv.messages.iter().collect();
|
||||
|
||||
// 1) 反向扫描 assistant tool_calls，找最近一条同名同参的 High 工具调用 → 拿到旧 tool_call_id
|
||||
// 反向：循环是「最近一次超时→重试」，命中通常是末尾附近，反向先停省全扫。
|
||||
let mut prev_tool_call_id: Option<String> = None;
|
||||
for msg in msgs.iter().rev() {
|
||||
if !matches!(msg.role, MessageRole::Assistant) {
|
||||
continue;
|
||||
}
|
||||
let Some(tcs) = msg.tool_calls.as_ref() else { continue };
|
||||
for tc in tcs {
|
||||
if tc.function.name != tool_name {
|
||||
continue;
|
||||
}
|
||||
// 旧调用的 args 是流式拼接的 JSON 字符串，解析失败跳过（不误判为命中）
|
||||
let Ok(old_args) = serde_json::from_str::<serde_json::Value>(&tc.function.arguments) else {
|
||||
continue;
|
||||
};
|
||||
if canonical_args_key(&old_args) == new_args_key {
|
||||
prev_tool_call_id = Some(tc.id.clone());
|
||||
break;
|
||||
}
|
||||
}
|
||||
if prev_tool_call_id.is_some() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// 2) 用旧 tool_call_id 找对应 tool_result。注意：审批拒绝/超时失败也属「已落定」，
|
||||
// 其 tool_result 内容同样回传（LLM 看到原反馈自行决定，不再逼用户二次审批）。
|
||||
// pending 占位（「需要用户审批，等待确认」）不命中——仍在审批中，走原流程。
|
||||
let old_id = prev_tool_call_id?;
|
||||
for msg in msgs.iter().rev() {
|
||||
if !matches!(msg.role, MessageRole::Tool) {
|
||||
continue;
|
||||
}
|
||||
if msg.tool_call_id.as_deref() != Some(old_id.as_str()) {
|
||||
continue;
|
||||
}
|
||||
// 命中旧 tool_result：排除 pending 占位（内容固定为 PENDING_APPROVAL_PLACEHOLDER）
|
||||
if msg.content == PENDING_APPROVAL_PLACEHOLDER {
|
||||
return None;
|
||||
}
|
||||
// SW-260618-16: 查审计表拿缓存来源真实 status(completed/rejected/failed),透传给
|
||||
// audit_tool_call 而非固定 completed(审计语义与结果内容一致,防"rejected/failed 结果
|
||||
// 记 completed"误导安全追溯)。审计记录缺失/查询失败 fallback completed(不阻塞去重,降级原行为)。
|
||||
let status = audit_repo
|
||||
.find_by_tool_call_id(old_id.as_str())
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|opt| opt.map(|rec| rec.status))
|
||||
.unwrap_or_else(|| "completed".to_string());
|
||||
return Some((msg.content.clone(), status));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// 把 JSON args 规范化为可比字符串：对象键按字典序排序后序列化，
|
||||
/// 键序不同的等价参数生成同一 key（防 LLM 两次生成键序不同误判为不同命令）。
|
||||
fn canonical_args_key(args: &serde_json::Value) -> String {
|
||||
let mut v = args.clone();
|
||||
sort_object_keys(&mut v);
|
||||
// 紧凑序列化（无空白），保证稳定可比
|
||||
serde_json::to_string(&v).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// 递归对 JSON 对象的键做字典序排序（就地），数组成员也递归排序。
|
||||
fn sort_object_keys(v: &mut serde_json::Value) {
|
||||
match v {
|
||||
serde_json::Value::Object(map) => {
|
||||
// BTreeMap 按键排序，重建 Object
|
||||
let mut entries: Vec<(String, serde_json::Value)> = map
|
||||
.iter()
|
||||
.map(|(k, val)| (k.clone(), val.clone()))
|
||||
.collect();
|
||||
entries.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
map.clear();
|
||||
for (k, mut val) in entries {
|
||||
sort_object_keys(&mut val);
|
||||
map.insert(k, val);
|
||||
}
|
||||
}
|
||||
serde_json::Value::Array(arr) => {
|
||||
for item in arr {
|
||||
sort_object_keys(item);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
// cache(audit/cache.rs):F-260616-05 高危工具去重缓存。
|
||||
// 第三批从本文件抽离,行为零变更。pub(super) use 供本文件 process_tool_calls 裸名调用
|
||||
// (find_cached_high_risk_result),PENDING_APPROVAL_PLACEHOLDER 供 process_tool_calls 占位回填共享。
|
||||
mod cache;
|
||||
pub(super) use cache::{find_cached_high_risk_result, PENDING_APPROVAL_PLACEHOLDER};
|
||||
|
||||
/// AE-2025-03（路径 B）：write_file 审批预览 diff 生成。
|
||||
///
|
||||
|
||||
Reference in New Issue
Block a user