mirror of
https://github.com/deepseek-ai/deepseek-recipe.git
synced 2026-10-28 17:17:08 +08:00
Merge 9fbb051caa35221199cd389196d12695fb681367 into 8cadfede7063c896b944e7bae05daa3549ae97ea
This commit is contained in:
commit
f34e4b08e3
@ -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`).
|
||||
|
||||
@ -59,6 +59,7 @@ impl ProtocolRequest for MessagesRequest {
|
||||
parsing_options,
|
||||
model: Some(self.model),
|
||||
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.
|
||||
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<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 [
|
||||
("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<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.
|
||||
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<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,
|
||||
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<String>,
|
||||
@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
@ -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,
|
||||
model: Some(self.model),
|
||||
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.
|
||||
pub model: Option<String>,
|
||||
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<ConversationRequest, ConversionError>;
|
||||
|
||||
/// 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,
|
||||
) -> <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.
|
||||
|
||||
@ -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<PreparedResponse, ConversionError> {
|
||||
let include_usage = req.include_usage();
|
||||
let request = req.convert(ConversionOptions::default())?;
|
||||
prepare_converted_response_with::<ChatCompletionRequest>(request, |generator| {
|
||||
generator.with_include_usage(include_usage)
|
||||
if request.completion_count <= 1 {
|
||||
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.
|
||||
|
||||
@ -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<ErrorResponse>);
|
||||
@ -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<ResponsesRequest>) -> Result<Response, ApiError> {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user