llm: switch Skald to agent-loop Model clients; drop llm-client (phase 1, D13)
The LLM call path now runs on the agent-loop crate's clients and trait: - core-api: BuiltLlmClient.client is Arc<dyn agent_loop::model::Model>; chatbot.rs (ChatbotClient + wire types) deleted; APP_NAME re-exported from agent-loop - providers (openai/anthropic/ollama/openrouter/requesty/declared) build OpenAiModel/AnthropicModel/OllamaModel with the model's wire id - LoggingModel decorator (llm/logging.rs) replaces LoggingChatbotClient; per-request correlation (session/stack/user) travels in the new ModelRequest.log field, never sent to providers - llm_call/llm_loop/compactor speak Model::complete + ModelResponse; retriability via Model::is_retriable (structured status, B6 rule now the crate's default); payload persistence reads RawMeta off ModelResponse/ModelError - crates/llm-client and skald-core/src/chatbot deleted Full workspace test suite green (incl. 162 skald-core + 32 agent-loop).
This commit is contained in:
Generated
+2
-17
@@ -599,6 +599,7 @@ checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
|
||||
name = "core-api"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"agent-loop",
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
"axum",
|
||||
@@ -2198,21 +2199,6 @@ version = "0.8.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
|
||||
|
||||
[[package]]
|
||||
name = "llm-client"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
"core-api",
|
||||
"futures-util",
|
||||
"reqwest 0.13.4",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lock_api"
|
||||
version = "0.4.14"
|
||||
@@ -4198,7 +4184,6 @@ dependencies = [
|
||||
"futures",
|
||||
"honcho-client",
|
||||
"indexmap 2.14.0",
|
||||
"llm-client",
|
||||
"mcp-client",
|
||||
"notify",
|
||||
"plugin-comfyui",
|
||||
@@ -4232,6 +4217,7 @@ name = "skald-core"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"agent-loop",
|
||||
"anyhow",
|
||||
"argon2",
|
||||
"async-trait",
|
||||
@@ -4248,7 +4234,6 @@ dependencies = [
|
||||
"indexmap 2.14.0",
|
||||
"libc",
|
||||
"libsqlite3-sys",
|
||||
"llm-client",
|
||||
"mcp-client",
|
||||
"notify",
|
||||
"os_info",
|
||||
|
||||
@@ -5,7 +5,6 @@ members = [
|
||||
"crates/skald-core",
|
||||
"crates/skald-setup",
|
||||
"crates/honcho-client",
|
||||
"crates/llm-client",
|
||||
"crates/core-api",
|
||||
"crates/mcp-client",
|
||||
"crates/plugin-tailscale-remote",
|
||||
@@ -74,7 +73,6 @@ tracing-appender = "0.2"
|
||||
chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
|
||||
notify = "8"
|
||||
honcho-client = { path = "crates/honcho-client" }
|
||||
llm-client = { path = "crates/llm-client" }
|
||||
core-api = { path = "crates/core-api" }
|
||||
mcp-client = { path = "crates/mcp-client" }
|
||||
plugin-tailscale-remote = { path = "crates/plugin-tailscale-remote" }
|
||||
|
||||
@@ -156,6 +156,7 @@ pub(crate) async fn run(
|
||||
conversation: params.conversation.clone(),
|
||||
frame,
|
||||
extras: handle.info.extras.clone(),
|
||||
log: None,
|
||||
};
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
|
||||
@@ -198,8 +198,12 @@ pub struct ModelRequest {
|
||||
pub conversation: ConversationId,
|
||||
pub frame: FrameId,
|
||||
/// Host free-form per-request extras (e.g. reasoning knobs resolved for
|
||||
/// this model). Merged last by the shipped clients.
|
||||
/// this model). Merged last by the shipped clients INTO THE REQUEST BODY.
|
||||
pub extras: Value,
|
||||
/// Host logging/telemetry correlation (session ids, user id, …).
|
||||
/// **Never** merged into the request body by the shipped clients — it
|
||||
/// exists for host decorators (e.g. a `LoggingModel`) only.
|
||||
pub log: Option<Value>,
|
||||
}
|
||||
|
||||
// ── Model ────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -4,6 +4,7 @@ version = "0.1.0"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
agent-loop = { path = "../agent-loop" }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
tokio = { version = "1", features = ["sync", "macros"] }
|
||||
|
||||
@@ -1,193 +0,0 @@
|
||||
use async_trait::async_trait;
|
||||
use serde_json::Value;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
/// A single message in a conversation.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Message {
|
||||
pub role: Role,
|
||||
pub content: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Role {
|
||||
System,
|
||||
User,
|
||||
Assistant,
|
||||
}
|
||||
|
||||
impl Message {
|
||||
pub fn system(content: impl Into<String>) -> Self {
|
||||
Self { role: Role::System, content: content.into() }
|
||||
}
|
||||
|
||||
pub fn user(content: impl Into<String>) -> Self {
|
||||
Self { role: Role::User, content: content.into() }
|
||||
}
|
||||
|
||||
pub fn assistant(content: impl Into<String>) -> Self {
|
||||
Self { role: Role::Assistant, content: content.into() }
|
||||
}
|
||||
}
|
||||
|
||||
/// Options for a single chat completion request.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ChatOptions {
|
||||
pub model: String,
|
||||
pub max_tokens: Option<u32>,
|
||||
pub temperature: Option<f32>,
|
||||
/// Session/stack IDs for request logging. Set by the LLM loop; ignored by
|
||||
/// providers — only the logging wrapper reads them.
|
||||
pub session_id: Option<i64>,
|
||||
pub stack_id: Option<i64>,
|
||||
/// The authenticated user driving this request. Correlates the metadata row
|
||||
/// in `system.db` with the payload in `{userid}.db`. Logging-only.
|
||||
pub user_id: Option<String>,
|
||||
/// UUID correlating the metadata row (`llm_requests`) with the payload row
|
||||
/// (`llm_request_payloads`). Generated by the LLM loop before the call.
|
||||
/// Logging-only.
|
||||
pub request_id: Option<String>,
|
||||
}
|
||||
|
||||
/// Raw HTTP metadata captured during a provider call.
|
||||
/// Sensitive header values (api_key) are redacted before storage.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct LlmRawMeta {
|
||||
pub request_headers: Option<Value>,
|
||||
pub request_body: Option<Value>,
|
||||
pub response_headers: Option<Value>,
|
||||
pub response_body: Option<Value>,
|
||||
}
|
||||
|
||||
/// The response from a chat completion (text only).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ChatResponse {
|
||||
pub content: String,
|
||||
pub input_tokens: Option<u32>,
|
||||
pub output_tokens: Option<u32>,
|
||||
/// True when the model stopped due to hitting the token limit.
|
||||
pub truncated: bool,
|
||||
/// Chain-of-thought produced by reasoning models (e.g. DeepSeek thinking mode).
|
||||
/// Must be echoed back in the assistant message on subsequent turns.
|
||||
pub reasoning_content: Option<String>,
|
||||
/// Tokens served from the provider's prompt cache (Anthropic: cache_read_input_tokens,
|
||||
/// OpenAI: prompt_tokens_details.cached_tokens). None when the provider does not
|
||||
/// report cache metrics.
|
||||
pub cache_read_tokens: Option<u32>,
|
||||
/// Tokens written into the provider's prompt cache (Anthropic only:
|
||||
/// cache_creation_input_tokens). None for providers that do not expose this.
|
||||
pub cache_creation_tokens: Option<u32>,
|
||||
/// Cost of the request in USD, when the provider reports it (OpenRouter
|
||||
/// returns it under `usage.cost`). None for providers that do not bill
|
||||
/// per-request or do not expose the figure.
|
||||
pub cost: Option<f64>,
|
||||
}
|
||||
|
||||
/// A single tool call requested by the LLM.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ToolCall {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub arguments: Value,
|
||||
}
|
||||
|
||||
/// An incremental piece of a streaming completion, pushed by providers that
|
||||
/// support SSE streaming. Purely best-effort UI feedback: the final `LlmTurn`
|
||||
/// remains the authoritative result.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum StreamDelta {
|
||||
/// Visible answer text.
|
||||
Text(String),
|
||||
/// Chain-of-thought / reasoning tokens (thinking models).
|
||||
Reasoning(String),
|
||||
}
|
||||
|
||||
/// Result of one LLM turn when tools are available.
|
||||
#[derive(Debug)]
|
||||
pub enum LlmTurn {
|
||||
Message(ChatResponse),
|
||||
ToolCalls {
|
||||
content: String,
|
||||
calls: Vec<ToolCall>,
|
||||
input_tokens: Option<u32>,
|
||||
output_tokens: Option<u32>,
|
||||
reasoning_content: Option<String>,
|
||||
cache_read_tokens: Option<u32>,
|
||||
cache_creation_tokens: Option<u32>,
|
||||
cost: Option<f64>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Stateless LLM client. Implementations hold only connection config (base URL,
|
||||
/// API key). No memory, no database, no session state.
|
||||
#[async_trait]
|
||||
pub trait ChatbotClient: Send + Sync {
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse>;
|
||||
|
||||
/// Extracts the request cost in USD from a provider's raw JSON response,
|
||||
/// when the provider reports it. OpenRouter (and other OpenAI-compatible
|
||||
/// gateways) return it under `usage.cost`; the default reads that path and
|
||||
/// yields None when absent. Providers with a different shape override this.
|
||||
fn extract_cost(&self, response: &Value) -> Option<f64> {
|
||||
response["usage"]["cost"].as_f64()
|
||||
}
|
||||
|
||||
/// Chat with tool support. Default implementation ignores tools and falls
|
||||
/// back to `chat()`.
|
||||
async fn chat_with_tools(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<LlmTurn> {
|
||||
let simple: Vec<Message> = messages
|
||||
.iter()
|
||||
.filter_map(|m| {
|
||||
let role = m["role"].as_str()?;
|
||||
let content = m["content"].as_str().unwrap_or("").to_string();
|
||||
match role {
|
||||
"system" => Some(Message::system(content)),
|
||||
"user" => Some(Message::user(content)),
|
||||
"assistant" => Some(Message::assistant(content)),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
let _ = tools;
|
||||
let resp = self.chat(&simple, options).await?;
|
||||
Ok(LlmTurn::Message(resp))
|
||||
}
|
||||
|
||||
/// Like `chat_with_tools` but also returns raw HTTP metadata for logging.
|
||||
/// Providers that make real HTTP calls should override this.
|
||||
async fn chat_with_tools_raw(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
self.chat_with_tools(messages, tools, options).await.map(|t| (t, None))
|
||||
}
|
||||
|
||||
/// Like `chat_with_tools_raw`, but the provider may push incremental
|
||||
/// [`StreamDelta`]s into `delta_tx` as tokens arrive (SSE streaming).
|
||||
/// Senders should use `try_send` and drop deltas when the channel is full —
|
||||
/// streaming is best-effort UI feedback and must never backpressure the
|
||||
/// HTTP read. The returned `LlmTurn` is always the complete, authoritative
|
||||
/// result. The default ignores the channel and falls back to the buffered
|
||||
/// call, so providers without streaming behave exactly as before.
|
||||
async fn chat_with_tools_raw_streaming(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: mpsc::Sender<StreamDelta>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let _ = delta_tx;
|
||||
self.chat_with_tools_raw(messages, tools, options).await
|
||||
}
|
||||
}
|
||||
@@ -1,11 +1,12 @@
|
||||
/// Application name, sent as `X-Title` HTTP header to LLM/image/audio providers.
|
||||
pub const APP_NAME: &str = "Skald";
|
||||
/// Lives in `agent-loop` (the LLM clients' home, blueprint D13); re-exported here
|
||||
/// so existing users don't change.
|
||||
pub use agent_loop::APP_NAME;
|
||||
|
||||
pub mod approval;
|
||||
pub mod bus;
|
||||
pub mod config_api;
|
||||
pub mod system_bus;
|
||||
pub mod chatbot;
|
||||
pub mod chat_hub;
|
||||
pub mod command;
|
||||
pub mod events;
|
||||
|
||||
@@ -3,7 +3,8 @@ use std::sync::Arc;
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
|
||||
use crate::chatbot::ChatbotClient;
|
||||
use agent_loop::model::Model;
|
||||
|
||||
use crate::image_generate::{ImageGenerate, ImageGenerateModelRecord};
|
||||
use crate::tts::{TextToSpeech, TtsModelRecord, RemoteTtsModelInfo};
|
||||
use crate::transcribe::{Transcribe, TranscribeModelRecord, RemoteTranscribeModelInfo};
|
||||
@@ -139,7 +140,8 @@ pub struct ProviderField {
|
||||
// ── BuiltLlmClient ────────────────────────────────────────────────────────────
|
||||
|
||||
pub struct BuiltLlmClient {
|
||||
pub client: Arc<dyn ChatbotClient>,
|
||||
/// A stateless `agent_loop` model client (blueprint D13).
|
||||
pub client: Arc<dyn Model>,
|
||||
pub prompt_cache: bool,
|
||||
}
|
||||
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
[package]
|
||||
name = "llm-client"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
core-api = { path = "../core-api" }
|
||||
reqwest = { version = "0.13", default-features = false, features = ["rustls-no-provider", "charset", "http2", "system-proxy", "json", "stream"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
async-trait = "0.1"
|
||||
anyhow = "1"
|
||||
tracing = "0.1"
|
||||
tokio = { version = "1", features = ["sync"] }
|
||||
futures-util = "0.3"
|
||||
@@ -1,811 +0,0 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use futures_util::StreamExt;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, info, trace, warn};
|
||||
|
||||
use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, error_response_body, headers_to_json, redact_key};
|
||||
|
||||
const DEFAULT_BASE_URL: &str = "https://api.anthropic.com";
|
||||
const ANTHROPIC_VERSION: &str = "2023-06-01";
|
||||
|
||||
pub struct AnthropicClient {
|
||||
base_url: String,
|
||||
api_key: String,
|
||||
/// Extra top-level request-body keys merged into every request (e.g. the
|
||||
/// `thinking` config for extended reasoning). See `apply_extra`.
|
||||
extra_body: Option<Value>,
|
||||
http: reqwest::Client,
|
||||
}
|
||||
|
||||
impl AnthropicClient {
|
||||
pub fn new(api_key: impl Into<String>) -> Self {
|
||||
Self::with_base_url(DEFAULT_BASE_URL, api_key)
|
||||
}
|
||||
|
||||
pub fn with_base_url(base_url: impl Into<String>, api_key: impl Into<String>) -> Self {
|
||||
Self {
|
||||
base_url: base_url.into(),
|
||||
api_key: api_key.into(),
|
||||
extra_body: None,
|
||||
http: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Like `new` but with extra request-body keys (e.g. `{"thinking": {...}}`).
|
||||
pub fn with_extra_body(api_key: impl Into<String>, extra_body: Option<Value>) -> Self {
|
||||
Self {
|
||||
base_url: DEFAULT_BASE_URL.to_string(),
|
||||
api_key: api_key.into(),
|
||||
extra_body,
|
||||
http: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Merges `extra_body` into `body` and enforces Anthropic's extended-thinking
|
||||
/// constraints: when `thinking` is enabled, `temperature` is not allowed and
|
||||
/// `max_tokens` must be strictly greater than `budget_tokens`.
|
||||
fn apply_extra(&self, body: &mut Value) {
|
||||
let Some(extra) = self.extra_body.as_ref().and_then(|v| v.as_object()) else { return };
|
||||
let Some(obj) = body.as_object_mut() else { return };
|
||||
for (k, v) in extra {
|
||||
obj.insert(k.clone(), v.clone());
|
||||
}
|
||||
if obj.get("thinking").map(|t| t["type"] == json!("enabled")).unwrap_or(false) {
|
||||
obj.remove("temperature");
|
||||
let budget = obj["thinking"]["budget_tokens"].as_i64().unwrap_or(0);
|
||||
let cur_max = obj.get("max_tokens").and_then(|v| v.as_i64()).unwrap_or(4096);
|
||||
if budget > 0 && cur_max <= budget {
|
||||
obj.insert("max_tokens".to_string(), json!(budget + 4096));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Converts OpenAI-format tool definitions to Anthropic format.
|
||||
/// OpenAI: { "type": "function", "function": { "name", "description", "parameters" } }
|
||||
/// Anthropic: { "name", "description", "input_schema" }
|
||||
///
|
||||
/// DTL (tool search): a top-level `defer_loading: true` on the OpenAI tool
|
||||
/// object is carried through to Anthropic's native `defer_loading` field. When
|
||||
/// any tool is deferred, the cache breakpoint is placed on the last
|
||||
/// **non-deferred** tool — a deferred tool cannot also carry `cache_control`
|
||||
/// (the API 400s), and at least one tool must stay non-deferred anyway.
|
||||
fn convert_tools(tools: &[Value]) -> Vec<Value> {
|
||||
let has_deferred = tools.iter().any(|t| t["defer_loading"].as_bool() == Some(true));
|
||||
let mut out: Vec<Value> = tools
|
||||
.iter()
|
||||
.filter_map(|t| {
|
||||
let func = &t["function"];
|
||||
let name = func["name"].as_str()?;
|
||||
let mut tool = json!({
|
||||
"name": name,
|
||||
"description": func["description"].as_str().unwrap_or(""),
|
||||
"input_schema": func["parameters"],
|
||||
});
|
||||
if t["defer_loading"].as_bool() == Some(true) {
|
||||
tool["defer_loading"] = json!(true);
|
||||
}
|
||||
Some(tool)
|
||||
})
|
||||
.collect();
|
||||
if has_deferred {
|
||||
if let Some(t) = out.iter_mut().rev().find(|t| t["defer_loading"].as_bool() != Some(true)) {
|
||||
t["cache_control"] = json!({ "type": "ephemeral" });
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Converts OpenAI-format message array to Anthropic format.
|
||||
///
|
||||
/// Key differences:
|
||||
/// - System messages are skipped (extracted separately).
|
||||
/// - Assistant messages with `tool_calls` become content arrays with `tool_use` blocks.
|
||||
/// - `tool` role messages are grouped into `user` messages with `tool_result` blocks.
|
||||
fn convert_messages(messages: &[Value]) -> Vec<Value> {
|
||||
let mut out: Vec<Value> = Vec::new();
|
||||
let mut i = 0;
|
||||
|
||||
while i < messages.len() {
|
||||
let msg = &messages[i];
|
||||
let role = msg["role"].as_str().unwrap_or("");
|
||||
|
||||
match role {
|
||||
"system" => { i += 1; }
|
||||
|
||||
"user" => {
|
||||
out.push(json!({
|
||||
"role": "user",
|
||||
"content": convert_user_content(&msg["content"]),
|
||||
}));
|
||||
i += 1;
|
||||
}
|
||||
|
||||
"assistant" => {
|
||||
if let Some(tool_calls) = msg["tool_calls"].as_array() {
|
||||
let mut content: Vec<Value> = Vec::new();
|
||||
|
||||
let text = msg["content"].as_str().unwrap_or("");
|
||||
if !text.is_empty() {
|
||||
content.push(json!({ "type": "text", "text": text }));
|
||||
}
|
||||
|
||||
for tc in tool_calls {
|
||||
let id = tc["id"].as_str().unwrap_or("");
|
||||
let name = tc["function"]["name"].as_str().unwrap_or("");
|
||||
let args_str = tc["function"]["arguments"].as_str().unwrap_or("{}");
|
||||
let input: Value = serde_json::from_str(args_str)
|
||||
.unwrap_or(Value::Object(Default::default()));
|
||||
|
||||
content.push(json!({
|
||||
"type": "tool_use",
|
||||
"id": id,
|
||||
"name": name,
|
||||
"input": input,
|
||||
}));
|
||||
}
|
||||
|
||||
out.push(json!({ "role": "assistant", "content": content }));
|
||||
} else {
|
||||
out.push(json!({
|
||||
"role": "assistant",
|
||||
"content": msg["content"].as_str().unwrap_or(""),
|
||||
}));
|
||||
}
|
||||
i += 1;
|
||||
}
|
||||
|
||||
"tool" => {
|
||||
// Group all consecutive tool-result messages into a single user message.
|
||||
let mut results: Vec<Value> = Vec::new();
|
||||
while i < messages.len() && messages[i]["role"].as_str() == Some("tool") {
|
||||
let tm = &messages[i];
|
||||
// DTL (custom tool search): a tool result carrying
|
||||
// `_tool_references` (set by the message builder on an
|
||||
// `activate_tools` result in AnthropicToolReference mode) becomes a
|
||||
// `content` array of `tool_reference` blocks, which the API expands
|
||||
// into the deferred tools' full definitions. Empty/absent → the
|
||||
// normal text result.
|
||||
let content: Value = match tm["_tool_references"].as_array() {
|
||||
Some(refs) if !refs.is_empty() => Value::Array(
|
||||
refs.iter()
|
||||
.filter_map(|r| r.as_str())
|
||||
.map(|name| json!({ "type": "tool_reference", "tool_name": name }))
|
||||
.collect(),
|
||||
),
|
||||
_ => Value::String(tm["content"].as_str().unwrap_or("").to_string()),
|
||||
};
|
||||
results.push(json!({
|
||||
"type": "tool_result",
|
||||
"tool_use_id": tm["tool_call_id"].as_str().unwrap_or(""),
|
||||
"content": content,
|
||||
}));
|
||||
i += 1;
|
||||
}
|
||||
out.push(json!({ "role": "user", "content": results }));
|
||||
}
|
||||
|
||||
_ => { i += 1; }
|
||||
}
|
||||
}
|
||||
|
||||
out
|
||||
}
|
||||
|
||||
/// Assembles the `/v1/messages` request body shared by the buffered and the
|
||||
/// streaming path (the caller adds `stream` on top).
|
||||
fn tools_body(&self, system: Option<Value>, messages: Vec<Value>, tools: Vec<Value>, options: &ChatOptions) -> Value {
|
||||
let max_tokens = options.max_tokens.unwrap_or(4096);
|
||||
let mut body = json!({
|
||||
"model": options.model,
|
||||
"max_tokens": max_tokens,
|
||||
"messages": messages,
|
||||
"tools": tools,
|
||||
});
|
||||
|
||||
if let Some(sys) = system { body["system"] = sys; }
|
||||
if let Some(t) = options.temperature { body["temperature"] = t.into(); }
|
||||
self.apply_extra(&mut body);
|
||||
body
|
||||
}
|
||||
|
||||
/// Collects ALL system-role messages (main prompt, mid-conversation summary,
|
||||
/// tail_reminder) into the single `system` parameter the Anthropic API accepts.
|
||||
///
|
||||
/// Returns a plain string in the common case. When any system message carries
|
||||
/// **structured** content (a text-block array, e.g. the static prompt tagged
|
||||
/// with `cache_control` when prompt caching is on), it returns the array form
|
||||
/// instead so the cache breakpoint survives into `system`. String-content
|
||||
/// messages become plain text blocks (no cache_control).
|
||||
fn merged_system(messages: &[Value]) -> Option<Value> {
|
||||
let sys: Vec<&Value> = messages
|
||||
.iter()
|
||||
.filter(|m| m["role"].as_str() == Some("system"))
|
||||
.collect();
|
||||
if sys.is_empty() { return None; }
|
||||
|
||||
if !sys.iter().any(|m| m["content"].is_array()) {
|
||||
let parts: Vec<&str> = sys.iter().filter_map(|m| m["content"].as_str()).collect();
|
||||
return if parts.is_empty() { None } else { Some(Value::String(parts.join("\n\n---\n\n"))) };
|
||||
}
|
||||
|
||||
let mut blocks: Vec<Value> = Vec::new();
|
||||
for m in &sys {
|
||||
match &m["content"] {
|
||||
Value::String(s) if !s.is_empty() => blocks.push(json!({ "type": "text", "text": s })),
|
||||
Value::Array(arr) => {
|
||||
for b in arr {
|
||||
if b["type"].as_str() == Some("text") {
|
||||
blocks.push(b.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
if blocks.is_empty() { None } else { Some(Value::Array(blocks)) }
|
||||
}
|
||||
|
||||
fn url(&self) -> String {
|
||||
format!("{}/v1/messages", self.base_url.trim_end_matches('/'))
|
||||
}
|
||||
|
||||
fn logged_headers(&self) -> Value {
|
||||
json!({
|
||||
"x-api-key": redact_key(&self.api_key),
|
||||
"anthropic-version": ANTHROPIC_VERSION,
|
||||
"content-type": "application/json",
|
||||
})
|
||||
}
|
||||
|
||||
/// Sends the request and returns the raw response **without** `error_for_status`,
|
||||
/// so the tool-calling paths can read the error body and attach the request
|
||||
/// payload to the `LlmError` (a `reqwest` status error discards the body). The
|
||||
/// plain `chat` path keeps its own `error_for_status`.
|
||||
async fn send_request(&self, body: &Value) -> reqwest::Result<reqwest::Response> {
|
||||
self.http
|
||||
.post(self.url())
|
||||
.header("x-api-key", &self.api_key)
|
||||
.header("anthropic-version", ANTHROPIC_VERSION)
|
||||
.header("X-Title", core_api::APP_NAME)
|
||||
.json(body)
|
||||
.send()
|
||||
.await
|
||||
}
|
||||
|
||||
/// Joined `thinking` blocks of a content array, if any (extended thinking).
|
||||
fn reasoning_of(content_blocks: &[Value]) -> Option<String> {
|
||||
let parts: Vec<&str> = content_blocks
|
||||
.iter()
|
||||
.filter(|b| b["type"].as_str() == Some("thinking"))
|
||||
.filter_map(|b| b["thinking"].as_str())
|
||||
.collect();
|
||||
if parts.is_empty() { None } else { Some(parts.join("\n")) }
|
||||
}
|
||||
|
||||
/// SSE streaming path behind `chat_with_tools_raw_streaming`. Anthropic
|
||||
/// streams typed events (`message_start` / `content_block_*` /
|
||||
/// `message_delta` / `message_stop`); text and thinking deltas are
|
||||
/// forwarded to `delta_tx` best-effort while the blocks are accumulated
|
||||
/// into the same `LlmTurn` the buffered path returns.
|
||||
async fn stream_chat(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: &mpsc::Sender<StreamDelta>,
|
||||
emitted: &mut bool,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let system = Self::merged_system(messages);
|
||||
let anthropic_messages = Self::convert_messages(messages);
|
||||
let anthropic_tools = Self::convert_tools(tools);
|
||||
let mut body = self.tools_body(system, anthropic_messages, anthropic_tools, options);
|
||||
body["stream"] = json!(true);
|
||||
|
||||
debug!(model = %options.model, tools = tools.len(), "anthropic: sending streaming chat_with_tools request");
|
||||
trace!(body = %body, "anthropic: streaming chat_with_tools request body");
|
||||
|
||||
let request_body = body.clone();
|
||||
let request_headers = self.logged_headers();
|
||||
|
||||
let http_resp = self.send_request(&body).await?;
|
||||
let response_headers = headers_to_json(http_resp.headers());
|
||||
let status = http_resp.status();
|
||||
if !status.is_success() {
|
||||
let resp_text = http_resp.text().await?;
|
||||
return Err(crate::LlmError {
|
||||
status: Some(status.as_u16()),
|
||||
message: format!(
|
||||
"anthropic: HTTP {status} from {url}\nbody: {resp_text}",
|
||||
url = self.url(),
|
||||
),
|
||||
raw_meta: Some(LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(error_response_body(resp_text)),
|
||||
}),
|
||||
}.into());
|
||||
}
|
||||
|
||||
/// One content block being accumulated by index.
|
||||
#[derive(Default)]
|
||||
struct Block {
|
||||
kind: String, // "text" | "thinking" | "tool_use"
|
||||
buf: String, // text/thinking content or input_json fragments
|
||||
id: String,
|
||||
name: String,
|
||||
}
|
||||
|
||||
let mut blocks: BTreeMap<u64, Block> = BTreeMap::new();
|
||||
let mut stop_reason: Option<String> = None;
|
||||
let mut usage = json!({});
|
||||
let mut sse = SseDecoder::new();
|
||||
let mut byte_stream = http_resp.bytes_stream();
|
||||
|
||||
let mut handle_payload = |payload: &str, emitted: &mut bool| -> anyhow::Result<()> {
|
||||
let Ok(v) = serde_json::from_str::<Value>(payload) else { return Ok(()) };
|
||||
match v["type"].as_str().unwrap_or("") {
|
||||
"message_start" => {
|
||||
if let Some(u) = v["message"]["usage"].as_object() {
|
||||
for (k, val) in u { usage[k.clone()] = val.clone(); }
|
||||
}
|
||||
}
|
||||
"content_block_start" => {
|
||||
let idx = v["index"].as_u64().unwrap_or(0);
|
||||
let cb = &v["content_block"];
|
||||
let block = blocks.entry(idx).or_default();
|
||||
block.kind = cb["type"].as_str().unwrap_or("").to_string();
|
||||
block.id = cb["id"].as_str().unwrap_or("").to_string();
|
||||
block.name = cb["name"].as_str().unwrap_or("").to_string();
|
||||
}
|
||||
"content_block_delta" => {
|
||||
let idx = v["index"].as_u64().unwrap_or(0);
|
||||
let delta = &v["delta"];
|
||||
match delta["type"].as_str().unwrap_or("") {
|
||||
"text_delta" => {
|
||||
if let Some(t) = delta["text"].as_str().filter(|t| !t.is_empty()) {
|
||||
blocks.entry(idx).or_default().buf.push_str(t);
|
||||
*emitted = true;
|
||||
let _ = delta_tx.try_send(StreamDelta::Text(t.to_string()));
|
||||
}
|
||||
}
|
||||
"thinking_delta" => {
|
||||
if let Some(t) = delta["thinking"].as_str().filter(|t| !t.is_empty()) {
|
||||
blocks.entry(idx).or_default().buf.push_str(t);
|
||||
*emitted = true;
|
||||
let _ = delta_tx.try_send(StreamDelta::Reasoning(t.to_string()));
|
||||
}
|
||||
}
|
||||
"input_json_delta" => {
|
||||
if let Some(j) = delta["partial_json"].as_str() {
|
||||
blocks.entry(idx).or_default().buf.push_str(j);
|
||||
}
|
||||
}
|
||||
// signature_delta and unknown deltas carry no displayable text.
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
"message_delta" => {
|
||||
if let Some(sr) = v["delta"]["stop_reason"].as_str() {
|
||||
stop_reason = Some(sr.to_string());
|
||||
}
|
||||
if let Some(u) = v["usage"].as_object() {
|
||||
for (k, val) in u { usage[k.clone()] = val.clone(); }
|
||||
}
|
||||
}
|
||||
"error" => {
|
||||
return Err(anyhow::anyhow!("anthropic: stream error event: {payload}"));
|
||||
}
|
||||
// content_block_stop / message_stop / ping: nothing to accumulate.
|
||||
_ => {}
|
||||
}
|
||||
Ok(())
|
||||
};
|
||||
|
||||
while let Some(chunk) = byte_stream.next().await {
|
||||
let chunk = chunk?;
|
||||
for payload in sse.feed(&chunk) {
|
||||
handle_payload(&payload, emitted)?;
|
||||
}
|
||||
}
|
||||
for payload in sse.finish() {
|
||||
handle_payload(&payload, emitted)?;
|
||||
}
|
||||
|
||||
let stop = stop_reason.as_deref().unwrap_or("");
|
||||
let input_tokens = usage["input_tokens"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = usage["output_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_read_tokens = usage["cache_read_input_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_creation_tokens = usage["cache_creation_input_tokens"].as_u64().map(|n| n as u32);
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, stop_reason = stop, "anthropic: streaming response completed");
|
||||
if stop == "max_tokens" {
|
||||
warn!(model = %options.model, ?output_tokens, "anthropic: response truncated (max_tokens reached)");
|
||||
}
|
||||
|
||||
let text_of = |kind: &str| -> String {
|
||||
blocks.values()
|
||||
.filter(|b| b.kind == kind)
|
||||
.map(|b| b.buf.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n")
|
||||
};
|
||||
let reasoning = text_of("thinking");
|
||||
let reasoning_content = if reasoning.is_empty() { None } else { Some(reasoning) };
|
||||
let tool_blocks: Vec<&Block> = blocks.values().filter(|b| b.kind == "tool_use").collect();
|
||||
|
||||
let turn = if !tool_blocks.is_empty() {
|
||||
let calls = tool_blocks
|
||||
.iter()
|
||||
.map(|b| ToolCall {
|
||||
id: b.id.clone(),
|
||||
name: b.name.clone(),
|
||||
arguments: serde_json::from_str(&b.buf).unwrap_or(Value::Object(Default::default())),
|
||||
})
|
||||
.collect();
|
||||
LlmTurn::ToolCalls { content: text_of("text"), calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens, cost: None }
|
||||
} else {
|
||||
let truncated = stop == "max_tokens";
|
||||
LlmTurn::Message(ChatResponse {
|
||||
content: text_of("text"), input_tokens, output_tokens, truncated,
|
||||
reasoning_content, cache_read_tokens, cache_creation_tokens, cost: None,
|
||||
})
|
||||
};
|
||||
|
||||
// Buffered-shaped response body for the payload log.
|
||||
let content_log: Vec<Value> = blocks.values().map(|b| match b.kind.as_str() {
|
||||
"tool_use" => json!({"type": "tool_use", "id": b.id, "name": b.name, "input": serde_json::from_str::<Value>(&b.buf).unwrap_or(json!({}))}),
|
||||
"thinking" => json!({"type": "thinking", "thinking": b.buf}),
|
||||
_ => json!({"type": "text", "text": b.buf}),
|
||||
}).collect();
|
||||
let raw_meta = LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(json!({
|
||||
"streamed": true,
|
||||
"content": content_log,
|
||||
"stop_reason": stop,
|
||||
"usage": usage,
|
||||
})),
|
||||
};
|
||||
|
||||
Ok((turn, Some(raw_meta)))
|
||||
}
|
||||
}
|
||||
|
||||
/// User content arrives either as a plain string or as an OpenAI-style parts
|
||||
/// array (text + `image_url` data URLs, produced when the resolved model has
|
||||
/// the `vision` capability). Strings pass through; parts become Anthropic
|
||||
/// blocks. Video and unknown parts are dropped with a warning — providers
|
||||
/// gate capabilities upstream, so this should only indicate a misconfigured
|
||||
/// model row.
|
||||
fn convert_user_content(content: &Value) -> Value {
|
||||
let Some(parts) = content.as_array() else {
|
||||
return Value::String(content.as_str().unwrap_or("").to_string());
|
||||
};
|
||||
let mut blocks = Vec::new();
|
||||
for p in parts {
|
||||
match p["type"].as_str().unwrap_or("") {
|
||||
"text" => blocks.push(json!({
|
||||
"type": "text",
|
||||
"text": p["text"].as_str().unwrap_or(""),
|
||||
})),
|
||||
"image_url" => {
|
||||
if let Some(block) = parse_data_image(&p["image_url"]) {
|
||||
blocks.push(block);
|
||||
}
|
||||
}
|
||||
"file" => {
|
||||
if let Some(block) = parse_data_document(&p["file"]) {
|
||||
blocks.push(block);
|
||||
}
|
||||
}
|
||||
other => tracing::warn!(part_type = other, "dropping content part unsupported by Anthropic"),
|
||||
}
|
||||
}
|
||||
Value::Array(blocks)
|
||||
}
|
||||
|
||||
/// `{"url": "data:<mime>;base64,<data>"}` (or the bare-string shorthand) → an
|
||||
/// Anthropic base64 image block. Only data URLs are supported.
|
||||
fn parse_data_image(image_url: &Value) -> Option<Value> {
|
||||
let url = image_url["url"].as_str().or_else(|| image_url.as_str())?;
|
||||
let (mime, data) = url.strip_prefix("data:")?.split_once(";base64,")?;
|
||||
Some(json!({
|
||||
"type": "image",
|
||||
"source": { "type": "base64", "media_type": mime, "data": data },
|
||||
}))
|
||||
}
|
||||
|
||||
/// `{"file_data": "data:application/pdf;base64,<data>"}` → an Anthropic base64
|
||||
/// `document` block (the native PDF input). Only base64 data URLs are supported;
|
||||
/// the OpenAI `file` part is what the media pipeline emits for a PDF.
|
||||
fn parse_data_document(file: &Value) -> Option<Value> {
|
||||
let url = file["file_data"].as_str()?;
|
||||
let (mime, data) = url.strip_prefix("data:")?.split_once(";base64,")?;
|
||||
Some(json!({
|
||||
"type": "document",
|
||||
"source": { "type": "base64", "media_type": mime, "data": data },
|
||||
}))
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChatbotClient for AnthropicClient {
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse> {
|
||||
// Merge all system-role messages into a single `system:` parameter.
|
||||
let system: Option<String> = {
|
||||
let parts: Vec<&str> = messages
|
||||
.iter()
|
||||
.filter(|m| m.role == Role::System)
|
||||
.map(|m| m.content.as_str())
|
||||
.collect();
|
||||
if parts.is_empty() { None } else { Some(parts.join("\n\n---\n\n")) }
|
||||
};
|
||||
|
||||
let msgs: Vec<Value> = messages
|
||||
.iter()
|
||||
.filter(|m| m.role != Role::System)
|
||||
.map(|m| {
|
||||
let role = match m.role {
|
||||
Role::User => "user",
|
||||
Role::Assistant => "assistant",
|
||||
Role::System => unreachable!(),
|
||||
};
|
||||
json!({ "role": role, "content": m.content })
|
||||
})
|
||||
.collect();
|
||||
|
||||
let max_tokens = options.max_tokens.unwrap_or(4096);
|
||||
let mut body = json!({
|
||||
"model": options.model,
|
||||
"max_tokens": max_tokens,
|
||||
"messages": msgs,
|
||||
});
|
||||
|
||||
if let Some(sys) = system { body["system"] = sys.into(); }
|
||||
if let Some(t) = options.temperature { body["temperature"] = t.into(); }
|
||||
self.apply_extra(&mut body);
|
||||
|
||||
let url = format!("{}/v1/messages", self.base_url.trim_end_matches('/'));
|
||||
debug!(model = %options.model, "anthropic: sending chat request");
|
||||
trace!(body = %body, "anthropic: chat request body");
|
||||
|
||||
let resp: Value = self
|
||||
.http
|
||||
.post(&url)
|
||||
.header("x-api-key", &self.api_key)
|
||||
.header("anthropic-version", ANTHROPIC_VERSION)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json()
|
||||
.await?;
|
||||
|
||||
let content = resp["content"]
|
||||
.as_array()
|
||||
.and_then(|arr| arr.iter().find(|b| b["type"].as_str() == Some("text")))
|
||||
.and_then(|block| block["text"].as_str())
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing content in Anthropic response"))?
|
||||
.to_string();
|
||||
|
||||
let input_tokens = resp["usage"]["input_tokens"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = resp["usage"]["output_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_read_tokens = resp["usage"]["cache_read_input_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_creation_tokens = resp["usage"]["cache_creation_input_tokens"].as_u64().map(|n| n as u32);
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, "anthropic: chat response received");
|
||||
|
||||
let cost = self.extract_cost(&resp);
|
||||
Ok(ChatResponse { content, input_tokens, output_tokens, truncated: false, reasoning_content: None, cache_read_tokens, cache_creation_tokens, cost })
|
||||
}
|
||||
|
||||
async fn chat_with_tools(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<LlmTurn> {
|
||||
self.chat_with_tools_raw(messages, tools, options).await.map(|(t, _)| t)
|
||||
}
|
||||
|
||||
async fn chat_with_tools_raw(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
// Mid-conversation system messages (compaction summaries, tail
|
||||
// reminders) are merged into the single `system:` parameter — they
|
||||
// must not be silently dropped.
|
||||
let system = Self::merged_system(messages);
|
||||
let anthropic_messages = Self::convert_messages(messages);
|
||||
let anthropic_tools = Self::convert_tools(tools);
|
||||
let body = self.tools_body(system, anthropic_messages, anthropic_tools, options);
|
||||
|
||||
debug!(model = %options.model, tools = tools.len(), "anthropic: sending chat_with_tools request");
|
||||
trace!(body = %body, "anthropic: chat_with_tools request body");
|
||||
|
||||
// Capture request metadata for logging.
|
||||
let request_body = body.clone();
|
||||
let request_headers = self.logged_headers();
|
||||
|
||||
let http_resp = self.send_request(&body).await?;
|
||||
|
||||
let response_headers = headers_to_json(http_resp.headers());
|
||||
let status = http_resp.status();
|
||||
let resp_text = http_resp.text().await?;
|
||||
if !status.is_success() {
|
||||
return Err(crate::LlmError {
|
||||
status: Some(status.as_u16()),
|
||||
message: format!(
|
||||
"anthropic: HTTP {status} from {url}\nbody: {resp_text}",
|
||||
url = self.url(),
|
||||
),
|
||||
raw_meta: Some(LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(error_response_body(resp_text)),
|
||||
}),
|
||||
}.into());
|
||||
}
|
||||
let resp: Value = serde_json::from_str(&resp_text)
|
||||
.map_err(|e| anyhow::anyhow!("anthropic: failed to parse response JSON: {e}\nbody: {resp_text}"))?;
|
||||
let response_body: Value = serde_json::from_str(&resp_text).unwrap_or(Value::Null);
|
||||
|
||||
let raw_meta = LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(response_body),
|
||||
};
|
||||
|
||||
let stop_reason = resp["stop_reason"].as_str().unwrap_or("");
|
||||
let input_tokens = resp["usage"]["input_tokens"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = resp["usage"]["output_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_read_tokens = resp["usage"]["cache_read_input_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_creation_tokens = resp["usage"]["cache_creation_input_tokens"].as_u64().map(|n| n as u32);
|
||||
let content_blocks = resp["content"].as_array().cloned().unwrap_or_default();
|
||||
let cost = self.extract_cost(&resp);
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, stop_reason, "anthropic: chat_with_tools response received");
|
||||
if stop_reason == "max_tokens" {
|
||||
warn!(model = %options.model, ?output_tokens, "anthropic: response truncated (max_tokens reached)");
|
||||
}
|
||||
|
||||
let has_tool_use = content_blocks.iter().any(|b| b["type"].as_str() == Some("tool_use"));
|
||||
let reasoning_content = Self::reasoning_of(&content_blocks);
|
||||
|
||||
// Check content blocks directly: Anthropic sometimes returns stop_reason "end_turn"
|
||||
// even when tool_use blocks are present, so stop_reason alone is not reliable.
|
||||
let turn = if stop_reason == "tool_use" || has_tool_use {
|
||||
let text: String = content_blocks
|
||||
.iter()
|
||||
.filter(|b| b["type"].as_str() == Some("text"))
|
||||
.filter_map(|b| b["text"].as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
|
||||
let calls: Vec<ToolCall> = content_blocks
|
||||
.iter()
|
||||
.filter(|b| b["type"].as_str() == Some("tool_use"))
|
||||
.map(|b| ToolCall {
|
||||
id: b["id"].as_str().unwrap_or("").to_string(),
|
||||
name: b["name"].as_str().unwrap_or("").to_string(),
|
||||
arguments: b["input"].clone(),
|
||||
})
|
||||
.collect();
|
||||
|
||||
LlmTurn::ToolCalls { content: text, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens, cost }
|
||||
} else {
|
||||
let content = content_blocks
|
||||
.iter()
|
||||
.find(|b| b["type"].as_str() == Some("text"))
|
||||
.and_then(|b| b["text"].as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
|
||||
let truncated = stop_reason == "max_tokens";
|
||||
LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens, cost })
|
||||
};
|
||||
|
||||
Ok((turn, Some(raw_meta)))
|
||||
}
|
||||
|
||||
async fn chat_with_tools_raw_streaming(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: mpsc::Sender<StreamDelta>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let mut emitted = false;
|
||||
match self.stream_chat(messages, tools, options, &delta_tx, &mut emitted).await {
|
||||
Ok(ok) => Ok(ok),
|
||||
// Pre-stream failure (nothing shown yet): retry buffered. A
|
||||
// mid-stream failure propagates to the model-fallback logic.
|
||||
Err(e) if !emitted => {
|
||||
debug!(model = %options.model, error = %e, "anthropic: streaming failed before any delta; retrying buffered");
|
||||
self.chat_with_tools_raw(messages, tools, options).await
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn reasoning_of_joins_thinking_blocks() {
|
||||
let blocks = vec![
|
||||
json!({"type": "thinking", "thinking": "first"}),
|
||||
json!({"type": "text", "text": "answer"}),
|
||||
json!({"type": "thinking", "thinking": "second"}),
|
||||
];
|
||||
assert_eq!(
|
||||
AnthropicClient::reasoning_of(&blocks),
|
||||
Some("first\nsecond".to_string())
|
||||
);
|
||||
assert_eq!(AnthropicClient::reasoning_of(&[]), None);
|
||||
assert_eq!(
|
||||
AnthropicClient::reasoning_of(&[json!({"type": "text", "text": "a"})]),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_content_string_passthrough() {
|
||||
let v = convert_user_content(&json!("hello"));
|
||||
assert_eq!(v, json!("hello"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_content_parts_become_anthropic_blocks() {
|
||||
let v = convert_user_content(&json!([
|
||||
{ "type": "text", "text": "what is this?" },
|
||||
{ "type": "image_url", "image_url": { "url": "data:image/png;base64,QUJD" } },
|
||||
]));
|
||||
assert_eq!(v, json!([
|
||||
{ "type": "text", "text": "what is this?" },
|
||||
{ "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": "QUJD" } },
|
||||
]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_content_drops_video_and_non_data_urls() {
|
||||
let v = convert_user_content(&json!([
|
||||
{ "type": "text", "text": "t" },
|
||||
{ "type": "video_url", "video_url": { "url": "data:video/mp4;base64,QUJD" } },
|
||||
{ "type": "image_url", "image_url": { "url": "https://example.com/x.png" } },
|
||||
]));
|
||||
assert_eq!(v, json!([{ "type": "text", "text": "t" }]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn user_content_file_part_becomes_document_block() {
|
||||
// The OpenAI `file` part (emitted by the media pipeline for a PDF) becomes
|
||||
// an Anthropic native `document` block.
|
||||
let v = convert_user_content(&json!([
|
||||
{ "type": "text", "text": "read this" },
|
||||
{ "type": "file", "file": { "filename": "a.pdf", "file_data": "data:application/pdf;base64,QUJD" } },
|
||||
]));
|
||||
assert_eq!(v, json!([
|
||||
{ "type": "text", "text": "read this" },
|
||||
{ "type": "document", "source": { "type": "base64", "media_type": "application/pdf", "data": "QUJD" } },
|
||||
]));
|
||||
|
||||
// A non-data file_data (or missing) is dropped, not forwarded.
|
||||
let v = convert_user_content(&json!([
|
||||
{ "type": "file", "file": { "filename": "a.pdf", "file_data": "https://example.com/a.pdf" } },
|
||||
]));
|
||||
assert_eq!(v, json!([]));
|
||||
}
|
||||
}
|
||||
@@ -1,168 +0,0 @@
|
||||
pub mod anthropic;
|
||||
pub mod lm_studio;
|
||||
pub mod ollama;
|
||||
pub mod openai;
|
||||
|
||||
// Re-export the trait and all associated types from core-api so existing
|
||||
// callers that import from `llm_client` continue to work unchanged.
|
||||
pub use core_api::chatbot::{
|
||||
ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, StreamDelta,
|
||||
ToolCall,
|
||||
};
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
/// Incremental SSE decoder: feed raw response bytes, get back the payload of
|
||||
/// every complete `data:` line seen (`[DONE]` included — callers decide).
|
||||
/// Buffers partial lines across chunks; `event:` lines and comments are
|
||||
/// skipped (both OpenAI and Anthropic put the event type inside the JSON).
|
||||
#[derive(Default)]
|
||||
pub struct SseDecoder {
|
||||
buf: Vec<u8>,
|
||||
}
|
||||
|
||||
impl SseDecoder {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
pub fn feed(&mut self, bytes: &[u8]) -> Vec<String> {
|
||||
self.buf.extend_from_slice(bytes);
|
||||
let mut out = Vec::new();
|
||||
while let Some(pos) = self.buf.iter().position(|&b| b == b'\n') {
|
||||
let line: Vec<u8> = self.buf.drain(..=pos).collect();
|
||||
if let Some(payload) = parse_sse_line(&line) {
|
||||
out.push(payload);
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Flush a trailing line not terminated by `\n` at end-of-stream.
|
||||
pub fn finish(&mut self) -> Vec<String> {
|
||||
let rest = std::mem::take(&mut self.buf);
|
||||
parse_sse_line(&rest).into_iter().collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// A complete SSE line is valid UTF-8 (a multibyte sequence never contains a
|
||||
/// `\n` byte), but decode lossily anyway — a corrupt line is skipped, not fatal.
|
||||
fn parse_sse_line(line: &[u8]) -> Option<String> {
|
||||
let line = String::from_utf8_lossy(line);
|
||||
let line = line.trim_end_matches('\r').trim();
|
||||
let data = line.strip_prefix("data:")?.trim_start();
|
||||
if data.is_empty() { None } else { Some(data.to_string()) }
|
||||
}
|
||||
|
||||
/// Converts a reqwest `HeaderMap` into a `serde_json::Value` object.
|
||||
pub fn headers_to_json(headers: &reqwest::header::HeaderMap) -> Value {
|
||||
let map: serde_json::Map<String, Value> = headers
|
||||
.iter()
|
||||
.map(|(k, v)| (
|
||||
k.as_str().to_string(),
|
||||
v.to_str().unwrap_or("<binary>").into(),
|
||||
))
|
||||
.collect();
|
||||
Value::Object(map)
|
||||
}
|
||||
|
||||
/// Turns a raw error-response body into a JSON `Value` for the payload log:
|
||||
/// the parsed JSON when the provider returned JSON (the common case — an
|
||||
/// `{"error": …}` object), else the raw text wrapped as a JSON string so a
|
||||
/// non-JSON body (HTML gateway page, plain text) is still preserved verbatim.
|
||||
pub fn error_response_body(text: String) -> Value {
|
||||
serde_json::from_str::<Value>(&text).unwrap_or(Value::String(text))
|
||||
}
|
||||
|
||||
/// Returns a redacted preview of an API key: first 7 chars + "***".
|
||||
pub fn redact_key(key: &str) -> String {
|
||||
if key.len() > 7 {
|
||||
format!("{}***", &key[..7])
|
||||
} else {
|
||||
"***".to_string()
|
||||
}
|
||||
}
|
||||
|
||||
/// A structured LLM call failure carrying the HTTP `status` of the response.
|
||||
///
|
||||
/// Clients that read the status themselves (rather than via `error_for_status`)
|
||||
/// return this so callers can classify retriability on the numeric code instead of
|
||||
/// substring-matching a formatted message — which mis-fires when a model id, token
|
||||
/// count or URL merely contains "401"/"404"/… (bug B6). Non-HTTP failures (network,
|
||||
/// JSON parse, cancellation) stay ordinary `anyhow` errors with no status.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct LlmError {
|
||||
/// HTTP status code, when the failure came from an HTTP response.
|
||||
pub status: Option<u16>,
|
||||
/// Human-readable detail (provider tag + body), used for logs and the UI.
|
||||
pub message: String,
|
||||
/// Request/response payload captured at the failing call, so the debug log
|
||||
/// can show what was actually sent even when the provider rejected it (e.g.
|
||||
/// a 400). `None` for failures with no HTTP round-trip (network, cancellation,
|
||||
/// parse) — those carry no body to surface.
|
||||
pub raw_meta: Option<LlmRawMeta>,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for LlmError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str(&self.message)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LlmError {}
|
||||
|
||||
/// Extracts the HTTP status of an LLM failure, if any: a structured
|
||||
/// [`LlmError::status`] first, else any `reqwest::Error` in the source chain (the
|
||||
/// clients that fail via `error_for_status()?`). Returns `None` for a non-HTTP
|
||||
/// error (network, parse, cancellation), which callers should treat as retriable.
|
||||
pub fn http_status(err: &anyhow::Error) -> Option<u16> {
|
||||
for cause in err.chain() {
|
||||
if let Some(le) = cause.downcast_ref::<LlmError>() {
|
||||
return le.status;
|
||||
}
|
||||
if let Some(re) = cause.downcast_ref::<reqwest::Error>() {
|
||||
if let Some(s) = re.status() {
|
||||
return Some(s.as_u16());
|
||||
}
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::SseDecoder;
|
||||
|
||||
#[test]
|
||||
fn sse_decoder_buffers_partial_lines_across_chunks() {
|
||||
let mut dec = SseDecoder::new();
|
||||
// A payload split mid-JSON across two chunks yields one complete line.
|
||||
assert!(dec.feed(br#"data: {"a": 1"#).is_empty());
|
||||
assert_eq!(dec.feed(b"}\r\n").len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sse_decoder_skips_events_comments_and_keeps_done() {
|
||||
let mut dec = SseDecoder::new();
|
||||
let out = dec.feed(b"event: message_start\n: ping\n\ndata: {\"type\":\"ping\"}\ndata: [DONE]\n");
|
||||
assert_eq!(out, vec!["{\"type\":\"ping\"}".to_string(), "[DONE]".to_string()]);
|
||||
assert!(dec.finish().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sse_decoder_finish_flushes_unterminated_tail() {
|
||||
let mut dec = SseDecoder::new();
|
||||
assert!(dec.feed(b"data: tail-without-newline").is_empty());
|
||||
assert_eq!(dec.finish(), vec!["tail-without-newline".to_string()]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sse_decoder_handles_multibyte_split() {
|
||||
let mut dec = SseDecoder::new();
|
||||
// "€" is 3 bytes in UTF-8; split across the chunk boundary.
|
||||
let payload = "data: {\"t\":\"€\"}\n".as_bytes();
|
||||
let (a, b) = payload.split_at(12);
|
||||
assert!(dec.feed(a).is_empty());
|
||||
assert_eq!(dec.feed(b), vec!["{\"t\":\"€\"}".to_string()]);
|
||||
}
|
||||
}
|
||||
@@ -1,65 +0,0 @@
|
||||
use async_trait::async_trait;
|
||||
use serde_json::Value;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta, openai::OpenAiClient};
|
||||
|
||||
/// LM Studio client.
|
||||
///
|
||||
/// LM Studio exposes an OpenAI-compatible `/v1` endpoint, so this is a thin
|
||||
/// wrapper that defaults to `http://localhost:1234/v1` and requires no API key.
|
||||
pub struct LmStudioClient {
|
||||
inner: OpenAiClient,
|
||||
}
|
||||
|
||||
impl LmStudioClient {
|
||||
/// `base_url` defaults to `http://localhost:1234/v1` if `None`.
|
||||
pub fn new(base_url: Option<impl Into<String>>) -> Self {
|
||||
let url = base_url
|
||||
.map(|u| u.into())
|
||||
.unwrap_or_else(|| "http://localhost:1234/v1".to_string());
|
||||
Self { inner: OpenAiClient::new(url, "", None, false) }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChatbotClient for LmStudioClient {
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse> {
|
||||
self.inner.chat(messages, options).await
|
||||
}
|
||||
|
||||
async fn chat_with_tools(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<LlmTurn> {
|
||||
self.inner.chat_with_tools(messages, tools, options).await
|
||||
}
|
||||
|
||||
async fn chat_with_tools_raw(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
self.inner.chat_with_tools_raw(messages, tools, options).await
|
||||
}
|
||||
|
||||
/// LM Studio is OpenAI-compatible: streaming forwards to the inner client.
|
||||
/// If a local build rejects `stream_options`, the inner pre-delta buffered
|
||||
/// retry covers it transparently.
|
||||
async fn chat_with_tools_raw_streaming(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: mpsc::Sender<StreamDelta>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
self.inner.chat_with_tools_raw_streaming(messages, tools, options, delta_tx).await
|
||||
}
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
use async_trait::async_trait;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use crate::{ChatOptions, ChatResponse, ChatbotClient, Message, Role};
|
||||
|
||||
/// Ollama client using the native `/api/chat` endpoint.
|
||||
///
|
||||
/// Defaults to `http://localhost:11434`. No API key required.
|
||||
pub struct OllamaClient {
|
||||
base_url: String,
|
||||
http: reqwest::Client,
|
||||
}
|
||||
|
||||
impl OllamaClient {
|
||||
/// `base_url` defaults to `http://localhost:11434` if `None`.
|
||||
pub fn new(base_url: Option<impl Into<String>>) -> Self {
|
||||
let url = base_url
|
||||
.map(|u| u.into())
|
||||
.unwrap_or_else(|| "http://localhost:11434".to_string());
|
||||
Self { base_url: url, http: reqwest::Client::new() }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChatbotClient for OllamaClient {
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse> {
|
||||
let msgs: Vec<Value> = messages
|
||||
.iter()
|
||||
.map(|m| {
|
||||
let role = match m.role {
|
||||
Role::System => "system",
|
||||
Role::User => "user",
|
||||
Role::Assistant => "assistant",
|
||||
};
|
||||
json!({ "role": role, "content": m.content })
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut options_obj = json!({});
|
||||
if let Some(t) = options.temperature { options_obj["temperature"] = t.into(); }
|
||||
if let Some(n) = options.max_tokens { options_obj["num_predict"] = n.into(); }
|
||||
|
||||
let body = json!({
|
||||
"model": options.model,
|
||||
"messages": msgs,
|
||||
"stream": false,
|
||||
"options": options_obj,
|
||||
});
|
||||
|
||||
let url = format!("{}/api/chat", self.base_url.trim_end_matches('/'));
|
||||
|
||||
let resp: Value = self
|
||||
.http
|
||||
.post(&url)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json()
|
||||
.await?;
|
||||
|
||||
let content = resp["message"]["content"]
|
||||
.as_str()
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing content in Ollama response"))?
|
||||
.to_string();
|
||||
|
||||
let input_tokens = resp["prompt_eval_count"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = resp["eval_count"].as_u64().map(|n| n as u32);
|
||||
|
||||
Ok(ChatResponse { content, input_tokens, output_tokens, truncated: false, reasoning_content: None, cache_read_tokens: None, cache_creation_tokens: None, cost: None })
|
||||
}
|
||||
}
|
||||
@@ -1,485 +0,0 @@
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use futures_util::StreamExt;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, info, trace, warn};
|
||||
|
||||
use crate::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, Role, SseDecoder, StreamDelta, ToolCall, error_response_body, headers_to_json, redact_key};
|
||||
use core_api::APP_NAME;
|
||||
|
||||
/// OpenAI ChatGPT client (also compatible with any OpenAI-spec endpoint).
|
||||
pub struct OpenAiClient {
|
||||
base_url: String,
|
||||
api_key: String,
|
||||
extra_params: Option<serde_json::Value>,
|
||||
/// When true, Anthropic-compatible prompt-caching hints are injected:
|
||||
/// - `anthropic-beta: prompt-caching-2024-07-31` header is sent.
|
||||
/// - The last tool definition is tagged with `cache_control: {"type":"ephemeral"}`.
|
||||
/// - System message content is expected to already be a content array with
|
||||
/// `cache_control` on the static block (set by `build_openai_messages`).
|
||||
/// Used for OpenRouter when routing to Anthropic models.
|
||||
enable_prompt_cache: bool,
|
||||
http: reqwest::Client,
|
||||
}
|
||||
|
||||
impl OpenAiClient {
|
||||
pub fn new(base_url: impl Into<String>, api_key: impl Into<String>, extra_params: Option<serde_json::Value>, enable_prompt_cache: bool) -> Self {
|
||||
Self {
|
||||
base_url: base_url.into(),
|
||||
api_key: api_key.into(),
|
||||
extra_params,
|
||||
enable_prompt_cache,
|
||||
http: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Merges `extra_params` (if any) into `body`. Only top-level object keys are merged.
|
||||
fn apply_extra(&self, body: &mut serde_json::Value) {
|
||||
if let Some(serde_json::Value::Object(extra)) = &self.extra_params {
|
||||
if let Some(b) = body.as_object_mut() {
|
||||
for (k, v) in extra {
|
||||
b.insert(k.clone(), v.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn url(&self) -> String {
|
||||
format!("{}/chat/completions", self.base_url.trim_end_matches('/'))
|
||||
}
|
||||
|
||||
/// Shared request body for the buffered and the streaming path. Caller adds
|
||||
/// `max_tokens`/`temperature`/`extra_params` afterwards via `finalize_body`.
|
||||
fn base_body(&self, model: &str, messages: &[Value], tools: &[Value]) -> Value {
|
||||
let mut body = json!({
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
});
|
||||
|
||||
if !tools.is_empty() {
|
||||
// When prompt caching is enabled, tag the last tool with cache_control
|
||||
// so the entire tools array is included in the Anthropic KV cache prefix.
|
||||
let tools_value: Value = if self.enable_prompt_cache {
|
||||
let mut tagged = tools.to_vec();
|
||||
if let Some(last) = tagged.last_mut() {
|
||||
last["cache_control"] = json!({"type": "ephemeral"});
|
||||
}
|
||||
tagged.into()
|
||||
} else {
|
||||
tools.into()
|
||||
};
|
||||
body["tools"] = tools_value;
|
||||
body["tool_choice"] = "auto".into();
|
||||
}
|
||||
body
|
||||
}
|
||||
|
||||
fn finalize_body(&self, mut body: Value, options: &ChatOptions) -> Value {
|
||||
if let Some(t) = options.max_tokens { body["max_tokens"] = t.into(); }
|
||||
if let Some(t) = options.temperature { body["temperature"] = t.into(); }
|
||||
self.apply_extra(&mut body);
|
||||
body
|
||||
}
|
||||
|
||||
/// Request metadata for logging (shared by buffered and streaming paths).
|
||||
fn logged_headers(&self) -> Value {
|
||||
let mut logged_headers = json!({
|
||||
"authorization": format!("Bearer {}", redact_key(&self.api_key)),
|
||||
"content-type": "application/json",
|
||||
});
|
||||
if self.enable_prompt_cache {
|
||||
logged_headers["anthropic-beta"] = "prompt-caching-2024-07-31".into();
|
||||
}
|
||||
logged_headers
|
||||
}
|
||||
|
||||
async fn send_request(&self, body: &Value) -> reqwest::Result<reqwest::Response> {
|
||||
let mut req = self.http.post(self.url()).bearer_auth(&self.api_key).header("X-Title", APP_NAME);
|
||||
if self.enable_prompt_cache {
|
||||
req = req.header("anthropic-beta", "prompt-caching-2024-07-31");
|
||||
}
|
||||
req.json(body).send().await
|
||||
}
|
||||
|
||||
/// SSE streaming path behind `chat_with_tools_raw_streaming`. Accumulates
|
||||
/// content/reasoning/tool-call fragments into the same `LlmTurn` the
|
||||
/// buffered path would return, while forwarding text/reasoning deltas to
|
||||
/// `delta_tx` (try_send, best-effort). `emitted` tracks whether any delta
|
||||
/// was pushed, so the caller can distinguish a pre-stream failure (safe to
|
||||
/// retry buffered) from a mid-stream one (partial output already shown).
|
||||
async fn stream_chat(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: &mpsc::Sender<StreamDelta>,
|
||||
emitted: &mut bool,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let mut body = self.base_body(&options.model, messages, tools);
|
||||
body["stream"] = json!(true);
|
||||
body["stream_options"] = json!({ "include_usage": true });
|
||||
let body = self.finalize_body(body, options);
|
||||
|
||||
debug!(model = %options.model, tools = tools.len(), prompt_cache = self.enable_prompt_cache, "openai: sending streaming chat_with_tools request");
|
||||
trace!(body = %body, "openai: streaming chat_with_tools request body");
|
||||
|
||||
let request_body = body.clone();
|
||||
let request_headers = self.logged_headers();
|
||||
|
||||
let http_resp = self.send_request(&body).await?;
|
||||
|
||||
let response_headers = headers_to_json(http_resp.headers());
|
||||
let status = http_resp.status();
|
||||
if !status.is_success() {
|
||||
let resp_text = http_resp.text().await?;
|
||||
return Err(crate::LlmError {
|
||||
status: Some(status.as_u16()),
|
||||
message: format!(
|
||||
"openai: HTTP {status} from {url}\nbody: {resp_text}",
|
||||
url = self.url(),
|
||||
),
|
||||
raw_meta: Some(LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(error_response_body(resp_text)),
|
||||
}),
|
||||
}.into());
|
||||
}
|
||||
|
||||
let mut content = String::new();
|
||||
let mut reasoning = String::new();
|
||||
// index → (id, name, arguments fragment buffer)
|
||||
let mut tool_calls: BTreeMap<u64, (String, String, String)> = BTreeMap::new();
|
||||
let mut finish_reason: Option<String> = None;
|
||||
let mut usage: Option<Value> = None;
|
||||
let mut sse = SseDecoder::new();
|
||||
let mut byte_stream = http_resp.bytes_stream();
|
||||
|
||||
// One SSE `data:` payload. Fragments update the accumulators; text and
|
||||
// reasoning also go out as deltas. Unparseable chunks are skipped —
|
||||
// the assembled turn stays consistent.
|
||||
let mut handle_payload = |payload: &str, emitted: &mut bool| {
|
||||
if payload == "[DONE]" {
|
||||
return;
|
||||
}
|
||||
let Ok(v) = serde_json::from_str::<Value>(payload) else { return };
|
||||
if let Some(u) = v.get("usage").filter(|u| !u.is_null()) {
|
||||
usage = Some(u.clone());
|
||||
}
|
||||
let Some(choice) = v["choices"].as_array().and_then(|a| a.first()) else { return };
|
||||
if let Some(fr) = choice["finish_reason"].as_str() {
|
||||
finish_reason = Some(fr.to_string());
|
||||
}
|
||||
let delta = &choice["delta"];
|
||||
if let Some(t) = delta["content"].as_str().filter(|t| !t.is_empty()) {
|
||||
content.push_str(t);
|
||||
*emitted = true;
|
||||
let _ = delta_tx.try_send(StreamDelta::Text(t.to_string()));
|
||||
}
|
||||
// Same normalization as the buffered path: DeepSeek uses
|
||||
// `reasoning_content`, MiniMax M3 and others `reasoning`.
|
||||
if let Some(t) = delta["reasoning_content"].as_str()
|
||||
.or_else(|| delta["reasoning"].as_str())
|
||||
.filter(|t| !t.is_empty())
|
||||
{
|
||||
reasoning.push_str(t);
|
||||
*emitted = true;
|
||||
let _ = delta_tx.try_send(StreamDelta::Reasoning(t.to_string()));
|
||||
}
|
||||
if let Some(tc_arr) = delta["tool_calls"].as_array() {
|
||||
for tc in tc_arr {
|
||||
let idx = tc["index"].as_u64().unwrap_or(0);
|
||||
let entry = tool_calls.entry(idx).or_default();
|
||||
if let Some(id) = tc["id"].as_str() { entry.0 = id.to_string(); }
|
||||
if let Some(n) = tc["function"]["name"].as_str() { entry.1 = n.to_string(); }
|
||||
if let Some(a) = tc["function"]["arguments"].as_str() { entry.2.push_str(a); }
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
while let Some(chunk) = byte_stream.next().await {
|
||||
let chunk = chunk?;
|
||||
for payload in sse.feed(&chunk) {
|
||||
handle_payload(&payload, emitted);
|
||||
}
|
||||
}
|
||||
for payload in sse.finish() {
|
||||
handle_payload(&payload, emitted);
|
||||
}
|
||||
|
||||
let finish = finish_reason.as_deref().unwrap_or("stop");
|
||||
let input_tokens = usage.as_ref().and_then(|u| u["prompt_tokens"].as_u64()).map(|n| n as u32);
|
||||
let output_tokens = usage.as_ref().and_then(|u| u["completion_tokens"].as_u64()).map(|n| n as u32);
|
||||
let cache_read_tokens = usage.as_ref()
|
||||
.and_then(|u| u["prompt_tokens_details"]["cached_tokens"].as_u64())
|
||||
.map(|n| n as u32);
|
||||
let cost = usage.as_ref().and_then(|u| u["cost"].as_f64());
|
||||
let reasoning_content = if reasoning.is_empty() { None } else { Some(reasoning) };
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, finish_reason = finish, "openai: streaming response completed");
|
||||
if finish == "length" {
|
||||
warn!(model = %options.model, ?output_tokens, "openai: response truncated (max_tokens reached)");
|
||||
}
|
||||
|
||||
// Reassemble the streamed message for the payload log, so a streamed call
|
||||
// leaves the same debugging trail as a buffered one — including
|
||||
// reasoning_content and tool_calls, which previously existed only as
|
||||
// transient deltas and never appeared in the logged body. Built here,
|
||||
// before `turn` consumes the accumulators (clones are cheap vs. the round-trip).
|
||||
let logged_tool_calls: Vec<Value> = tool_calls.iter()
|
||||
.map(|(_idx, (id, name, args))| json!({
|
||||
"id": id,
|
||||
"type": "function",
|
||||
"function": { "name": name, "arguments": args },
|
||||
}))
|
||||
.collect();
|
||||
let mut logged_message = json!({ "role": "assistant", "content": content.clone() });
|
||||
if let Some(rc) = &reasoning_content {
|
||||
logged_message["reasoning_content"] = rc.clone().into();
|
||||
}
|
||||
if !logged_tool_calls.is_empty() {
|
||||
logged_message["tool_calls"] = Value::Array(logged_tool_calls);
|
||||
}
|
||||
|
||||
let turn = if !tool_calls.is_empty() {
|
||||
let calls = tool_calls
|
||||
.into_values()
|
||||
.map(|(id, name, args)| ToolCall {
|
||||
id,
|
||||
name,
|
||||
arguments: serde_json::from_str(&args).unwrap_or(Value::Object(Default::default())),
|
||||
})
|
||||
.collect();
|
||||
LlmTurn::ToolCalls { content, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost }
|
||||
} else {
|
||||
let truncated = finish == "length";
|
||||
LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost })
|
||||
};
|
||||
|
||||
// Synthesize a buffered-shaped response body for the payload log, so a
|
||||
// streamed call leaves the same debugging trail as a buffered one.
|
||||
let response_body = json!({
|
||||
"streamed": true,
|
||||
"choices": [{ "finish_reason": finish, "message": logged_message }],
|
||||
"usage": usage,
|
||||
});
|
||||
let raw_meta = LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(response_body),
|
||||
};
|
||||
|
||||
Ok((turn, Some(raw_meta)))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChatbotClient for OpenAiClient {
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse> {
|
||||
let msgs: Vec<Value> = messages
|
||||
.iter()
|
||||
.map(|m| {
|
||||
let role = match m.role {
|
||||
Role::System => "system",
|
||||
Role::User => "user",
|
||||
Role::Assistant => "assistant",
|
||||
};
|
||||
json!({ "role": role, "content": m.content })
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut body = json!({
|
||||
"model": options.model,
|
||||
"messages": msgs,
|
||||
});
|
||||
|
||||
if let Some(t) = options.max_tokens { body["max_tokens"] = t.into(); }
|
||||
if let Some(t) = options.temperature { body["temperature"] = t.into(); }
|
||||
self.apply_extra(&mut body);
|
||||
|
||||
debug!(model = %options.model, "openai: sending chat request");
|
||||
trace!(body = %body, "openai: chat request body");
|
||||
|
||||
let resp: Value = self
|
||||
.http
|
||||
.post(self.url())
|
||||
.bearer_auth(&self.api_key)
|
||||
.header("X-Title", APP_NAME)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await?
|
||||
.error_for_status()?
|
||||
.json()
|
||||
.await?;
|
||||
|
||||
let content = match resp["choices"][0]["message"]["content"].as_str() {
|
||||
Some(s) => s.to_string(),
|
||||
None => {
|
||||
warn!(raw_response = %resp, "openai: chat() response has null content");
|
||||
String::new()
|
||||
}
|
||||
};
|
||||
|
||||
let input_tokens = resp["usage"]["prompt_tokens"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = resp["usage"]["completion_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_read_tokens = resp["usage"]["prompt_tokens_details"]["cached_tokens"].as_u64().map(|n| n as u32);
|
||||
let truncated = resp["choices"][0]["finish_reason"].as_str() == Some("length");
|
||||
let cost = self.extract_cost(&resp);
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, ?cost, truncated, "openai: chat response received");
|
||||
|
||||
Ok(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content: None, cache_read_tokens, cache_creation_tokens: None, cost })
|
||||
}
|
||||
|
||||
async fn chat_with_tools(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<LlmTurn> {
|
||||
self.chat_with_tools_raw(messages, tools, options).await.map(|(t, _)| t)
|
||||
}
|
||||
|
||||
async fn chat_with_tools_raw(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let body = self.finalize_body(self.base_body(&options.model, messages, tools), options);
|
||||
|
||||
debug!(model = %options.model, tools = tools.len(), prompt_cache = self.enable_prompt_cache, "openai: sending chat_with_tools request");
|
||||
trace!(body = %body, "openai: chat_with_tools request body");
|
||||
|
||||
// Capture request metadata for logging.
|
||||
let request_body = body.clone();
|
||||
let request_headers = self.logged_headers();
|
||||
|
||||
let http_resp = self.send_request(&body).await?;
|
||||
|
||||
let response_headers = headers_to_json(http_resp.headers());
|
||||
let status = http_resp.status();
|
||||
let resp_text = http_resp.text().await?;
|
||||
|
||||
if !status.is_success() {
|
||||
return Err(crate::LlmError {
|
||||
status: Some(status.as_u16()),
|
||||
message: format!(
|
||||
"openai: HTTP {status} from {url}\nbody: {resp_text}",
|
||||
url = self.url(),
|
||||
),
|
||||
raw_meta: Some(LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(error_response_body(resp_text)),
|
||||
}),
|
||||
}.into());
|
||||
}
|
||||
|
||||
let resp: Value = serde_json::from_str(&resp_text)
|
||||
.map_err(|e| anyhow::anyhow!("openai: failed to parse response JSON: {e}\nbody: {resp_text}"))?;
|
||||
let response_body: Value = serde_json::from_str(&resp_text).unwrap_or(Value::Null);
|
||||
|
||||
let raw_meta = LlmRawMeta {
|
||||
request_headers: Some(request_headers),
|
||||
request_body: Some(request_body),
|
||||
response_headers: Some(response_headers),
|
||||
response_body: Some(response_body),
|
||||
};
|
||||
|
||||
let input_tokens = resp["usage"]["prompt_tokens"].as_u64().map(|n| n as u32);
|
||||
let output_tokens = resp["usage"]["completion_tokens"].as_u64().map(|n| n as u32);
|
||||
let cache_read_tokens = resp["usage"]["prompt_tokens_details"]["cached_tokens"].as_u64().map(|n| n as u32);
|
||||
let cost = self.extract_cost(&resp);
|
||||
|
||||
let choice = &resp["choices"][0];
|
||||
let message = &choice["message"];
|
||||
let finish = choice["finish_reason"].as_str().unwrap_or("stop");
|
||||
info!(model = %options.model, ?input_tokens, ?output_tokens, finish_reason = finish, "openai: chat_with_tools response received");
|
||||
if finish == "length" {
|
||||
warn!(model = %options.model, ?output_tokens, "openai: response truncated (max_tokens reached)");
|
||||
}
|
||||
|
||||
// Thinking/reasoning content varies by provider:
|
||||
// - DeepSeek: "reasoning_content" (must be echoed back on subsequent turns, even as "")
|
||||
// - MiniMax M3 and others: "reasoning"
|
||||
// We normalize to a single field and echo under both names in message_builder.
|
||||
let reasoning_content = message["reasoning_content"].as_str()
|
||||
.or_else(|| message["reasoning"].as_str())
|
||||
.map(str::to_string);
|
||||
|
||||
let tool_calls_array = message["tool_calls"].as_array().filter(|a| !a.is_empty());
|
||||
|
||||
// Some models (e.g. Qwen via OpenRouter) return finish_reason "stop" even when
|
||||
// tool_calls are present, so check the array directly rather than relying on finish_reason.
|
||||
let turn = if finish == "tool_calls" || tool_calls_array.is_some() {
|
||||
let content = message["content"].as_str().unwrap_or("").to_string();
|
||||
|
||||
let calls = tool_calls_array
|
||||
.ok_or_else(|| anyhow::anyhow!("finish_reason=tool_calls but tool_calls array missing or empty"))?
|
||||
.iter()
|
||||
.map(|tc| {
|
||||
let id = tc["id"].as_str().unwrap_or("").to_string();
|
||||
let name = tc["function"]["name"].as_str().unwrap_or("").to_string();
|
||||
let args: Value = tc["function"]["arguments"]
|
||||
.as_str()
|
||||
.and_then(|s| serde_json::from_str(s).ok())
|
||||
.unwrap_or(Value::Object(Default::default()));
|
||||
ToolCall { id, name, arguments: args }
|
||||
})
|
||||
.collect();
|
||||
|
||||
LlmTurn::ToolCalls { content, calls, input_tokens, output_tokens, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost }
|
||||
} else {
|
||||
// content can be null for thinking/reasoning models or when finish_reason="length".
|
||||
// Fall back to empty string rather than erroring — the partial response is still
|
||||
// useful and a hard error breaks the session.
|
||||
let content = match message["content"].as_str() {
|
||||
Some(s) => s.to_string(),
|
||||
None => {
|
||||
tracing::warn!(
|
||||
finish_reason = finish,
|
||||
?input_tokens,
|
||||
?output_tokens,
|
||||
raw_message = %message,
|
||||
"OpenAI response has null content",
|
||||
);
|
||||
String::new()
|
||||
}
|
||||
};
|
||||
let truncated = finish == "length";
|
||||
LlmTurn::Message(ChatResponse { content, input_tokens, output_tokens, truncated, reasoning_content, cache_read_tokens, cache_creation_tokens: None, cost })
|
||||
};
|
||||
|
||||
Ok((turn, Some(raw_meta)))
|
||||
}
|
||||
|
||||
async fn chat_with_tools_raw_streaming(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: mpsc::Sender<StreamDelta>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let mut emitted = false;
|
||||
match self.stream_chat(messages, tools, options, &delta_tx, &mut emitted).await {
|
||||
Ok(ok) => Ok(ok),
|
||||
// Nothing was ever streamed: some OpenAI-compatible providers reject
|
||||
// `stream`/`stream_options` outright — retry buffered so they keep
|
||||
// working exactly as before. A mid-stream failure (deltas already
|
||||
// shown) instead propagates to the model-fallback logic.
|
||||
Err(e) if !emitted => {
|
||||
debug!(model = %options.model, error = %e, "openai: streaming failed before any delta; retrying buffered");
|
||||
self.chat_with_tools_raw(messages, tools, options).await
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -78,6 +78,6 @@ base64 = "0.22"
|
||||
sha2 = "0.10"
|
||||
notify = "8"
|
||||
honcho-client = { path = "../honcho-client" }
|
||||
llm-client = { path = "../llm-client" }
|
||||
agent-loop = { path = "../agent-loop" }
|
||||
core-api = { path = "../core-api" }
|
||||
mcp-client = { path = "../mcp-client" }
|
||||
|
||||
@@ -1,166 +0,0 @@
|
||||
//! Transparent logging wrapper for any [`ChatbotClient`].
|
||||
//!
|
||||
//! [`LoggingChatbotClient`] intercepts every `chat_with_tools_raw` call, captures
|
||||
//! the raw HTTP request/response from the inner provider, persists a **metadata-only**
|
||||
//! row to `llm_requests` in `system.db` (fire-and-forget), then returns the raw data
|
||||
//! to the caller so it can write the **payload** to the user's own database.
|
||||
//!
|
||||
//! The split keeps conversation content (payloads) behind the user key while
|
||||
//! metadata (cost, tokens, timing) stays in the admin-readable registry.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use serde_json::Value;
|
||||
use sqlx::SqlitePool;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::warn;
|
||||
|
||||
use crate::db::llm_requests;
|
||||
|
||||
use super::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Message, StreamDelta};
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
pub struct LoggingChatbotClient {
|
||||
inner: Arc<dyn ChatbotClient>,
|
||||
pool: Arc<SqlitePool>,
|
||||
model_name: String,
|
||||
}
|
||||
|
||||
impl LoggingChatbotClient {
|
||||
pub fn new(
|
||||
inner: Arc<dyn ChatbotClient>,
|
||||
pool: Arc<SqlitePool>,
|
||||
model_name: impl Into<String>,
|
||||
) -> Self {
|
||||
Self { inner, pool, model_name: model_name.into() }
|
||||
}
|
||||
|
||||
/// Shared logging tail of both raw entry points: writes the metadata-only
|
||||
/// row to `system.db` (fire-and-forget), then passes the result through.
|
||||
async fn log_and_return(
|
||||
&self,
|
||||
options: &ChatOptions,
|
||||
duration: Duration,
|
||||
result: anyhow::Result<(LlmTurn, Option<LlmRawMeta>)>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let duration_ms = duration.as_millis() as i64;
|
||||
|
||||
let session_id = options.session_id;
|
||||
let stack_id = options.stack_id;
|
||||
let user_id = options.user_id.clone();
|
||||
let request_id = options.request_id.clone();
|
||||
let model_name = self.model_name.clone();
|
||||
let pool = Arc::clone(&self.pool);
|
||||
|
||||
match result {
|
||||
Ok((turn, meta)) => {
|
||||
let (input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens) = match &turn {
|
||||
LlmTurn::Message(r) => (r.input_tokens, r.output_tokens, r.cache_read_tokens, r.cache_creation_tokens),
|
||||
LlmTurn::ToolCalls { input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, .. } =>
|
||||
(*input_tokens, *output_tokens, *cache_read_tokens, *cache_creation_tokens),
|
||||
};
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
|
||||
request_id,
|
||||
user_id,
|
||||
session_id,
|
||||
stack_id,
|
||||
model_name,
|
||||
error_text: None,
|
||||
input_tokens: input_tokens.map(|n| n as i64),
|
||||
output_tokens: output_tokens.map(|n| n as i64),
|
||||
duration_ms,
|
||||
cache_read_tokens: cache_read_tokens.map(|n| n as i64),
|
||||
cache_creation_tokens: cache_creation_tokens.map(|n| n as i64),
|
||||
}).await {
|
||||
warn!(error = %e, "llm_requests: failed to insert log row");
|
||||
}
|
||||
});
|
||||
|
||||
Ok((turn, meta))
|
||||
}
|
||||
|
||||
Err(e) => {
|
||||
let error_text = e.to_string();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(log_err) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
|
||||
request_id,
|
||||
user_id,
|
||||
session_id,
|
||||
stack_id,
|
||||
model_name,
|
||||
error_text: Some(error_text),
|
||||
input_tokens: None,
|
||||
output_tokens: None,
|
||||
duration_ms,
|
||||
cache_read_tokens: None,
|
||||
cache_creation_tokens: None,
|
||||
}).await {
|
||||
warn!(error = %log_err, "llm_requests: failed to insert error log row");
|
||||
}
|
||||
});
|
||||
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ChatbotClient for LoggingChatbotClient {
|
||||
/// Passthrough — logging only applies to the tool-calling path.
|
||||
async fn chat(
|
||||
&self,
|
||||
messages: &[Message],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<ChatResponse> {
|
||||
self.inner.chat(messages, options).await
|
||||
}
|
||||
|
||||
/// Passthrough that drops the raw meta. Used by callers that do not need
|
||||
/// payload capture (e.g. the compactor).
|
||||
async fn chat_with_tools(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<LlmTurn> {
|
||||
let (turn, _) = self.chat_with_tools_raw(messages, tools, options).await?;
|
||||
Ok(turn)
|
||||
}
|
||||
|
||||
/// Intercepts the call, delegates to `inner.chat_with_tools_raw` to capture
|
||||
/// HTTP wire data, writes a **metadata-only** row to `system.db`, then returns
|
||||
/// the raw data so the caller can persist payloads to the user's own database.
|
||||
async fn chat_with_tools_raw(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let start = Instant::now();
|
||||
let result = self.inner.chat_with_tools_raw(messages, tools, options).await;
|
||||
self.log_and_return(options, start.elapsed(), result).await
|
||||
}
|
||||
|
||||
/// Streaming twin of `chat_with_tools_raw`: forwards `delta_tx` untouched to
|
||||
/// the inner client (deltas are not logged — only the final turn is), then
|
||||
/// applies the same metadata logging. Without this override the trait
|
||||
/// default would silently fall back to the buffered call.
|
||||
async fn chat_with_tools_raw_streaming(
|
||||
&self,
|
||||
messages: &[Value],
|
||||
tools: &[Value],
|
||||
options: &ChatOptions,
|
||||
delta_tx: mpsc::Sender<StreamDelta>,
|
||||
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
|
||||
let start = Instant::now();
|
||||
let result = self.inner.chat_with_tools_raw_streaming(messages, tools, options, delta_tx).await;
|
||||
self.log_and_return(options, start.elapsed(), result).await
|
||||
}
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
pub mod logging;
|
||||
|
||||
// Re-export from the independent llm-client crate.
|
||||
pub use llm_client::{
|
||||
ChatOptions, ChatResponse, ChatbotClient, LlmError, LlmRawMeta, LlmTurn, Message, StreamDelta,
|
||||
ToolCall, anthropic, http_status, lm_studio, ollama, openai,
|
||||
};
|
||||
@@ -52,7 +52,6 @@ use tracing::{debug, info, warn};
|
||||
use core_api::{ConfigProperty, ConfigSet, PropertyType};
|
||||
|
||||
use crate::chat_event_bus::{ChatEventBus, CompactionEvent};
|
||||
use crate::chatbot::ChatOptions;
|
||||
use crate::config::CompactionConfig;
|
||||
use crate::config_store::GlobalConfigManager;
|
||||
use crate::db::{chat_history, chat_llm_tools, chat_summaries};
|
||||
@@ -353,25 +352,28 @@ impl ContextCompactor {
|
||||
json!({ "role": "user", "content": conversation_text }),
|
||||
];
|
||||
|
||||
let options = ChatOptions {
|
||||
let request = agent_loop::model::ModelRequest {
|
||||
messages: messages_payload,
|
||||
tools: Vec::new(),
|
||||
model: llm.model.clone(),
|
||||
max_tokens: None,
|
||||
temperature: Some(0.3),
|
||||
session_id: Some(session_id),
|
||||
stack_id: Some(stack_id),
|
||||
user_id: None,
|
||||
request_id: None,
|
||||
request_id: uuid::Uuid::new_v4().to_string(),
|
||||
conversation: agent_loop::ids::ConversationId::new(format!("session:{session_id}")),
|
||||
frame: agent_loop::ids::FrameId(stack_id),
|
||||
extras: serde_json::Value::Null,
|
||||
log: Some(json!({ "session_id": session_id, "stack_id": stack_id })),
|
||||
};
|
||||
|
||||
let turn = llm.client.chat_with_tools(&messages_payload, &[], &options).await
|
||||
let resp = llm.client.complete(&request, None).await
|
||||
.map_err(|e| {
|
||||
warn!(stack_id, error = %e, "compactor: LLM call failed");
|
||||
e
|
||||
})?;
|
||||
|
||||
let summary_text = match turn {
|
||||
crate::chatbot::LlmTurn::Message(resp) => resp.content,
|
||||
crate::chatbot::LlmTurn::ToolCalls { content, .. } => {
|
||||
let summary_text = match resp {
|
||||
agent_loop::model::ModelResponse::Message { content, .. } => content,
|
||||
agent_loop::model::ModelResponse::ToolCalls { content, .. } => {
|
||||
warn!(stack_id, "compactor: unexpected tool calls in summary response, using content");
|
||||
content
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
//! DB operations for the `llm_requests` table (metadata only).
|
||||
//!
|
||||
//! Every `chat_with_tools` call is logged here by the
|
||||
//! [`crate::chatbot::logging::LoggingChatbotClient`] wrapper.
|
||||
//! [`crate::llm::logging::LoggingModel`] decorator.
|
||||
//! Payloads (request/response bodies + headers) live in `llm_request_payloads`
|
||||
//! in the owner bucket (`{userid}.db`), correlated by `request_id`.
|
||||
//! Rows are retained for `llm.request_log.retention_days` days (default 14).
|
||||
|
||||
@@ -14,7 +14,6 @@ pub mod agents;
|
||||
pub mod approval;
|
||||
pub mod chat_event_bus;
|
||||
pub mod chat_hub;
|
||||
pub mod chatbot;
|
||||
pub mod clarification;
|
||||
pub mod command;
|
||||
pub mod compactor;
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
//! Transparent logging decorator for any [`agent_loop::model::Model`].
|
||||
//!
|
||||
//! [`LoggingModel`] intercepts every `complete` call, measures the duration,
|
||||
//! and persists a **metadata-only** row to `llm_requests` in `system.db`
|
||||
//! (fire-and-forget). Per-request correlation (session/stack/user id) travels
|
||||
//! in [`ModelRequest::log`], set by the caller; the payload (request/response
|
||||
//! bodies) is returned to the caller inside [`ModelResponse::raw`] /
|
||||
//! [`ModelError::raw`] so it can be written to the user's own database.
|
||||
//!
|
||||
//! The split keeps conversation content (payloads) behind the user key while
|
||||
//! metadata (cost, tokens, timing) stays in the admin-readable registry.
|
||||
//! (Successor of `chatbot::logging::LoggingChatbotClient`, blueprint D13.)
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use sqlx::SqlitePool;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::warn;
|
||||
|
||||
use agent_loop::model::{Model, ModelError, ModelRequest, ModelResponse, StreamDelta};
|
||||
|
||||
use crate::db::llm_requests;
|
||||
|
||||
pub struct LoggingModel {
|
||||
inner: Arc<dyn Model>,
|
||||
pool: Arc<SqlitePool>,
|
||||
model_name: String,
|
||||
}
|
||||
|
||||
impl LoggingModel {
|
||||
pub fn new(inner: Arc<dyn Model>, pool: Arc<SqlitePool>, model_name: impl Into<String>) -> Self {
|
||||
Self { inner, pool, model_name: model_name.into() }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Model for LoggingModel {
|
||||
async fn complete(
|
||||
&self,
|
||||
req: &ModelRequest,
|
||||
deltas: Option<mpsc::Sender<StreamDelta>>,
|
||||
) -> Result<ModelResponse, ModelError> {
|
||||
let start = Instant::now();
|
||||
let result = self.inner.complete(req, deltas).await;
|
||||
let duration_ms = start.elapsed().as_millis() as i64;
|
||||
|
||||
// Per-request correlation set by the caller (llm_call / compactor).
|
||||
let log = req.log.clone().unwrap_or_default();
|
||||
let session_id = log["session_id"].as_i64();
|
||||
let stack_id = log["stack_id"].as_i64();
|
||||
let user_id = log["user_id"].as_str().map(str::to_string);
|
||||
let request_id = Some(req.request_id.clone());
|
||||
let model_name = self.model_name.clone();
|
||||
let pool = Arc::clone(&self.pool);
|
||||
|
||||
match &result {
|
||||
Ok(resp) => {
|
||||
let usage = resp.usage();
|
||||
let (input_tokens, output_tokens, cache_read, cache_write) = (
|
||||
usage.input_tokens.map(|n| n as i64),
|
||||
usage.output_tokens.map(|n| n as i64),
|
||||
usage.cache_read.map(|n| n as i64),
|
||||
usage.cache_write.map(|n| n as i64),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
|
||||
request_id,
|
||||
user_id,
|
||||
session_id,
|
||||
stack_id,
|
||||
model_name,
|
||||
error_text: None,
|
||||
input_tokens,
|
||||
output_tokens,
|
||||
duration_ms,
|
||||
cache_read_tokens: cache_read,
|
||||
cache_creation_tokens: cache_write,
|
||||
}).await {
|
||||
warn!(error = %e, "llm_requests: failed to insert log row");
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
let error_text = e.to_string();
|
||||
tokio::spawn(async move {
|
||||
if let Err(log_err) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
|
||||
request_id,
|
||||
user_id,
|
||||
session_id,
|
||||
stack_id,
|
||||
model_name,
|
||||
error_text: Some(error_text),
|
||||
input_tokens: None,
|
||||
output_tokens: None,
|
||||
duration_ms,
|
||||
cache_read_tokens: None,
|
||||
cache_creation_tokens: None,
|
||||
}).await {
|
||||
warn!(error = %log_err, "llm_requests: failed to insert error log row");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
fn is_retriable(&self, err: &ModelError) -> bool {
|
||||
self.inner.is_retriable(err)
|
||||
}
|
||||
}
|
||||
@@ -8,11 +8,11 @@ use sqlx::SqlitePool;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::chatbot::ChatbotClient;
|
||||
use crate::chatbot::logging::LoggingChatbotClient;
|
||||
use agent_loop::model::Model;
|
||||
use core_api::provider::LlmStrength;
|
||||
use crate::provider::{ApiProvider, ProviderRegistry, ReasoningMode};
|
||||
|
||||
use super::logging::LoggingModel;
|
||||
use super::providers::RemoteLlmModelInfo;
|
||||
use super::{ClientStatus, LlmEntry, LlmModelInfo, LlmModelRecord, LlmProviderInfo, LlmProviderRecord};
|
||||
use super::db;
|
||||
@@ -512,8 +512,8 @@ fn build_entry(
|
||||
let prompt_cache = built.prompt_cache;
|
||||
let extra = model.extra_params.clone();
|
||||
|
||||
let client: Arc<dyn ChatbotClient> = match log_pool {
|
||||
Some(pool) => Arc::new(LoggingChatbotClient::new(inner, pool, &model.name)),
|
||||
let client: Arc<dyn Model> = match log_pool {
|
||||
Some(pool) => Arc::new(LoggingModel::new(inner, pool, &model.name)),
|
||||
None => inner,
|
||||
};
|
||||
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
pub(crate) mod db;
|
||||
pub mod logging;
|
||||
pub mod manager;
|
||||
pub mod providers;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::chatbot::ChatbotClient;
|
||||
use agent_loop::model::Model;
|
||||
|
||||
use crate::provider::ServiceType;
|
||||
|
||||
pub use core_api::provider::{LlmProviderRecord, LlmModelRecord, LlmStrength, ReasoningMode};
|
||||
@@ -13,7 +15,7 @@ pub use manager::{LlmManager, sort_models_for_agent};
|
||||
/// A resolved, ready-to-use LLM client with its associated metadata.
|
||||
#[derive(Clone)]
|
||||
pub struct LlmEntry {
|
||||
pub client: Arc<dyn ChatbotClient>,
|
||||
pub client: Arc<dyn Model>,
|
||||
pub model: String,
|
||||
pub model_db_id: i64,
|
||||
pub strength: Option<LlmStrength>,
|
||||
|
||||
@@ -2,7 +2,7 @@ use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result, anyhow};
|
||||
|
||||
use crate::chatbot::anthropic::AnthropicClient;
|
||||
use agent_loop::models::AnthropicModel;
|
||||
use crate::llm::{LlmModelRecord, LlmProviderRecord};
|
||||
use crate::llm::providers::{RemoteLlmModelInfo, extra_with_reasoning};
|
||||
use crate::provider::{ApiProvider, BuiltLlmClient, ProviderField, ProviderUiMeta, ReasoningMode, ServiceType};
|
||||
@@ -111,7 +111,7 @@ impl ApiProvider for AnthropicProvider {
|
||||
// stays uncached, as before.
|
||||
let prompt_cache = model.capabilities.iter().any(|c| c == "tool_search");
|
||||
Ok(BuiltLlmClient {
|
||||
client: Arc::new(AnthropicClient::with_extra_body(key, extra)),
|
||||
client: Arc::new(AnthropicModel::with_extra_body(key, model.model_id.clone(), extra)),
|
||||
prompt_cache,
|
||||
})
|
||||
})())
|
||||
|
||||
@@ -18,7 +18,7 @@ use std::sync::Arc;
|
||||
use anyhow::{anyhow, Context, Result};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::chatbot::openai::OpenAiClient;
|
||||
use agent_loop::models::OpenAiModel;
|
||||
use crate::llm::providers::{extra_with_reasoning, RemoteLlmModelInfo};
|
||||
use crate::llm::{LlmModelRecord, LlmProviderRecord};
|
||||
use crate::provider::{
|
||||
@@ -575,7 +575,7 @@ impl ApiProvider for DeclaredProvider {
|
||||
let extra = extra_with_reasoning(self, model);
|
||||
let prompt_cache = self.spec.prompt_cache;
|
||||
Ok(BuiltLlmClient {
|
||||
client: Arc::new(OpenAiClient::new(self.base_url(record), key, extra, prompt_cache)),
|
||||
client: Arc::new(OpenAiModel::with_options(self.base_url(record), key, model.model_id.clone(), extra, prompt_cache)),
|
||||
prompt_cache,
|
||||
})
|
||||
})())
|
||||
|
||||
@@ -15,7 +15,7 @@ use anyhow::{anyhow, Context, Result};
|
||||
|
||||
use core_api::provider::{ApiProvider, BuiltLlmClient, LlmModelRecord, LlmProviderRecord};
|
||||
|
||||
use crate::chatbot::openai::OpenAiClient;
|
||||
use agent_loop::models::OpenAiModel;
|
||||
|
||||
/// Computes the `extra_params` an OpenAI-compatible client should be built with,
|
||||
/// given a model's stored `extra_params` and its selected reasoning value. The
|
||||
@@ -75,7 +75,7 @@ pub(crate) async fn fetch_openai_models(
|
||||
.ok_or_else(|| anyhow!("unexpected {who} response shape"))
|
||||
}
|
||||
|
||||
/// Builds an `OpenAiClient` for an OpenAI-compatible provider: requires the
|
||||
/// Builds an `OpenAiModel` for an OpenAI-compatible provider: requires the
|
||||
/// provider record's `api_key` and merges the model's stored `extra_params`
|
||||
/// with the provider-translated reasoning fragment (see `extra_with_reasoning`).
|
||||
pub(crate) fn build_openai_llm(
|
||||
@@ -89,7 +89,7 @@ pub(crate) fn build_openai_llm(
|
||||
.with_context(|| format!("provider '{}': api_key required for {}", record.name, provider.type_id()))?;
|
||||
let extra = extra_with_reasoning(provider, model);
|
||||
Ok(BuiltLlmClient {
|
||||
client: Arc::new(OpenAiClient::new(base_url, key, extra, prompt_cache)),
|
||||
client: Arc::new(OpenAiModel::with_options(base_url, key, model.model_id.clone(), extra, prompt_cache)),
|
||||
prompt_cache,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ use std::sync::Arc;
|
||||
|
||||
use anyhow::{Result, anyhow};
|
||||
|
||||
use crate::chatbot::ollama::OllamaClient;
|
||||
use agent_loop::models::OllamaModel;
|
||||
use crate::llm::{LlmModelRecord, LlmProviderRecord};
|
||||
use crate::llm::providers::RemoteLlmModelInfo;
|
||||
use crate::provider::{ApiProvider, BuiltLlmClient, ProviderField, ProviderUiMeta, ServiceType};
|
||||
@@ -101,9 +101,9 @@ impl ApiProvider for OllamaProvider {
|
||||
Ok(Some(Self::parse_model_info(&resp, model_id)))
|
||||
}
|
||||
|
||||
fn build_llm(&self, record: &LlmProviderRecord, _model: &LlmModelRecord) -> Option<Result<BuiltLlmClient>> {
|
||||
fn build_llm(&self, record: &LlmProviderRecord, model: &LlmModelRecord) -> Option<Result<BuiltLlmClient>> {
|
||||
Some(Ok(BuiltLlmClient {
|
||||
client: Arc::new(OllamaClient::new(record.base_url.as_deref())),
|
||||
client: Arc::new(OllamaModel::new(record.base_url.as_deref(), model.model_id.clone())),
|
||||
prompt_cache: false,
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -3,17 +3,21 @@
|
||||
//! Extracted from `run_agent_turn`: on a retriable error (5xx / network) it retries
|
||||
//! up to `MAX_LLM_ATTEMPTS` models in priority order, rebuilding the message list
|
||||
//! when the replacement model has a different `prompt_cache` setting, and emits
|
||||
//! `ModelFallback` / `LlmFailed` along the way.
|
||||
//! `ModelFallback` / `LlmFailed` along the way. The call itself goes through the
|
||||
//! `agent_loop::model::Model` trait (blueprint D13) — clients and protocols live
|
||||
//! in the `agent-loop` crate.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
|
||||
use serde_json::Value;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, warn};
|
||||
|
||||
use crate::chatbot::{ChatOptions, LlmError, LlmTurn, StreamDelta};
|
||||
use agent_loop::ids::{ConversationId, FrameId};
|
||||
use agent_loop::model::{ModelRequest, ModelResponse, StreamDelta};
|
||||
|
||||
use crate::db::llm_request_payloads;
|
||||
use crate::events::{ServerEvent, TokenDeltaKind};
|
||||
use crate::llm::{LlmEntry, LlmStrength};
|
||||
@@ -24,8 +28,9 @@ use super::interface_tools::AgentRunConfig;
|
||||
|
||||
/// Outcome of one round's LLM call.
|
||||
pub(super) enum RoundLlm {
|
||||
/// The model responded (message or tool calls).
|
||||
Turn(LlmTurn),
|
||||
/// The model responded (message or tool calls). Boxed: `ModelResponse`
|
||||
/// dwarfs the other variants.
|
||||
Turn(Box<ModelResponse>),
|
||||
/// The turn was cancelled (`/stop`) while the request was in flight.
|
||||
Cancelled,
|
||||
/// All fallback attempts were exhausted, or an error is non-retriable.
|
||||
@@ -59,15 +64,6 @@ impl ChatSessionHandler {
|
||||
// a fallback across DTL modes must re-shape (deferred candidates or not).
|
||||
let cur_tool_defs = config.all_tool_defs(cur_llm.dtl);
|
||||
let request_id = uuid::Uuid::new_v4().to_string();
|
||||
let options = ChatOptions {
|
||||
model: cur_llm.model.clone(),
|
||||
max_tokens: None,
|
||||
temperature: None,
|
||||
session_id: Some(self.session_id),
|
||||
stack_id: Some(stack_id),
|
||||
user_id: Some(self.user_id.clone()),
|
||||
request_id: Some(request_id.clone()),
|
||||
};
|
||||
|
||||
// Tell the model, in read_file's description, which media formats it can
|
||||
// open directly — keyed on the model actually serving this attempt, so a
|
||||
@@ -80,6 +76,23 @@ impl ChatSessionHandler {
|
||||
// the fallback reassignment below. On cancel we drop the future
|
||||
// (aborting the request) and return immediately.
|
||||
let client = cur_llm.client.clone();
|
||||
let request = ModelRequest {
|
||||
messages: messages.clone(),
|
||||
tools: defs.to_vec(),
|
||||
model: cur_llm.model.clone(),
|
||||
max_tokens: None,
|
||||
temperature: None,
|
||||
request_id: request_id.clone(),
|
||||
conversation: ConversationId::new(format!("session:{}", self.session_id)),
|
||||
frame: FrameId(stack_id),
|
||||
extras: Value::Null,
|
||||
// Correlation for the LoggingModel decorator (never sent).
|
||||
log: Some(json!({
|
||||
"session_id": self.session_id,
|
||||
"stack_id": stack_id,
|
||||
"user_id": self.user_id,
|
||||
})),
|
||||
};
|
||||
// Streaming side-channel: providers that support SSE push deltas here;
|
||||
// the forwarder re-emits them as `TokenDelta` events on the turn bus.
|
||||
// Best-effort — the round's final events remain authoritative.
|
||||
@@ -87,7 +100,7 @@ impl ChatSessionHandler {
|
||||
let forwarder = spawn_delta_forwarder(delta_rx, em.sender());
|
||||
let call_result = tokio::select! {
|
||||
_ = token.cancelled() => return RoundLlm::Cancelled,
|
||||
r = client.chat_with_tools_raw_streaming(messages.as_slice(), defs, &options, delta_tx) => r,
|
||||
r = client.complete(&request, Some(delta_tx)) => r,
|
||||
};
|
||||
// The client's sender dropped with the completed future: the forwarder
|
||||
// drains any queued deltas and exits, so every `TokenDelta` precedes the
|
||||
@@ -95,39 +108,39 @@ impl ChatSessionHandler {
|
||||
forwarder.await.ok();
|
||||
|
||||
let e = match call_result {
|
||||
Ok((turn, meta)) => {
|
||||
Ok(resp) => {
|
||||
self.llm_manager.mark_success(cur_name).await;
|
||||
// Persist the payload (request/response bodies + headers) to the
|
||||
// user's own database. Fire-and-forget — a failed write must not
|
||||
// break the turn. The metadata row is already written by the
|
||||
// logging wrapper to system.db with the same request_id.
|
||||
if let Some(meta) = meta {
|
||||
// LoggingModel decorator to system.db with the same request_id.
|
||||
if let Some(meta) = resp.raw() {
|
||||
let pool = Arc::clone(&self.db);
|
||||
let rid = request_id.clone();
|
||||
tokio::spawn(async move {
|
||||
let row = llm_request_payloads::PayloadRow {
|
||||
request_id: rid,
|
||||
request_json: meta.request_body.map(|v| v.to_string()).unwrap_or_default(),
|
||||
request_headers: meta.request_headers.map(|v| v.to_string()),
|
||||
response_json: meta.response_body.map(|v| v.to_string()),
|
||||
response_headers: meta.response_headers.map(|v| v.to_string()),
|
||||
request_json: meta.request_body.as_ref().map(|v| v.to_string()).unwrap_or_default(),
|
||||
request_headers: meta.request_headers.as_ref().map(|v| v.to_string()),
|
||||
response_json: meta.response_body.as_ref().map(|v| v.to_string()),
|
||||
response_headers: meta.response_headers.as_ref().map(|v| v.to_string()),
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = llm_request_payloads::insert(&pool, row).await {
|
||||
tracing::warn!(error = %e, "llm_request_payloads: failed to insert");
|
||||
}
|
||||
});
|
||||
}
|
||||
return RoundLlm::Turn(turn);
|
||||
return RoundLlm::Turn(Box::new(resp));
|
||||
}
|
||||
Err(e) => e,
|
||||
};
|
||||
|
||||
// Persist the payload even on failure so the debug log shows the request
|
||||
// that was rejected (e.g. a provider 400). Only the HTTP clients attach a
|
||||
// body (`LlmError::raw_meta`); a network/parse/cancel error carries none.
|
||||
// that was rejected (e.g. a provider 400). Only HTTP failures attach a
|
||||
// body (`ModelError::raw`); a network/parse/cancel error carries none.
|
||||
// Fire-and-forget, keyed on the same `request_id` as the metadata row the
|
||||
// logging wrapper wrote to system.db.
|
||||
if let Some(meta) = e.downcast_ref::<LlmError>().and_then(|le| le.raw_meta.as_ref()) {
|
||||
// LoggingModel decorator wrote to system.db.
|
||||
if let Some(meta) = e.raw.as_ref() {
|
||||
let row = llm_request_payloads::PayloadRow {
|
||||
request_id: request_id.clone(),
|
||||
request_json: meta.request_body.as_ref().map(|v| v.to_string()).unwrap_or_default(),
|
||||
@@ -147,10 +160,10 @@ impl ChatSessionHandler {
|
||||
self.llm_manager.mark_failure(cur_name, &e.to_string()).await;
|
||||
|
||||
let can_fallback = tried_this_round.len() < MAX_LLM_ATTEMPTS
|
||||
&& is_retriable_llm_error(&e);
|
||||
&& client.is_retriable(&e);
|
||||
if !can_fallback {
|
||||
em.llm_failed(tried_this_round.clone(), e.to_string()).await;
|
||||
return RoundLlm::Failed(e);
|
||||
return RoundLlm::Failed(e.into());
|
||||
}
|
||||
|
||||
let excluded: Vec<&str> = tried_this_round.iter().map(String::as_str).collect();
|
||||
@@ -179,7 +192,7 @@ impl ChatSessionHandler {
|
||||
}
|
||||
Err(_) => {
|
||||
em.llm_failed(tried_this_round.clone(), e.to_string()).await;
|
||||
return RoundLlm::Failed(e);
|
||||
return RoundLlm::Failed(e.into());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -206,21 +219,6 @@ fn spawn_delta_forwarder(
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether an LLM error is worth retrying on a different model.
|
||||
///
|
||||
/// Classifies on the real HTTP status ([`crate::chatbot::http_status`]), not a
|
||||
/// substring of the message — a model id or token count containing "404"/"401" no
|
||||
/// longer mis-classifies (bug B6). A non-HTTP failure (network, parse) has no status
|
||||
/// and is retriable, matching the previous default.
|
||||
fn is_retriable_llm_error(e: &anyhow::Error) -> bool {
|
||||
// Never retry these client errors — the request itself is unauthorized, not
|
||||
// found, or unprocessable. 400 is intentionally NOT listed: some providers
|
||||
// reject valid requests that others accept (e.g. DeepSeek requires a
|
||||
// reasoning_content echo, OpenAI does not), so retrying elsewhere can succeed.
|
||||
// 429 and 5xx stay retriable (a different model / provider may serve the call).
|
||||
!matches!(crate::chatbot::http_status(e), Some(401 | 403 | 404 | 422))
|
||||
}
|
||||
|
||||
fn first_line(s: &str) -> String {
|
||||
s.lines().next().unwrap_or(s).to_string()
|
||||
}
|
||||
@@ -246,38 +244,40 @@ fn media_annotated_tools(tool_defs: &[Value], capabilities: &[String]) -> Option
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::is_retriable_llm_error;
|
||||
use crate::chatbot::LlmError;
|
||||
use agent_loop::model::{Model, ModelError, ModelRequest, ModelResponse, StreamDelta};
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
fn http_err(status: u16, message: &str) -> anyhow::Error {
|
||||
LlmError { status: Some(status), message: message.to_string(), ..Default::default() }.into()
|
||||
struct Dummy;
|
||||
#[async_trait]
|
||||
impl agent_loop::model::Model for Dummy {
|
||||
async fn complete(
|
||||
&self,
|
||||
_req: &ModelRequest,
|
||||
_d: Option<mpsc::Sender<StreamDelta>>,
|
||||
) -> Result<ModelResponse, ModelError> {
|
||||
unreachable!()
|
||||
}
|
||||
}
|
||||
|
||||
/// Retriability classification lives on the `Model` trait default (the crate
|
||||
/// owns the protocols, blueprint D13): 401/403/404/422 don't retry,
|
||||
/// 400/429/5xx/network do. Classification keys on the structured status,
|
||||
/// never on the message string (bug B6 regression).
|
||||
#[test]
|
||||
fn client_errors_are_not_retried() {
|
||||
fn retriability_keys_on_structured_status() {
|
||||
let m = Dummy;
|
||||
for code in [401, 403, 404, 422] {
|
||||
assert!(!is_retriable_llm_error(&http_err(code, "nope")), "{code} must not retry");
|
||||
assert!(!m.is_retriable(&ModelError::new(Some(code), "nope")), "{code} must not retry");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn server_rate_limit_and_400_retry() {
|
||||
for code in [400, 429, 500, 502, 503] {
|
||||
assert!(is_retriable_llm_error(&http_err(code, "retry")), "{code} must retry");
|
||||
assert!(m.is_retriable(&ModelError::new(Some(code), "retry")), "{code} must retry");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_http_errors_retry() {
|
||||
assert!(is_retriable_llm_error(&anyhow::anyhow!("connection reset by peer")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn status_digits_in_the_message_do_not_mislead() {
|
||||
// Regression for B6: the old substring check read any "404"/"401" in the text
|
||||
// as a client error. A 500 whose body mentions "1401 tokens" / "code 404" must
|
||||
// still retry — classification keys on the structured status, not the string.
|
||||
let e = http_err(500, "provider error: too many (1401) tokens, see code 404 in docs");
|
||||
assert!(is_retriable_llm_error(&e));
|
||||
// A 500 whose body mentions "1401 tokens" / "code 404" must still retry.
|
||||
assert!(m.is_retriable(&ModelError::new(
|
||||
Some(500),
|
||||
"provider error: too many (1401) tokens, see code 404 in docs"
|
||||
)));
|
||||
assert!(m.is_retriable(&ModelError::new(None, "connection reset by peer")));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, trace};
|
||||
|
||||
use crate::chat_event_bus::ToolCallEvent;
|
||||
use crate::chatbot::{LlmTurn, ToolCall};
|
||||
use agent_loop::model::{ModelResponse, ToolCall};
|
||||
use crate::db::{chat_history, chat_llm_tools};
|
||||
use crate::events::ServerEvent;
|
||||
use crate::tools::{
|
||||
@@ -139,36 +139,37 @@ impl ChatSessionHandler {
|
||||
RoundLlm::Failed(e) => return Err(e),
|
||||
};
|
||||
|
||||
match turn_result {
|
||||
LlmTurn::Message(resp) => {
|
||||
match *turn_result {
|
||||
ModelResponse::Message { content, reasoning, usage, .. } => {
|
||||
let message_id = chat_history::append(
|
||||
pool, stack_id, &chat_history::Role::Assistant, &resp.content, false,
|
||||
resp.reasoning_content.as_deref(),
|
||||
pool, stack_id, &chat_history::Role::Assistant, &content, false,
|
||||
reasoning.as_deref(),
|
||||
).await?;
|
||||
if let (Some(i), Some(o)) = (resp.input_tokens, resp.output_tokens) {
|
||||
chat_history::set_usage(pool, message_id, i, o, 0, resp.cost).await?;
|
||||
if let (Some(i), Some(o)) = (usage.input_tokens, usage.output_tokens) {
|
||||
chat_history::set_usage(pool, message_id, i, o, 0, usage.cost_usd).await?;
|
||||
}
|
||||
return Ok(TurnOutcome::Final {
|
||||
content: resp.content,
|
||||
content,
|
||||
message_id,
|
||||
input_tokens: resp.input_tokens,
|
||||
output_tokens: resp.output_tokens,
|
||||
truncated: resp.truncated,
|
||||
reasoning_content: resp.reasoning_content,
|
||||
input_tokens: usage.input_tokens,
|
||||
output_tokens: usage.output_tokens,
|
||||
truncated: usage.truncated,
|
||||
reasoning_content: reasoning,
|
||||
tool_calls: all_tool_calls,
|
||||
});
|
||||
}
|
||||
|
||||
LlmTurn::ToolCalls { content: assistant_text, calls, input_tokens, output_tokens, reasoning_content, cost, .. } => {
|
||||
ModelResponse::ToolCalls { content: assistant_text, calls, usage, reasoning, .. } => {
|
||||
let (input_tokens, output_tokens) = (usage.input_tokens, usage.output_tokens);
|
||||
let message_id = chat_history::append(
|
||||
pool, stack_id, &chat_history::Role::Assistant, &assistant_text, false,
|
||||
reasoning_content.as_deref(),
|
||||
reasoning.as_deref(),
|
||||
).await?;
|
||||
if let (Some(i), Some(o)) = (input_tokens, output_tokens) {
|
||||
chat_history::set_usage(pool, message_id, i, o, 0, cost).await?;
|
||||
chat_history::set_usage(pool, message_id, i, o, 0, usage.cost_usd).await?;
|
||||
}
|
||||
if !assistant_text.trim().is_empty() || input_tokens.is_some() {
|
||||
em.thinking(message_id, assistant_text, input_tokens, output_tokens, reasoning_content).await;
|
||||
em.thinking(message_id, assistant_text, input_tokens, output_tokens, reasoning).await;
|
||||
}
|
||||
|
||||
// A homogeneous batch of ≥2 synchronous sub-agent calls is fanned
|
||||
|
||||
Reference in New Issue
Block a user