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

优化: AI模块(provider重试兼容+aichat交互+工具卡片可读化+diff高亮)

Browse Source
This commit is contained in:
绝尘
2026-06-16 02:33:16 +08:00
parent d2cb38cdac
commit 10e4945e5a
26 changed files with 2337 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/df-ai/Cargo.toml
View File

@@ -7,7 +7,7 @@ edition = "2021"
df-core = { path = "../df-core" }
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["sync"] }
tokio = { workspace = true, features = ["sync", "time"] }
async-trait = { workspace = true }
anyhow = { workspace = true }
tracing = { workspace = true }

149
crates/df-ai/src/anthropic_compat.rs
View File

@@ -11,19 +11,23 @@ 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,
};
use crate::retry::{
retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable,
};
// ============================================================
// Anthropic API 请求/响应结构体
// ============================================================
/// Anthropic 请求体
#[derive(Debug, Serialize)]
#[derive(Debug, Clone, Serialize)]
struct AnthropicRequest {
model: String,
messages: Vec<serde_json::Value>,
@@ -40,7 +44,7 @@ struct AnthropicRequest {
}
/// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters）
#[derive(Debug, Serialize)]
#[derive(Debug, Clone, Serialize)]
struct AnthropicToolDef {
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -406,68 +410,93 @@ impl LlmProvider for AnthropicCompatProvider {
debug!(model = %body.model, "Anthropic 同步调用");
let resp = self
.auth_headers(self.client.post(self.messages_url()))
.json(&body)
.send()
.await?;
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);
}
let resp: AnthropicResponse = resp.json().await?;
// content 块中拼接 text，收集 tool_use
let mut text = String::new();
let mut tool_calls: Vec<ToolCall> = Vec::new();
for block in resp.content {
match block.block_type.as_str() {
"text" => {
if let Some(t) = block.text {
text.push_str(&t);
// 指数退避重试(B-260616-07): 包裹 send + 状态码判定。
// 同时补 FR-R4 遗漏: Anthropic 同步路径此前无单请求 timeout(建连后挂起会无限 hang)，
// 此处加 60s timeout，与 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))
.json(&body);
let resp = match rb.send().await {
Ok(r) => r,
Err(e) => {
if is_reqwest_error_retryable(&e) {
return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e));
}
return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", 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);
}
let resp: AnthropicResponse = match resp.json().await {
Ok(r) => r,
Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)),
};
// content 块中拼接 text，收集 tool_use
let mut text = String::new();
let mut tool_calls: Vec<ToolCall> = Vec::new();
for block in resp.content {
match block.block_type.as_str() {
"text" => {
if let Some(t) = block.text {
text.push_str(&t);
}
}
"tool_use" => {
let id = match block.id {
Some(id) if !id.is_empty() => id,
_ => {
warn!(
name = ?block.name,
"Anthropic tool_use 块缺少 id，已跳过（空 id 会回传空 tool_use_id 触发 500）"
);
continue;
}
};
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 块类型，已忽略"),
}
}
"tool_use" => {
// tool_use 必须带 id：后续 tool_result 用它回引，缺 id 的 tool_call
// 会拖出空 tool_use_id 的 tool_result，触发 GLM 500 卡死会话。
// id 缺失/为空时跳过该 tool_use 解析并告警。
let id = match block.id {
Some(id) if !id.is_empty() => id,
_ => {
warn!(
name = ?block.name,
"Anthropic tool_use 块缺少 id，已跳过（空 id 会回传空 tool_use_id 触发 500）"
);
continue;
}
};
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,
};
AttemptOutcome::Ok(CompletionResponse {
text,
model: resp.model,
usage,
tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) },
})
}
}
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,
};
Ok(CompletionResponse {
text,
model: resp.model,
usage,
tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) },
})
.await
}
async fn stream(&self, request: CompletionRequest) -> anyhow::Result<StreamResult> {

1
crates/df-ai/src/lib.rs
View File

@@ -6,6 +6,7 @@ pub mod context;
pub mod coordinator;
pub mod openai_compat;
pub mod provider;
mod retry;
use provider::LlmProvider;

126
crates/df-ai/src/openai_compat.rs
View File

@@ -16,13 +16,16 @@ use crate::provider::{
CompletionRequest, CompletionResponse, LlmProvider, StreamChunk, StreamResult,
TokenUsage, ToolCall, ToolCallDelta,
};
use crate::retry::{
retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable,
};
// ============================================================
// OpenAI API 请求/响应结构体
// ============================================================
/// OpenAI 兼容请求体
#[derive(Debug, Serialize)]
#[derive(Debug, Clone, Serialize)]
struct OpenAiRequest {
model: String,
messages: Vec<OpenAiMessage>,
@@ -41,7 +44,7 @@ struct OpenAiRequest {
}
/// OpenAI 消息格式
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OpenAiMessage {
role: String,
content: String,
@@ -398,60 +401,75 @@ impl LlmProvider for OpenAICompatProvider {
debug!(model = %openai_req.model, "OpenAI 同步调用");
// 同步路径整体超时(FR-R4)：client 仅配 connect_timeout，无总 timeout（防误砍流式长生成）。
// complete 为一次性请求，挂死会让整轮对话静默卡住，此处用 RequestBuilder 的单请求 timeout 兜底，
// 超时返回 is_timeout 错误（不静默挂），且不影响 stream() 流式路径。
let resp = self
.client
.post(self.chat_url())
.header("Authorization", format!("Bearer {}", self.api_key))
.header("Content-Type", "application/json")
.timeout(Duration::from_secs(60))
.json(&openai_req)
.send()
.await
.map_err(|e| {
if e.is_timeout() {
error!(model = %openai_req.model, "OpenAI 同步调用超时(60s)");
anyhow::anyhow!("LLM 同步调用超时(60s)，请检查网络或模型可用性: {}", e)
} else {
anyhow::anyhow!(e)
// 指数退避重试(B-260616-07): 包裹 send + 状态码判定。
// 单请求 60s timeout 保持不变(FR-R4)，重试是额外层: 3 次 × 60s 最坏 180s，
// 由 retry_with_backoff 内部 30s 总预算主动止损。
let label = format!("OpenAI[{}]", openai_req.model);
retry_with_backoff(&label, move |_| {
let client = self.client.clone();
let url = self.chat_url();
let api_key = self.api_key.clone();
let openai_req = openai_req.clone();
async move {
// send
let resp = client
.post(url)
.header("Authorization", format!("Bearer {}", api_key))
.header("Content-Type", "application/json")
.timeout(Duration::from_secs(60))
.json(&openai_req)
.send()
.await;
let resp = match resp {
Ok(r) => r,
Err(e) => {
if is_reqwest_error_retryable(&e) {
return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e));
}
return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", e));
}
};
// 状态码判定
if !resp.status().is_success() {
let status = resp.status().as_u16();
let body = resp.text().await.unwrap_or_default();
let msg = format!("LLM API 错误 {}: {}", status, body);
if is_status_retryable(status) {
warn!(%status, "OpenAI 同步调用可重试状态码");
return AttemptOutcome::Retryable(msg);
}
error!(%status, %body, "LLM API 调用失败(不可重试)");
return AttemptOutcome::Fatal(msg);
}
})?;
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
error!(%status, %body, "LLM API 调用失败");
anyhow::bail!("LLM API 错误 {}: {}", status, body);
}
let body: OpenAiResponse = resp.json().await?;
let choice = body
.choices
.into_iter()
.next()
.ok_or_else(|| anyhow::anyhow!("LLM 响应无 choices"))?;
let text = choice.message.content.unwrap_or_default();
let tool_calls = choice.message.tool_calls.map(Self::parse_tool_calls);
let usage = body.usage.map(|u| TokenUsage {
prompt_tokens: u.prompt_tokens,
completion_tokens: u.completion_tokens,
total_tokens: u.total_tokens,
}).unwrap_or(TokenUsage {
prompt_tokens: 0,
completion_tokens: 0,
total_tokens: 0,
});
Ok(CompletionResponse {
text,
model: body.model,
usage,
tool_calls,
// body 解析: 解析错属 Fatal(响应已成功送达，重试也会因同样格式失败)
let body: OpenAiResponse = match resp.json().await {
Ok(b) => b,
Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)),
};
let choice = match body.choices.into_iter().next() {
Some(c) => c,
None => return AttemptOutcome::Fatal("LLM 响应无 choices".to_string()),
};
let text = choice.message.content.unwrap_or_default();
let tool_calls = choice.message.tool_calls.map(Self::parse_tool_calls);
let usage = body.usage.map(|u| TokenUsage {
prompt_tokens: u.prompt_tokens,
completion_tokens: u.completion_tokens,
total_tokens: u.total_tokens,
}).unwrap_or(TokenUsage {
prompt_tokens: 0,
completion_tokens: 0,
total_tokens: 0,
});
AttemptOutcome::Ok(CompletionResponse {
text,
model: body.model,
usage,
tool_calls,
})
}
})
.await
}
async fn stream(&self, request: CompletionRequest) -> anyhow::Result<StreamResult> {

257
crates/df-ai/src/retry.rs Normal file
View File

@@ -0,0 +1,257 @@
//! complete() 指数退避重试 — 仅非流式同步路径，不动 stream()
//!
//! 背景: complete() 是 LLM 一次性请求，网络波动/瞬态 5xx/429 会让整轮对话静默失败。
//! 各调用方(title.rs/knowledge_inject.rs/project.rs/ai_node.rs)虽已有人工兜底降级，
//! 但重试能在不侵入上层的前提下吸收瞬态错误，对低频后台调用零风险。
//!
//! 策略:
//! - 最多 3 次尝试(初次 + 2 次重试)，指数退避 1s → 2s → 4s + ±20% jitter
//! - 总挂钟预算 30s(含 sleep + 各次请求耗时)，超预算直接放弃重试返回最后错误
//! - 仅对可重试错误重试: timeout / connect / 5xx / 429; 不重试 4xx(非429) / 参数错 / 鉴权错
//! - 不引入 reqwest-retry / tower: complete 路径简单，手写 loop 零依赖、可调试、不与 SSE/双层 Semaphore 耦合
//!
//! 流式 stream() 路径不进入此模块(流式长生成不能整体 timeout，重试会丢增量)。
//!
//! 重试期间不释放 Semaphore permit(已在调用方持有)，对并发池有挤占 —— 但 complete 调用低频可接受。
use std::future::Future;
use std::time::{Duration, SystemTime};
use tracing::warn;
/// 最多尝试次数(含初次)。B-260616-07: 3 次 = 初次 + 2 次重试。
///
/// 配置化 TODO: 未来接入 per-provider config(`AiProviderRecord.config` JSON)或全局开关
/// (`useSetting('df-ai-max-retries')`)时改为读取配置。当前低频后台调用，常量足够。
pub const MAX_COMPLETE_ATTEMPTS: u32 = 3;
/// 退避基准(秒): 第 n 次重试前 sleep `BASE_BACKOFF_SECS * 2^(n-1)`。
/// 即 1s → 2s → 4s。
const BASE_BACKOFF_SECS: u64 = 1;
/// 整体重试挂钟预算(含 sleep 与各次请求耗时)。超过即放弃，返回最后错误。
/// 与单请求 60s timeout 配合: 最坏 3 次 × 60s = 180s，30s 预算会在此之前主动止损。
const MAX_TOTAL_BUDGET: Duration = Duration::from_secs(30);
/// jitter 上限(相对 base 的 ±比例)。避免重试风暴对齐。
const JITTER_RATIO: f64 = 0.2;
/// 一次尝试的分类结果 —— 在 anyhow 不透明化前决定是否值得重试。
#[derive(Debug)]
pub(super) enum AttemptOutcome<T> {
/// 成功，携带结果。
Ok(T),
/// 可重试错误(timeout / connect / 5xx / 429)。携带人类可读错误描述。
Retryable(String),
/// 不可重试错误(4xx 非 429 / 鉴权 / 参数错 / 解析错)。携带描述，直接放弃。
Fatal(String),
}
/// 判定 reqwest::Error 是否可重试(仅 timeout / connect 类)。
///
/// body 解析错(reqwest::Error::is_decode)与请求构造错(is_builder)属 Fatal，不重试。
pub(super) fn is_reqwest_error_retryable(e: &reqwest::Error) -> bool {
e.is_timeout() || e.is_connect() || (e.is_request() && !e.is_builder())
}
/// 判定 HTTP 状态码是否可重试: 5xx 与 429 重试，其余(含 4xx)不重试。
pub(super) fn is_status_retryable(status: u16) -> bool {
status == 429 || (500..600).contains(&status)
}
/// 指数退避 + jitter: 返回第 `attempt`(1-based)次重试前应 sleep 的时长。
/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s，各 ±20% jitter。
///
/// jitter 用 SystemTime 纳秒取模生成(无依赖)，避免多客户端同步重试风暴。
/// 以毫秒粒度计算后向下取整(避免秒级截断把 0.9s 砍成 0)。
fn backoff_delay(attempt: u32) -> Duration {
let base_ms = BASE_BACKOFF_SECS.saturating_mul(1u64 << (attempt - 1)) * 1000;
// 纳秒 → [0, 2000) 区间，再映射到 [-1.0, +1.0) 比例
let nanos = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map(|d| d.subsec_nanos() as u64)
.unwrap_or(0);
let jitter_ratio = (nanos % 2000) as f64 / 1000.0 - 1.0; // [-1.0, 1.0)
let factor = 1.0 + jitter_ratio * JITTER_RATIO; // [0.8, 1.2]
let ms = (base_ms as f64 * factor).max(0.0) as u64;
Duration::from_millis(ms)
}
/// 对一个返回 `AttemptOutcome` 的异步操作做指数退避重试。
///
/// `attempt_fn` 每次被调用执行一次尝试并返回分类结果。
/// 达到 `MAX_COMPLETE_ATTEMPTS` 或超出 `MAX_TOTAL_BUDGET` 或遇 Fatal 即终止。
/// 终止时若仍失败，返回最后一次的 Retryable/Fatal 描述作为 `anyhow::Error`。
///
/// 调用方约定: `attempt_fn` 内部应保留完整请求重发能力(每次重建 RequestBuilder)，
/// 因为 reqwest::RequestBuilder 一次 `.send()` 后不可复用。
pub(super) async fn retry_with_backoff<T, F, Fut>(
label: &str,
attempt_fn: F,
) -> anyhow::Result<T>
where
F: Fn(u32) -> Fut,
Fut: Future<Output = AttemptOutcome<T>>,
{
let deadline = tokio::time::Instant::now() + MAX_TOTAL_BUDGET;
let mut last_err = String::from("未尝试");
for attempt in 1..=MAX_COMPLETE_ATTEMPTS {
match attempt_fn(attempt).await {
AttemptOutcome::Ok(v) => return Ok(v),
AttemptOutcome::Fatal(msg) => {
// 不可重试: 立即放弃，不浪费预算
return Err(anyhow::anyhow!("{}", msg));
}
AttemptOutcome::Retryable(msg) => {
last_err = msg;
let is_last = attempt == MAX_COMPLETE_ATTEMPTS;
if is_last {
warn!(
label = label,
attempt = attempt,
"重试已达上限({}次)，放弃", MAX_COMPLETE_ATTEMPTS
);
break;
}
let now = tokio::time::Instant::now();
if now >= deadline {
warn!(
label = label,
attempt = attempt,
"重试预算({:?})已耗尽，放弃", MAX_TOTAL_BUDGET
);
break;
}
let delay = backoff_delay(attempt).min(deadline.saturating_duration_since(now));
warn!(
label = label,
attempt = attempt,
delay_ms = delay.as_millis() as u64,
err = %last_err,
"可重试错误，退避后重试"
);
tokio::time::sleep(delay).await;
}
}
}
Err(anyhow::anyhow!("{} 重试耗尽: {}", label, last_err))
}
// ============================================================
// 单测(纯逻辑，不发 HTTP)
// ============================================================
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
#[test]
fn backoff_is_monotonic_within_jitter() {
// base 序列: 1, 2, 4 秒; jitter ±20% 后仍在 [0.8,4.8] 范围
for attempt in 1..=3 {
let d = backoff_delay(attempt);
let base_ms = (BASE_BACKOFF_SECS * (1u64 << (attempt - 1))) * 1000;
let lo = (base_ms as f64 * (1.0 - JITTER_RATIO)) as u64;
let hi = (base_ms as f64 * (1.0 + JITTER_RATIO)) as u64;
assert!(
d.as_millis() as u64 >= lo && d.as_millis() as u64 <= hi,
"attempt={} delay={:?} 应在 [{},{}]ms",
attempt,
d,
lo,
hi
);
}
}
#[test]
fn status_classification() {
assert!(is_status_retryable(429));
assert!(is_status_retryable(500));
assert!(is_status_retryable(503));
assert!(!is_status_retryable(400));
assert!(!is_status_retryable(401));
assert!(!is_status_retryable(403));
assert!(!is_status_retryable(404));
}
/// 初次成功 → 不重试
#[tokio::test]
async fn first_attempt_ok() {
let calls = Arc::new(AtomicU32::new(0));
let calls_clone = calls.clone();
let r = retry_with_backoff("test", move |_| {
let c = calls_clone.clone();
async move {
c.fetch_add(1, Ordering::SeqCst);
AttemptOutcome::Ok(42u32)
}
})
.await
.unwrap();
assert_eq!(r, 42);
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
/// 第二次成功 → 恰好 2 次调用
#[tokio::test]
async fn retry_once_then_ok() {
let calls = Arc::new(AtomicU32::new(0));
let calls_clone = calls.clone();
let r = retry_with_backoff("test", move |_| {
let c = calls_clone.clone();
async move {
let n = c.fetch_add(1, Ordering::SeqCst) + 1;
if n == 1 {
AttemptOutcome::Retryable("timeout".into())
} else {
AttemptOutcome::Ok(7u32)
}
}
})
.await
.unwrap();
assert_eq!(r, 7);
assert_eq!(calls.load(Ordering::SeqCst), 2);
}
/// 全部 Retryable → 用尽 3 次，返回错误
#[tokio::test]
async fn all_retryable_exhausts() {
let calls = Arc::new(AtomicU32::new(0));
let calls_clone = calls.clone();
let r = retry_with_backoff("test", move |_| {
let c = calls_clone.clone();
async move {
c.fetch_add(1, Ordering::SeqCst);
AttemptOutcome::<u32>::Retryable("503".into())
}
})
.await;
assert!(r.is_err());
assert!(r.unwrap_err().to_string().contains("重试耗尽"));
assert_eq!(calls.load(Ordering::SeqCst), MAX_COMPLETE_ATTEMPTS);
}
/// Fatal 立即放弃，不重试
#[tokio::test]
async fn fatal_aborts_immediately() {
let calls = Arc::new(AtomicU32::new(0));
let calls_clone = calls.clone();
let r = retry_with_backoff("test", move |_| {
let c = calls_clone.clone();
async move {
c.fetch_add(1, Ordering::SeqCst);
AttemptOutcome::<u32>::Fatal("401".into())
}
})
.await;
assert!(r.is_err());
assert!(r.unwrap_err().to_string().contains("401"));
assert_eq!(calls.load(Ordering::SeqCst), 1, "Fatal 不应触发重试");
}
}