core/engine/src/llm/client.py - main - alexefan136/flowstack

core/engine/src/llm/client.py
633 строки23 KB
Alexander Efanov
upd fix
29 июл 2026, 23:26
29 июл 2026, 23:26d8ac8ee
"""LLM клиент с поддержкой OpenAI-compatible API (Cloud.ru, OpenAI, etc.).

Возможности:
- Chat completions (streaming + non-streaming)
- Function Calling (tools, tool_choice, parallel_tool_calls)
- Structure Output (response_format / guided_json)
- Reasoning models (reasoning_content — DeepSeek-R1, V4-Pro, thinking)
- Usage tracking в стриме (stream_options.include_usage)
- max_completion_tokens (современный стандарт) с fallback на max_tokens
"""

from __future__ import annotations

from collections.abc import AsyncIterator, Sequence
from typing import Any

import httpx
import structlog
from openai import APIConnectionError, APIStatusError, APITimeoutError, AsyncOpenAI
from openai.types.chat import ChatCompletionMessageParam

from .config import LLMConfig, LLMSettings, get_settings

logger = structlog.get_logger()

# Провайдеры, которые используют legacy max_tokens (вместо max_completion_tokens)
_LEGACY_MAX_TOKENS_PROVIDERS = ("anthropic", "yandex")


def extract_text_content(msg: ChatCompletionMessageParam) -> str:
    """Извлечь текст из сообщения любого типа.

    Обрабатывает:
    - Простые текстовые сообщения (content: str)
    - Multimodal сообщения (content: list[ChatCompletionContentPartParam])
    """
    content = msg.get("content")
    if content is None:
        return ""

    if isinstance(content, str):
        return content

    if isinstance(content, list):
        texts: list[str] = []
        for part in content:
            if isinstance(part, dict):
                text_value = part.get("text")
                if isinstance(text_value, str):
                    texts.append(text_value)
        return "\n".join(texts)

    return str(content)


def _extract_error_body(error: Exception) -> str | None:
    """Извлечь тело ответа из ошибки OpenAI API."""
    if isinstance(error, APIStatusError):
        try:
            response = error.response  # httpx.Response
            body = response.text
            # Cloud.ru возвращает HTML при 404 — детектим
            if body and body.strip().startswith(("<!DOCTYPE", "<html")):
                return "[HTML response — вероятно, Cloud.ru error page или неверный URL]"
            return body[:500] if body else None
        except Exception:
            return None
    return None


def _serialize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]:
    """Сериализовать tool_calls из OpenAI SDK в унифицированный dict."""
    result: list[dict[str, Any]] = []
    for tc in tool_calls:
        result.append(
            {
                "id": tc.id,
                "type": getattr(tc, "type", "function"),
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                },
            }
        )
    return result


