M3: OpenAI chat and responses codecs, provider, registry wiring
Adds both OpenAI codecs — chat completions and the newer responses API — plus the OpenAiProvider and registry wiring, so the same session can run against OpenAI in addition to Anthropic.
This commit is contained in:
@@ -19,7 +19,7 @@ use harness_core::tool::ToolRegistry;
|
|||||||
use harness_core::types::{
|
use harness_core::types::{
|
||||||
Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId,
|
Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId,
|
||||||
};
|
};
|
||||||
use harness_providers::{AnthropicProvider, ProviderRegistry};
|
use harness_providers::{AnthropicProvider, OpenAiProvider, ProviderRegistry};
|
||||||
use tokio::task::JoinHandle;
|
use tokio::task::JoinHandle;
|
||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
|
|
||||||
@@ -122,6 +122,21 @@ impl EngineHandle {
|
|||||||
{
|
{
|
||||||
providers.register(Arc::new(AnthropicProvider::new(key)));
|
providers.register(Arc::new(AnthropicProvider::new(key)));
|
||||||
}
|
}
|
||||||
|
if let Some(key) = config
|
||||||
|
.providers
|
||||||
|
.get("openai")
|
||||||
|
.and_then(|p| p.api_key.clone())
|
||||||
|
{
|
||||||
|
let provider = match config
|
||||||
|
.providers
|
||||||
|
.get("openai")
|
||||||
|
.and_then(|p| p.base_url.clone())
|
||||||
|
{
|
||||||
|
Some(base_url) => OpenAiProvider::with_base_url(key, base_url),
|
||||||
|
None => OpenAiProvider::new(key),
|
||||||
|
};
|
||||||
|
providers.register(Arc::new(provider));
|
||||||
|
}
|
||||||
|
|
||||||
let data_dir = dirs::data_dir()
|
let data_dir = dirs::data_dir()
|
||||||
.unwrap_or_else(std::env::temp_dir)
|
.unwrap_or_else(std::env::temp_dir)
|
||||||
|
|||||||
@@ -1 +1,3 @@
|
|||||||
pub mod anthropic;
|
pub mod anthropic;
|
||||||
|
pub mod openai_chat;
|
||||||
|
pub mod openai_responses;
|
||||||
|
|||||||
@@ -0,0 +1,502 @@
|
|||||||
|
//! Request builder + SSE decoder for OpenAI's `/chat/completions` streaming API.
|
||||||
|
//!
|
||||||
|
//! Also used as the fallback codec for OpenAI-compatible endpoints (including Copilot models
|
||||||
|
//! whose `supported_endpoints` list `/chat/completions`).
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
|
||||||
|
use async_stream::try_stream;
|
||||||
|
use eventsource_stream::Eventsource;
|
||||||
|
use futures::Stream;
|
||||||
|
use harness_core::llm::{
|
||||||
|
FinishReason, LlmEvent, LlmEventStream, LlmRequest, ProviderError, ReasoningEffort, Role,
|
||||||
|
WireContent,
|
||||||
|
};
|
||||||
|
use harness_core::types::TokenUsage;
|
||||||
|
use serde_json::{json, Value};
|
||||||
|
|
||||||
|
fn effort_str(effort: ReasoningEffort) -> &'static str {
|
||||||
|
match effort {
|
||||||
|
ReasoningEffort::Low => "low",
|
||||||
|
ReasoningEffort::Medium => "medium",
|
||||||
|
ReasoningEffort::High => "high",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn role_str(role: Role) -> &'static str {
|
||||||
|
match role {
|
||||||
|
Role::System => "system",
|
||||||
|
Role::User => "user",
|
||||||
|
Role::Assistant => "assistant",
|
||||||
|
Role::Tool => "tool",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Flattens our grouped `WireMessage`s into the flat chat-completions message list. Tool
|
||||||
|
/// results become their own `role: "tool"` messages (one per result), which is what the API
|
||||||
|
/// expects regardless of how the engine grouped them.
|
||||||
|
fn build_messages(system: &[String], messages: &[harness_core::llm::WireMessage]) -> Vec<Value> {
|
||||||
|
let mut out: Vec<Value> = Vec::new();
|
||||||
|
|
||||||
|
if !system.is_empty() {
|
||||||
|
out.push(json!({"role": "system", "content": system.join("\n\n")}));
|
||||||
|
}
|
||||||
|
|
||||||
|
for m in messages {
|
||||||
|
// Tool results are always emitted as standalone `tool` messages.
|
||||||
|
for c in &m.content {
|
||||||
|
if let WireContent::ToolResult {
|
||||||
|
call_id, output, ..
|
||||||
|
} = c
|
||||||
|
{
|
||||||
|
out.push(json!({
|
||||||
|
"role": "tool",
|
||||||
|
"tool_call_id": call_id,
|
||||||
|
"content": output,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut text = String::new();
|
||||||
|
let mut images: Vec<Value> = Vec::new();
|
||||||
|
let mut tool_calls: Vec<Value> = Vec::new();
|
||||||
|
for c in &m.content {
|
||||||
|
match c {
|
||||||
|
WireContent::Text { text: t } => {
|
||||||
|
if !text.is_empty() {
|
||||||
|
text.push('\n');
|
||||||
|
}
|
||||||
|
text.push_str(t);
|
||||||
|
}
|
||||||
|
WireContent::ToolCall {
|
||||||
|
call_id,
|
||||||
|
name,
|
||||||
|
input,
|
||||||
|
} => tool_calls.push(json!({
|
||||||
|
"id": call_id,
|
||||||
|
"type": "function",
|
||||||
|
"function": {"name": name, "arguments": input.to_string()},
|
||||||
|
})),
|
||||||
|
WireContent::Image { mime_type, data } => images.push(json!({
|
||||||
|
"type": "image_url",
|
||||||
|
"image_url": {"url": format!("data:{mime_type};base64,{data}")},
|
||||||
|
})),
|
||||||
|
WireContent::ToolResult { .. } => {} // handled above
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A message that carried nothing but tool results contributes no further entry.
|
||||||
|
if text.is_empty() && images.is_empty() && tool_calls.is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut msg = json!({"role": role_str(m.role)});
|
||||||
|
if images.is_empty() {
|
||||||
|
msg["content"] = json!(text);
|
||||||
|
} else {
|
||||||
|
let mut parts = vec![json!({"type": "text", "text": text})];
|
||||||
|
parts.extend(images);
|
||||||
|
msg["content"] = json!(parts);
|
||||||
|
}
|
||||||
|
if !tool_calls.is_empty() {
|
||||||
|
msg["tool_calls"] = json!(tool_calls);
|
||||||
|
}
|
||||||
|
out.push(msg);
|
||||||
|
}
|
||||||
|
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn build_request(req: &LlmRequest) -> Value {
|
||||||
|
let mut body = json!({
|
||||||
|
"model": req.model,
|
||||||
|
"messages": build_messages(&req.system, &req.messages),
|
||||||
|
"stream": true,
|
||||||
|
"stream_options": {"include_usage": true},
|
||||||
|
});
|
||||||
|
|
||||||
|
if !req.tools.is_empty() {
|
||||||
|
body["tools"] = Value::Array(
|
||||||
|
req.tools
|
||||||
|
.iter()
|
||||||
|
.map(|t| {
|
||||||
|
json!({
|
||||||
|
"type": "function",
|
||||||
|
"function": {
|
||||||
|
"name": t.name,
|
||||||
|
"description": t.description,
|
||||||
|
"parameters": t.parameters,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
})
|
||||||
|
.collect(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if let Some(temp) = req.temperature {
|
||||||
|
body["temperature"] = json!(temp);
|
||||||
|
}
|
||||||
|
if let Some(max) = req.max_tokens {
|
||||||
|
body["max_completion_tokens"] = json!(max);
|
||||||
|
}
|
||||||
|
if let Some(effort) = req.reasoning.as_ref().and_then(|r| r.effort) {
|
||||||
|
body["reasoning_effort"] = json!(effort_str(effort));
|
||||||
|
}
|
||||||
|
body
|
||||||
|
}
|
||||||
|
|
||||||
|
fn map_finish_reason(reason: &str) -> FinishReason {
|
||||||
|
match reason {
|
||||||
|
"stop" => FinishReason::Stop,
|
||||||
|
"tool_calls" | "function_call" => FinishReason::ToolCalls,
|
||||||
|
"length" => FinishReason::Length,
|
||||||
|
"content_filter" => FinishReason::ContentFilter,
|
||||||
|
other => FinishReason::Unknown(other.to_string()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct ToolAccum {
|
||||||
|
call_id: String,
|
||||||
|
name: String,
|
||||||
|
args: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Decodes a chat-completions SSE byte stream into normalized `LlmEvent`s. Tool-call deltas
|
||||||
|
/// are accumulated by their `index` and flushed as `ToolCall`s once the stream ends.
|
||||||
|
pub fn decode<S, E>(byte_stream: S) -> LlmEventStream
|
||||||
|
where
|
||||||
|
S: Stream<Item = Result<bytes::Bytes, E>> + Send + 'static,
|
||||||
|
E: std::error::Error + Send + Sync + 'static,
|
||||||
|
{
|
||||||
|
let events = byte_stream.eventsource();
|
||||||
|
Box::pin(try_stream! {
|
||||||
|
futures::pin_mut!(events);
|
||||||
|
let mut usage = TokenUsage::default();
|
||||||
|
let mut reason = FinishReason::Stop;
|
||||||
|
let mut text_open = false;
|
||||||
|
let mut reasoning_open = false;
|
||||||
|
let mut tools: HashMap<u64, ToolAccum> = HashMap::new();
|
||||||
|
let mut tool_order: Vec<u64> = Vec::new();
|
||||||
|
|
||||||
|
while let Some(item) = futures::StreamExt::next(&mut events).await {
|
||||||
|
let event = item.map_err(|e| ProviderError::Decode(e.to_string()))?;
|
||||||
|
let data = event.data.trim();
|
||||||
|
if data.is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if data == "[DONE]" {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let value: Value = serde_json::from_str(data)
|
||||||
|
.map_err(|e| ProviderError::Decode(format!("{e}: {data}")))?;
|
||||||
|
|
||||||
|
if let Some(u) = value.get("usage").filter(|u| u.is_object()) {
|
||||||
|
usage.input = u["prompt_tokens"].as_u64().unwrap_or(usage.input);
|
||||||
|
usage.output = u["completion_tokens"].as_u64().unwrap_or(usage.output);
|
||||||
|
usage.reasoning = u["completion_tokens_details"]["reasoning_tokens"]
|
||||||
|
.as_u64()
|
||||||
|
.unwrap_or(usage.reasoning);
|
||||||
|
usage.cache_read = u["prompt_tokens_details"]["cached_tokens"]
|
||||||
|
.as_u64()
|
||||||
|
.unwrap_or(usage.cache_read);
|
||||||
|
}
|
||||||
|
|
||||||
|
let choice = &value["choices"][0];
|
||||||
|
let delta = &choice["delta"];
|
||||||
|
|
||||||
|
if let Some(rc) = delta["reasoning_content"].as_str().filter(|s| !s.is_empty()) {
|
||||||
|
if !reasoning_open {
|
||||||
|
reasoning_open = true;
|
||||||
|
yield LlmEvent::ReasoningStart { id: "reasoning".into() };
|
||||||
|
}
|
||||||
|
yield LlmEvent::ReasoningDelta { id: "reasoning".into(), text: rc.to_string() };
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(text) = delta["content"].as_str().filter(|s| !s.is_empty()) {
|
||||||
|
if reasoning_open {
|
||||||
|
reasoning_open = false;
|
||||||
|
yield LlmEvent::ReasoningEnd { id: "reasoning".into(), signature: None };
|
||||||
|
}
|
||||||
|
if !text_open {
|
||||||
|
text_open = true;
|
||||||
|
yield LlmEvent::TextStart { id: "0".into() };
|
||||||
|
}
|
||||||
|
yield LlmEvent::TextDelta { id: "0".into(), text: text.to_string() };
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(calls) = delta["tool_calls"].as_array() {
|
||||||
|
for call in calls {
|
||||||
|
let index = call["index"].as_u64().unwrap_or(0);
|
||||||
|
let entry = tools.entry(index).or_insert_with(|| {
|
||||||
|
tool_order.push(index);
|
||||||
|
ToolAccum::default()
|
||||||
|
});
|
||||||
|
if let Some(id) = call["id"].as_str().filter(|s| !s.is_empty()) {
|
||||||
|
entry.call_id = id.to_string();
|
||||||
|
}
|
||||||
|
if let Some(name) = call["function"]["name"].as_str().filter(|s| !s.is_empty()) {
|
||||||
|
entry.name = name.to_string();
|
||||||
|
yield LlmEvent::ToolInputStart {
|
||||||
|
call_id: entry.call_id.clone(),
|
||||||
|
name: entry.name.clone(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
if let Some(args) = call["function"]["arguments"].as_str().filter(|s| !s.is_empty()) {
|
||||||
|
entry.args.push_str(args);
|
||||||
|
yield LlmEvent::ToolInputDelta {
|
||||||
|
call_id: entry.call_id.clone(),
|
||||||
|
json: args.to_string(),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(fr) = choice["finish_reason"].as_str() {
|
||||||
|
reason = map_finish_reason(fr);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if reasoning_open {
|
||||||
|
yield LlmEvent::ReasoningEnd { id: "reasoning".into(), signature: None };
|
||||||
|
}
|
||||||
|
if text_open {
|
||||||
|
yield LlmEvent::TextEnd { id: "0".into() };
|
||||||
|
}
|
||||||
|
for index in tool_order {
|
||||||
|
if let Some(entry) = tools.remove(&index) {
|
||||||
|
let input: Value = if entry.args.trim().is_empty() {
|
||||||
|
json!({})
|
||||||
|
} else {
|
||||||
|
serde_json::from_str(&entry.args).unwrap_or(Value::Null)
|
||||||
|
};
|
||||||
|
yield LlmEvent::ToolCall { call_id: entry.call_id, name: entry.name, input };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
yield LlmEvent::Finish { reason, usage };
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use futures::StreamExt;
|
||||||
|
use harness_core::llm::{Initiator, ReasoningOpts, ToolSchema, WireMessage};
|
||||||
|
|
||||||
|
fn sse_stream(raw: &'static str) -> LlmEventStream {
|
||||||
|
let chunks: Vec<Result<bytes::Bytes, std::io::Error>> =
|
||||||
|
vec![Ok(bytes::Bytes::from_static(raw.as_bytes()))];
|
||||||
|
decode(futures::stream::iter(chunks))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn req(model: &str) -> LlmRequest {
|
||||||
|
LlmRequest {
|
||||||
|
model: model.into(),
|
||||||
|
system: vec![],
|
||||||
|
messages: vec![],
|
||||||
|
tools: vec![],
|
||||||
|
temperature: None,
|
||||||
|
max_tokens: None,
|
||||||
|
reasoning: None,
|
||||||
|
initiator: Initiator::User,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_text_only_response() {
|
||||||
|
let raw = concat!(
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"content\":\"Hel\"},\"finish_reason\":null}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"content\":\"lo\"},\"finish_reason\":null}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
|
||||||
|
"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":5}}\n\n",
|
||||||
|
"data: [DONE]\n\n",
|
||||||
|
);
|
||||||
|
let events: Vec<LlmEvent> = sse_stream(raw).map(|e| e.unwrap()).collect().await;
|
||||||
|
assert_eq!(
|
||||||
|
events,
|
||||||
|
vec![
|
||||||
|
LlmEvent::TextStart { id: "0".into() },
|
||||||
|
LlmEvent::TextDelta {
|
||||||
|
id: "0".into(),
|
||||||
|
text: "Hel".into()
|
||||||
|
},
|
||||||
|
LlmEvent::TextDelta {
|
||||||
|
id: "0".into(),
|
||||||
|
text: "lo".into()
|
||||||
|
},
|
||||||
|
LlmEvent::TextEnd { id: "0".into() },
|
||||||
|
LlmEvent::Finish {
|
||||||
|
reason: FinishReason::Stop,
|
||||||
|
usage: TokenUsage {
|
||||||
|
input: 10,
|
||||||
|
output: 5,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_tool_call_accumulated_by_index() {
|
||||||
|
let raw = concat!(
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"read\",\"arguments\":\"\"}}]}}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"file\\\"\"}}]}}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\":\\\"a.txt\\\"}\"}}]}}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":8}}\n\n",
|
||||||
|
"data: [DONE]\n\n",
|
||||||
|
);
|
||||||
|
let events: Vec<LlmEvent> = sse_stream(raw).map(|e| e.unwrap()).collect().await;
|
||||||
|
assert_eq!(
|
||||||
|
events,
|
||||||
|
vec![
|
||||||
|
LlmEvent::ToolInputStart {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolInputDelta {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
json: "{\"file\"".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolInputDelta {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
json: ":\"a.txt\"}".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolCall {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into(),
|
||||||
|
input: json!({"file": "a.txt"}),
|
||||||
|
},
|
||||||
|
LlmEvent::Finish {
|
||||||
|
reason: FinishReason::ToolCalls,
|
||||||
|
usage: TokenUsage {
|
||||||
|
input: 1,
|
||||||
|
output: 8,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_reasoning_content_before_text() {
|
||||||
|
let raw = concat!(
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"hmm\"}}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n",
|
||||||
|
"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
|
||||||
|
"data: [DONE]\n\n",
|
||||||
|
);
|
||||||
|
let events: Vec<LlmEvent> = sse_stream(raw).map(|e| e.unwrap()).collect().await;
|
||||||
|
assert_eq!(
|
||||||
|
events,
|
||||||
|
vec![
|
||||||
|
LlmEvent::ReasoningStart {
|
||||||
|
id: "reasoning".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ReasoningDelta {
|
||||||
|
id: "reasoning".into(),
|
||||||
|
text: "hmm".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ReasoningEnd {
|
||||||
|
id: "reasoning".into(),
|
||||||
|
signature: None
|
||||||
|
},
|
||||||
|
LlmEvent::TextStart { id: "0".into() },
|
||||||
|
LlmEvent::TextDelta {
|
||||||
|
id: "0".into(),
|
||||||
|
text: "answer".into()
|
||||||
|
},
|
||||||
|
LlmEvent::TextEnd { id: "0".into() },
|
||||||
|
LlmEvent::Finish {
|
||||||
|
reason: FinishReason::Stop,
|
||||||
|
usage: TokenUsage::default()
|
||||||
|
},
|
||||||
|
]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_wraps_tools_in_function_envelope() {
|
||||||
|
let mut r = req("gpt-4o");
|
||||||
|
r.tools = vec![ToolSchema {
|
||||||
|
name: "read".into(),
|
||||||
|
description: "reads a file".into(),
|
||||||
|
parameters: json!({"type": "object"}),
|
||||||
|
}];
|
||||||
|
let body = build_request(&r);
|
||||||
|
assert_eq!(body["tools"][0]["type"], "function");
|
||||||
|
assert_eq!(body["tools"][0]["function"]["name"], "read");
|
||||||
|
assert_eq!(
|
||||||
|
body["tools"][0]["function"]["parameters"],
|
||||||
|
json!({"type": "object"})
|
||||||
|
);
|
||||||
|
assert_eq!(body["stream_options"]["include_usage"], true);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_prepends_system_message() {
|
||||||
|
let mut r = req("gpt-4o");
|
||||||
|
r.system = vec!["env".into(), "agent".into()];
|
||||||
|
r.messages = vec![WireMessage {
|
||||||
|
role: Role::User,
|
||||||
|
content: vec![WireContent::Text { text: "hi".into() }],
|
||||||
|
}];
|
||||||
|
let body = build_request(&r);
|
||||||
|
let msgs = body["messages"].as_array().unwrap();
|
||||||
|
assert_eq!(msgs[0]["role"], "system");
|
||||||
|
assert_eq!(msgs[0]["content"], "env\n\nagent");
|
||||||
|
assert_eq!(msgs[1]["role"], "user");
|
||||||
|
assert_eq!(msgs[1]["content"], "hi");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_expands_tool_results_to_tool_messages() {
|
||||||
|
let mut r = req("gpt-4o");
|
||||||
|
r.messages = vec![WireMessage {
|
||||||
|
role: Role::Tool,
|
||||||
|
content: vec![WireContent::ToolResult {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
output: "contents".into(),
|
||||||
|
is_error: false,
|
||||||
|
}],
|
||||||
|
}];
|
||||||
|
let body = build_request(&r);
|
||||||
|
let msgs = body["messages"].as_array().unwrap();
|
||||||
|
assert_eq!(msgs.len(), 1);
|
||||||
|
assert_eq!(msgs[0]["role"], "tool");
|
||||||
|
assert_eq!(msgs[0]["tool_call_id"], "call_1");
|
||||||
|
assert_eq!(msgs[0]["content"], "contents");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_emits_assistant_tool_calls() {
|
||||||
|
let mut r = req("gpt-4o");
|
||||||
|
r.messages = vec![WireMessage {
|
||||||
|
role: Role::Assistant,
|
||||||
|
content: vec![WireContent::ToolCall {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into(),
|
||||||
|
input: json!({"file": "a.txt"}),
|
||||||
|
}],
|
||||||
|
}];
|
||||||
|
let body = build_request(&r);
|
||||||
|
let msgs = body["messages"].as_array().unwrap();
|
||||||
|
assert_eq!(msgs[0]["tool_calls"][0]["id"], "call_1");
|
||||||
|
assert_eq!(msgs[0]["tool_calls"][0]["function"]["name"], "read");
|
||||||
|
assert_eq!(
|
||||||
|
msgs[0]["tool_calls"][0]["function"]["arguments"],
|
||||||
|
"{\"file\":\"a.txt\"}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_sets_reasoning_effort() {
|
||||||
|
let mut r = req("o3");
|
||||||
|
r.reasoning = Some(ReasoningOpts {
|
||||||
|
effort: Some(ReasoningEffort::High),
|
||||||
|
budget_tokens: None,
|
||||||
|
});
|
||||||
|
let body = build_request(&r);
|
||||||
|
assert_eq!(body["reasoning_effort"], "high");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,402 @@
|
|||||||
|
//! Request builder + SSE decoder for OpenAI's `/responses` streaming API.
|
||||||
|
//!
|
||||||
|
//! The responses API is the preferred surface for `gpt-*` / `o-*` models: it carries
|
||||||
|
//! reasoning items natively and reports reasoning-token usage separately.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
|
||||||
|
use async_stream::try_stream;
|
||||||
|
use eventsource_stream::Eventsource;
|
||||||
|
use futures::Stream;
|
||||||
|
use harness_core::llm::{
|
||||||
|
FinishReason, LlmEvent, LlmEventStream, LlmRequest, ProviderError, ReasoningEffort, Role,
|
||||||
|
WireContent,
|
||||||
|
};
|
||||||
|
use harness_core::types::TokenUsage;
|
||||||
|
use serde_json::{json, Value};
|
||||||
|
|
||||||
|
fn effort_str(effort: ReasoningEffort) -> &'static str {
|
||||||
|
match effort {
|
||||||
|
ReasoningEffort::Low => "low",
|
||||||
|
ReasoningEffort::Medium => "medium",
|
||||||
|
ReasoningEffort::High => "high",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Builds the `input` item list. Text/image content become `message` items; tool calls and
|
||||||
|
/// their results become `function_call` / `function_call_output` items (the responses API
|
||||||
|
/// keeps these as top-level items rather than nesting them inside messages).
|
||||||
|
fn build_input(messages: &[harness_core::llm::WireMessage]) -> Vec<Value> {
|
||||||
|
let mut out: Vec<Value> = Vec::new();
|
||||||
|
for m in messages {
|
||||||
|
let (role, text_type) = match m.role {
|
||||||
|
Role::Assistant => ("assistant", "output_text"),
|
||||||
|
_ => ("user", "input_text"),
|
||||||
|
};
|
||||||
|
let mut content_parts: Vec<Value> = Vec::new();
|
||||||
|
for c in &m.content {
|
||||||
|
match c {
|
||||||
|
WireContent::Text { text } => {
|
||||||
|
content_parts.push(json!({"type": text_type, "text": text}));
|
||||||
|
}
|
||||||
|
WireContent::Image { mime_type, data } => {
|
||||||
|
content_parts.push(json!({
|
||||||
|
"type": "input_image",
|
||||||
|
"image_url": format!("data:{mime_type};base64,{data}"),
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
WireContent::ToolCall {
|
||||||
|
call_id,
|
||||||
|
name,
|
||||||
|
input,
|
||||||
|
} => out.push(json!({
|
||||||
|
"type": "function_call",
|
||||||
|
"call_id": call_id,
|
||||||
|
"name": name,
|
||||||
|
"arguments": input.to_string(),
|
||||||
|
})),
|
||||||
|
WireContent::ToolResult {
|
||||||
|
call_id, output, ..
|
||||||
|
} => out.push(json!({
|
||||||
|
"type": "function_call_output",
|
||||||
|
"call_id": call_id,
|
||||||
|
"output": output,
|
||||||
|
})),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !content_parts.is_empty() {
|
||||||
|
out.push(json!({"type": "message", "role": role, "content": content_parts}));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn build_request(req: &LlmRequest) -> Value {
|
||||||
|
let mut body = json!({
|
||||||
|
"model": req.model,
|
||||||
|
"input": build_input(&req.messages),
|
||||||
|
"stream": true,
|
||||||
|
"store": false,
|
||||||
|
});
|
||||||
|
|
||||||
|
if !req.system.is_empty() {
|
||||||
|
body["instructions"] = json!(req.system.join("\n\n"));
|
||||||
|
}
|
||||||
|
if !req.tools.is_empty() {
|
||||||
|
body["tools"] = Value::Array(
|
||||||
|
req.tools
|
||||||
|
.iter()
|
||||||
|
.map(|t| {
|
||||||
|
json!({
|
||||||
|
"type": "function",
|
||||||
|
"name": t.name,
|
||||||
|
"description": t.description,
|
||||||
|
"parameters": t.parameters,
|
||||||
|
})
|
||||||
|
})
|
||||||
|
.collect(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if let Some(temp) = req.temperature {
|
||||||
|
body["temperature"] = json!(temp);
|
||||||
|
}
|
||||||
|
if let Some(max) = req.max_tokens {
|
||||||
|
body["max_output_tokens"] = json!(max);
|
||||||
|
}
|
||||||
|
if let Some(effort) = req.reasoning.as_ref().and_then(|r| r.effort) {
|
||||||
|
body["reasoning"] = json!({"effort": effort_str(effort), "summary": "auto"});
|
||||||
|
}
|
||||||
|
body
|
||||||
|
}
|
||||||
|
|
||||||
|
fn map_status(status: Option<&str>, had_tool_call: bool) -> FinishReason {
|
||||||
|
match status {
|
||||||
|
Some("completed") if had_tool_call => FinishReason::ToolCalls,
|
||||||
|
Some("completed") => FinishReason::Stop,
|
||||||
|
Some("incomplete") => FinishReason::Length,
|
||||||
|
Some(other) => FinishReason::Unknown(other.to_string()),
|
||||||
|
None if had_tool_call => FinishReason::ToolCalls,
|
||||||
|
None => FinishReason::Stop,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Tracks which normalized stream an `item_id` belongs to so deltas route correctly.
|
||||||
|
enum ItemKind {
|
||||||
|
Text,
|
||||||
|
Reasoning,
|
||||||
|
FunctionCall { call_id: String, name: String },
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Decodes a `/responses` SSE byte stream into normalized `LlmEvent`s.
|
||||||
|
pub fn decode<S, E>(byte_stream: S) -> LlmEventStream
|
||||||
|
where
|
||||||
|
S: Stream<Item = Result<bytes::Bytes, E>> + Send + 'static,
|
||||||
|
E: std::error::Error + Send + Sync + 'static,
|
||||||
|
{
|
||||||
|
let events = byte_stream.eventsource();
|
||||||
|
Box::pin(try_stream! {
|
||||||
|
futures::pin_mut!(events);
|
||||||
|
let mut items: HashMap<String, ItemKind> = HashMap::new();
|
||||||
|
let mut fn_args: HashMap<String, String> = HashMap::new();
|
||||||
|
let mut usage = TokenUsage::default();
|
||||||
|
let mut had_tool_call = false;
|
||||||
|
|
||||||
|
while let Some(item) = futures::StreamExt::next(&mut events).await {
|
||||||
|
let event = item.map_err(|e| ProviderError::Decode(e.to_string()))?;
|
||||||
|
if event.data.trim().is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let value: Value = serde_json::from_str(&event.data)
|
||||||
|
.map_err(|e| ProviderError::Decode(format!("{e}: {}", event.data)))?;
|
||||||
|
let kind = value["type"].as_str().unwrap_or_default();
|
||||||
|
|
||||||
|
match kind {
|
||||||
|
"response.output_item.added" => {
|
||||||
|
let item = &value["item"];
|
||||||
|
let id = item["id"].as_str().unwrap_or_default().to_string();
|
||||||
|
match item["type"].as_str().unwrap_or_default() {
|
||||||
|
"message" => {
|
||||||
|
items.insert(id.clone(), ItemKind::Text);
|
||||||
|
yield LlmEvent::TextStart { id };
|
||||||
|
}
|
||||||
|
"reasoning" => {
|
||||||
|
items.insert(id.clone(), ItemKind::Reasoning);
|
||||||
|
yield LlmEvent::ReasoningStart { id };
|
||||||
|
}
|
||||||
|
"function_call" => {
|
||||||
|
let call_id = item["call_id"].as_str().unwrap_or_default().to_string();
|
||||||
|
let name = item["name"].as_str().unwrap_or_default().to_string();
|
||||||
|
fn_args.insert(id.clone(), String::new());
|
||||||
|
items.insert(id, ItemKind::FunctionCall { call_id: call_id.clone(), name: name.clone() });
|
||||||
|
had_tool_call = true;
|
||||||
|
yield LlmEvent::ToolInputStart { call_id, name };
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"response.output_text.delta" => {
|
||||||
|
let id = value["item_id"].as_str().unwrap_or_default().to_string();
|
||||||
|
let text = value["delta"].as_str().unwrap_or_default().to_string();
|
||||||
|
yield LlmEvent::TextDelta { id, text };
|
||||||
|
}
|
||||||
|
"response.reasoning_summary_text.delta" => {
|
||||||
|
let id = value["item_id"].as_str().unwrap_or_default().to_string();
|
||||||
|
let text = value["delta"].as_str().unwrap_or_default().to_string();
|
||||||
|
yield LlmEvent::ReasoningDelta { id, text };
|
||||||
|
}
|
||||||
|
"response.function_call_arguments.delta" => {
|
||||||
|
let id = value["item_id"].as_str().unwrap_or_default().to_string();
|
||||||
|
let delta = value["delta"].as_str().unwrap_or_default();
|
||||||
|
if let (Some(buf), Some(ItemKind::FunctionCall { call_id, .. })) =
|
||||||
|
(fn_args.get_mut(&id), items.get(&id))
|
||||||
|
{
|
||||||
|
buf.push_str(delta);
|
||||||
|
yield LlmEvent::ToolInputDelta { call_id: call_id.clone(), json: delta.to_string() };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"response.output_item.done" => {
|
||||||
|
let id = value["item"]["id"].as_str().unwrap_or_default().to_string();
|
||||||
|
match items.remove(&id) {
|
||||||
|
Some(ItemKind::Text) => yield LlmEvent::TextEnd { id },
|
||||||
|
Some(ItemKind::Reasoning) => yield LlmEvent::ReasoningEnd { id, signature: None },
|
||||||
|
Some(ItemKind::FunctionCall { call_id, name }) => {
|
||||||
|
let raw = fn_args.remove(&id).unwrap_or_default();
|
||||||
|
let input: Value = if raw.trim().is_empty() {
|
||||||
|
json!({})
|
||||||
|
} else {
|
||||||
|
serde_json::from_str(&raw).unwrap_or(Value::Null)
|
||||||
|
};
|
||||||
|
yield LlmEvent::ToolCall { call_id, name, input };
|
||||||
|
}
|
||||||
|
None => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"response.completed" | "response.incomplete" => {
|
||||||
|
let response = &value["response"];
|
||||||
|
let u = &response["usage"];
|
||||||
|
usage.input = u["input_tokens"].as_u64().unwrap_or(0);
|
||||||
|
usage.output = u["output_tokens"].as_u64().unwrap_or(0);
|
||||||
|
usage.reasoning = u["output_tokens_details"]["reasoning_tokens"].as_u64().unwrap_or(0);
|
||||||
|
usage.cache_read = u["input_tokens_details"]["cached_tokens"].as_u64().unwrap_or(0);
|
||||||
|
let status = response["status"].as_str();
|
||||||
|
yield LlmEvent::Finish { reason: map_status(status, had_tool_call), usage };
|
||||||
|
}
|
||||||
|
"error" | "response.failed" => {
|
||||||
|
let message = value["response"]["error"]["message"]
|
||||||
|
.as_str()
|
||||||
|
.or_else(|| value["message"].as_str())
|
||||||
|
.unwrap_or("unknown error")
|
||||||
|
.to_string();
|
||||||
|
Err(ProviderError::Http { status: 0, body: message })?;
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use futures::StreamExt;
|
||||||
|
use harness_core::llm::{Initiator, ReasoningOpts, ToolSchema, WireMessage};
|
||||||
|
|
||||||
|
fn sse_stream(raw: &'static str) -> LlmEventStream {
|
||||||
|
let chunks: Vec<Result<bytes::Bytes, std::io::Error>> =
|
||||||
|
vec![Ok(bytes::Bytes::from_static(raw.as_bytes()))];
|
||||||
|
decode(futures::stream::iter(chunks))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn req(model: &str) -> LlmRequest {
|
||||||
|
LlmRequest {
|
||||||
|
model: model.into(),
|
||||||
|
system: vec![],
|
||||||
|
messages: vec![],
|
||||||
|
tools: vec![],
|
||||||
|
temperature: None,
|
||||||
|
max_tokens: None,
|
||||||
|
reasoning: None,
|
||||||
|
initiator: Initiator::User,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_text_response() {
|
||||||
|
let raw = concat!(
|
||||||
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n",
|
||||||
|
"data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"delta\":\"Hello\"}\n\n",
|
||||||
|
"data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"msg_1\"}}\n\n",
|
||||||
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}}\n\n",
|
||||||
|
);
|
||||||
|
let events: Vec<LlmEvent> = sse_stream(raw).map(|e| e.unwrap()).collect().await;
|
||||||
|
assert_eq!(
|
||||||
|
events,
|
||||||
|
vec![
|
||||||
|
LlmEvent::TextStart { id: "msg_1".into() },
|
||||||
|
LlmEvent::TextDelta {
|
||||||
|
id: "msg_1".into(),
|
||||||
|
text: "Hello".into()
|
||||||
|
},
|
||||||
|
LlmEvent::TextEnd { id: "msg_1".into() },
|
||||||
|
LlmEvent::Finish {
|
||||||
|
reason: FinishReason::Stop,
|
||||||
|
usage: TokenUsage {
|
||||||
|
input: 10,
|
||||||
|
output: 5,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_function_call_with_reasoning_tokens() {
|
||||||
|
let raw = concat!(
|
||||||
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"read\"}}\n\n",
|
||||||
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"file\\\":\"}\n\n",
|
||||||
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"\\\"a.txt\\\"}\"}\n\n",
|
||||||
|
"data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"fc_1\"}}\n\n",
|
||||||
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":3,\"output_tokens\":9,\"output_tokens_details\":{\"reasoning_tokens\":4}}}}\n\n",
|
||||||
|
);
|
||||||
|
let events: Vec<LlmEvent> = sse_stream(raw).map(|e| e.unwrap()).collect().await;
|
||||||
|
assert_eq!(
|
||||||
|
events,
|
||||||
|
vec![
|
||||||
|
LlmEvent::ToolInputStart {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolInputDelta {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
json: "{\"file\":".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolInputDelta {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
json: "\"a.txt\"}".into()
|
||||||
|
},
|
||||||
|
LlmEvent::ToolCall {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into(),
|
||||||
|
input: json!({"file": "a.txt"}),
|
||||||
|
},
|
||||||
|
LlmEvent::Finish {
|
||||||
|
reason: FinishReason::ToolCalls,
|
||||||
|
usage: TokenUsage {
|
||||||
|
input: 3,
|
||||||
|
output: 9,
|
||||||
|
reasoning: 4,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn failed_response_surfaces_as_err() {
|
||||||
|
let raw = "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"boom\"}}}\n\n";
|
||||||
|
let events: Vec<Result<LlmEvent, ProviderError>> = sse_stream(raw).collect().await;
|
||||||
|
assert!(
|
||||||
|
matches!(events.last(), Some(Err(ProviderError::Http { body, .. })) if body == "boom")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_puts_system_in_instructions() {
|
||||||
|
let mut r = req("gpt-5");
|
||||||
|
r.system = vec!["env".into(), "agent".into()];
|
||||||
|
let body = build_request(&r);
|
||||||
|
assert_eq!(body["instructions"], "env\n\nagent");
|
||||||
|
assert!(body.get("messages").is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_flattens_tool_calls_and_results_to_items() {
|
||||||
|
let mut r = req("gpt-5");
|
||||||
|
r.messages = vec![
|
||||||
|
WireMessage {
|
||||||
|
role: Role::Assistant,
|
||||||
|
content: vec![WireContent::ToolCall {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
name: "read".into(),
|
||||||
|
input: json!({"file": "a.txt"}),
|
||||||
|
}],
|
||||||
|
},
|
||||||
|
WireMessage {
|
||||||
|
role: Role::Tool,
|
||||||
|
content: vec![WireContent::ToolResult {
|
||||||
|
call_id: "call_1".into(),
|
||||||
|
output: "contents".into(),
|
||||||
|
is_error: false,
|
||||||
|
}],
|
||||||
|
},
|
||||||
|
];
|
||||||
|
let body = build_request(&r);
|
||||||
|
let input = body["input"].as_array().unwrap();
|
||||||
|
assert_eq!(input[0]["type"], "function_call");
|
||||||
|
assert_eq!(input[0]["call_id"], "call_1");
|
||||||
|
assert_eq!(input[0]["arguments"], "{\"file\":\"a.txt\"}");
|
||||||
|
assert_eq!(input[1]["type"], "function_call_output");
|
||||||
|
assert_eq!(input[1]["output"], "contents");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn build_request_uses_flat_function_tool_shape_and_reasoning() {
|
||||||
|
let mut r = req("o3");
|
||||||
|
r.tools = vec![ToolSchema {
|
||||||
|
name: "read".into(),
|
||||||
|
description: "reads a file".into(),
|
||||||
|
parameters: json!({"type": "object"}),
|
||||||
|
}];
|
||||||
|
r.reasoning = Some(ReasoningOpts {
|
||||||
|
effort: Some(ReasoningEffort::Medium),
|
||||||
|
budget_tokens: None,
|
||||||
|
});
|
||||||
|
let body = build_request(&r);
|
||||||
|
assert_eq!(body["tools"][0]["type"], "function");
|
||||||
|
assert_eq!(body["tools"][0]["name"], "read");
|
||||||
|
assert_eq!(body["reasoning"]["effort"], "medium");
|
||||||
|
assert_eq!(body["reasoning"]["summary"], "auto");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,6 +1,8 @@
|
|||||||
pub mod anthropic;
|
pub mod anthropic;
|
||||||
pub mod codec;
|
pub mod codec;
|
||||||
|
pub mod openai;
|
||||||
pub mod registry;
|
pub mod registry;
|
||||||
|
|
||||||
pub use anthropic::AnthropicProvider;
|
pub use anthropic::AnthropicProvider;
|
||||||
|
pub use openai::OpenAiProvider;
|
||||||
pub use registry::ProviderRegistry;
|
pub use registry::ProviderRegistry;
|
||||||
|
|||||||
@@ -0,0 +1,173 @@
|
|||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use harness_core::llm::{LlmEventStream, LlmRequest, Provider, ProviderError};
|
||||||
|
use harness_core::types::ModelInfo;
|
||||||
|
use tokio_util::sync::CancellationToken;
|
||||||
|
|
||||||
|
use crate::codec::{openai_chat, openai_responses};
|
||||||
|
|
||||||
|
const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1";
|
||||||
|
|
||||||
|
/// Which OpenAI wire format to speak for a given model.
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
enum ApiFlavor {
|
||||||
|
Responses,
|
||||||
|
Chat,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `gpt-*` and `o-*` models use the responses API; everything else (including
|
||||||
|
/// OpenAI-compatible third-party endpoints) falls back to chat completions.
|
||||||
|
fn flavor_for(model: &str) -> ApiFlavor {
|
||||||
|
let is_native = model.starts_with("gpt-")
|
||||||
|
|| model.starts_with("o1")
|
||||||
|
|| model.starts_with("o3")
|
||||||
|
|| model.starts_with("o4")
|
||||||
|
|| model == "o1"
|
||||||
|
|| model == "o3";
|
||||||
|
if is_native {
|
||||||
|
ApiFlavor::Responses
|
||||||
|
} else {
|
||||||
|
ApiFlavor::Chat
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct OpenAiProvider {
|
||||||
|
api_key: String,
|
||||||
|
base_url: String,
|
||||||
|
client: reqwest::Client,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl OpenAiProvider {
|
||||||
|
pub fn new(api_key: impl Into<String>) -> Self {
|
||||||
|
Self {
|
||||||
|
api_key: api_key.into(),
|
||||||
|
base_url: DEFAULT_BASE_URL.to_string(),
|
||||||
|
client: reqwest::Client::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn with_base_url(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
|
||||||
|
Self {
|
||||||
|
api_key: api_key.into(),
|
||||||
|
base_url: base_url.into(),
|
||||||
|
client: reqwest::Client::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn classify_error(
|
||||||
|
status: reqwest::StatusCode,
|
||||||
|
body: String,
|
||||||
|
retry_after: Option<Duration>,
|
||||||
|
) -> ProviderError {
|
||||||
|
match status.as_u16() {
|
||||||
|
400 if body.contains("context_length_exceeded")
|
||||||
|
|| body.contains("maximum context length") =>
|
||||||
|
{
|
||||||
|
ProviderError::ContextOverflow
|
||||||
|
}
|
||||||
|
401 | 403 => ProviderError::Auth(body),
|
||||||
|
429 => ProviderError::RateLimited { retry_after },
|
||||||
|
s if (500..600).contains(&s) => ProviderError::Overloaded,
|
||||||
|
s => ProviderError::Http { status: s, body },
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Provider for OpenAiProvider {
|
||||||
|
fn id(&self) -> &str {
|
||||||
|
"openai"
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn list_models(&self) -> Result<Vec<ModelInfo>, ProviderError> {
|
||||||
|
// models.dev metadata is layered on top by the caller (see `modelsdev.rs`).
|
||||||
|
Ok(Vec::new())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn stream(
|
||||||
|
&self,
|
||||||
|
req: LlmRequest,
|
||||||
|
cancel: CancellationToken,
|
||||||
|
) -> Result<LlmEventStream, ProviderError> {
|
||||||
|
let (path, body) = match flavor_for(&req.model) {
|
||||||
|
ApiFlavor::Responses => ("/responses", openai_responses::build_request(&req)),
|
||||||
|
ApiFlavor::Chat => ("/chat/completions", openai_chat::build_request(&req)),
|
||||||
|
};
|
||||||
|
let url = format!("{}{}", self.base_url, path);
|
||||||
|
|
||||||
|
let send = self
|
||||||
|
.client
|
||||||
|
.post(&url)
|
||||||
|
.header("authorization", format!("Bearer {}", self.api_key))
|
||||||
|
.json(&body)
|
||||||
|
.send();
|
||||||
|
|
||||||
|
let response = tokio::select! {
|
||||||
|
result = send => result.map_err(|e| ProviderError::Network(e.to_string()))?,
|
||||||
|
_ = cancel.cancelled() => return Err(ProviderError::Cancelled),
|
||||||
|
};
|
||||||
|
|
||||||
|
if !response.status().is_success() {
|
||||||
|
let status = response.status();
|
||||||
|
let retry_after = response
|
||||||
|
.headers()
|
||||||
|
.get("retry-after")
|
||||||
|
.and_then(|v| v.to_str().ok())
|
||||||
|
.and_then(|s| s.parse::<u64>().ok())
|
||||||
|
.map(Duration::from_secs);
|
||||||
|
let body = response.text().await.unwrap_or_default();
|
||||||
|
return Err(Self::classify_error(status, body, retry_after));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(match flavor_for(&req.model) {
|
||||||
|
ApiFlavor::Responses => openai_responses::decode(response.bytes_stream()),
|
||||||
|
ApiFlavor::Chat => openai_chat::decode(response.bytes_stream()),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn routes_gpt_and_o_series_to_responses() {
|
||||||
|
assert_eq!(flavor_for("gpt-4o"), ApiFlavor::Responses);
|
||||||
|
assert_eq!(flavor_for("gpt-5"), ApiFlavor::Responses);
|
||||||
|
assert_eq!(flavor_for("o1"), ApiFlavor::Responses);
|
||||||
|
assert_eq!(flavor_for("o3-mini"), ApiFlavor::Responses);
|
||||||
|
assert_eq!(flavor_for("o4-mini"), ApiFlavor::Responses);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn routes_other_models_to_chat() {
|
||||||
|
assert_eq!(flavor_for("llama-3.1-70b"), ApiFlavor::Chat);
|
||||||
|
assert_eq!(flavor_for("deepseek-chat"), ApiFlavor::Chat);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn id_is_openai() {
|
||||||
|
assert_eq!(OpenAiProvider::new("k").id(), "openai");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn classifies_context_overflow_from_400_body() {
|
||||||
|
assert!(matches!(
|
||||||
|
OpenAiProvider::classify_error(
|
||||||
|
reqwest::StatusCode::BAD_REQUEST,
|
||||||
|
"context_length_exceeded".into(),
|
||||||
|
None,
|
||||||
|
),
|
||||||
|
ProviderError::ContextOverflow
|
||||||
|
));
|
||||||
|
assert!(matches!(
|
||||||
|
OpenAiProvider::classify_error(
|
||||||
|
reqwest::StatusCode::BAD_REQUEST,
|
||||||
|
"some other error".into(),
|
||||||
|
None,
|
||||||
|
),
|
||||||
|
ProviderError::Http { status: 400, .. }
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user