client.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/deepseek/client.py
 
"""
DeepSeek API client with streaming chat and automatic retry logic.
 
Purpose
-------
Provides :class:`DeepSeekClient` which wraps the OpenAI-compatible
SDK, handles streaming of reasoning content, text, tool calls, and
token usage, and retries transient errors with exponential backoff.
 
Moved from ``oc_chat/core/client.py`` in Phase C.2b. The old path
remains as a compat re-export shim until C.3f.
 
Responsibilities
----------------
- :class:`DeepSeekClient` — concrete :class:`BaseClient` for DeepSeek;
  exposes ``stream_chat``, ``one_shot``, ``list_models``.
- :func:`_is_retryable` — decide whether an SDK exception is worth
  retrying (auth errors no, transient HTTP yes).
- :func:`_deep_usage_dict` — defensively normalise pydantic
  ``CompletionUsage`` objects into plain nested dicts so consumers
  can rely on ``isinstance(field, dict)``.
- Streaming chunk shaping: yield ``("reasoning", str)``,
  ``("content", str)``, ``("tool_call", dict)``, ``("usage", dict)``
  in API arrival order, then a terminal ``("finish_reason", str)``.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Up to 3 attempts on transient errors with delays of 1/2/4 seconds.
- Authentication failures fail fast — never retried.
- ``stream_chat`` is a generator; iteration completes when the SDK's
  stream ends.
- Usage objects are always converted to plain dicts before being
  yielded (no pydantic models leak past this client).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- openai (API client for chat completions and streaming)
- option_c_agent_contracts (lazy; SupportsCancellation type of the stream_chat cancellation token)
- httpx (lazy; HTTP client with trust_env off, handed to the OpenAI SDK)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §5 Streaming chunk tuples — concrete impl
of the stream_chat tuple contract.
"""
from __future__ import annotations
 
import time
from collections.abc import Iterator
from typing import Any, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
from openai import OpenAI, AuthenticationError, APIStatusError
 
if TYPE_CHECKING:
    from option_c_agent_contracts import SupportsCancellation
 
_RETRY_DELAYS = (1, 2, 4)  # seconds between attempts (3 attempts total)
_RETRYABLE_STATUS = {429, 500, 502, 503, 524}
 
 
def _is_retryable(exc: Exception) -> bool:
    """Return True if *exc* is a transient error worth retrying."""
    if isinstance(exc, AuthenticationError):
        return False  # wrong key — retrying won't help
    if isinstance(exc, APIStatusError):
        return exc.status_code in _RETRYABLE_STATUS
    return True  # network errors, timeouts, etc.
 
 
def _deep_usage_dict(usage) -> dict:
    """Convert a pydantic ``CompletionUsage`` (or any object) to a
    plain nested dict, recursively ensuring pydantic model fields
    (``prompt_tokens_details``, ``completion_tokens_details``, etc.)
    become plain dicts rather than remaining as pydantic model objects.
 
    This prevents silent ``isinstance(field, dict)`` failures in consumers.
    """
    # Preferred: pydantic v2's ``model_dump`` already deep-converts.
    if hasattr(usage, "model_dump"):
        raw = usage.model_dump()
    else:
        # Fallback: ``dict()`` may leave nested pydantic models as-is.
        raw = dict(usage)
 
    # Recursively walk the result and coerce any remaining model objects.
    return _ensure_plain(raw)
 
 
