ZerdaZerda

BaseLLMInterface

On this page

llm.BaseLLMInterface — abstract base class (abc.ABC) that every provider interface inherits from.

Purpose

Defines the shared contract that lets calling code treat any provider interchangeably: construct with (model_name, api_key, **kwargs), call generate() / generate_async() / astream(), and clean up with cleanup() / acleanup() (the latter via a default no-op).

Constructor

Python · 4 lines
def __init__(self, model_name: str, api_key: str, **kwargs):
    self.model_name = model_name
    self.api_key = api_key
    self.kwargs = kwargs

No validation, no network calls, no default values applied — a plain attribute assignment. Subclasses call super().__init__(model_name, api_key, **kwargs) first, then perform their own SDK client setup (which is where import errors, missing credentials, etc. surface).

ParameterTypeRequiredNotes
model_namestrYes (no default at this level; subclasses supply their own default)Stored verbatim on self.model_name.
api_keystrYes (no default at this level)Stored verbatim on self.api_key. Note: AWSInterface accepts this parameter for interface consistency but does not use it — AWS auth goes through boto3's standard credential chain instead.
**kwargs—NoStored as self.kwargs, a plain dict. Not automatically used elsewhere in the base class; subclasses may read specific keys out of their own **kwargs (e.g. OllamaInterface reads base_url, auto_start, server_start_wait).

Abstract methods

Subclasses must implement both of these or Python will refuse to instantiate them (standard ABC behavior):

