Yijia-Xiao
e3bc872982
Clear the deferred full-repo lint backlog so the whole tree passes the strict ruff select (E,W,F,I,B,UP,C4,SIM). Mechanical fixes dominate: import sorting, pep585/604 annotations, dropped dead imports, and whitespace. The few semantic changes are behavior-preserving: declare __all__ on the agent_utils and alpha_vantage re-export hubs; expand 'from x import *' to explicit names; use immutable tuple defaults instead of mutable list defaults; contextlib.suppress for try/except/pass; and narrow an over-broad assertRaises.
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
import warnings
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
|
|
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: str | None = 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
|