diff --git a/README.md b/README.md index c0365ec..d19e16d 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,11 @@ inference, tool execution, and HTTP transport must be provided externally. DeepSeek V4.1 preprocessing with OpenCV. - **Tool definitions:** Function tools; the Responses API also supports tool namespaces and the `apply_patch` custom tool. +- **Multiple completions:** Chat Completions `n > 1`. The adapter validates + and threads `n` through as `ConversationRequest::completion_count`; the + caller runs one inference per completion and builds each completion's + events with [`ProtocolRequest::chunk_generator_for_choice`], tagging every + chunk with its `choices[].index`. ## Not yet supported @@ -36,7 +41,6 @@ inference, tool execution, and HTTP transport must be provided externally. - Document content, audio/video input, and file retrieval by `file_id`. - Server tool execution, such as `web_search`. - JSON Schema and regex output constraints, or enforcement of tool `strict` settings. -- Multiple completions per Chat Completions request (`n > 1`). - Responses custom tool definitions other than `apply_patch`. - Responses conversation storage and context retrieval through `previous_response_id`. - Responses encrypted thinking content (`encrypted_content`). diff --git a/deepseek-recipe/src/protocol/anthropic/messages/request/convert.rs b/deepseek-recipe/src/protocol/anthropic/messages/request/convert.rs index b95a9e2..fbb1054 100644 --- a/deepseek-recipe/src/protocol/anthropic/messages/request/convert.rs +++ b/deepseek-recipe/src/protocol/anthropic/messages/request/convert.rs @@ -59,6 +59,7 @@ impl ProtocolRequest for MessagesRequest { parsing_options, model: Some(self.model), stream: self.stream.unwrap_or(false), + completion_count: 1, }) } diff --git a/deepseek-recipe/src/protocol/openai/chat_completion/request/convert.rs b/deepseek-recipe/src/protocol/openai/chat_completion/request/convert.rs index 20c0ebc..16556c6 100644 --- a/deepseek-recipe/src/protocol/openai/chat_completion/request/convert.rs +++ b/deepseek-recipe/src/protocol/openai/chat_completion/request/convert.rs @@ -28,6 +28,13 @@ use super::schema::{ /// Maximum number of stop sequences accepted by the adapter. const MAX_STOP_SEQUENCES: usize = 16; +/// Maximum number of parallel completions accepted for `n`. +/// +/// Matches the ceiling OpenAI's own Chat Completions API documents. The +/// caller is responsible for actually running this many inferences; this +/// bound only guards against pathological request values. +const MAX_COMPLETIONS: u8 = 128; + impl ProtocolRequest for ChatCompletionRequest { type Response = ChatCompletionResponse; @@ -35,6 +42,7 @@ impl ProtocolRequest for ChatCompletionRequest { let (thinking_mode, reasoning_effort) = self.resolve_thinking(options); let mut parsing_options = self.parsing_options(thinking_mode); let inference_options = self.inference_options()?; + let completion_count = validate_completion_count(self.n)?; validate_api_params( self.stop.as_ref(), self.stream, @@ -60,6 +68,7 @@ impl ProtocolRequest for ChatCompletionRequest { parsing_options, model: Some(self.model), stream: self.stream.unwrap_or(false), + completion_count, }) } @@ -67,6 +76,15 @@ impl ProtocolRequest for ChatCompletionRequest { request: &ConversationRequest, id: String, model: String, + ) -> ChatCompletionChunkGenerator { + Self::chunk_generator_for_choice(request, id, model, 0) + } + + fn chunk_generator_for_choice( + request: &ConversationRequest, + id: String, + model: String, + choice_index: u32, ) -> ChatCompletionChunkGenerator { ChatCompletionChunkGenerator::new( id, @@ -74,6 +92,7 @@ impl ProtocolRequest for ChatCompletionRequest { !request.stream, request.conversation.thinking_mode, ) + .with_choice_index(choice_index) } } @@ -98,11 +117,6 @@ impl ChatCompletionRequest { } fn inference_options(&self) -> Result { - if self.n.is_some_and(|value| value != 1) { - return Err(ConversionError::bad_request( - "Invalid n value (currently only n = 1 is supported)", - )); - } for (name, value) in [ ("frequency_penalty", self.frequency_penalty), ("presence_penalty", self.presence_penalty), @@ -481,6 +495,20 @@ fn convert_response_format( } } +/// Validate `n` and resolve it to the number of completions to generate. +/// +/// Absent `n` defaults to a single completion, matching the OpenAI API. +fn validate_completion_count(n: Option) -> Result { + match n { + None => Ok(1), + Some(0) => Err(ConversionError::bad_request("n must be at least 1")), + Some(value) if value > MAX_COMPLETIONS => Err(ConversionError::bad_request(format!( + "n must be at most {MAX_COMPLETIONS}" + ))), + Some(value) => Ok(u32::from(value)), + } +} + /// Validate stop sequences and streaming options. fn validate_api_params( stop: Option<&ChatCompletionStopSequences>, @@ -502,3 +530,75 @@ fn validate_api_params( } Ok(()) } + +#[cfg(test)] +mod tests { + use crate::request::{ConversionOptions, ProtocolRequest}; + + use super::ChatCompletionRequest; + + /// A minimal, otherwise-valid request with the given `n`. + fn request_with_n(n: Option) -> ChatCompletionRequest { + serde_json::from_value(serde_json::json!({ + "messages": [{"role": "user", "content": "hi"}], + "model": "deepseek-chat", + "frequency_penalty": null, + "max_tokens": null, + "n": n, + "presence_penalty": null, + "seed": null, + "stop": null, + "stream": null, + "stream_options": null, + "temperature": null, + "top_p": null, + "response_format": null, + "tools": null, + "tool_choice": null, + "reasoning_effort": null, + "thinking": null, + "parallel_tool_calls": null, + })) + .expect("fixture request should deserialize") + } + + #[test] + fn n_absent_defaults_to_one_completion() { + let request = request_with_n(None) + .convert(ConversionOptions::default()) + .unwrap(); + assert_eq!(request.completion_count, 1); + } + + #[test] + fn n_one_is_accepted_as_before() { + let request = request_with_n(Some(1)) + .convert(ConversionOptions::default()) + .unwrap(); + assert_eq!(request.completion_count, 1); + } + + #[test] + fn n_greater_than_one_is_now_accepted_and_threaded_through() { + let request = request_with_n(Some(4)) + .convert(ConversionOptions::default()) + .unwrap(); + assert_eq!(request.completion_count, 4); + } + + #[test] + fn n_zero_is_rejected() { + let error = request_with_n(Some(0)) + .convert(ConversionOptions::default()) + .unwrap_err(); + assert!(error.to_string().contains("at least 1")); + } + + #[test] + fn n_above_the_bound_is_rejected() { + let error = request_with_n(Some(255)) + .convert(ConversionOptions::default()) + .unwrap_err(); + assert!(error.to_string().contains("at most")); + } +} diff --git a/deepseek-recipe/src/protocol/openai/chat_completion/response/chunk_generator.rs b/deepseek-recipe/src/protocol/openai/chat_completion/response/chunk_generator.rs index 5a40dbd..4d39d3e 100644 --- a/deepseek-recipe/src/protocol/openai/chat_completion/response/chunk_generator.rs +++ b/deepseek-recipe/src/protocol/openai/chat_completion/response/chunk_generator.rs @@ -20,6 +20,9 @@ pub struct ChatCompletionChunkGenerator { created: u64, thinking_mode: bool, include_usage: bool, + /// Which of the request's `n` parallel completions this generator's + /// output belongs to. Tagged on every emitted chunk's `choices[].index`. + choice_index: u32, tool_call_index: u32, prompt_usage: PromptUsage, system_fingerprint: Option, @@ -27,7 +30,10 @@ pub struct ChatCompletionChunkGenerator { impl ChatCompletionChunkGenerator { /// The caller supplies a unique response ID and public model name. Tool call - /// IDs combine the response ID and per-response tool index. + /// IDs combine the response ID, choice index, and per-response tool index. + /// + /// Defaults to choice index `0`; use [`Self::with_choice_index`] for + /// requests with `n > 1`. pub fn new(id: String, model: String, include_usage: bool, thinking_mode: bool) -> Self { Self { id, @@ -38,6 +44,7 @@ impl ChatCompletionChunkGenerator { .unwrap_or(0), thinking_mode, include_usage, + choice_index: 0, tool_call_index: 0, prompt_usage: PromptUsage::default(), system_fingerprint: None, @@ -50,6 +57,16 @@ impl ChatCompletionChunkGenerator { self } + /// Tag this generator's output as belonging to a specific completion. + /// + /// The caller creates one generator per completion when `n > 1`, each with + /// a distinct `choice_index` (`0..n`), and drives each through its own + /// [`crate::stream::StreamProcessor`]. + pub fn with_choice_index(mut self, choice_index: u32) -> Self { + self.choice_index = choice_index; + self + } + fn new_chunk( &self, delta: ChatCompletionMessageDelta, @@ -70,7 +87,7 @@ impl ChatCompletionChunkGenerator { model: self.model.clone(), system_fingerprint: self.system_fingerprint.clone(), choices: vec![ChatCompletionChunkChoice { - index: 0, + index: self.choice_index, delta, logprobs: None, finish_reason, @@ -130,7 +147,7 @@ impl ChunkGenerator for ChatCompletionChunkGenerator { tool_calls: Some(vec![ChatCompletionToolCallDelta::Start { index, call: ChatCompletionResponseToolCall { - id: format!("call_{}_{index}", self.id), + id: format!("call_{}_{}_{index}", self.id, self.choice_index), r#type: ChatCompletionToolCallType::Function, function: ChatCompletionResponseFunctionCall { name: tool_name, @@ -167,3 +184,79 @@ impl ChunkGenerator for ChatCompletionChunkGenerator { } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// A generator built for one choice of an `n > 1` request must tag every + /// chunk it produces with that choice's index, not the default `0`. + #[tokio::test] + async fn with_choice_index_tags_every_emitted_chunk() { + let mut generator = + ChatCompletionChunkGenerator::new("resp_1".into(), "deepseek-chat".into(), true, false) + .with_choice_index(2); + + let start_chunks = generator + .generate(OutputChunk::Start { + system_fingerprint: None, + usage: PromptUsage::default(), + }) + .await; + let text_chunks = generator + .generate(OutputChunk::Raw { + content: "hi".into(), + }) + .await; + + for chunk in start_chunks.iter().chain(text_chunks.iter()) { + for choice in &chunk.choices { + assert_eq!(choice.index, 2); + } + } + } + + /// Two generators for different choices of the same response must not + /// mint colliding tool call ids, since both start from the same response + /// id and the same per-generator tool call counter. + #[tokio::test] + async fn tool_call_ids_are_unique_across_choices() { + let mut choice0 = + ChatCompletionChunkGenerator::new("resp_1".into(), "deepseek-chat".into(), true, false) + .with_choice_index(0); + let mut choice1 = + ChatCompletionChunkGenerator::new("resp_1".into(), "deepseek-chat".into(), true, false) + .with_choice_index(1); + + let chunks0 = choice0 + .generate(OutputChunk::ToolCall { + tool_name: "search".into(), + arguments: String::new(), + }) + .await; + let chunks1 = choice1 + .generate(OutputChunk::ToolCall { + tool_name: "search".into(), + arguments: String::new(), + }) + .await; + + assert_ne!(tool_call_id(&chunks0), tool_call_id(&chunks1)); + } + + fn tool_call_id(chunks: &[ChatCompletionChunk]) -> String { + for chunk in chunks { + for choice in &chunk.choices { + let Some(tool_calls) = &choice.delta.tool_calls else { + continue; + }; + for call in tool_calls { + if let ChatCompletionToolCallDelta::Start { call, .. } = call { + return call.id.clone(); + } + } + } + } + panic!("no tool call start delta found"); + } +} diff --git a/deepseek-recipe/src/protocol/openai/chat_completion/response/schema.rs b/deepseek-recipe/src/protocol/openai/chat_completion/response/schema.rs index 5c75962..a927bfc 100644 --- a/deepseek-recipe/src/protocol/openai/chat_completion/response/schema.rs +++ b/deepseek-recipe/src/protocol/openai/chat_completion/response/schema.rs @@ -307,3 +307,55 @@ impl AppendDelta for ChatCompletionResponseMessage { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn text_chunk(index: u32, content: &str) -> ChatCompletionChunk { + ChatCompletionChunk { + id: "resp_1".to_string(), + object: CHAT_COMPLETION_CHUNK_OBJECT, + created: 0, + model: "deepseek-chat".to_string(), + system_fingerprint: None, + choices: vec![ChatCompletionChunkChoice { + index, + delta: ChatCompletionMessageDelta { + content: Some(Some(content.to_string())), + ..Default::default() + }, + logprobs: None, + finish_reason: None, + }], + usage: None, + } + } + + /// Regression test for the `n > 1` non-streaming path: the accumulator + /// keys choices by `index`, so interleaved chunks from several + /// completions must land in separate, independently-growing choices + /// rather than clobbering each other. + #[test] + fn chunks_with_different_indices_accumulate_into_separate_choices() { + let mut response = + ChatCompletionResponse::new("resp_1".into(), "deepseek-chat".into(), 0, 0, 0); + response.append(text_chunk(0, "Hello")); + response.append(text_chunk(1, "Hi")); + response.append(text_chunk(0, " world")); + + assert_eq!(response.choices.len(), 2); + let choice0 = response + .choices + .iter() + .find(|choice| choice.index == 0) + .expect("choice 0 present"); + let choice1 = response + .choices + .iter() + .find(|choice| choice.index == 1) + .expect("choice 1 present"); + assert_eq!(choice0.message.content.as_deref(), Some("Hello world")); + assert_eq!(choice1.message.content.as_deref(), Some("Hi")); + } +} diff --git a/deepseek-recipe/src/protocol/openai/responses/request/convert.rs b/deepseek-recipe/src/protocol/openai/responses/request/convert.rs index d05786f..c6de534 100644 --- a/deepseek-recipe/src/protocol/openai/responses/request/convert.rs +++ b/deepseek-recipe/src/protocol/openai/responses/request/convert.rs @@ -585,6 +585,7 @@ impl ProtocolRequest for ResponsesRequest { parsing_options, model: Some(self.model), stream: self.stream.unwrap_or(false), + completion_count: 1, }) } diff --git a/deepseek-recipe/src/request/mod.rs b/deepseek-recipe/src/request/mod.rs index 78f23f0..c38a5e1 100644 --- a/deepseek-recipe/src/request/mod.rs +++ b/deepseek-recipe/src/request/mod.rs @@ -38,6 +38,14 @@ pub struct ConversationRequest { /// Original model name, when exposed by the adapter. The caller resolves it. pub model: Option, pub stream: bool, + /// Number of parallel completions the caller should generate for this + /// conversation. + /// + /// Always `1` for protocols that don't expose a multi-completion parameter. + /// Chat Completions surfaces this from its `n` request field. The caller + /// runs inference once per completion and tags each with a distinct + /// `choice_index` via [`ProtocolRequest::chunk_generator_for_choice`]. + pub completion_count: u32, } impl ConversationRequest { @@ -49,6 +57,7 @@ impl ConversationRequest { parsing_options: ParsingOptions::default(), model: None, stream: false, + completion_count: 1, } } } @@ -68,11 +77,38 @@ pub trait ProtocolRequest: 'static + Send + Sync { fn convert(self, options: ConversionOptions) -> Result; /// Construct protocol events using the converted request's resolved settings. + /// + /// Equivalent to [`Self::chunk_generator_for_choice`] with `choice_index` + /// `0`. Protocols that never produce more than one completion can keep + /// implementing only this method. fn chunk_generator( request: &ConversationRequest, id: String, model: String, ) -> ::ChunkGenerator; + + /// Construct protocol events for one completion of a `completion_count > 1` + /// request. + /// + /// `choice_index` identifies which of the caller's parallel completions + /// this generator's output belongs to; the generator must tag every event + /// it produces with that index. The caller runs one inference per + /// completion, drives each through its own [`crate::stream::StreamProcessor`] + /// built from the generator returned here, and merges (streaming) or + /// accumulates (complete response) the results. + /// + /// The default implementation ignores `choice_index` and delegates to + /// [`Self::chunk_generator`], which is correct for every protocol except + /// Chat Completions today. + fn chunk_generator_for_choice( + request: &ConversationRequest, + id: String, + model: String, + choice_index: u32, + ) -> ::ChunkGenerator { + let _ = choice_index; + Self::chunk_generator(request, id, model) + } } /// A failure while converting protocol input into a conversation request. diff --git a/server-rs/src/lib.rs b/server-rs/src/lib.rs index b7afefe..b02f72b 100644 --- a/server-rs/src/lib.rs +++ b/server-rs/src/lib.rs @@ -11,6 +11,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use deepseek_recipe::anthropic::MessagesRequest; pub use deepseek_recipe::error_response::ErrorResponse; +use deepseek_recipe::openai::chat_completion::response::ChatCompletionResponse; use deepseek_recipe::openai::responses::response::ResponsesResponse; use deepseek_recipe::openai::{ChatCompletionRequest, ResponsesRequest}; pub use deepseek_recipe::request::ConversionError; @@ -27,7 +28,7 @@ use deepseek_recipe_image::{ ImageError, ImageQuota, ImageResolver, OpenCvImagePreprocessor, ReqwestImageFetcher, }; use serde::{Serialize, de::DeserializeOwned}; -use tokio_stream::{Stream, StreamExt}; +use tokio_stream::{Stream, StreamExt, StreamMap}; const MOCK_ID: &str = "mock-id"; const MOCK_MODEL: &str = "deepseek-flash"; @@ -135,6 +136,10 @@ where /// Prepare Chat Completions SSE using the request's usage inclusion setting. /// +/// When the request's `n` asks for more than one completion, runs one mock +/// inference per completion and interleaves their chunks as they arrive, each +/// tagged with its own `choices[].index`. +/// /// # Errors /// /// Returns the protocol conversion error before producing response events, or a @@ -145,10 +150,131 @@ pub async fn prepare_chat_completion_response( ) -> Result { let include_usage = req.include_usage(); let request = req.convert(ConversionOptions::default())?; - prepare_converted_response_with::(request, |generator| { - generator.with_include_usage(include_usage) + if request.completion_count <= 1 { + return prepare_converted_response_with::(request, |generator| { + generator.with_include_usage(include_usage) + }) + .await; + } + prepare_multi_completion_response(request, include_usage).await +} + +/// Prepare Chat Completions SSE for a `n > 1` request. +/// +/// Each of the `n` completions gets its own mock inference stream and its own +/// [`StreamProcessor`], built with a distinct `choice_index` so downstream +/// chunks carry the right `choices[].index`. A [`StreamMap`] merges the `n` +/// processors so chunks are emitted in true arrival order rather than one +/// completion at a time, mirroring how a real backend would interleave +/// parallel generations. +/// +/// # Errors +/// +/// Returns a conversion error with status code 400 or 500 when an image cannot +/// be resolved or preprocessed. +async fn prepare_multi_completion_response( + request: ConversationRequest, + include_usage: bool, +) -> Result { + let rendered = DeepseekV41Encoding::new().render_conversation(&request.conversation); + let resolver = default_resolver()?; + let mut quota = ImageQuota::new(); + let multi_modal_data = resolver + .resolve(&rendered.image_sources, &mut quota) + .await + .map_err(image_error)?; + tracing::debug!( + prompt = rendered.prompt, + image_count = multi_modal_data.images.len(), + completion_count = request.completion_count, + "rendered prompt for multi-completion request" + ); + + let mut processors = StreamMap::new(); + for choice_index in 0..request.completion_count { + let generator = ChatCompletionRequest::chunk_generator_for_choice( + &request, + MOCK_ID.into(), + MOCK_MODEL.into(), + choice_index, + ) + .with_include_usage(include_usage); + let processor = StreamProcessor::new(generator, request.parsing_options.clone()); + processors.insert(choice_index, Box::pin(processor.process(mock_inference()))); + } + + let events = processors.map(|(_choice_index, chunk)| { + let chunk = chunk.map_err(|error| serde_json::Error::io(std::io::Error::other(error)))?; + let data = serde_json::to_string(&chunk)?; + Ok( + match ::chunk_event_type(&chunk) { + Some(event_type) => format!("event: {event_type}\ndata: {data}\n\n"), + None => format!("data: {data}\n\n"), + }, + ) + }); + let done = ::done_message() + .map(|message| Ok(format!("data: {message}\n\n"))); + + Ok(PreparedResponse { + prompt: rendered.prompt, + multi_modal_data, + events: Box::pin(events.chain(tokio_stream::iter(done))), }) - .await +} + +/// Aggregate a `n > 1` Chat Completions request into a single JSON response. +/// +/// Every completion's chunks append into the same [`ChatCompletionResponse`]; +/// its accumulator already keys choices by `index`, so no merge logic is +/// needed beyond running each completion's chunks through `.append()`. +/// +/// # Errors +/// +/// Returns a conversion error with status code 400 or 500 when an image cannot +/// be resolved or preprocessed, or if a completion's output fails to process. +pub async fn complete_chat_completion_response( + req: ChatCompletionRequest, +) -> Result { + let include_usage = req.include_usage(); + let request = req.convert(ConversionOptions::default())?; + if request.completion_count <= 1 { + return complete_response_with::(request, |generator| { + generator.with_include_usage(include_usage) + }) + .await; + } + + let rendered = DeepseekV41Encoding::new().render_conversation(&request.conversation); + let resolver = default_resolver()?; + let mut quota = ImageQuota::new(); + resolver + .resolve(&rendered.image_sources, &mut quota) + .await + .map_err(image_error)?; + + let created = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_secs()) + .unwrap_or(0); + let mut response = + ChatCompletionResponse::new(MOCK_ID.into(), MOCK_MODEL.into(), created, 0, 0); + for choice_index in 0..request.completion_count { + let generator = ChatCompletionRequest::chunk_generator_for_choice( + &request, + MOCK_ID.into(), + MOCK_MODEL.into(), + choice_index, + ) + .with_include_usage(include_usage); + let processor = StreamProcessor::new(generator, request.parsing_options.clone()); + let mut chunks = pin!(processor.process(mock_inference())); + while let Some(chunk) = chunks.next().await { + let chunk = chunk.map_err(|error| ConversionError::internal(error.to_string()))?; + response.append(chunk); + } + } + Ok(response) } /// Prepare Responses SSE using the request's custom tool declarations. diff --git a/server-rs/src/main.rs b/server-rs/src/main.rs index 7a9fcca..4e1defb 100644 --- a/server-rs/src/main.rs +++ b/server-rs/src/main.rs @@ -13,9 +13,9 @@ use deepseek_recipe::response::ProtocolResponse; use deepseek_recipe::stream::ChunkGenerator; use serde::{Serialize, de::DeserializeOwned}; use server_rs::{ - ConversionError, ErrorResponse, PreparedResponse, complete_response, - complete_responses_response, prepare_chat_completion_response, prepare_converted_response, - prepare_responses_response, + ConversionError, ErrorResponse, PreparedResponse, complete_chat_completion_response, + complete_response, complete_responses_response, prepare_chat_completion_response, + prepare_converted_response, prepare_responses_response, }; type ApiError = (StatusCode, Json); @@ -47,7 +47,10 @@ async fn chat_completion_handler( .map(sse_response) .map_err(http_error); } - api_handler(Json(request)).await + complete_chat_completion_response(request) + .await + .map(|response| Json(response).into_response()) + .map_err(http_error) } async fn responses_handler(Json(request): Json) -> Result {