117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""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)
|