ZerdaZerda

OpenAIInterface

On this page

llm.OpenAIInterface — sync & async OpenAI Chat Completions interface with native token streaming.

Constructor

Python · 5 lines
OpenAIInterface(
    model_name: str = "gpt-4",
    api_key: Optional[str] = None,
    **kwargs,
)

Creates both an openai.OpenAI (sync) and openai.AsyncOpenAI (async) client eagerly, both with the same api_key. If the openai package isn't installed, raises ImportError("install openai: pip install openai").

ParameterTypeDefaultNotes
model_namestr"gpt-4"Any OpenAI chat-completions-compatible model id.
api_keyOptional[str]NonePassed straight to both openai.OpenAI(api_key=...) and openai.AsyncOpenAI(api_key=...). If None, the openai SDK's own environment-variable fallback (OPENAI_API_KEY) applies.
**kwargs——Stored on self.kwargs via the base class; not otherwise consumed by this constructor.

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

Synchronous call to chat.completions.create.

Parameters

NameTypeRequiredDefaultNotes
messagesOptional[list[dict]]One of messages/prompt requiredNoneStandard OpenAI chat message list, e.g. [{"role": "user", "content": "..."}]. Takes priority over prompt if both are given.
promptOptional[str]One of messages/prompt requiredNoneConvenience single-turn shortcut; internally wrapped into [{"role": "user", "content": prompt}].
max_tokensintNo2048 (from llm_config)Forwarded as max_tokens.
temperaturefloatNo0.3 (from llm_config)Forwarded as temperature.
toolsOptional[list[dict]]NoNoneTool definitions in the unified schema; converted via llm.tools.to_openai_tools into chat.completions.create's native tools=[{"type": "function", "function": {...}}] shape before the call.
tool_choiceOptional[Union[str, dict]]NoNoneForwarded as-is to chat.completions.create(tool_choice=...) when given — accepts OpenAI's own values ("auto", "required", "none", or {"type": "function", "function": {"name": "..."}}).
**kwargs—No—Forwarded directly into chat.completions.create(...), so any OpenAI-specific parameter (e.g. top_p, presence_penalty, stop) works without this class knowing about it.

Returns: If the model doesn't call a tool, r.choices[0].message.content — a plain str, exactly as without tools. If the model calls one or more tools, a dict: {"content": ..., "tool_calls": [{"id", "name", "arguments"}, ...], "stop_reason": r.choices[0].finish_reason}, built by llm.tools.parse_openai_tool_calls + tool_calls_to_dicts — arguments is already parsed from OpenAI's JSON-string form into a dict. On failure, a dict: {"error": str(e)} — note the return type is not uniform; calling code should check isinstance(result, dict) and then distinguish an error dict ("error" key) from a tool-call dict ("tool_calls" key) if it needs to. See Tool Calling for the full contract.

Raises: nothing directly — a missing prompt/messages does trigger _build_messages to raise ValueError("Either 'prompt' or 'messages' must be provided."), but that call happens inside this method's own try/except, so the ValueError is caught along with every other exception from the OpenAI SDK, logged via logger.error(...), and converted into the {"error": ...} return value — it does not propagate to the caller. (Verified against the source and empirically: OpenAIInterface(api_key="...").generate() with no arguments returns {"error": "Either 'prompt' or 'messages' must be provided."} rather than raising.)

Example

Python · 8 lines
from zerda.llm import OpenAIInterface

llm = OpenAIInterface(api_key="sk-...")
result = llm.generate(prompt="Summarize RAG in one sentence.")
if isinstance(result, dict):
    print("Error:", result["error"])
else:
    print(result)

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

Identical contract to generate, but calls await self._async_client.chat.completions.create(...). Same return-shape caveat applies, including the tool-call dict shape.

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

Native incremental streaming — passes stream=True to chat.completions.create and yields each chunk's delta.content as it arrives:

Python · 2 lines
async for token in llm.astream(prompt="Write a haiku about the desert."):
    print(token, end="", flush=True)

On error, yields a single string f"Error: {e}" instead of raising — the generator does not raise; the error surfaces as stream content.

astream does not accept tools/tool_choice as named parameters and does not parse tool-call deltas — a tools= kwarg would be forwarded via **kwargs straight into the streaming chat.completions.create call, but any tool-call chunks in the resulting stream are not reconstructed into the unified tool_calls shape. Use generate/generate_async for tool calling.

cleanup() / acleanup()

Python · 2 lines
def cleanup(self):        # closes self._sync_client
async def acleanup(self): # closes self._async_client

Both wrap the close call in try/except, logging a warning (not raising) if it fails. Call cleanup() when you're done with the sync client, and either await llm.acleanup() or use async with OpenAIInterface(...) as llm: for the async client.

Thread safety

Not explicitly documented; relies on the underlying openai SDK client's own thread-safety guarantees, which this wrapper does not add to or verify.

Related pages