"""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)