class LLMClientError(Exception):
    """Исключение LLM клиента с детальной информацией."""

    def __init__(
        self,
        message: str,
        status_code: int = 500,
        raw_response: str | None = None,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.raw_response = raw_response


class LLMClient:
    """Универсальный LLM клиент для OpenAI-compatible API.

    Детализированные таймауты:
    - connect_timeout: установление соединения (30 сек)
    - read_timeout: чтение ответа (300 сек для длинных запросов)
    - write/pool: отправка и пул (30 сек)
    """

    def __init__(self, settings: LLMSettings | None = None) -> None:
        self.settings = settings or get_settings()
        self._client: AsyncOpenAI | None = None
        self._http_client: httpx.AsyncClient | None = None

    def _create_http_client(self) -> httpx.AsyncClient:
        """Создать httpx клиент с правильными таймаутами."""
        timeout = self.settings.get_httpx_timeout()

        # Cloud.ru часто работает через корпоративные прокси (Kaspersky) —
        # отключаем SSL verification для cloud_ru и yandex
        verify_ssl = self.settings.provider not in ("cloud_ru", "yandex")

        return httpx.AsyncClient(
            timeout=timeout,
            verify=verify_ssl,
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
            ),
        )

    @property
    def client(self) -> AsyncOpenAI:
        """Ленивая инициализация OpenAI клиента."""
        if self._client is None:
            if not self.settings.is_configured():
                raise LLMClientError(
                    f"LLM provider '{self.settings.provider}' не настроен. "
                    f"Установите переменную окружения LLM_API_KEY.",
                    status_code=503,
                )

            base_url = self.settings.get_base_url()
            verify_ssl = self.settings.provider not in ("cloud_ru", "yandex")

            if self._http_client is None:
                self._http_client = self._create_http_client()

            self._client = AsyncOpenAI(
                api_key=self.settings.api_key,
                base_url=base_url,
                max_retries=self.settings.max_retries,
                http_client=self._http_client,
            )

            logger.info(
                "llm.client.initialized",
                provider=self.settings.provider,
                base_url=base_url,
                model=self.settings.model,
                verify_ssl=verify_ssl,
                connect_timeout=self.settings.connect_timeout,
                read_timeout=self.settings.read_timeout,
                max_retries=self.settings.max_retries,
            )

        return self._client

    # =========================================================================
    # BUILD REQUEST KWARGS
    # =========================================================================

    def _build_request_kwargs(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        config: LLMConfig,
        *,
        stream: bool,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        parallel_tool_calls: bool | None = None,
        response_format: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Собрать kwargs для chat.completions.create().

        Учитывает:
        - max_completion_tokens vs max_tokens (зависит от провайдера)
        - stream_options.include_usage (для получения usage в стриме)
        - tools / tool_choice / parallel_tool_calls (function calling)
        - response_format (structure output)
        """
        kwargs: dict[str, Any] = {
            "model": config.model,
            "messages": list(messages),
            "temperature": config.temperature,
            "top_p": config.top_p,
            "stream": stream,
        }

        # ✅ max_completion_tokens (современный стандарт) vs max_tokens (legacy)
        if self.settings.provider in _LEGACY_MAX_TOKENS_PROVIDERS:
            kwargs["max_tokens"] = config.max_tokens
        else:
            kwargs["max_completion_tokens"] = config.max_tokens

        # ✅ Usage в стриме (Cloud.ru/OpenAI поддерживают)
        if stream:
            kwargs["stream_options"] = {"include_usage": True}

        # ✅ Function calling
        if tools:
            kwargs["tools"] = tools
            if tool_choice is not None:
                kwargs["tool_choice"] = tool_choice
            if parallel_tool_calls is not None:
                kwargs["parallel_tool_calls"] = parallel_tool_calls

        # ✅ Structure output
        if response_format is not None:
            kwargs["response_format"] = response_format

        return kwargs

    # =========================================================================
    # NON-STREAMING
    # =========================================================================

    async def chat_completion(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        config: LLMConfig | None = None,
        *,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        parallel_tool_calls: bool | None = None,
        response_format: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Обычный (не-стриминг) запрос к LLM.

        Поддерживает function calling (tools) и structure output (response_format).

        Returns:
            {
                "content": str,
                "model": str,
                "tokens_prompt": int,
                "tokens_completion": int,
                "tokens_total": int,
                "finish_reason": str,
                "reasoning_content": str,   # только для reasoning-моделей
                "tool_calls": list[dict],   # только если модель вызвала инструменты
            }
        """
        config = config or LLMConfig.from_settings(self.settings)

        logger.info(
            "llm.request",
            model=config.model,
            messages_count=len(messages),
            stream=False,
            tools_count=len(tools) if tools else 0,
            has_response_format=response_format is not None,
        )

        kwargs = self._build_request_kwargs(
            messages,
            config,
            stream=False,
            tools=tools,
            tool_choice=tool_choice,
            parallel_tool_calls=parallel_tool_calls,
            response_format=response_format,
        )

        try:
            response = await self.client.chat.completions.create(**kwargs)

            choice = response.choices[0]
            message = choice.message
            usage = response.usage

            # ✅ Reasoning content (DeepSeek-R1, V4-Pro, thinking-модели)
            reasoning_content = getattr(message, "reasoning_content", None)

            result: dict[str, Any] = {
                "content": message.content or "",
                "model": config.model,
                "tokens_prompt": usage.prompt_tokens if usage else 0,
                "tokens_completion": usage.completion_tokens if usage else 0,
                "tokens_total": usage.total_tokens if usage else 0,
                "finish_reason": choice.finish_reason,
            }

            if reasoning_content:
                result["reasoning_content"] = reasoning_content

            # ✅ Tool calls (function calling)
            if message.tool_calls:
                result["tool_calls"] = _serialize_tool_calls(message.tool_calls)

            logger.info(
                "llm.response.success",
                model=config.model,
                tokens_total=result["tokens_total"],
                finish_reason=result["finish_reason"],
                tool_calls_count=len(result.get("tool_calls", [])),
                has_reasoning=bool(reasoning_content),
            )

            return result

        except APIStatusError as e:
            raw_response = _extract_error_body(e)
            error_msg = str(e)
            if raw_response:
                error_msg = f"{error_msg}\nRaw: {raw_response}"

            logger.error(
                "llm.request.failed",
                error=str(e),
                status_code=e.status_code,
                model=config.model,
                raw_response=raw_response[:200] if raw_response else None,
            )

            raise LLMClientError(
                message=error_msg,
                status_code=e.status_code,
                raw_response=raw_response,
            ) from e

        except APITimeoutError as e:
            logger.error(
                "llm.request.timeout",
                error=str(e),
                model=config.model,
                read_timeout=self.settings.read_timeout,
            )
            raise LLMClientError(
                message=f"LLM timeout: {e} (read timeout: {self.settings.read_timeout}s)",
                status_code=504,
            ) from e

        except APIConnectionError as e:
            logger.error(
                "llm.request.connection_error",
                error=str(e),
                model=config.model,
                connect_timeout=self.settings.connect_timeout,
            )
            raise LLMClientError(
                message=f"LLM connection error: {e} (connect timeout: {self.settings.connect_timeout}s)",
                status_code=503,
            ) from e

        except Exception as e:
            logger.error(
                "llm.request.unexpected_error",
                error=str(e),
                model=config.model,
            )
            raise LLMClientError(
                message=str(e),
                status_code=500,
            ) from e

    # =========================================================================
    # STREAMING
    # =========================================================================

    async def chat_completion_stream(
        self,
        messages: Sequence[ChatCompletionMessageParam],
        config: LLMConfig | None = None,
        *,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        parallel_tool_calls: bool | None = None,
        response_format: dict[str, Any] | None = None,
    ) -> AsyncIterator[dict[str, Any]]:
        """Стриминг запрос к LLM.

        Yield-ает события:
            {"type": "reasoning", "content": str, ...}   # цепочка рассуждений
            {"type": "content", "content": str, ...}      # основной контент
            {"type": "tool_calls", "tool_calls": [...]}   # вызовы инструментов
            {"type": "done", "tokens_*": int, "finish_reason": str, ...}
            {"type": "error", "error": str}
        """
        config = config or LLMConfig.from_settings(self.settings, stream=True)

        logger.info(
            "llm.request.stream",
            model=config.model,
            messages_count=len(messages),
            tools_count=len(tools) if tools else 0,
        )

        kwargs = self._build_request_kwargs(
            messages,
            config,
            stream=True,
            tools=tools,
            tool_choice=tool_choice,
            parallel_tool_calls=parallel_tool_calls,
            response_format=response_format,
        )

        try:
            stream = await self.client.chat.completions.create(**kwargs)

            full_content = ""
            full_reasoning = ""
            tokens_prompt = 0
            tokens_completion = 0
            finish_reason: str | None = None

            # ✅ Аккумуляция tool_calls по index (приходят инкрементально)
            tool_calls_acc: dict[int, dict[str, Any]] = {}

            async for chunk in stream:
                # ✅ Usage (приходит в финальном chunk благодаря include_usage)
                if hasattr(chunk, "usage") and chunk.usage:
                    tokens_prompt = chunk.usage.prompt_tokens or 0
                    tokens_completion = chunk.usage.completion_tokens or 0

                if not chunk.choices:
                    continue

                choice = chunk.choices[0]
                delta = choice.delta

                if choice.finish_reason:
                    finish_reason = choice.finish_reason

                # ✅ Reasoning content (chain-of-thought)
                reasoning_piece = getattr(delta, "reasoning_content", None)
                if reasoning_piece:
                    full_reasoning += reasoning_piece
                    yield {
                        "type": "reasoning",
                        "content": reasoning_piece,
                        "model": config.model,
                    }

                # ✅ Основной контент
                if delta.content:
                    full_content += delta.content
                    yield {
                        "type": "content",
                        "content": delta.content,
                        "model": config.model,
                    }

                # ✅ Tool calls (инкрементальная аккумуляция)
                if delta.tool_calls:
                    for tc_delta in delta.tool_calls:
                        idx = tc_delta.index
                        if idx not in tool_calls_acc:
                            tool_calls_acc[idx] = {
                                "id": "",
                                "type": "function",
                                "function": {"name": "", "arguments": ""},
                            }
                        acc = tool_calls_acc[idx]
                        if tc_delta.id:
                            acc["id"] = tc_delta.id
                        if tc_delta.type:
                            acc["type"] = tc_delta.type
                        if tc_delta.function:
                            if tc_delta.function.name:
                                acc["function"]["name"] += tc_delta.function.name
                            if tc_delta.function.arguments:
                                acc["function"]["arguments"] += tc_delta.function.arguments

            # Фоллбэк: если usage не вернулся от провайдера
            if tokens_prompt == 0:
                prompt_texts = [extract_text_content(m) for m in messages]
                prompt_text = "\n".join(t for t in prompt_texts if t)
                tokens_prompt = len(prompt_text) // 4 if prompt_text else 0
                tokens_completion = len(full_content) // 4 if full_content else 0

            # ✅ Yield tool_calls если модель вызвала инструменты
            if tool_calls_acc:
                tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc.keys())]
                yield {
                    "type": "tool_calls",
                    "tool_calls": tool_calls,
                    "model": config.model,
                }

            # ✅ Финальное событие
            done: dict[str, Any] = {
                "type": "done",
                "model": config.model,
                "tokens_prompt": tokens_prompt,
                "tokens_completion": tokens_completion,
                "tokens_total": tokens_prompt + tokens_completion,
                "finish_reason": finish_reason or ("tool_calls" if tool_calls_acc else "stop"),
            }
            if full_reasoning:
                done["reasoning_content"] = full_reasoning

            yield done

            logger.info(
                "llm.response.stream.success",
                model=config.model,
                tokens_total=tokens_prompt + tokens_completion,
                finish_reason=done["finish_reason"],
                tool_calls_count=len(tool_calls_acc),
                has_reasoning=bool(full_reasoning),
            )

        except APIStatusError as e:
            raw_response = _extract_error_body(e)
            error_msg = str(e)
            if raw_response:
                error_msg = f"{error_msg}\nRaw: {raw_response}"

            logger.error(
                "llm.request.stream.failed",
                error=str(e),
                status_code=e.status_code,
                model=config.model,
                raw_response=raw_response[:200] if raw_response else None,
            )

            yield {
                "type": "error",
                "error": error_msg,
                "model": config.model,
            }

        except (APITimeoutError, APIConnectionError) as e:
            logger.error(
                "llm.request.stream.connection_error",
                error=str(e),
                model=config.model,
            )
            yield {
                "type": "error",
                "error": f"Connection error: {e}",
                "model": config.model,
            }

        except Exception as e:
            logger.error(
                "llm.request.stream.unexpected_error",
                error=str(e),
                model=config.model,
            )
            yield {
                "type": "error",
                "error": str(e),
                "model": config.model,
            }

    # =========================================================================
    # EMBEDDINGS (для RAG — Cloud.ru поддерживает через OpenAI-compatible)
    # =========================================================================

    async def create_embedding(
        self,
        input_text: str | list[str],
        model: str | None = None,
    ) -> dict[str, Any]:
        """Создать эмбеддинги (для RAG).

        Cloud.ru: BAAI/bge-m3, Qwen/Qwen3-Embedding-0.6B, etc.

        Returns:
            {"embeddings": list[list[float]], "model": str, "tokens": int}
        """
        embedding_model = model or "BAAI/bge-m3"

        logger.info(
            "llm.embedding.request",
            model=embedding_model,
            input_count=len(input_text) if isinstance(input_text, list) else 1,
        )

        try:
            response = await self.client.embeddings.create(
                model=embedding_model,
                input=input_text,
            )

            embeddings = [item.embedding for item in response.data]
            tokens = response.usage.total_tokens if response.usage else 0

            return {
                "embeddings": embeddings,
                "model": embedding_model,
                "tokens": tokens,
            }

        except APIStatusError as e:
            raw_response = _extract_error_body(e)
            logger.error("llm.embedding.failed", error=str(e), status_code=e.status_code)
            raise LLMClientError(
                message=str(e),
                status_code=e.status_code,
                raw_response=raw_response,
            ) from e

        except Exception as e:
            logger.error("llm.embedding.unexpected_error", error=str(e))
            raise LLMClientError(message=str(e), status_code=500) from e

    # =========================================================================
    # LIFECYCLE
    # =========================================================================

    async def close(self) -> None:
        """Закрыть HTTP клиент."""
        if self._http_client:
            await self._http_client.aclose()
            self._http_client = None
        if self._client:
            self._client = None


_llm_client: LLMClient | None = None


def get_llm_client(settings: LLMSettings | None = None) -> LLMClient:
    """Получить singleton LLM клиент."""
    global _llm_client
    if _llm_client is None:
        _llm_client = LLMClient(settings)
    return _llm_client