orchestrata_ultimate.providers.adapters.deepseek_adapter

Reference › orchestrata_ultimate › providers › adapters › deepseek_adapter

deepseek_adapter is the adapter for the DeepSeek API, which is OpenAI-compatible (specification 011, requirement 032). DeepSeekAdapter extends BaseAdapter (base) and implements complete, get_models and supports_capability. Its responses carry tool calls as raw provider dictionaries, which the tool loop in the base agent (base) parses. client registers it when its key is configured, and the default profile assigns most volks to DeepSeek models.

At a glance

Class diagram

Dependency diagram

API

Classes

DeepSeekAdapter

Bases: BaseAdapter · Decorators: traces(spec='011', story='US-001', reqs=['REQ-011-032']) · Line 26

DeepSeek API provider adapter (OpenAI-compatible).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/adapters/deepseek_adapter.py

"""DeepSeek provider adapter.

SPEC-011 FR-007: Implements IProvider for DeepSeek API.
Uses OpenAI-compatible API format.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    Message,
    ModelCapability,
    ModelInfo,
    RawProviderResponse,
)
from orchestrata_ultimate.providers.adapters.base import (
    BaseAdapter,
    strip_tools_when_choice_none,
)

logger = logging.getLogger(__name__)


@traces(spec="011", story="US-001", reqs=["REQ-011-032"])
class DeepSeekAdapter(BaseAdapter):
    """DeepSeek API provider adapter (OpenAI-compatible)."""

    def _get_headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._config.api_key}",
            "Content-Type": "application/json",
        }

    @traces(
        spec="103",
        story="US-001",
        reqs=["REQ-103-005", "REQ-103-006", "REQ-103-007"],
    )
    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to DeepSeek."""
        client = await self._ensure_client()
        # Build OpenAI-compatible messages, preserving tool-calling fields when
        # present (assistant tool_calls, and tool-result messages). `tools` /
        # `tool_choice`, when used, arrive via `parameters` and are forwarded by
        # the `**call_params` spread below.
        api_messages: list[dict[str, Any]] = []
        for m in messages:
            msg: dict[str, Any] = {"role": m.role, "content": m.content or ""}
            if m.tool_calls:
                msg["tool_calls"] = m.tool_calls
            if m.tool_call_id:
                msg["tool_call_id"] = m.tool_call_id
            if m.name:
                msg["name"] = m.name
            # Thinking models (deepseek-v4-pro) require the reasoning_content of a
            # prior assistant turn to be passed back, or the API returns 400.
            if getattr(m, "reasoning_content", None):
                msg["reasoning_content"] = m.reasoning_content
            api_messages.append(msg)
        # Dogfood run-3e (F12): DeepSeek v4 ignored tool_choice="none" while
        # the `tools` schema was still attached to the request and emitted
        # its internal tool-call markup as text instead of a finished
        # answer. Omit `tools`/`tool_choice` entirely when tool_choice
        # resolves to "none" -- mirrors anthropic_adapter.py's handling (see
        # strip_tools_when_choice_none's docstring in base.py).
        call_params = strip_tools_when_choice_none(parameters)
        payload = {
            "model": model,
            "messages": api_messages,
            **call_params,
        }
        response = await client.post("/chat/completions", json=payload)
        response.raise_for_status()
        data = response.json()
        choice = data.get("choices", [{}])[0]
        message = choice.get("message", {})
        usage = data.get("usage", {})
        # DeepSeek reasoner may have reasoning_content; tool calls when requested.
        content = message.get("content") or ""
        reasoning = message.get("reasoning_content", "")
        tool_calls = message.get("tool_calls") or []
        return RawProviderResponse(
            content=content,
            input_tokens=usage.get("prompt_tokens", 0),
            output_tokens=usage.get("completion_tokens", 0),
            model=data.get("model", model),
            tool_calls=tool_calls,
            metadata={
                "id": data.get("id", ""),
                "finish_reason": choice.get("finish_reason", ""),
                "reasoning_content": reasoning,
            },
        )

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models."""
        return [
            ModelInfo(
                name=model_name,
                provider="deepseek",
                capabilities=pricing.capabilities,
                context_window=pricing.context_window,
                pricing=pricing,
            )
            for model_name, pricing in self._config.pricing.items()
        ]

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if any model supports the capability."""
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False