boos/BoosAPI
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: show conversation content in usage logs

Browse Source
This commit is contained in:
boos
2026-06-06 19:36:42 +08:00
parent ef0f91e683
commit cd843465b2
10 changed files with 566 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/proxy/mod.rs
+327 -39
View File
@@ -45,6 +45,13 @@ pub enum DownstreamSource {
Global,
}
#[derive(Clone, Debug, Default)]
struct ProxyLogContent {
summary: Option<String>,
request_content: Option<String>,
response_content: Option<String>,
}
// ── Route definitions ─────────────────────────────────────────────────
/// Build the /v1/ proxy router
@@ -101,17 +108,20 @@ async fn messages_handler(
let canonical = crate::transformers::anthropic::messages::parse_messages_request(&body)
.map_err(AppError::BadRequest)?;
let model = canonical.model.clone();
let log_content = proxy_log_content_from_request(&body);
if canonical.stream {
let chunks =
execute_messages_stream_with_conductor(state, key_id, model, canonical).await?;
execute_messages_stream_with_conductor(state, key_id, model, canonical, log_content)
.await?;
let sse =
Sse::new(stream::iter(chunks).map(|chunk| {
Ok::<_, Infallible>(axum::response::sse::Event::default().data(chunk))
}));
Ok(sse.into_response())
} else {
let result = execute_messages_with_conductor(state, key_id, model, canonical).await?;
let result =
execute_messages_with_conductor(state, key_id, model, canonical, log_content).await?;
Ok(Json(result).into_response())
}
}
@@ -372,6 +382,7 @@ async fn images_generations(
let selected = TokenRouter::new(state.db.clone())
.resolve_selected(&model)
.await?;
let request_log_body = body.clone();
let body = body_with_selected_model(body, &selected);
let result = executors::execute_image_generation(
state.clone(),
@@ -394,6 +405,7 @@ async fn images_generations(
cost,
1,
false,
proxy_log_content_from_request_response(&request_log_body, Some(&payload)),
)
.await;
record_billing(&state, key_id, &model, 0, 0, cost).await;
@@ -410,6 +422,7 @@ async fn images_generations(
Some(&err.to_string()),
1,
false,
proxy_log_content_from_request(&request_log_body),
)
.await;
Err(err)
@@ -437,6 +450,7 @@ async fn images_edits(
let mut fields: Vec<executors::MultipartField> = Vec::new();
let mut model: Option<String> = None;
let mut log_fields = serde_json::Map::new();
while let Some(field) = multipart
.next_field()
@@ -481,6 +495,7 @@ async fn images_edits(
if name == "model" {
model = Some(text.clone());
}
log_fields.insert(name.clone(), Value::String(text.clone()));
fields.push(executors::MultipartField {
name,
bytes: None,
@@ -492,6 +507,7 @@ async fn images_edits(
}
let model = model.ok_or_else(|| AppError::BadRequest("model is required".into()))?;
let request_log_body = Value::Object(log_fields);
let started_at = Instant::now();
let selected = TokenRouter::new(state.db.clone())
.resolve_selected(&model)
@@ -528,6 +544,7 @@ async fn images_edits(
cost,
1,
false,
proxy_log_content_from_request_response(&request_log_body, Some(&payload)),
)
.await;
record_billing(&state, key_id, &model, 0, 0, cost).await;
@@ -544,6 +561,7 @@ async fn images_edits(
Some(&err.to_string()),
1,
false,
proxy_log_content_from_request(&request_log_body),
)
.await;
Err(err)
@@ -665,6 +683,168 @@ fn estimate_request_cost(
}
}
fn push_text_part(parts: &mut Vec<String>, text: &str) {
let normalized = text.split_whitespace().collect::<Vec<_>>().join(" ");
if !normalized.is_empty() {
parts.push(normalized);
}
}
fn collect_text_from_value(value: &Value, parts: &mut Vec<String>) {
match value {
Value::String(text) => push_text_part(parts, text),
Value::Array(items) => {
for item in items {
collect_text_from_value(item, parts);
}
}
Value::Object(object) => {
if let Some(text) = object
.get("text")
.or_else(|| object.get("content"))
.or_else(|| object.get("prompt"))
.or_else(|| object.get("input"))
.and_then(Value::as_str)
{
push_text_part(parts, text);
}
if let Some(parts_value) = object
.get("parts")
.or_else(|| object.get("content"))
.or_else(|| object.get("input"))
{
if !parts_value.is_string() {
collect_text_from_value(parts_value, parts);
}
}
}
_ => {}
}
}
fn trim_chars(value: &str, max_chars: usize) -> String {
let mut output = String::new();
for (index, ch) in value.chars().enumerate() {
if index >= max_chars {
output.push('…');
return output;
}
output.push(ch);
}
output
}
fn join_content_parts(parts: Vec<String>, max_chars: usize) -> Option<String> {
let mut deduped = Vec::new();
for part in parts {
let clean = part.trim();
if clean.is_empty() || deduped.iter().any(|item: &String| item == clean) {
continue;
}
deduped.push(clean.to_string());
}
if deduped.is_empty() {
return None;
}
Some(trim_chars(&deduped.join("\n\n"), max_chars))
}
fn extract_request_content(body: &Value) -> Option<String> {
let mut parts = Vec::new();
if let Some(prompt) = body.get("prompt").and_then(Value::as_str) {
push_text_part(&mut parts, prompt);
}
if let Some(input) = body.get("input") {
collect_text_from_value(input, &mut parts);
}
if let Some(messages) = body.get("messages").and_then(Value::as_array) {
for message in messages {
let role = message
.get("role")
.and_then(Value::as_str)
.unwrap_or_default();
if matches!(role, "assistant" | "tool") {
continue;
}
if let Some(content) = message.get("content") {
collect_text_from_value(content, &mut parts);
}
}
}
if let Some(system) = body.get("system") {
collect_text_from_value(system, &mut parts);
}
if let Some(contents) = body.get("contents").and_then(Value::as_array) {
for content in contents {
collect_text_from_value(content, &mut parts);
}
}
join_content_parts(parts, 4000)
}
fn extract_response_content(response: &Value) -> Option<String> {
let mut parts = Vec::new();
if let Some(output) = response.get("output") {
collect_text_from_value(output, &mut parts);
}
if let Some(content) = response.get("content") {
collect_text_from_value(content, &mut parts);
}
if let Some(text) = response.get("text").and_then(Value::as_str) {
push_text_part(&mut parts, text);
}
if let Some(choices) = response.get("choices").and_then(Value::as_array) {
for choice in choices {
if let Some(message) = choice.get("message") {
collect_text_from_value(message.get("content").unwrap_or(message), &mut parts);
}
if let Some(text) = choice.get("text").and_then(Value::as_str) {
push_text_part(&mut parts, text);
}
}
}
join_content_parts(parts, 4000)
}
fn proxy_log_content_from_request_response(
request: &Value,
response: Option<&Value>,
) -> ProxyLogContent {
let request_content = extract_request_content(request);
let response_content = response.and_then(extract_response_content);
let summary = request_content
.as_deref()
.or(response_content.as_deref())
.map(|text| trim_chars(text, 240));
ProxyLogContent {
summary,
request_content,
response_content,
}
}
fn proxy_log_content_from_request(request: &Value) -> ProxyLogContent {
proxy_log_content_from_request_response(request, None)
}
fn proxy_log_content_with_response(
mut content: ProxyLogContent,
response: Option<&Value>,
) -> ProxyLogContent {
content.response_content = response.and_then(extract_response_content);
content.summary = content
.request_content
.as_deref()
.or(content.response_content.as_deref())
.map(|text| trim_chars(text, 240));
content
}
async fn record_proxy_log_success(
state: &AppState,
key_id: Option<i64>,
@@ -676,14 +856,15 @@ async fn record_proxy_log_success(
cost: f64,
attempts: usize,
is_stream: bool,
content: ProxyLogContent,
) {
if let Err(err) = sqlx::query(
"INSERT INTO proxy_logs (
route_id, channel_id, account_id, downstream_api_key_id,
model_requested, model_actual, status, http_status,
latency_ms, prompt_tokens, completion_tokens, total_tokens, estimated_cost,
retry_count, is_stream
) VALUES (?, ?, ?, ?, ?, ?, 'success', 200, ?, ?, ?, ?, ?, ?, ?)",
retry_count, is_stream, content_summary, request_content, response_content
) VALUES (?, ?, ?, ?, ?, ?, 'success', 200, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(selected.route_id)
.bind(selected.channel_id)
@@ -698,6 +879,9 @@ async fn record_proxy_log_success(
.bind(cost)
.bind(attempts.saturating_sub(1) as i64)
.bind(if is_stream { 1 } else { 0 })
.bind(content.summary.as_deref())
.bind(content.request_content.as_deref())
.bind(content.response_content.as_deref())
.execute(&*state.db)
.await
{
@@ -720,13 +904,15 @@ async fn record_proxy_log_failure(
error_message: Option<&str>,
attempts: usize,
is_stream: bool,
content: ProxyLogContent,
) {
if let Err(err) = sqlx::query(
"INSERT INTO proxy_logs (
route_id, channel_id, account_id, downstream_api_key_id,
model_requested, model_actual, status, http_status,
latency_ms, error_message, retry_count, is_stream
) VALUES (?, ?, ?, ?, ?, ?, 'error', ?, ?, ?, ?, ?)",
latency_ms, error_message, retry_count, is_stream,
content_summary, request_content, response_content
) VALUES (?, ?, ?, ?, ?, ?, 'error', ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(selected.and_then(|item| item.route_id))
.bind(selected.map(|item| item.channel_id))
@@ -739,6 +925,9 @@ async fn record_proxy_log_failure(
.bind(error_message.unwrap_or("proxy request failed"))
.bind(attempts.saturating_sub(1) as i64)
.bind(if is_stream { 1 } else { 0 })
.bind(content.summary.as_deref())
.bind(content.request_content.as_deref())
.bind(content.response_content.as_deref())
.execute(&*state.db)
.await
{
@@ -879,6 +1068,7 @@ async fn execute_messages_with_conductor(
key_id: Option<i64>,
model: String,
request: crate::transformers::canonical::CanonicalRequest,
log_content: ProxyLogContent,
) -> AppResult<Value> {
let started_at = Instant::now();
let conductor = build_conductor(&state);
@@ -935,6 +1125,7 @@ async fn execute_messages_with_conductor(
cost,
attempts,
false,
proxy_log_content_with_response(log_content, Some(&response)),
)
.await;
record_billing(
@@ -965,6 +1156,7 @@ async fn execute_messages_with_conductor(
raw_error_text.as_deref(),
attempts,
false,
log_content,
)
.await;
Err(AppError::UpstreamError(format!(
@@ -981,6 +1173,7 @@ async fn execute_messages_stream_with_conductor(
key_id: Option<i64>,
model: String,
request: crate::transformers::canonical::CanonicalRequest,
log_content: ProxyLogContent,
) -> AppResult<Vec<String>> {
let started_at = Instant::now();
let conductor = build_conductor(&state);
@@ -1055,6 +1248,7 @@ async fn execute_messages_stream_with_conductor(
raw_error_text.as_deref(),
attempts,
true,
log_content,
)
.await;
return Err(AppError::UpstreamError(format!(
@@ -1077,6 +1271,7 @@ async fn execute_messages_stream_with_conductor(
cost,
attempts,
true,
log_content,
)
.await;
record_billing(&billing_state, key_id, &billing_model, 0, 0, cost).await;
@@ -1090,6 +1285,7 @@ async fn execute_chat_completions_with_conductor(
body: Value,
) -> AppResult<Value> {
let started_at = Instant::now();
let request_log_body = body.clone();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
@@ -1145,6 +1341,7 @@ async fn execute_chat_completions_with_conductor(
cost,
attempts,
false,
proxy_log_content_from_request_response(&request_log_body, Some(&response)),
)
.await;
record_billing(
@@ -1175,6 +1372,7 @@ async fn execute_chat_completions_with_conductor(
raw_error_text.as_deref(),
attempts,
false,
proxy_log_content_from_request(&request_log_body),
)
.await;
Err(AppError::UpstreamError(format!(
@@ -1193,6 +1391,7 @@ async fn execute_chat_completions_stream_with_conductor(
body: Value,
) -> AppResult<Vec<String>> {
let started_at = Instant::now();
let request_log_body = body.clone();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
@@ -1273,6 +1472,7 @@ async fn execute_chat_completions_stream_with_conductor(
raw_error_text.as_deref(),
attempts,
true,
proxy_log_content_from_request(&request_log_body),
)
.await;
return Err(AppError::UpstreamError(format!(
@@ -1318,6 +1518,7 @@ async fn execute_chat_completions_stream_with_conductor(
cost,
attempts,
true,
proxy_log_content_from_request(&request_log_body),
)
.await;
record_billing(
@@ -1332,29 +1533,14 @@ async fn execute_chat_completions_stream_with_conductor(
Ok(chunks)
}
fn conductor_result_to_app_result(result: ExecuteResult) -> AppResult<Value> {
match result {
ExecuteResult::Success { response, .. } => Ok(response),
ExecuteResult::Failure {
reason,
status,
raw_error_text,
attempts,
..
} => Err(AppError::UpstreamError(format!(
"Proxy conductor failed after {attempts} attempts ({reason:?}, status: {:?}): {}",
status,
raw_error_text.unwrap_or_else(|| "no upstream error text".to_string())
))),
}
}
async fn execute_responses_with_conductor(
state: AppState,
key_id: Option<i64>,
model: String,
body: Value,
) -> AppResult<Value> {
let started_at = Instant::now();
let request_log_body = body.clone();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
@@ -1392,15 +1578,62 @@ async fn execute_responses_with_conductor(
})
.await;
let response = conductor_result_to_app_result(result)?;
let (selected, response, attempts) = match result {
ExecuteResult::Success {
selected,
response,
attempts,
} => (selected, response, attempts),
ExecuteResult::Failure {
reason,
selected,
status,
raw_error_text,
attempts,
} => {
record_proxy_log_failure(
&billing_state,
key_id,
selected.as_ref(),
&billing_model,
status,
Some(elapsed_ms_since(started_at)),
raw_error_text.as_deref(),
attempts,
false,
proxy_log_content_from_request(&request_log_body),
)
.await;
return Err(AppError::UpstreamError(format!(
"Proxy conductor failed after {attempts} attempts ({reason:?}, status: {:?}): {}",
status,
raw_error_text.unwrap_or_else(|| "no upstream error text".to_string())
)));
}
};
let (prompt, completion) = extract_usage(&response);
let cost = estimate_request_cost(&selected, prompt, completion);
record_proxy_log_success(
&billing_state,
key_id,
&selected,
&billing_model,
Some(elapsed_ms_since(started_at)),
prompt,
completion,
cost,
attempts,
false,
proxy_log_content_from_request_response(&request_log_body, Some(&response)),
)
.await;
record_billing(
&billing_state,
key_id,
&billing_model,
prompt,
completion,
0.0,
cost,
)
.await;
Ok(response)
@@ -1412,6 +1645,8 @@ async fn execute_responses_stream_with_conductor(
model: String,
body: Value,
) -> AppResult<Vec<String>> {
let started_at = Instant::now();
let request_log_body = body.clone();
let conductor = build_conductor(&state);
let billing_state = state.clone();
let billing_model = model.clone();
@@ -1447,19 +1682,57 @@ async fn execute_responses_stream_with_conductor(
})
.await;
let chunks = match conductor_result_to_app_result(result)? {
Value::Array(chunks) => chunks
.into_iter()
.map(|chunk| {
chunk.as_str().map(str::to_string).ok_or_else(|| {
AppError::UpstreamError("Invalid streaming response from conductor".into())
})
})
.collect::<AppResult<Vec<String>>>()?,
_ => {
return Err(AppError::UpstreamError(
"Invalid streaming response from conductor".into(),
))
let (selected, attempts, chunks) = match result {
ExecuteResult::Success {
selected,
response,
attempts,
} => match response {
Value::Array(chunks) => (
selected,
attempts,
chunks
.into_iter()
.map(|chunk| {
chunk.as_str().map(str::to_string).ok_or_else(|| {
AppError::UpstreamError(
"Invalid streaming response from conductor".into(),
)
})
})
.collect::<AppResult<Vec<String>>>()?,
),
_ => {
return Err(AppError::UpstreamError(
"Invalid streaming response from conductor".into(),
))
}
},
ExecuteResult::Failure {
reason,
selected,
status,
raw_error_text,
attempts,
} => {
record_proxy_log_failure(
&billing_state,
key_id,
selected.as_ref(),
&billing_model,
status,
Some(elapsed_ms_since(started_at)),
raw_error_text.as_deref(),
attempts,
true,
proxy_log_content_from_request(&request_log_body),
)
.await;
return Err(AppError::UpstreamError(format!(
"Proxy conductor failed after {attempts} attempts ({reason:?}, status: {:?}): {}",
status,
raw_error_text.unwrap_or_else(|| "no upstream error text".to_string())
)));
}
};
@@ -1485,13 +1758,28 @@ async fn execute_responses_stream_with_conductor(
}
})
.unwrap_or((0, 0));
let cost = estimate_request_cost(&selected, prompt, completion);
record_proxy_log_success(
&billing_state,
key_id,
&selected,
&billing_model,
Some(elapsed_ms_since(started_at)),
prompt,
completion,
cost,
attempts,
true,
proxy_log_content_from_request(&request_log_body),
)
.await;
record_billing(
&billing_state,
key_id,
&billing_model,
prompt,
completion,
0.0,
cost,
)
.await;
Ok(chunks)