AIRegulation-DocAnalysis/backend/app/services/llm/base_client.py

"""Provide service-layer logic for base client.

P0-0: ``LLMResponse`` now carries an optional ``tool_calls`` list so that any
downstream code (agents, pipelines) can inspect and dispatch tool invocations
without touching the provider-specific adapter layer.
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum

from app.services.llm.tool_types import Tool, ToolCall  # noqa: F401 – re-exported for callers
# Keep provider-specific behavior explicit so debugging stays straightforward.



class LLMProvider(Enum):
    """Define the L L M Provider enumeration."""
    DEEPSEEK = "deepseek"
    QWEN = "qwen"
    QWEN_VL = "qwen_vl"


@dataclass
class LLMResponse:
    """Represent the L L M Response type."""
    content: str
    model: str
    usage: Dict[str, int] = field(default_factory=dict)
    finish_reason: str = "stop"
    latency_ms: int = 0
    error: Optional[str] = None
    # P0-0: populated when the model returns tool-call(s) instead of plain text.
    tool_calls: List[ToolCall] = field(default_factory=list)

    @property
    def is_success(self) -> bool:
        """Return whether success for the L L M Response instance."""
        return self.error is None


@dataclass
class LLMConfig:
    """Define configuration for l l m config."""
    provider: LLMProvider
    model: str
    api_key: str
    base_url: str
    max_tokens: int = 4096
    temperature: float = 0.7
    top_p: float = 0.9
    timeout: int = 300  # Keep provider-specific behavior explicit so debugging stays straightforward.


class BaseLLMClient(ABC):
    """Represent the Base L L M Client type."""

    def __init__(self, config: LLMConfig):
        """Initialize the Base L L M Client instance."""
        self.config = config
        self._client = None

    @abstractmethod
    def _init_client(self):
        """Handle init client for this module for the Base L L M Client instance."""
        pass

    @abstractmethod
    def chat(
        self,
        messages: List[Dict[str, str]],
        max_tokens: Optional[int] = None,
        temperature: Optional[float] = None,
        tools: Optional[List["Tool"]] = None,
        **kwargs
    ) -> LLMResponse:
        """Handle chat for the Base L L M Client instance.

        Args:
            messages: OpenAI-format message list.
            max_tokens: Override config max_tokens when set.
            temperature: Override config temperature when set.
            tools: Optional list of Tool definitions to offer the model.
                   When provided, the model may respond with tool_calls in the
                   returned LLMResponse instead of (or in addition to) content.
        """
        pass

    def complete(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        max_tokens: Optional[int] = None,
        temperature: Optional[float] = None,
        **kwargs
    ) -> LLMResponse:
        """Handle complete for the Base L L M Client instance."""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        return self.chat(messages, max_tokens, temperature, **kwargs)

    @abstractmethod
    def get_available_models(self) -> List[str]:
        """Return available models for the Base L L M Client instance."""
        pass

    def estimate_tokens(self, text: str) -> int:
        """Handle estimate tokens for the Base L L M Client instance."""
        # Keep provider-specific behavior explicit so debugging stays straightforward.
        chinese_chars = sum(1 for c in text if '一' <= c <= '鿿')
        other_chars = len(text) - chinese_chars
        return int(chinese_chars * 1.5 + other_chars * 0.25)