deepseek-recipe/docs/streaming.md

Streaming response

English | 中文

StreamProcessor converts backend output into Messages, Chat Completions, or Responses events. The examples below use mock inference and print Chat Completions chunks. The application supplies the backend and HTTP transport. In thinking mode, the processor counts reasoning_tokens for usage only when a tokenizer is attached; see Use with tokenizer.

Rust

Add the dependencies to your application:

cargo add deepseek-recipe@0.2 serde_json@1 tokio-stream@0.1
cargo add tokio@1 --features macros,rt-multi-thread

Save this as src/main.rs and run cargo run:

use deepseek_recipe::openai::ChatCompletionRequest;
use deepseek_recipe::request::{ConversionOptions, ProtocolRequest};
use deepseek_recipe::stream::{
    InferenceChunk, InferenceFinishReason, PromptUsage, StreamProcessor,
};
use serde_json::json;
use tokio_stream::{StreamExt, iter};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let body = json!({
        "model": "DeepSeek-V4.1-Flash",
        "messages": [{"role": "user", "content": "Hello"}],
        "thinking": {"type": "disabled"},
    });

    // The chunk generator uses the converted request's resolved settings.
    let request: ChatCompletionRequest = serde_json::from_value(body)?;
    let converted = request.convert(ConversionOptions::default())?;
    let generator = ChatCompletionRequest::chunk_generator(
        &converted,
        "id-1".into(),
        "DeepSeek-V4.1-Flash".into(),
    );
    let processor = StreamProcessor::new(generator, converted.parsing_options);

    // Mock the backend's inference chunks.
    let inference = iter(vec![
        InferenceChunk::Ready {
            system_fingerprint: Some("fp-1".into()),
            prompt_usage: PromptUsage {
                prompt_tokens: 4,
                prompt_cache_hit_tokens: 0,
            },
        },
        InferenceChunk::Text {
            content: "Hello!".into(),
            token_count: 2,
        },
        InferenceChunk::Finish {
            finish_reason: InferenceFinishReason::Stop,
        },
    ]);

    // Convert them into protocol chunks.
    let mut chunks = std::pin::pin!(processor.into_protocol_stream(inference));
    while let Some(chunk) = chunks.next().await {
        println!("{}", serde_json::to_string(&chunk?)?);
    }
    Ok(())
}

Python

Install the package with python3 -m pip install deepseek-recipe. push processes one backend chunk; finish completes response processing.

from deepseek_recipe import (
    ChatCompletionRequest,
    ConversionOptions,
    InferenceChunk,
    InferenceFinishReason,
    PromptUsage,
    StreamProcessor,
)

request = ChatCompletionRequest({
    "model": "DeepSeek-V4.1-Flash",
    "messages": [{"role": "user", "content": "Hello"}],
    "thinking": {"type": "disabled"},
})
converted = request.convert(ConversionOptions())
generator = ChatCompletionRequest.chunk_generator(converted, "id-1", "DeepSeek-V4.1-Flash")
processor = StreamProcessor(generator, converted.parsing_options)

# Mock the backend's inference chunks.
chunks = []
for chunk in [
    InferenceChunk.ready(system_fingerprint="fp-1", prompt_usage=PromptUsage(prompt_tokens=4)),
    InferenceChunk.text("Hello!", token_count=2),
    InferenceChunk.finish(InferenceFinishReason.Stop),
]:
    chunks.extend(processor.push(chunk))
chunks.extend(processor.finish())

for chunk in chunks:
    print(chunk.to_json())
processor.close()

See the project README for supported scope and the Rust API documentation for API contracts.