63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Any, Optional
|
|
import warnings
|
|
|
|
|
|
def normalize_content(response):
|
|
"""Normalize LLM response content to a plain string.
|
|
|
|
Multiple providers (OpenAI Responses API, Google Gemini 3) return content
|
|
as a list of typed blocks, e.g. [{'type': 'reasoning', ...}, {'type': 'text', 'text': '...'}].
|
|
Downstream agents expect response.content to be a string. This extracts
|
|
and joins the text blocks, discarding reasoning/metadata blocks.
|
|
"""
|
|
content = response.content
|
|
if isinstance(content, list):
|
|
texts = [
|
|
item.get("text", "") if isinstance(item, dict) and item.get("type") == "text"
|
|
else item if isinstance(item, str) else ""
|
|
for item in content
|
|
]
|
|
response.content = "\n".join(t for t in texts if t)
|
|
return response
|
|
|
|
|
|
class BaseLLMClient(ABC):
|
|
"""Abstract base class for LLM clients."""
|
|
|
|
def __init__(self, model: str, base_url: Optional[str] = None, **kwargs):
|
|
self.model = model
|
|
self.base_url = base_url
|
|
self.kwargs = kwargs
|
|
|
|
def get_provider_name(self) -> str:
|
|
"""Return the provider name used in warning messages."""
|
|
provider = getattr(self, "provider", None)
|
|
if provider:
|
|
return str(provider)
|
|
return self.__class__.__name__.removesuffix("Client").lower()
|
|
|
|
def warn_if_unknown_model(self) -> None:
|
|
"""Warn when the model is outside the known list for the provider."""
|
|
if self.validate_model():
|
|
return
|
|
|
|
warnings.warn(
|
|
(
|
|
f"Model '{self.model}' is not in the known model list for "
|
|
f"provider '{self.get_provider_name()}'. Continuing anyway."
|
|
),
|
|
RuntimeWarning,
|
|
stacklevel=2,
|
|
)
|
|
|
|
@abstractmethod
|
|
def get_llm(self) -> Any:
|
|
"""Return the configured LLM instance."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def validate_model(self) -> bool:
|
|
"""Validate that the model is supported by this client."""
|
|
pass
|