DeepSeek plugin

DeepSeek is available through the OpenAI-compatible plugin in genkit-openai.

Installation

uv add genkit-openai

Configuration

Point OpenAI at DeepSeek’s API:

from genkit import Genkit
from genkit_openai import OpenAI
import os

ai = Genkit(
    plugins=[
        OpenAI(
            base_url='https://api.deepseek.com/v1',
            api_key=os.getenv('DEEPSEEK_API_KEY'),
        ),
    ],
)

Get an API key from your DeepSeek account settings and pass it as api_key= when you initialize the plugin—for example from the DEEPSEEK_API_KEY environment variable. Don’t embed API keys directly in code.

Usage

Use the openai_model() helper to reference a DeepSeek model.

from genkit import Genkit
from genkit_openai import OpenAI, openai_model
import os

ai = Genkit(
    plugins=[OpenAI(base_url='https://api.deepseek.com/v1', api_key=os.getenv('DEEPSEEK_API_KEY'))],
)

@ai.flow()
async def deepseek_flow(subject: str) -> str:
    """Generate information about a subject using DeepSeek.

    Args:
        subject: The subject to generate information about.

    Returns:
        Information about the subject.
    """
    response = await ai.generate(
        model=openai_model('deepseek-v4-flash'),
        prompt=f'Tell me something about {subject}.',
    )
    return response.text

Available Models:

The DeepSeek plugin provides access to several models:

• deepseek-v4-flash: Fast, cost-effective model with reasoning capabilities that closely approach V4-Pro, supporting both thinking and non-thinking modes
• deepseek-v4-pro: Flagship model with the strongest reasoning and agent capabilities, supporting both thinking and non-thinking modes

Both models support a large context window. Prefer the explicit V4 model IDs above over older deepseek-chat / deepseek-reasoner aliases.

Advanced usage

Reasoning Model

DeepSeek thinking mode shows step-by-step reasoning, making it ideal for complex logic, math, and coding problems:

@ai.flow()
async def reasoning_flow(problem: str) -> str:
    """Solve a problem using DeepSeek's reasoning model.

    Args:
        problem: The problem to solve.

    Returns:
        The solution with reasoning steps.
    """
    response = await ai.generate(
        model=openai_model('deepseek-v4-pro'),
        prompt=f'Solve this problem step by step: {problem}',
    )
    return response.text

Example with a classic reasoning problem:

response = await ai.generate(
    model=openai_model('deepseek-v4-pro'),
    prompt='What is heavier, one kilo of steel or one kilo of feathers?',
)
print(response.text)  # Shows reasoning steps before the answer

Tool Calling

DeepSeek models support tool calling, allowing them to use functions you define:

from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    """Input for weather tool."""
    location: str = Field(description='City name')

@ai.tool()
async def get_weather(input: WeatherInput) -> str:
    """Get the current weather for a location."""
    # In a real implementation, call a weather API
    return f'22°C and sunny in {input.location}'

@ai.flow()
async def weather_flow(location: str) -> str:
    """Get weather information using DeepSeek with tool calling.

    Args:
        location: The location to get weather for.

    Returns:
        Weather information for the location.
    """
    response = await ai.generate(
        model=openai_model('deepseek-v4-flash'),
        prompt=f'What is the weather in {location}?',
        tools=[get_weather],
    )
    return response.text

Streaming

The plugin supports streaming responses for real-time output:

from genkit import ActionRunContext

@ai.flow()
async def streaming_flow(topic: str, ctx: ActionRunContext) -> str:
    """Generate content with streaming output.

    Args:
        topic: Topic to generate content about.
        ctx: Action context for streaming chunks.

    Returns:
        The complete generated content.
    """
    stream_response = ai.generate_stream(
        model=openai_model('deepseek-v4-flash'),
        prompt=f'Tell me about {topic}',
    )
    async for chunk in stream_response.stream:
        ctx.send_chunk(chunk.text)
    return (await stream_response.response).text

Multi-turn Chat

Maintain conversation context across multiple turns:

from genkit import Message, Part, Role, TextPart

@ai.flow()
async def chat_flow() -> str:
    """Example of multi-turn conversation with context.

    Returns:
        The final response.
    """
    history = []

    # First message
    response1 = await ai.generate(
        model=openai_model('deepseek-v4-flash'),
        prompt='I love Japanese food, especially ramen.',
        system='You are a helpful assistant.',
    )

    # Build conversation history
    history.append(Message(
        role=Role.USER,
        content=[Part(root=TextPart(text='I love Japanese food, especially ramen.'))]
    ))
    if response1.message:
        history.append(response1.message)

    # Follow-up using context
    response2 = await ai.generate(
        model=openai_model('deepseek-v4-flash'),
        messages=[
            *history,
            Message(
                role=Role.USER,
                content=[Part(root=TextPart(text='What food did I mention?'))]
            ),
        ],
        system='You are a helpful assistant.',
    )
    return response2.text

Structured Output

Generate structured data using Pydantic models:

from pydantic import BaseModel, Field

class BookRecommendation(BaseModel):
    """A book recommendation."""
    title: str = Field(description='Book title')
    author: str = Field(description='Book author')
    genre: str = Field(description='Primary genre')
    summary: str = Field(description='Brief summary')
    why_recommended: str = Field(description='Why this book is recommended')

@ai.flow()
async def recommend_book(preferences: str) -> BookRecommendation:
    """Get a book recommendation with structured output.

    Args:
        preferences: User's reading preferences.

    Returns:
        A structured book recommendation.
    """
    response = await ai.generate(
        model=openai_model('deepseek-v4-flash'),
        prompt=f'Recommend a book for someone who likes: {preferences}',
        output_schema=BookRecommendation,
    )
    return response.output

Passthrough configuration

You can pass configuration options that are not defined in the plugin’s custom configuration schema. This permits you to access new models and features without having to update your Genkit version.

from genkit import Genkit
from genkit_openai import OpenAI, openai_model
import os

ai = Genkit(plugins=[OpenAI(base_url='https://api.deepseek.com/v1', api_key=os.getenv('DEEPSEEK_API_KEY'))])

response = await ai.generate(
    prompt='Tell me a cool story',
    model=openai_model('deepseek-new'),  # hypothetical new model
    config={
        'new_feature_parameter': ...,  # hypothetical config needed for new model
    },
)

Genkit passes this configuration as-is to the DeepSeek API giving you access to the new model features. Note that the field name and types are not validated by Genkit and should match the DeepSeek API specification to work.