generate(messages=None, prompt=None, max_tokens, temperature, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Python · 23 lines
@abstractmethod
def generate(
    self,
    messages: Optional[list] = None,
    prompt: Optional[str] = None,
    max_tokens: int = llm_config_.max_token,
    temperature: float = llm_config_.temperature,
    tools: Optional[list] = None,
    tool_choice: Optional[Union[str, dict]] = None,
    **kwargs,
) -> Union[str, dict]:
    """generate text. Provide either `prompt` (a string) or `messages`
    (a list of {"role": ..., "content": ...} dicts, OpenAI-style).

    tools: optional list of tool definitions in the unified schema
        (see `llm.tools` module docstring), e.g.
        [{"name": "...", "description": "...", "input_schema": {...}}].
        If the model decides to call a tool, the return value becomes a
        dict: {"content": str | None, "tool_calls": [...], "stop_reason": ...}
        instead of a plain string.
    tool_choice: optional, forwarded as-is to the underlying provider
        (accepted values are provider-specific).
    """

Synchronous text generation. messages is the first parameter — call with keyword arguments (llm.generate(prompt=...) / llm.generate(messages=...)) rather than positionally, since a bare positional string is treated as messages, not prompt. Every one of the seven concrete subclasses implements this exact signature; what differs by provider is the payload shape sent downstream, whether a missing prompt/messages raises uncaught or is folded into the return value (see Architecture § prompt vs messages), and how tools/tool_choice are converted to that provider's native schema (see Tool Calling). tools and tool_choice are declared here on the abstract signature but, as with every other parameter, each subclass is free to ignore what it can't support — OllamaInterface, for instance, accepts tool_choice for signature symmetry but never actually forwards it anywhere, since Ollama's API has no equivalent concept.

generate_async(messages=None, prompt=None, max_tokens, temperature, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Same shape as generate, but async def — the async counterpart every subclass must implement. Same tools/tool_choice contract.

Concrete methods (inherited as-is unless overridden)

astream(messages=None, prompt=None, max_tokens, temperature, **kwargs) -> AsyncIterator[str]

Python · 7 lines
async def astream(self, messages=None, prompt=None, max_tokens=..., temperature=..., **kwargs):
    response = await self.generate_async(messages=messages, prompt=prompt, max_tokens=max_tokens, temperature=temperature, **kwargs)
    if isinstance(response, dict):
        response = response.get("error", "")
    for word in response.split(" "):
        yield word + " "
        await asyncio.sleep(0)

The isinstance(response, dict) check exists because generate_async can return a dict on two separate occasions now: providers whose error path returns {"error": ...}, and — since tools support was added — providers whose model called a tool, returning {"content": ..., "tool_calls": [...], "stop_reason": ...}. Without the isinstance check, response.split(" ") on either dict shape would raise AttributeError instead of yielding text. Note this default implementation only special-cases the error shape (response.get("error", "")): if a tool-call dict reached this code path, it has no "error" key, so .get("error", "") silently returns "" and the method yields nothing at all, rather than surfacing the tool call in any form. In practice this default astream is not actually reached by any of the seven concrete subclasses in this package — every one of them overrides astream with its own implementation — but the same "does not accept tools, does not reconstruct tool-call deltas" caveat applies to every one of those overrides too; see Tool Calling § Streaming limitation.

This is not real token streaming. The default implementation waits for the entire generate_async response and then yields it word-by-word, sleeping 0 seconds between words purely to yield control back to the event loop (not for pacing/throttling). Providers with genuine incremental streaming (OpenAI, Anthropic, Groq, AWS Bedrock) override this method entirely to stream real deltas from the provider's API. Gemini's override also falls back to this same word-split pattern (it does not override with true incremental streaming). Mistral's override attempts native streaming and falls back to the word-split pattern only if the native stream call raises.

[!TIP] Don't assume astream implies lower time-to-first-token Unless you're using OpenAI, Anthropic, Groq, or AWS Bedrock (or Mistral succeeds on its native path), astream() still waits for the full response internally before yielding anything — it changes how output is delivered to your code, not how fast the underlying API responds.

validate_connection(test_prompt="test", max_tokens=10, temperature=0.7, async_mode=False) -> dict

Python
def validate_connection(self, test_prompt="test", max_tokens=10, temperature=0.7, async_mode=False) -> dict:

Fires a small test generation and reports the outcome as a dict rather than raising.

Parameters

NameTypeDefaultNotes
test_promptstr"test"The prompt sent as a connectivity check.
max_tokensint10Kept small deliberately, since this is just a health check.
temperaturefloat0.7Independent of the class's normal generation default.
async_modeboolFalseIf True, runs generate_async via asyncio.run(...) internally — do not call with async_mode=True from inside an already-running event loop, since asyncio.run will raise RuntimeError: asyncio.run() cannot be called from a running event loop in that case. From synchronous code this is safe.

Internally, validate_connection calls self.generate(prompt=test_prompt, ...) / self.generate_async(prompt=test_prompt, ...) with prompt passed by keyword. This matters now that messages is the first positional parameter on every provider: an earlier version of this method called generate positionally (self.generate(test_prompt, ...)), which — after messages became the first parameter everywhere — would have silently sent the test string in as messages instead of prompt on every provider. That's fixed; test_prompt is always passed as the prompt= keyword.

Returns: dict with keys:

  • success: bool
  • reason: str — "Connection successful", "Empty response", or the caught exception's string.
  • response: str | None — the generated text if successful, else None.

Raises: Nothing — every exception from the underlying generate/generate_async call is caught internally and reported in the returned dict.

Example

Python · 6 lines
from zerda.llm import OpenAIInterface

llm = OpenAIInterface(api_key="sk-...")
check = llm.validate_connection()
if not check["success"]:
    raise RuntimeError(f"LLM unavailable: {check['reason']}")

acleanup()

Python · 3 lines
async def acleanup(self):
    """Override in subclasses to release async resources (e.g., aiohttp sessions)"""
    pass

No-op by default. Subclasses that open async SDK sessions (OpenAI, Anthropic, Groq, AWS Bedrock, Mistral) override this to close them.

__aenter__ / __aexit__

Python · 6 lines
async def __aenter__(self):
    return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
    await self.acleanup()
    return False

Enables async with SomeInterface(...) as llm: — on exit, acleanup() runs regardless of whether an exception occurred (__aexit__ returns False, so exceptions are not suppressed).

Thread safety / async safety

Not documented or enforced in the implementation — there is no locking, and no explicit statement of thread safety in the source. Given that most subclasses store SDK client objects as plain instance attributes, sharing a single interface instance across threads or concurrent async tasks follows the thread-safety characteristics of the underlying SDK client, which this package does not control or verify.

  • Architecture for the full contract diagram.
  • Tool Calling for the unified tools/tool_choice contract every subclass implements against this base signature.
  • Any concrete provider page (e.g. OpenAIInterface) for how the abstract methods are implemented.

Related pages