mirror of
https://github.com/deepseek-ai/deepseek-recipe.git
synced 2026-10-28 17:17:08 +08:00
feat(chat-completions): support n > 1 (multiple completions)
Threads n through ConversationRequest as completion_count, tags each chunk generator with a choice_index, and merges/accumulates results by index. Adds a default ProtocolRequest::chunk_generator_for_choice so the Messages and Responses adapters are unaffected. Fixes a latent tool-call-id collision across choices as part of the same change (ids now include choice_index, not just response id and local tool index). Signed-off-by: Selby Mashiki <mashikiselby@gmail.com>
This commit is contained in:
parent
8cadfede70
commit
9fbb051caa
@ -29,6 +29,11 @@ inference, tool execution, and HTTP transport must be provided externally.
|
|||||||
DeepSeek V4.1 preprocessing with OpenCV.
|
DeepSeek V4.1 preprocessing with OpenCV.
|
||||||
- **Tool definitions:** Function tools; the Responses API also supports tool
|
- **Tool definitions:** Function tools; the Responses API also supports tool
|
||||||
namespaces and the `apply_patch` custom 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
|
## 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`.
|
- Document content, audio/video input, and file retrieval by `file_id`.
|
||||||
- Server tool execution, such as `web_search`.
|
- Server tool execution, such as `web_search`.
|
||||||
- JSON Schema and regex output constraints, or enforcement of tool `strict` settings.
|
- 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 custom tool definitions other than `apply_patch`.
|
||||||
- Responses conversation storage and context retrieval through `previous_response_id`.
|
- Responses conversation storage and context retrieval through `previous_response_id`.
|
||||||
- Responses encrypted thinking content (`encrypted_content`).
|
- Responses encrypted thinking content (`encrypted_content`).
|
||||||
|
|||||||
@ -59,6 +59,7 @@ impl ProtocolRequest for MessagesRequest {
|
|||||||
parsing_options,
|
parsing_options,
|
||||||
model: Some(self.model),
|
model: Some(self.model),
|
||||||
stream: self.stream.unwrap_or(false),
|
stream: self.stream.unwrap_or(false),
|
||||||
|
completion_count: 1,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -28,6 +28,13 @@ use super::schema::{
|
|||||||
/// Maximum number of stop sequences accepted by the adapter.
|
/// Maximum number of stop sequences accepted by the adapter.
|
||||||
const MAX_STOP_SEQUENCES: usize = 16;
|
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 {
|
impl ProtocolRequest for ChatCompletionRequest {
|
||||||
type Response = ChatCompletionResponse;
|
type Response = ChatCompletionResponse;
|
||||||
|
|
||||||
@ -35,6 +42,7 @@ impl ProtocolRequest for ChatCompletionRequest {
|
|||||||
let (thinking_mode, reasoning_effort) = self.resolve_thinking(options);
|
let (thinking_mode, reasoning_effort) = self.resolve_thinking(options);
|
||||||
let mut parsing_options = self.parsing_options(thinking_mode);
|
let mut parsing_options = self.parsing_options(thinking_mode);
|
||||||
let inference_options = self.inference_options()?;
|
let inference_options = self.inference_options()?;
|
||||||
|
let completion_count = validate_completion_count(self.n)?;
|
||||||
validate_api_params(
|
validate_api_params(
|
||||||
self.stop.as_ref(),
|
self.stop.as_ref(),
|
||||||
self.stream,
|
self.stream,
|
||||||
@ -60,6 +68,7 @@ impl ProtocolRequest for ChatCompletionRequest {
|
|||||||
parsing_options,
|
parsing_options,
|
||||||
model: Some(self.model),
|
model: Some(self.model),
|
||||||
stream: self.stream.unwrap_or(false),
|
stream: self.stream.unwrap_or(false),
|
||||||
|
completion_count,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -67,6 +76,15 @@ impl ProtocolRequest for ChatCompletionRequest {
|
|||||||
request: &ConversationRequest,
|
request: &ConversationRequest,
|
||||||
id: String,
|
id: String,
|
||||||
model: 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 {
|
||||||
ChatCompletionChunkGenerator::new(
|
ChatCompletionChunkGenerator::new(
|
||||||
id,
|
id,
|
||||||
@ -74,6 +92,7 @@ impl ProtocolRequest for ChatCompletionRequest {
|
|||||||
!request.stream,
|
!request.stream,
|
||||||
request.conversation.thinking_mode,
|
request.conversation.thinking_mode,
|
||||||
)
|
)
|
||||||
|
.with_choice_index(choice_index)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -98,11 +117,6 @@ impl ChatCompletionRequest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn inference_options(&self) -> Result<InferenceOptions, ConversionError> {
|
fn inference_options(&self) -> Result<InferenceOptions, ConversionError> {
|
||||||
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 [
|
for (name, value) in [
|
||||||
("frequency_penalty", self.frequency_penalty),
|
("frequency_penalty", self.frequency_penalty),
|
||||||
("presence_penalty", self.presence_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<u8>) -> Result<u32, ConversionError> {
|
||||||
|
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.
|
/// Validate stop sequences and streaming options.
|
||||||
fn validate_api_params(
|
fn validate_api_params(
|
||||||
stop: Option<&ChatCompletionStopSequences>,
|
stop: Option<&ChatCompletionStopSequences>,
|
||||||
@ -502,3 +530,75 @@ fn validate_api_params(
|
|||||||
}
|
}
|
||||||
Ok(())
|
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<u8>) -> 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"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -20,6 +20,9 @@ pub struct ChatCompletionChunkGenerator {
|
|||||||
created: u64,
|
created: u64,
|
||||||
thinking_mode: bool,
|
thinking_mode: bool,
|
||||||
include_usage: 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,
|
tool_call_index: u32,
|
||||||
prompt_usage: PromptUsage,
|
prompt_usage: PromptUsage,
|
||||||
system_fingerprint: Option<String>,
|
system_fingerprint: Option<String>,
|
||||||
@ -27,7 +30,10 @@ pub struct ChatCompletionChunkGenerator {
|
|||||||
|
|
||||||
impl ChatCompletionChunkGenerator {
|
impl ChatCompletionChunkGenerator {
|
||||||
/// The caller supplies a unique response ID and public model name. Tool call
|
/// 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 {
|
pub fn new(id: String, model: String, include_usage: bool, thinking_mode: bool) -> Self {
|
||||||
Self {
|
Self {
|
||||||
id,
|
id,
|
||||||
@ -38,6 +44,7 @@ impl ChatCompletionChunkGenerator {
|
|||||||
.unwrap_or(0),
|
.unwrap_or(0),
|
||||||
thinking_mode,
|
thinking_mode,
|
||||||
include_usage,
|
include_usage,
|
||||||
|
choice_index: 0,
|
||||||
tool_call_index: 0,
|
tool_call_index: 0,
|
||||||
prompt_usage: PromptUsage::default(),
|
prompt_usage: PromptUsage::default(),
|
||||||
system_fingerprint: None,
|
system_fingerprint: None,
|
||||||
@ -50,6 +57,16 @@ impl ChatCompletionChunkGenerator {
|
|||||||
self
|
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(
|
fn new_chunk(
|
||||||
&self,
|
&self,
|
||||||
delta: ChatCompletionMessageDelta,
|
delta: ChatCompletionMessageDelta,
|
||||||
@ -70,7 +87,7 @@ impl ChatCompletionChunkGenerator {
|
|||||||
model: self.model.clone(),
|
model: self.model.clone(),
|
||||||
system_fingerprint: self.system_fingerprint.clone(),
|
system_fingerprint: self.system_fingerprint.clone(),
|
||||||
choices: vec![ChatCompletionChunkChoice {
|
choices: vec![ChatCompletionChunkChoice {
|
||||||
index: 0,
|
index: self.choice_index,
|
||||||
delta,
|
delta,
|
||||||
logprobs: None,
|
logprobs: None,
|
||||||
finish_reason,
|
finish_reason,
|
||||||
@ -130,7 +147,7 @@ impl ChunkGenerator for ChatCompletionChunkGenerator {
|
|||||||
tool_calls: Some(vec![ChatCompletionToolCallDelta::Start {
|
tool_calls: Some(vec![ChatCompletionToolCallDelta::Start {
|
||||||
index,
|
index,
|
||||||
call: ChatCompletionResponseToolCall {
|
call: ChatCompletionResponseToolCall {
|
||||||
id: format!("call_{}_{index}", self.id),
|
id: format!("call_{}_{}_{index}", self.id, self.choice_index),
|
||||||
r#type: ChatCompletionToolCallType::Function,
|
r#type: ChatCompletionToolCallType::Function,
|
||||||
function: ChatCompletionResponseFunctionCall {
|
function: ChatCompletionResponseFunctionCall {
|
||||||
name: tool_name,
|
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");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -307,3 +307,55 @@ impl AppendDelta<ChatCompletionMessageDelta> 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"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -585,6 +585,7 @@ impl ProtocolRequest for ResponsesRequest {
|
|||||||
parsing_options,
|
parsing_options,
|
||||||
model: Some(self.model),
|
model: Some(self.model),
|
||||||
stream: self.stream.unwrap_or(false),
|
stream: self.stream.unwrap_or(false),
|
||||||
|
completion_count: 1,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -38,6 +38,14 @@ pub struct ConversationRequest {
|
|||||||
/// Original model name, when exposed by the adapter. The caller resolves it.
|
/// Original model name, when exposed by the adapter. The caller resolves it.
|
||||||
pub model: Option<String>,
|
pub model: Option<String>,
|
||||||
pub stream: bool,
|
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 {
|
impl ConversationRequest {
|
||||||
@ -49,6 +57,7 @@ impl ConversationRequest {
|
|||||||
parsing_options: ParsingOptions::default(),
|
parsing_options: ParsingOptions::default(),
|
||||||
model: None,
|
model: None,
|
||||||
stream: false,
|
stream: false,
|
||||||
|
completion_count: 1,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -68,11 +77,38 @@ pub trait ProtocolRequest: 'static + Send + Sync {
|
|||||||
fn convert(self, options: ConversionOptions) -> Result<ConversationRequest, ConversionError>;
|
fn convert(self, options: ConversionOptions) -> Result<ConversationRequest, ConversionError>;
|
||||||
|
|
||||||
/// Construct protocol events using the converted request's resolved settings.
|
/// 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(
|
fn chunk_generator(
|
||||||
request: &ConversationRequest,
|
request: &ConversationRequest,
|
||||||
id: String,
|
id: String,
|
||||||
model: String,
|
model: String,
|
||||||
) -> <Self::Response as ProtocolResponse>::ChunkGenerator;
|
) -> <Self::Response as ProtocolResponse>::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,
|
||||||
|
) -> <Self::Response as ProtocolResponse>::ChunkGenerator {
|
||||||
|
let _ = choice_index;
|
||||||
|
Self::chunk_generator(request, id, model)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A failure while converting protocol input into a conversation request.
|
/// A failure while converting protocol input into a conversation request.
|
||||||
|
|||||||
@ -11,6 +11,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
|
|||||||
|
|
||||||
use deepseek_recipe::anthropic::MessagesRequest;
|
use deepseek_recipe::anthropic::MessagesRequest;
|
||||||
pub use deepseek_recipe::error_response::ErrorResponse;
|
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::responses::response::ResponsesResponse;
|
||||||
use deepseek_recipe::openai::{ChatCompletionRequest, ResponsesRequest};
|
use deepseek_recipe::openai::{ChatCompletionRequest, ResponsesRequest};
|
||||||
pub use deepseek_recipe::request::ConversionError;
|
pub use deepseek_recipe::request::ConversionError;
|
||||||
@ -27,7 +28,7 @@ use deepseek_recipe_image::{
|
|||||||
ImageError, ImageQuota, ImageResolver, OpenCvImagePreprocessor, ReqwestImageFetcher,
|
ImageError, ImageQuota, ImageResolver, OpenCvImagePreprocessor, ReqwestImageFetcher,
|
||||||
};
|
};
|
||||||
use serde::{Serialize, de::DeserializeOwned};
|
use serde::{Serialize, de::DeserializeOwned};
|
||||||
use tokio_stream::{Stream, StreamExt};
|
use tokio_stream::{Stream, StreamExt, StreamMap};
|
||||||
|
|
||||||
const MOCK_ID: &str = "mock-id";
|
const MOCK_ID: &str = "mock-id";
|
||||||
const MOCK_MODEL: &str = "deepseek-flash";
|
const MOCK_MODEL: &str = "deepseek-flash";
|
||||||
@ -135,6 +136,10 @@ where
|
|||||||
|
|
||||||
/// Prepare Chat Completions SSE using the request's usage inclusion setting.
|
/// 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
|
/// # Errors
|
||||||
///
|
///
|
||||||
/// Returns the protocol conversion error before producing response events, or a
|
/// Returns the protocol conversion error before producing response events, or a
|
||||||
@ -145,10 +150,131 @@ pub async fn prepare_chat_completion_response(
|
|||||||
) -> Result<PreparedResponse, ConversionError> {
|
) -> Result<PreparedResponse, ConversionError> {
|
||||||
let include_usage = req.include_usage();
|
let include_usage = req.include_usage();
|
||||||
let request = req.convert(ConversionOptions::default())?;
|
let request = req.convert(ConversionOptions::default())?;
|
||||||
prepare_converted_response_with::<ChatCompletionRequest>(request, |generator| {
|
if request.completion_count <= 1 {
|
||||||
generator.with_include_usage(include_usage)
|
return prepare_converted_response_with::<ChatCompletionRequest>(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<PreparedResponse, ConversionError> {
|
||||||
|
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 <ChatCompletionResponse as ProtocolResponse>::chunk_event_type(&chunk) {
|
||||||
|
Some(event_type) => format!("event: {event_type}\ndata: {data}\n\n"),
|
||||||
|
None => format!("data: {data}\n\n"),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
});
|
||||||
|
let done = <ChatCompletionResponse as ProtocolResponse>::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<ChatCompletionResponse, ConversionError> {
|
||||||
|
let include_usage = req.include_usage();
|
||||||
|
let request = req.convert(ConversionOptions::default())?;
|
||||||
|
if request.completion_count <= 1 {
|
||||||
|
return complete_response_with::<ChatCompletionRequest>(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.
|
/// Prepare Responses SSE using the request's custom tool declarations.
|
||||||
|
|||||||
@ -13,9 +13,9 @@ use deepseek_recipe::response::ProtocolResponse;
|
|||||||
use deepseek_recipe::stream::ChunkGenerator;
|
use deepseek_recipe::stream::ChunkGenerator;
|
||||||
use serde::{Serialize, de::DeserializeOwned};
|
use serde::{Serialize, de::DeserializeOwned};
|
||||||
use server_rs::{
|
use server_rs::{
|
||||||
ConversionError, ErrorResponse, PreparedResponse, complete_response,
|
ConversionError, ErrorResponse, PreparedResponse, complete_chat_completion_response,
|
||||||
complete_responses_response, prepare_chat_completion_response, prepare_converted_response,
|
complete_response, complete_responses_response, prepare_chat_completion_response,
|
||||||
prepare_responses_response,
|
prepare_converted_response, prepare_responses_response,
|
||||||
};
|
};
|
||||||
|
|
||||||
type ApiError = (StatusCode, Json<ErrorResponse>);
|
type ApiError = (StatusCode, Json<ErrorResponse>);
|
||||||
@ -47,7 +47,10 @@ async fn chat_completion_handler(
|
|||||||
.map(sse_response)
|
.map(sse_response)
|
||||||
.map_err(http_error);
|
.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<ResponsesRequest>) -> Result<Response, ApiError> {
|
async fn responses_handler(Json(request): Json<ResponsesRequest>) -> Result<Response, ApiError> {
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user