def _ensure_plain(value):
    """Recursively convert pydantic / non-primitive objects to plain dicts/lists."""
    func = "_ensure_plain"
    if isinstance(value, dict):
        return {k: _ensure_plain(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_ensure_plain(v) for v in value]
    if value is None or isinstance(value, (str, int, float, bool, bytes)):
        return value
    # Pydantic model (model_dump wasn't called at a deeper level)
    if hasattr(value, "model_dump"):
        return _ensure_plain(value.model_dump())
    # Try __dict__ as last resort
    try:
        return _ensure_plain(value.__dict__)
    except AttributeError as e:
        # VALIDATION_ERROR — slots-only or built-in object; coerce via str()
        _dbg("OC_CHAT.CLIENT", func, f"VALIDATION_ERROR: {type(value).__name__}: {type(e).__name__}: {e}", 3)
        return str(value)
 
 
class DeepSeekClient:
    """Streaming client for the DeepSeek chat-completion API.
 
    Handles retries internally (``max_retries=0`` on the SDK so we
    control backoff), yields typed chunks via :meth:`stream_chat`, and
    provides a convenience :meth:`one_shot` for non-streaming requests
    (e.g. context compression summaries).
    """
 
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
        func = "__init__"
        import httpx
        # M8.0 WU-4c (F72/F79): inject an httpx client with trust_env=False so the
        # transport-trust env family (HTTP(S)_PROXY/ALL_PROXY/SSL_CERT_*/
        # REQUESTS_CA_BUNDLE) cannot MITM or redirect the provider transcript.
        self._client = OpenAI(
            api_key=api_key, base_url=base_url, max_retries=0,
            http_client=httpx.Client(trust_env=False),
        )
        _dbg("OC_CHAT.CLIENT", func, f"DeepSeekClient base_url={base_url}", 3)
 
    def list_models(self) -> list[str]:
        """Return a list of model IDs available from the API."""
        models = self._client.models.list()
        return [m.id for m in models]
 
    def stream_chat(
        self,
        messages: list[dict],
        model: str,
        effort: str | None = None,
        extra_body: dict | None = None,
        tools: list[dict] | None = None,
        max_tokens: int | None = None,
        *,
        cancellation: "SupportsCancellation | None" = None,
    ) -> Iterator[tuple[str, Any]]:
        """Stream a chat completion, yielding typed ``(chunk_type, data)`` tuples.
 
        Yields ``(chunk_type, data)`` tuples:
 
        * ``("reasoning", str)`` — chain-of-thought token
        * ``("content",   str)`` — answer token
        * ``("tool_call", dict)`` — completed tool call ``{id, name, arguments}``
        * ``("usage",     dict)`` — token usage (final chunk)
        * ``("finish_reason", str)`` — why generation stopped, yielded last.
          ``"length"`` means the model was cut off at ``max_tokens``; without
          it a truncation is indistinguishable from a normal completion.
 
        Retries up to 3 times with exponential backoff on transient
        errors.  Raises immediately on authentication errors (HTTP 401).
 
        If *cancellation* is provided, the token is polled before yielding
        each chunk; a cancelled token raises :class:`CancellationError` and
        the stream is abandoned (Phase O3c).
        """
        from collections import OrderedDict
        kwargs = OrderedDict()
        kwargs["model"] = model
        if tools:
            kwargs["tools"] = tools
        kwargs["stream"] = True
        kwargs["messages"] = messages
        if effort is not None:
            kwargs["reasoning_effort"] = effort
        if extra_body:
            kwargs["extra_body"] = extra_body
        if max_tokens is not None:
            kwargs["max_tokens"] = max_tokens
 
        last_exc = None
        for attempt, delay in enumerate((*_RETRY_DELAYS, None)):
            try:
                stream = self._client.chat.completions.create(**kwargs)
                break
            except Exception as exc:
                last_exc = exc
                if not _is_retryable(exc) or delay is None:
                    raise
                print(f"\n[Retrying in {delay}s — {type(exc).__name__}]", flush=True)
                time.sleep(delay)
        else:
            raise last_exc  # all retries exhausted
 
        # Accumulate tool call fragments (arrive across multiple chunks)
        tool_calls_acc: dict[int, dict] = {}
        finish_reason: str | None = None
 
        for chunk in stream:
            if cancellation is not None:
                cancellation.raise_if_cancelled()
            # Usage is usually on the last (empty) chunk
            if hasattr(chunk, "usage") and chunk.usage is not None:
                yield ("usage", _deep_usage_dict(chunk.usage))
 
            if not chunk.choices:
                continue
 
            choice = chunk.choices[0]
            # Carried on the terminal chunk only. Keep the last non-empty value:
            # the stream ends with it, and it is the sole signal distinguishing
            # "the model finished" from "the model ran out of budget".
            if getattr(choice, "finish_reason", None):
                finish_reason = choice.finish_reason
 
            delta = choice.delta
 
            # DeepSeek chain-of-thought field
            if hasattr(delta, "reasoning_content") and delta.reasoning_content:
                yield ("reasoning", delta.reasoning_content)
 
            if delta.content:
                yield ("content", delta.content)
 
            # Tool call fragments
            if hasattr(delta, "tool_calls") and delta.tool_calls:
                for tc in delta.tool_calls:
                    idx = tc.index
                    if idx not in tool_calls_acc:
                        tool_calls_acc[idx] = {"id": "", "name": "", "arguments": ""}
                    if getattr(tc, "id", None):
                        tool_calls_acc[idx]["id"] = tc.id
                    if tc.function:
                        if tc.function.name:
                            tool_calls_acc[idx]["name"] = tc.function.name
                        if tc.function.arguments:
                            tool_calls_acc[idx]["arguments"] += tc.function.arguments
 
        # Yield completed tool calls in index order
        for idx in sorted(tool_calls_acc):
            yield ("tool_call", tool_calls_acc[idx])
 
        # Last, so a consumer that reads it has the full stream in hand — in
        # particular whether the truncation landed mid-tool_call (leaving
        # ``arguments`` an unparseable JSON fragment).
        if finish_reason is not None:
            yield ("finish_reason", finish_reason)
 
    def one_shot(
        self,
        messages: list,
        model: str,
        *,
        effort: str = None,
        extra_body: dict = None,
        max_tokens: int = None,
    ) -> str:
        """Non-streaming single call with retry (used for context compression summaries).
 
        Accepts the same shaping kwargs as :meth:`stream_chat` so callers
        (e.g. context compression in ``oc_chat/core/context.py``) can pass
        ``effort="low"`` and a small ``max_tokens`` cap for cheap summary
        calls.
        """
        from collections import OrderedDict
        kwargs: "OrderedDict[str, object]" = OrderedDict()
        kwargs["model"] = model
        kwargs["stream"] = False
        kwargs["messages"] = messages
        if effort is not None:
            kwargs["reasoning_effort"] = effort
        if extra_body:
            kwargs["extra_body"] = extra_body
        if max_tokens is not None:
            kwargs["max_tokens"] = max_tokens
 
        last_exc = None
        for attempt, delay in enumerate((*_RETRY_DELAYS, None)):
            try:
                resp = self._client.chat.completions.create(**kwargs)
                return resp.choices[0].message.content or ""
            except Exception as exc:
                last_exc = exc
                if not _is_retryable(exc) or delay is None:
                    raise
                time.sleep(delay)
        raise last_exc