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

修复: DeepSeek reasoning_content 全链路透传(跨6 crate补字段+映射+agentic loop累积回传)

Browse Source
This commit is contained in:
绝尘
2026-06-17 18:08:04 +08:00
parent 1cd7652a34
commit 74003bc04f
11 changed files with 251 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -109,7 +109,7 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
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 }
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("");
@@ -128,7 +128,7 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
total_tokens: inp as u32,
});
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
}
// 消息增量：output_tokens 是累计值（非增量），直接覆盖 completion + 重算 total
"message_delta" => {
@@ -138,14 +138,14 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
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 }
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 };
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") {
@@ -162,10 +162,11 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
}]),
usage: None,
error: None,
reasoning_content: None,
};
}
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
}
// 工具块开始：带 id + name
"content_block_start" => {
@@ -194,10 +195,11 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
}]),
usage: None,
error: None,
reasoning_content: None,
};
}
}
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None }
StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None }
}
// 消息结束：带出累积 usage
"message_stop" => StreamChunk {
@@ -206,16 +208,17 @@ pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUs
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) }
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 },
_ => StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, reasoning_content: None },
}
}
@@ -541,6 +544,7 @@ impl LlmProvider for AnthropicCompatProvider {
model: resp.model,
usage,
tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) },
reasoning_content: None,
})
}
})
@@ -808,6 +812,7 @@ mod tests {
stream: false,
tools: None,
tool_choice: None,
reasoning_content: None,
};
let body = provider.convert_request(req);
// user 消息 content 应为数组形态
@@ -839,6 +844,7 @@ mod tests {
stream: false,
tools: None,
tool_choice: None,
reasoning_content: None,
};
let body = provider.convert_request(req);
let user_msg = body

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

@@ -44,6 +44,9 @@ struct OpenAiRequest {
/// 流式时请求末 chunk 携带 usage（OpenAI 官方 + DeepSeek/GLM 兼容）
#[serde(skip_serializing_if = "Option::is_none")]
stream_options: Option<serde_json::Value>,
/// DeepSeek thinking 模式
#[serde(skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
}
/// OpenAI 消息格式
@@ -58,6 +61,9 @@ struct OpenAiMessage {
tool_call_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
tool_calls: Option<Vec<serde_json::Value>>,
/// DeepSeek thinking 模式推理内容
#[serde(skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
}
/// OpenAI 同步响应
@@ -78,6 +84,9 @@ struct OpenAiChoice {
struct OpenAiMessageResp {
content: Option<String>,
tool_calls: Option<Vec<OpenAiToolCallResp>>,
/// DeepSeek thinking 模式响应中的推理内容
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -120,6 +129,9 @@ struct OpenAiStreamChoice {
struct OpenAiStreamDelta {
content: Option<String>,
tool_calls: Option<Vec<OpenAiStreamToolCall>>,
/// DeepSeek thinking 模式流式推理内容
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -158,6 +170,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
tool_calls: None,
usage: usage_accum.take(),
error: None,
reasoning_content: None,
};
}
@@ -195,6 +208,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
tool_calls,
usage: None,
error: None,
reasoning_content: choice.delta.reasoning_content,
}
} else {
// choices 为空 = usage-only chunk，不输出文本（usage 已累积）
@@ -204,6 +218,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
tool_calls: None,
usage: None,
error: None,
reasoning_content: None,
}
}
}
@@ -215,6 +230,7 @@ pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>)
tool_calls: None,
usage: None,
error: None,
reasoning_content: None,
}
}
}
@@ -371,6 +387,7 @@ impl OpenAICompatProvider {
content,
tool_call_id: m.tool_call_id,
tool_calls,
reasoning_content: m.reasoning_content,
}
})
.collect();
@@ -389,6 +406,7 @@ impl OpenAICompatProvider {
stream: req.stream,
tools,
tool_choice: req.tool_choice,
reasoning_content: req.reasoning_content,
// 流式请求末 chunk 带 usage（同步调用 complete 不需要）
stream_options: if req.stream {
Some(serde_json::json!({ "include_usage": true }))
@@ -509,6 +527,7 @@ impl LlmProvider for OpenAICompatProvider {
model: body.model,
usage,
tool_calls,
reasoning_content: choice.message.reasoning_content,
})
}
})
@@ -755,6 +774,7 @@ mod tests {
stream: false,
tools: None,
tool_choice: None,
reasoning_content: None,
};
let out = provider.convert_request(req);
let msg = &out.messages[0];
@@ -783,6 +803,7 @@ mod tests {
stream: false,
tools: None,
tool_choice: None,
reasoning_content: None,
};
let out = provider.convert_request(req);
let msg = &out.messages[0];