GenarrativeAI/Genarrative
5
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

Auto-open draft result after foundation completes

Browse Source
This commit is contained in:
高物
2026-04-25 10:52:39 +08:00
parent 35c2bce6f1
commit 03acbc5cb1
31 changed files with 36472 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

320
server-rs/crates/platform-llm/src/lib.rs
View File

@@ -1,4 +1,12 @@
use std::{error::Error, fmt, str as std_str, time::Duration};
use std::{
env,
error::Error,
fmt, fs,
path::PathBuf,
str as std_str,
sync::atomic::{AtomicU64, Ordering},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use log::{debug, warn};
use reqwest::{Client, StatusCode};
@@ -10,6 +18,9 @@ pub const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 30_000;
pub const DEFAULT_MAX_RETRIES: u32 = 1;
pub const DEFAULT_RETRY_BACKOFF_MS: u64 = 500;
pub const CHAT_COMPLETIONS_PATH: &str = "/chat/completions";
const DEFAULT_LLM_RAW_LOG_DIR: &str = "logs/llm-raw";
static LLM_RAW_LOG_SEQUENCE: AtomicU64 = AtomicU64::new(1);
// 冻结平台来源，避免上层继续散落 provider 字符串。
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -113,6 +124,17 @@ struct ChatCompletionsRequestBody<'a> {
max_tokens: Option<u32>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct LlmRawFailureInputLog<'a> {
provider: &'static str,
model: &'a str,
stream: bool,
attempt: u32,
max_tokens: Option<u32>,
messages: &'a [LlmMessage],
}
#[derive(Deserialize)]
struct ChatCompletionsResponseEnvelope {
id: Option<String>,
@@ -156,6 +178,7 @@ struct ChatCompletionsContentPart {
#[derive(Default)]
struct OpenAiCompatibleSseParser {
buffer: String,
raw_text: String,
}
#[derive(Debug)]
@@ -382,12 +405,31 @@ impl LlmClient {
request.validate()?;
let resolved_model = request.resolved_model(self.config.model()).to_string();
let response = self.execute_request(&request, false).await?;
let raw_text = response
.text()
.await
.map_err(|error| map_stream_read_error(error, 1))?;
let raw_text = response.text().await.map_err(|error| {
let llm_error = map_stream_read_error(error, 1);
log_llm_raw_failure(
&self.config,
&request,
false,
1,
"read_response_failed",
llm_error.to_string().as_str(),
);
llm_error
})?;
parse_chat_completions_response(self.config.provider(), &resolved_model, raw_text.as_str())
.map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
false,
1,
"parse_response_failed",
raw_text.as_str(),
);
error
})
}
pub async fn request_single_message_text(
@@ -422,10 +464,18 @@ impl LlmClient {
let mut undecoded_chunk_bytes = Vec::new();
loop {
let next_chunk = response
.chunk()
.await
.map_err(|error| map_stream_read_error(error, 1))?;
let next_chunk = response.chunk().await.map_err(|error| {
let llm_error = map_stream_read_error(error, 1);
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"read_stream_failed",
parser.raw_text().as_str(),
);
llm_error
})?;
let Some(chunk) = next_chunk else {
break;
@@ -433,12 +483,33 @@ impl LlmClient {
undecoded_chunk_bytes.extend_from_slice(chunk.as_ref());
let (chunk_text, remaining_bytes) =
decode_utf8_stream_chunk(undecoded_chunk_bytes.as_slice())?;
decode_utf8_stream_chunk(undecoded_chunk_bytes.as_slice()).map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"decode_stream_failed",
parser.raw_text().as_str(),
);
error
})?;
undecoded_chunk_bytes = remaining_bytes;
if chunk_text.is_empty() {
continue;
}
for event in parser.push_chunk(chunk_text.as_ref())? {
let stream_events = parser.push_chunk(chunk_text.as_ref()).map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"parse_stream_failed",
parser.raw_text().as_str(),
);
error
})?;
for event in stream_events {
if let Some(delta_text) = event.delta_text
&& !delta_text.is_empty()
{
@@ -460,10 +531,29 @@ impl LlmClient {
if !undecoded_chunk_bytes.is_empty() {
let trailing_text =
std_str::from_utf8(undecoded_chunk_bytes.as_slice()).map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"decode_stream_failed",
parser.raw_text().as_str(),
);
LlmError::Deserialize(format!("解析 LLM 流式 UTF-8 响应失败：{error}"))
})?;
if !trailing_text.is_empty() {
for event in parser.push_chunk(trailing_text)? {
let trailing_events = parser.push_chunk(trailing_text).map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"parse_stream_failed",
parser.raw_text().as_str(),
);
error
})?;
for event in trailing_events {
if let Some(delta_text) = event.delta_text
&& !delta_text.is_empty()
{
@@ -483,7 +573,18 @@ impl LlmClient {
}
}
for event in parser.finish()? {
let remaining_events = parser.finish().map_err(|error| {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"parse_stream_failed",
parser.raw_text().as_str(),
);
error
})?;
for event in remaining_events {
if let Some(delta_text) = event.delta_text
&& !delta_text.is_empty()
{
@@ -503,6 +604,14 @@ impl LlmClient {
let content = accumulated_text.trim().to_string();
if content.is_empty() {
log_llm_raw_failure(
&self.config,
&request,
true,
1,
"empty_stream_response",
parser.raw_text().as_str(),
);
return Err(LlmError::EmptyResponse);
}
@@ -591,6 +700,14 @@ impl LlmClient {
continue;
}
log_llm_raw_failure(
&self.config,
request,
stream,
attempt,
"upstream_status_failed",
raw_text.as_str(),
);
return Err(LlmError::Upstream {
status_code: status.as_u16(),
message,
@@ -607,7 +724,16 @@ impl LlmClient {
continue;
}
return Err(LlmError::Timeout { attempts: attempt });
let error = LlmError::Timeout { attempts: attempt };
log_llm_raw_failure(
&self.config,
request,
stream,
attempt,
"request_timeout",
error.to_string().as_str(),
);
return Err(error);
}
Err(error) if error.is_connect() => {
let message = error.to_string();
@@ -622,13 +748,31 @@ impl LlmClient {
continue;
}
return Err(LlmError::Connectivity {
let error = LlmError::Connectivity {
attempts: attempt,
message,
});
};
log_llm_raw_failure(
&self.config,
request,
stream,
attempt,
"request_connectivity_failed",
error.to_string().as_str(),
);
return Err(error);
}
Err(error) => {
return Err(LlmError::Transport(error.to_string()));
let error = LlmError::Transport(error.to_string());
log_llm_raw_failure(
&self.config,
request,
stream,
attempt,
"request_transport_failed",
error.to_string().as_str(),
);
return Err(error);
}
}
}
@@ -652,11 +796,16 @@ impl LlmClient {
impl OpenAiCompatibleSseParser {
fn push_chunk(&mut self, chunk: &str) -> Result<Vec<ParsedStreamEvent>, LlmError> {
self.raw_text.push_str(chunk);
self.buffer.push_str(chunk);
self.buffer = self.buffer.replace("\r\n", "\n");
self.drain_complete_events()
}
fn raw_text(&self) -> String {
self.raw_text.clone()
}
fn finish(&mut self) -> Result<Vec<ParsedStreamEvent>, LlmError> {
if self.buffer.trim().is_empty() {
return Ok(Vec::new());
@@ -691,6 +840,87 @@ fn normalize_non_empty(value: String, error_message: &str) -> Result<String, Llm
Ok(trimmed)
}
fn log_llm_raw_failure(
config: &LlmConfig,
request: &LlmTextRequest,
stream: bool,
attempt: u32,
failure_stage: &str,
raw_output: &str,
) {
if let Err(error) =
write_llm_raw_failure(config, request, stream, attempt, failure_stage, raw_output)
{
warn!(
"LLM 失败原文日志落盘失败，主错误流程继续执行: failure_stage={}, error={}",
failure_stage, error
);
}
}
fn write_llm_raw_failure(
config: &LlmConfig,
request: &LlmTextRequest,
stream: bool,
attempt: u32,
failure_stage: &str,
raw_output: &str,
) -> Result<(), String> {
let log_dir = env::var("LLM_RAW_LOG_DIR")
.map(PathBuf::from)
.unwrap_or_else(|_| PathBuf::from(DEFAULT_LLM_RAW_LOG_DIR));
fs::create_dir_all(&log_dir).map_err(|error| format!("创建日志目录失败：{error}"))?;
let prefix = build_llm_raw_log_prefix(failure_stage);
let model = request.resolved_model(config.model());
let input_log = LlmRawFailureInputLog {
provider: config.provider().as_str(),
model,
stream,
attempt,
max_tokens: request.max_tokens,
messages: request.messages.as_slice(),
};
let input_text = serde_json::to_string_pretty(&input_log)
.map_err(|error| format!("序列化模型输入日志失败：{error}"))?;
fs::write(log_dir.join(format!("{prefix}.input.json")), input_text)
.map_err(|error| format!("写入模型输入日志失败：{error}"))?;
fs::write(log_dir.join(format!("{prefix}.output.txt")), raw_output)
.map_err(|error| format!("写入模型输出日志失败：{error}"))?;
Ok(())
}
fn build_llm_raw_log_prefix(failure_stage: &str) -> String {
let millis = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.unwrap_or_default();
let sequence = LLM_RAW_LOG_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let safe_stage = sanitize_log_file_segment(failure_stage);
format!("{millis}-{}-{sequence:06}-{safe_stage}", std::process::id())
}
fn sanitize_log_file_segment(value: &str) -> String {
let sanitized = value
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() || character == '-' || character == '_' {
character
} else {
'_'
}
})
.collect::<String>();
if sanitized.is_empty() {
"unknown".to_string()
} else {
sanitized
}
}
fn parse_chat_completions_response(
provider: LlmProvider,
fallback_model: &str,
@@ -1028,6 +1258,62 @@ mod tests {
assert_eq!(response.response_id.as_deref(), Some("req_stream_01"));
}
#[tokio::test]
async fn request_text_writes_raw_failure_logs_after_parse_error() {
let log_dir = std::env::temp_dir().join(format!(
"platform-llm-raw-log-test-{}",
build_llm_raw_log_prefix("parse_error")
));
unsafe {
std::env::set_var("LLM_RAW_LOG_DIR", &log_dir);
}
let server_url = spawn_mock_server(vec![MockResponse {
status_line: "200 OK",
content_type: "application/json; charset=utf-8",
body: "不是合法 JSON".to_string(),
extra_headers: Vec::new(),
}]);
let client = build_test_client(server_url, 0);
let error = client
.request_single_message_text("系统原文", "用户原文")
.await
.expect_err("invalid json should fail");
assert!(matches!(error, LlmError::Deserialize(_)));
let mut input_logs = Vec::new();
let mut output_logs = Vec::new();
for entry in fs::read_dir(&log_dir).expect("log dir should exist") {
let path = entry.expect("log entry should be readable").path();
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or_default()
.to_string();
if file_name.ends_with(".input.json") {
input_logs.push(path);
} else if file_name.ends_with(".output.txt") {
output_logs.push(path);
}
}
assert_eq!(input_logs.len(), 1);
assert_eq!(output_logs.len(), 1);
let input_text = fs::read_to_string(&input_logs[0]).expect("input log should be readable");
let output_text =
fs::read_to_string(&output_logs[0]).expect("output log should be readable");
assert!(input_text.contains("系统原文"));
assert!(input_text.contains("用户原文"));
assert!(!input_text.contains("test-key"));
assert_eq!(output_text, "不是合法 JSON");
unsafe {
std::env::remove_var("LLM_RAW_LOG_DIR");
}
fs::remove_dir_all(log_dir).expect("log dir should be removed");
}
fn build_test_client(base_url: String, max_retries: u32) -> LlmClient {
let config = LlmConfig::new(
LlmProvider::Ark,