Layer 1 - LLM Abstraction

This layer is about abstracting the function of an LLM, so that you can think about prompting and output and tool use in a way that does not tie you to a specific LLM, its calling conventions, and the quirks of its specific library.

At this layer we have:

• LLMBroker: This is the main entrypoint to the layer. It leverages an LLM specific Gateway, and is the primary interface for interacting with the LLM on the other side. The LLMBroker correctly handles text generation, structured output, and tool use.

• ChatSession: This is a simple class that wraps the LLMBroker and provides a conversational interface to the LLM with context size management. It is a good starting point for building a chatbot.

• OllamaGateway, OpenAIGateway, OMLXGateway: These are out-of-the-box adapters that will interact with models available through Ollama, OpenAI and oMLX.

Note: - OpenAIGateway supports environment-variable defaults. If api_key is not provided, it uses OPENAI_API_KEY. If base_url is not provided, it uses OPENAI_API_ENDPOINT. - Explicit constructor arguments always take precedence over environment variables.

• LLMGateway: This is the abstract class that all LLM adapters must inherit from. It provides a common interface and isolation point for interacting with LLMs.

• MessageBuilder: This is a utility class for constructing messages with text, images, and file contents using a fluent interface.

Architecture Overview

The following diagram illustrates how the key classes in Layer 1 relate to each other:

classDiagram
    class LLMBroker {
        +model: str
        +adapter: LLMGateway
        +tokenizer: TokenizerGateway
        +tracer: TracerSystem
        +generate(messages, tools, temperature) str
        +generate_object(messages, object_model) BaseModel
    }

    class ChatSession {
        +broker: LLMBroker
        +messages: List[LLMMessage]
        +chat(message) str
        +clear_history()
    }

    class LLMGateway {
        <<abstract>>
        +complete(model, messages, tools) LLMGatewayResponse
        +calculate_embeddings(text, model) List[float]
    }

    class OllamaGateway {
        +complete(model, messages, tools) LLMGatewayResponse
        +calculate_embeddings(text, model) List[float]
    }

    class OpenAIGateway {
        +complete(model, messages, tools) LLMGatewayResponse
        +calculate_embeddings(text, model) List[float]
    }

    class TokenizerGateway {
        +encode(text) List
        +decode(tokens) str
    }

    class TracerSystem {
        <<interface>>
        +record_llm_call()
        +record_llm_response()
        +record_tool_call()
    }

    class LLMGatewayResponse {
        +content: str
        +tool_calls: List[LLMToolCall]
        +object: BaseModel
    }

    class LLMMessage {
        +role: MessageRole
        +content: str
        +tool_calls: List[LLMToolCall]
    }

    class MessageBuilder {
        +add_text(text) MessageBuilder
        +add_image(path) MessageBuilder
        +add_file(path) MessageBuilder
        +build() LLMMessage
    }

    ChatSession --> LLMBroker : wraps
    LLMBroker --> LLMGateway : uses via adapter
    LLMBroker --> TokenizerGateway : uses
    LLMBroker --> TracerSystem : uses (optional)
    OllamaGateway --|> LLMGateway : extends
    OpenAIGateway --|> LLMGateway : extends
    LLMGateway --> LLMGatewayResponse : returns
    LLMBroker --> LLMMessage : sends/receives
    MessageBuilder --> LLMMessage : builds
    LLMGatewayResponse --> LLMMessage : contains

Working with Embeddings

Mojentic provides embeddings functionality through the calculate_embeddings method in both the OllamaGateway and OpenAIGateway classes. Embeddings are vector representations of text that capture semantic meaning, making them useful for similarity comparisons, clustering, and other NLP tasks.

Usage Example

from mojentic.llm.gateways import OllamaGateway, OpenAIGateway

# Initialize the gateways
ollama_gateway = OllamaGateway()
openai_gateway = OpenAIGateway(api_key="your-api-key")

# Calculate embeddings using Ollama
text = "This is a sample text for embeddings."
ollama_embeddings = ollama_gateway.calculate_embeddings(
    text=text,
    model="mxbai-embed-large"  # Default model for Ollama
)

# Calculate embeddings using OpenAI
openai_embeddings = openai_gateway.calculate_embeddings(
    text=text,
    model="text-embedding-3-large"  # Default model for OpenAI
)

# Use the embeddings for similarity comparison, clustering, etc.
print(f"Ollama embeddings dimension: {len(ollama_embeddings)}")
print(f"OpenAI embeddings dimension: {len(openai_embeddings)}")

OpenAI splits long text into parts of at most 8191 tokens. It weights each part's embedding by its token count, then normalizes the mean to length 1.

Important Notes

• Available Models:
• For Ollama: Models like "mxbai-embed-large" (default), "nomic-embed-text" are commonly used

• For OpenAI: Models like "text-embedding-3-large" (default), "text-embedding-3-small", "text-embedding-ada-002" are available

• Embedding Dimensions:

• Different models produce embeddings with different dimensions
• Ollama's "mxbai-embed-large" typically produces 1024-dimensional embeddings

• OpenAI's "text-embedding-3-large" typically produces 3072-dimensional embeddings

• Performance Considerations:

• Embedding generation is generally faster and less resource-intensive than text generation
• Local embedding models (via Ollama) may be more cost-effective for high-volume applications

Working with Images

Mojentic supports sending images to LLMs using the MessageBuilder class. This allows you to perform image analysis, OCR, and other vision-based tasks with a clean, fluent interface.

Usage Example

from mojentic.llm import LLMBroker
from mojentic.llm import MessageBuilder
from pathlib import Path

# Initialize the LLM broker
llm = LLMBroker(model="gemma3:27b")  # Use an image-capable model

# Build a message with an image
message = MessageBuilder("Describe what you see in this image.") \
    .add_image(Path.cwd() / "images" / "example.jpg") \
    .build()

# Generate a response
result = llm.generate(messages=[message])

print(result)

Important Notes

• Image-Capable Models: You must use an image-capable model to process images. Not all models support image analysis.
• For Ollama: Models like "gemma3", "llava", and "bakllava" support image analysis

• For OpenAI: Models like "gpt-4-vision-preview" and "gpt-4o" support image analysis

• Image Formats: Supported image formats include JPEG, PNG, GIF, and WebP.

• Implementation Details:

• The MessageBuilder handles the appropriate encoding of images for different LLM providers
• For Ollama: Images are passed as file paths (handled internally by MessageBuilder)

• For OpenAI: Images are base64-encoded and included in the message content (handled internally by MessageBuilder)

• Performance Considerations: Image analysis may require more tokens and processing time than text-only requests.

Building Blocks

mojentic.llm.LLMBroker

This class is responsible for managing interaction with a Large Language Model. It abstracts the user from the specific mechanics of the LLM and provides a common interface for generating responses.

Source code in src/mojentic/llm/llm_broker.py

class LLMBroker:
    """
    This class is responsible for managing interaction with a Large Language Model. It abstracts
    the user
    from the specific mechanics of the LLM and provides a common interface for generating responses.
    """

    adapter: LLMGateway
    tokenizer: TokenizerGateway
    model: str
    tracer: TracerSystem | None
    tool_runner: ToolRunner

    def __init__(
        self,
        model: str,
        gateway: LLMGateway | None = None,
        tokenizer: TokenizerGateway | None = None,
        tracer: TracerSystem | None = None,
        tool_runner: ToolRunner | None = None,
        tool_context: ToolRunContext | None = None,
    ):
        """
        Create an instance of the LLMBroker.

        Parameters
        ----------
        model
            The name of the model to use.
        gateway
            The gateway to use for communication with the LLM. If None, a gateway is created that
            will utilize a local
            Ollama server.
        tokenizer
            The gateway to use for tokenization. This is used to log approximate token counts for
            the LLM calls. If
            None, tiktoken's `cl100k_base` tokenizer is used.
        tracer
            Optional tracer system to record LLM calls and responses.
        tool_runner
            Strategy for executing tool calls returned by the LLM. Defaults to
            :class:`SerialToolRunner` for backward compatibility. Pass
            :class:`AsyncParallelToolRunner` (used by the realtime broker) or a
            custom :class:`ToolRunner` to change concurrency policy.
        """
        self.model = model

        # Use null_tracer if no tracer is provided
        from mojentic.tracer import null_tracer

        self.tracer = tracer or null_tracer

        if tokenizer is None:
            self.tokenizer = TokenizerGateway()
        else:
            self.tokenizer = tokenizer
        if gateway is None:
            self.adapter = OllamaGateway()
        else:
            self.adapter = gateway

        self.tool_runner = tool_runner or SerialToolRunner()
        self.tool_context = tool_context

    def generate(
        self,
        messages: list[LLMMessage],
        tools=None,
        config: CompletionConfig | None = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
        correlation_id: str | None = None,
    ) -> str:
        """
        Generate a text response from the LLM.

        Parameters
        ----------
        messages : LLMMessage
            A list of messages to send to the LLM.
        tools : List[Tool]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be
            called and the output
            will be included in the response.
        config : Optional[CompletionConfig]
            Configuration object for LLM completion (recommended). If provided with individual
            kwargs, a DeprecationWarning is emitted.
        temperature : Optional[float]
            The temperature to use for the response. Deprecated: use config.
        num_ctx : Optional[int]
            The number of context tokens to use. Deprecated: use config.
        max_tokens : Optional[int]
            The maximum number of tokens to generate. Deprecated: use config.
        correlation_id : str
            UUID string that is copied from cause-to-affect for tracing events. A new UUID is
            generated when omitted.

        Returns
        -------
        str
            The response from the LLM.

        Raises
        ------
        MaxToolIterationsExceededError
            If tool calls exceed config.max_tool_iterations.
        """
        # Handle config vs individual kwargs
        if config is not None and any(
            param is not None for param in [temperature, num_ctx, max_tokens]
        ):
            warnings.warn(
                "Both config and individual kwargs provided. Using config and ignoring kwargs. "
                "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        elif config is None:
            # Build config from individual kwargs
            config = CompletionConfig(
                temperature=temperature,
                num_ctx=num_ctx,
                max_tokens=max_tokens,
            )

        correlation_id = _ensure_correlation_id(correlation_id)
        while True:
            if (
                config.max_tool_iterations is not None
                and config.max_tool_iterations <= 0
            ):
                raise MaxToolIterationsExceededError(
                    f"Tool call iterations exceeded the maximum budget for model '{self.model}'. "
                    f"Increase config.max_tool_iterations to allow more rounds."
                )
            result = self.generate_response(messages, tools, config, correlation_id)
            if not result.tool_calls or tools is None:
                return result.content
            paired = self._dispatch_tool_batch(
                result.tool_calls,
                tools,
                caller="LLMBroker",
                correlation_id=correlation_id,
            )
            messages.append(
                LLMMessage(
                    role=MessageRole.Assistant,
                    content=result.content,
                    thinking=(
                        result.thinking
                        if config.retain_reasoning
                        and result.finish_reason in ("stop", "tool_calls")
                        else None
                    ),
                    tool_calls=result.tool_calls,
                )
            )
            messages.extend(
                LLMMessage(
                    role=MessageRole.Tool,
                    content=self._serialize_outcome(outcome),
                    tool_calls=[call],
                )
                for call, outcome in paired
            )
            config = config.model_copy(
                update={
                    "max_tool_iterations": _next_iteration(config.max_tool_iterations)
                }
            )

    def generate_response(
        self,
        messages: list[LLMMessage],
        tools=None,
        config: CompletionConfig | None = None,
        correlation_id: str | None = None,
    ) -> LLMGatewayResponse:
        """Return one native model response without executing tools or changing history.

        The caller assembles context and decides when to dispatch returned calls.
        Provider failures propagate; no retry or tool iteration occurs here. A new
        correlation ID is generated when none is supplied.
        """
        config = config or CompletionConfig()
        correlation_id = _ensure_correlation_id(correlation_id)
        approximate_tokens = len(
            self.tokenizer.encode(self._content_to_count(messages))
        )
        logger.info(f"Requesting llm response with approx {approximate_tokens} tokens")

        # Convert messages to serializable dict for audit
        messages_for_tracer = [m.model_dump() for m in messages]

        # Record LLM call in tracer
        tools_for_tracer = (
            [{"name": t.name, "description": t.description} for t in tools]
            if tools
            else None
        )
        self.tracer.record_llm_call(
            self.model,
            messages_for_tracer,
            config.temperature,
            tools=tools_for_tracer,
            source=type(self),
            correlation_id=correlation_id,
        )

        # Measure call duration for audit
        start_time = time.time()

        result: LLMGatewayResponse = self.adapter.complete(
            model=self.model,
            messages=messages,
            tools=tools,
            config=config,
            temperature=config.temperature,
            num_ctx=config.num_ctx,
            max_tokens=config.max_tokens,
        )

        call_duration_ms = (time.time() - start_time) * 1000

        # Record LLM response in tracer
        tool_calls_for_tracer = (
            [tc.model_dump() for tc in result.tool_calls] if result.tool_calls else None
        )
        self.tracer.record_llm_response(
            self.model,
            result.content,
            tool_calls=tool_calls_for_tracer,
            call_duration_ms=call_duration_ms,
            source=type(self),
            correlation_id=correlation_id,
            **_response_evidence(result),
        )

        return result

    @property
    def last_stream_metrics(self) -> ClientTiming | None:
        """Latest dispatched stream evidence, including failure and local close."""
        return getattr(self.adapter, "last_stream_metrics", None)

    @last_stream_metrics.setter
    def last_stream_metrics(self, value: ClientTiming | None) -> None:
        self.adapter.last_stream_metrics = value

    def generate_stream(
        self,
        messages: list[LLMMessage],
        tools=None,
        config: CompletionConfig | None = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
        correlation_id: str | None = None,
    ) -> Iterator[str]:
        """
        Generate a streaming text response from the LLM.

        This method mirrors generate() but yields content chunks as they arrive from the LLM,
        providing a better user experience for long-running requests. When tool calls are
        detected, tools are executed and the LLM is called recursively, with the new response
        also being streamed.

        Parameters
        ----------
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        tools : List[Tool]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be
            called and the output will be included in the response.
        config : Optional[CompletionConfig]
            Configuration object for LLM completion (recommended). If provided with individual
            kwargs, a DeprecationWarning is emitted.
        temperature : Optional[float]
            The temperature to use for the response. Deprecated: use config.
        num_ctx : Optional[int]
            The number of context tokens to use. Deprecated: use config.
        max_tokens : Optional[int]
            The maximum number of tokens to generate. Deprecated: use config.
        correlation_id : str
            UUID string that is copied from cause-to-affect for tracing events. A new UUID is
            generated when omitted.

        Yields
        ------
        str
            Content chunks as they arrive from the LLM.

        Raises
        ------
        MaxToolIterationsExceededError
            If tool calls exceed config.max_tool_iterations.
        """
        # Handle config vs individual kwargs
        if config is not None and any(
            param is not None for param in [temperature, num_ctx, max_tokens]
        ):
            warnings.warn(
                "Both config and individual kwargs provided. Using config and ignoring kwargs. "
                "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        elif config is None:
            # Build config from individual kwargs
            config = CompletionConfig(
                temperature=temperature,
                num_ctx=num_ctx,
                max_tokens=max_tokens,
            )

        correlation_id = _ensure_correlation_id(correlation_id)
        while True:
            if (
                config.max_tool_iterations is not None
                and config.max_tool_iterations <= 0
            ):
                raise MaxToolIterationsExceededError(
                    f"Tool call iterations exceeded the maximum budget for model '{self.model}'. "
                    f"Increase config.max_tool_iterations to allow more recursion."
                )
            # Check if gateway supports streaming
            if not hasattr(self.adapter, "complete_stream"):
                raise NotImplementedError(
                    f"Gateway {type(self.adapter).__name__} does not support streaming"
                )

            approximate_tokens = len(
                self.tokenizer.encode(self._content_to_count(messages))
            )
            logger.info(
                f"Requesting streaming llm response with approx {approximate_tokens} tokens"
            )

            # Convert messages to serializable dict for audit
            messages_for_tracer = [m.model_dump() for m in messages]

            # Record LLM call in tracer
            tools_for_tracer = (
                [{"name": t.name, "description": t.description} for t in tools]
                if tools
                else None
            )
            self.tracer.record_llm_call(
                self.model,
                messages_for_tracer,
                config.temperature,
                tools=tools_for_tracer,
                source=type(self),
                correlation_id=correlation_id,
            )

            # Measure call duration for audit
            start_time = time.time()

            # Accumulate content and tool calls from stream
            accumulated_content = ""
            accumulated_thinking = ""
            stream_completed = False
            accumulated_tool_calls = []

            stream = self.adapter.complete_stream(
                model=self.model,
                messages=messages,
                tools=tools,
                config=config,
                temperature=config.temperature,
                num_ctx=config.num_ctx,
                max_tokens=config.max_tokens,
            )

            with ExitStack() as resources:
                close = getattr(stream, "close", None)
                if close is not None:
                    resources.callback(close)
                for chunk in stream:
                    stream_completed = stream_completed or getattr(
                        chunk, "completed", False
                    )
                    if config.retain_reasoning and getattr(chunk, "thinking", None):
                        accumulated_thinking += chunk.thinking
                    # Handle content chunks
                    if hasattr(chunk, "content") and chunk.content:
                        accumulated_content += chunk.content
                        yield chunk.content

                    # Handle tool calls if present
                    if hasattr(chunk, "tool_calls") and chunk.tool_calls:
                        accumulated_tool_calls.extend(chunk.tool_calls)

            call_duration_ms = (time.time() - start_time) * 1000

            # Record LLM response in tracer
            tool_calls_for_tracer = (
                [
                    tc.model_dump() if hasattr(tc, "model_dump") else tc
                    for tc in accumulated_tool_calls
                ]
                if accumulated_tool_calls
                else None
            )
            self.tracer.record_llm_response(
                self.model,
                accumulated_content,
                tool_calls=tool_calls_for_tracer,
                call_duration_ms=call_duration_ms,
                source=type(self),
                correlation_id=correlation_id,
            )

            # Process tool calls if any were accumulated
            if accumulated_tool_calls and tools is not None:
                logger.info("Tool call requested in streaming response")
                normalized_calls = [
                    self._normalize_streamed_tool_call(tc)
                    for tc in accumulated_tool_calls
                ]
                paired = self._dispatch_tool_batch(
                    normalized_calls,
                    tools,
                    caller="LLMBroker.generate_stream",
                    correlation_id=correlation_id,
                )

                messages.append(
                    LLMMessage(
                        role=MessageRole.Assistant,
                        content=accumulated_content,
                        thinking=(
                            accumulated_thinking or None
                            if config.retain_reasoning and stream_completed
                            else None
                        ),
                        tool_calls=normalized_calls,
                    )
                )
                messages.extend(
                    LLMMessage(
                        role=MessageRole.Tool,
                        content=self._serialize_outcome(outcome),
                        tool_calls=[call],
                    )
                    for call, outcome in paired
                )
                config = config.model_copy(
                    update={
                        "max_tool_iterations": _next_iteration(
                            config.max_tool_iterations
                        )
                    }
                )
            else:
                return

    def generate_stream_events(
        self,
        messages: list[LLMMessage],
        config: CompletionConfig | None = None,
        correlation_id: str | None = None,
    ) -> Iterator[StreamEvent]:
        """
        Stream one model turn as events that end with terminal completion evidence.

        Use this when incomplete output must never be mistaken for a result. The stream
        yields :class:`StreamContent` events and ends with exactly one terminal event:
        :class:`StreamCompleted` when the provider finished the turn normally, or
        :class:`StreamError` otherwise. Content yielded before a :class:`StreamError` is
        evidence of what the provider sent, not a usable result.

        The broker sends one request with no tools, forces zero tool iterations and never
        retries. Call ``close()`` on the generator to cancel the HTTP request. Use
        ``contextlib.closing`` around a loop that may break early. A retained generator
        stays open after ``break``. A gateway without
        ``complete_stream_events`` yields a single ``STREAM_EVENTS_UNSUPPORTED`` error
        without sending a request.

        Parameters
        ----------
        messages : List[LLMMessage]
            The messages to send.
        config : Optional[CompletionConfig]
            Configuration for the request. ``max_tool_iterations`` is forced to zero.
        correlation_id : Optional[str]
            UUID string that is copied from cause-to-affect for tracing events. A new UUID is
            generated when omitted.

        Yields
        ------
        StreamEvent
            Content events followed by exactly one terminal event.
        """
        complete_stream_events = getattr(self.adapter, "complete_stream_events", None)
        if complete_stream_events is None:
            yield StreamError(
                reason=StreamErrorReason.STREAM_EVENTS_UNSUPPORTED,
                detail=type(self.adapter).__name__,
            )
            return

        config = (config or CompletionConfig()).model_copy(
            update={"max_tool_iterations": 0}
        )
        correlation_id = _ensure_correlation_id(correlation_id)
        self.tracer.record_llm_call(
            self.model,
            [m.model_dump() for m in messages],
            config.temperature,
            tools=None,
            source=type(self),
            correlation_id=correlation_id,
        )
        start_time = time.time()
        content = ""
        events = iter(
            complete_stream_events(model=self.model, messages=messages, config=config)
        )
        try:
            for event in events:
                if isinstance(event, StreamContent):
                    content += event.text
                    yield event
                    continue
                self.last_stream_metrics = event.metadata
                self._record_stream_events_response(
                    content, event, start_time, correlation_id
                )
                yield event
                return
            terminal = StreamError(reason=StreamErrorReason.INCOMPLETE_STREAM)
            self._record_stream_events_response(
                content, terminal, start_time, correlation_id
            )
            yield terminal
        finally:
            close = getattr(events, "close", None)
            if close is not None:
                close()

    def _record_stream_events_response(
        self,
        content: str,
        terminal: TerminalStreamEvent,
        start_time: float,
        correlation_id: str | None,
    ) -> None:
        evidence = terminal.metadata
        self.tracer.record_llm_response(
            self.model,
            content,
            call_duration_ms=(time.time() - start_time) * 1000,
            source=type(self),
            correlation_id=correlation_id,
            usage=evidence.usage if evidence else None,
            provider_model=evidence.provider_model if evidence else None,
            finish_reason=evidence.finish_reason if evidence else None,
            metadata=evidence.metadata if evidence else None,
        )

    @staticmethod
    def _normalize_streamed_tool_call(tool_call):
        """Convert raw provider tool-call payloads into LLMToolCall objects."""
        if isinstance(tool_call, LLMToolCall):
            return tool_call
        call = tool_call if hasattr(tool_call, "name") else tool_call.function
        return LLMToolCall(
            id=getattr(tool_call, "id", None), name=call.name, arguments=call.arguments
        )

    def _dispatch_tool_batch(self, tool_calls, tools, caller, correlation_id):
        """
        Execute a batch of tool calls through the configured tool_runner.

        Returns a list of ``(tool_call, outcome)`` pairs in input order for
        every requested call, including unknown names and errors.
        Per-call tracer events are emitted for every outcome.
        """
        from mojentic.llm.tools.runner import ToolCallExecution as _ToolCallExecution

        dispatched = list(tool_calls)
        executions = [
            _ToolCallExecution(
                id=getattr(call, "id", None) or f"call-{idx}",
                name=call.name,
                args=call.arguments,
            )
            for idx, call in enumerate(dispatched)
        ]

        if not executions:
            return []

        context = (
            self.tool_context.model_copy(
                update={"correlation_id": correlation_id, "source": caller}
            )
            if self.tool_context is not None
            else None
        )
        outcomes = self.tool_runner.run_batch(executions, tools, context)
        if inspect.isawaitable(outcomes):
            outcomes = _run_async_outcomes(outcomes)
        paired = list(zip(dispatched, outcomes))
        for tool_call, outcome in paired:
            self.tracer.record_tool_call(
                tool_call.name,
                tool_call.arguments,
                outcome.result if outcome.ok else {"error": str(outcome.error)},
                caller=caller,
                call_duration_ms=outcome.duration_ms,
                source=type(self),
                correlation_id=correlation_id,
            )
            if outcome.ok:
                logger.info("Function output", output=outcome.result)
            else:
                logger.warning(
                    "Tool execution failed",
                    function=tool_call.name,
                    error=str(outcome.error),
                )
        return paired

    @staticmethod
    def _serialize_outcome(outcome) -> str:
        if outcome.ok:
            return json.dumps(outcome.result)
        return json.dumps({"error": str(outcome.error)})

    def _content_to_count(self, messages: list[LLMMessage]):
        content = ""
        for message in messages:
            if message.content:
                content += message.content
        return content

    def generate_object(
        self,
        messages: list[LLMMessage],
        object_model: type[BaseModel],
        config: CompletionConfig | None = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
        correlation_id: str | None = None,
    ) -> BaseModel:
        """
        Generate a structured response from the LLM and return it as an object.

        Parameters
        ----------
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        object_model : BaseModel
            The class of the model to use for the structured response data.
        config : Optional[CompletionConfig]
            Configuration object for LLM completion (recommended). If provided with individual
            kwargs, a DeprecationWarning is emitted.
        temperature : Optional[float]
            The temperature to use for the response. Deprecated: use config.
        num_ctx : Optional[int]
            The number of context tokens to use. Deprecated: use config.
        max_tokens : Optional[int]
            The maximum number of tokens to generate. Deprecated: use config.
        correlation_id : str
            UUID string that is copied from cause-to-affect for tracing events. A new UUID is
            generated when omitted.

        Returns
        -------
        BaseModel
            An instance of the model class provided containing the structured response data.
        """
        # Handle config vs individual kwargs
        if config is not None and any(
            param is not None for param in [temperature, num_ctx, max_tokens]
        ):
            warnings.warn(
                "Both config and individual kwargs provided. Using config and ignoring kwargs. "
                "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        elif config is None:
            # Build config from individual kwargs
            config = CompletionConfig(
                temperature=temperature,
                num_ctx=num_ctx,
                max_tokens=max_tokens,
            )
        correlation_id = _ensure_correlation_id(correlation_id)
        approximate_tokens = len(
            self.tokenizer.encode(self._content_to_count(messages))
        )
        logger.info(f"Requesting llm response with approx {approximate_tokens} tokens")

        # Convert messages to serializable dict for audit
        messages_for_tracer = [m.model_dump() for m in messages]

        # Record LLM call in tracer
        self.tracer.record_llm_call(
            self.model,
            messages_for_tracer,
            config.temperature,
            tools=None,
            source=type(self),
            correlation_id=correlation_id,
        )

        # Measure call duration for audit
        start_time = time.time()

        result = self.adapter.complete(
            model=self.model,
            messages=messages,
            object_model=object_model,
            config=config,
            temperature=config.temperature,
            num_ctx=config.num_ctx,
            max_tokens=config.max_tokens,
        )

        call_duration_ms = (time.time() - start_time) * 1000

        # Record LLM response in tracer with object representation
        # Convert object to string for tracer
        object_str = (
            str(result.object.model_dump())
            if hasattr(result.object, "model_dump")
            else str(result.object)
        )
        self.tracer.record_llm_response(
            self.model,
            f"Structured response: {object_str}",
            call_duration_ms=call_duration_ms,
            source=type(self),
            correlation_id=correlation_id,
            **_response_evidence(result),
        )

        return result.object

__init__(model, gateway=None, tokenizer=None, tracer=None, tool_runner=None, tool_context=None)

Create an instance of the LLMBroker.

Parameters:

Name	Type	Description	Default
model	str	The name of the model to use.	required
gateway	LLMGateway | None	The gateway to use for communication with the LLM. If None, a gateway is created that will utilize a local Ollama server.	None
tokenizer	TokenizerGateway | None	The gateway to use for tokenization. This is used to log approximate token counts for the LLM calls. If None, tiktoken's cl100k_base tokenizer is used.	None
tracer	TracerSystem | None	Optional tracer system to record LLM calls and responses.	None
tool_runner	ToolRunner | None	Strategy for executing tool calls returned by the LLM. Defaults to :class:SerialToolRunner for backward compatibility. Pass :class:AsyncParallelToolRunner (used by the realtime broker) or a custom :class:ToolRunner to change concurrency policy.	None

Source code in src/mojentic/llm/llm_broker.py

def __init__(
    self,
    model: str,
    gateway: LLMGateway | None = None,
    tokenizer: TokenizerGateway | None = None,
    tracer: TracerSystem | None = None,
    tool_runner: ToolRunner | None = None,
    tool_context: ToolRunContext | None = None,
):
    """
    Create an instance of the LLMBroker.

    Parameters
    ----------
    model
        The name of the model to use.
    gateway
        The gateway to use for communication with the LLM. If None, a gateway is created that
        will utilize a local
        Ollama server.
    tokenizer
        The gateway to use for tokenization. This is used to log approximate token counts for
        the LLM calls. If
        None, tiktoken's `cl100k_base` tokenizer is used.
    tracer
        Optional tracer system to record LLM calls and responses.
    tool_runner
        Strategy for executing tool calls returned by the LLM. Defaults to
        :class:`SerialToolRunner` for backward compatibility. Pass
        :class:`AsyncParallelToolRunner` (used by the realtime broker) or a
        custom :class:`ToolRunner` to change concurrency policy.
    """
    self.model = model

    # Use null_tracer if no tracer is provided
    from mojentic.tracer import null_tracer

    self.tracer = tracer or null_tracer

    if tokenizer is None:
        self.tokenizer = TokenizerGateway()
    else:
        self.tokenizer = tokenizer
    if gateway is None:
        self.adapter = OllamaGateway()
    else:
        self.adapter = gateway

    self.tool_runner = tool_runner or SerialToolRunner()
    self.tool_context = tool_context

generate(messages, tools=None, config=None, temperature=None, num_ctx=None, max_tokens=None, correlation_id=None)

Generate a text response from the LLM.

Parameters:

Name	Type	Description	Default
messages	LLMMessage	A list of messages to send to the LLM.	required
tools	List[Tool]	A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response.	None
config	Optional[CompletionConfig]	Configuration object for LLM completion (recommended). If provided with individual kwargs, a DeprecationWarning is emitted.	None
temperature	Optional[float]	The temperature to use for the response. Deprecated: use config.	None
num_ctx	Optional[int]	The number of context tokens to use. Deprecated: use config.	None
max_tokens	Optional[int]	The maximum number of tokens to generate. Deprecated: use config.	None
correlation_id	str	UUID string that is copied from cause-to-affect for tracing events. A new UUID is generated when omitted.	None

Returns:

Type	Description
str	The response from the LLM.

Raises:

Type	Description
MaxToolIterationsExceededError	If tool calls exceed config.max_tool_iterations.

Source code in src/mojentic/llm/llm_broker.py

def generate(
    self,
    messages: list[LLMMessage],
    tools=None,
    config: CompletionConfig | None = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
    correlation_id: str | None = None,
) -> str:
    """
    Generate a text response from the LLM.

    Parameters
    ----------
    messages : LLMMessage
        A list of messages to send to the LLM.
    tools : List[Tool]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be
        called and the output
        will be included in the response.
    config : Optional[CompletionConfig]
        Configuration object for LLM completion (recommended). If provided with individual
        kwargs, a DeprecationWarning is emitted.
    temperature : Optional[float]
        The temperature to use for the response. Deprecated: use config.
    num_ctx : Optional[int]
        The number of context tokens to use. Deprecated: use config.
    max_tokens : Optional[int]
        The maximum number of tokens to generate. Deprecated: use config.
    correlation_id : str
        UUID string that is copied from cause-to-affect for tracing events. A new UUID is
        generated when omitted.

    Returns
    -------
    str
        The response from the LLM.

    Raises
    ------
    MaxToolIterationsExceededError
        If tool calls exceed config.max_tool_iterations.
    """
    # Handle config vs individual kwargs
    if config is not None and any(
        param is not None for param in [temperature, num_ctx, max_tokens]
    ):
        warnings.warn(
            "Both config and individual kwargs provided. Using config and ignoring kwargs. "
            "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    elif config is None:
        # Build config from individual kwargs
        config = CompletionConfig(
            temperature=temperature,
            num_ctx=num_ctx,
            max_tokens=max_tokens,
        )

    correlation_id = _ensure_correlation_id(correlation_id)
    while True:
        if (
            config.max_tool_iterations is not None
            and config.max_tool_iterations <= 0
        ):
            raise MaxToolIterationsExceededError(
                f"Tool call iterations exceeded the maximum budget for model '{self.model}'. "
                f"Increase config.max_tool_iterations to allow more rounds."
            )
        result = self.generate_response(messages, tools, config, correlation_id)
        if not result.tool_calls or tools is None:
            return result.content
        paired = self._dispatch_tool_batch(
            result.tool_calls,
            tools,
            caller="LLMBroker",
            correlation_id=correlation_id,
        )
        messages.append(
            LLMMessage(
                role=MessageRole.Assistant,
                content=result.content,
                thinking=(
                    result.thinking
                    if config.retain_reasoning
                    and result.finish_reason in ("stop", "tool_calls")
                    else None
                ),
                tool_calls=result.tool_calls,
            )
        )
        messages.extend(
            LLMMessage(
                role=MessageRole.Tool,
                content=self._serialize_outcome(outcome),
                tool_calls=[call],
            )
            for call, outcome in paired
        )
        config = config.model_copy(
            update={
                "max_tool_iterations": _next_iteration(config.max_tool_iterations)
            }
        )

generate_object(messages, object_model, config=None, temperature=None, num_ctx=None, max_tokens=None, correlation_id=None)

Generate a structured response from the LLM and return it as an object.

Parameters:

Name	Type	Description	Default
messages	List[LLMMessage]	A list of messages to send to the LLM.	required
object_model	BaseModel	The class of the model to use for the structured response data.	required
config	Optional[CompletionConfig]	Configuration object for LLM completion (recommended). If provided with individual kwargs, a DeprecationWarning is emitted.	None
temperature	Optional[float]	The temperature to use for the response. Deprecated: use config.	None
num_ctx	Optional[int]	The number of context tokens to use. Deprecated: use config.	None
max_tokens	Optional[int]	The maximum number of tokens to generate. Deprecated: use config.	None
correlation_id	str	UUID string that is copied from cause-to-affect for tracing events. A new UUID is generated when omitted.	None

Returns:

Type	Description
BaseModel	An instance of the model class provided containing the structured response data.

Source code in src/mojentic/llm/llm_broker.py

def generate_object(
    self,
    messages: list[LLMMessage],
    object_model: type[BaseModel],
    config: CompletionConfig | None = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
    correlation_id: str | None = None,
) -> BaseModel:
    """
    Generate a structured response from the LLM and return it as an object.

    Parameters
    ----------
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    object_model : BaseModel
        The class of the model to use for the structured response data.
    config : Optional[CompletionConfig]
        Configuration object for LLM completion (recommended). If provided with individual
        kwargs, a DeprecationWarning is emitted.
    temperature : Optional[float]
        The temperature to use for the response. Deprecated: use config.
    num_ctx : Optional[int]
        The number of context tokens to use. Deprecated: use config.
    max_tokens : Optional[int]
        The maximum number of tokens to generate. Deprecated: use config.
    correlation_id : str
        UUID string that is copied from cause-to-affect for tracing events. A new UUID is
        generated when omitted.

    Returns
    -------
    BaseModel
        An instance of the model class provided containing the structured response data.
    """
    # Handle config vs individual kwargs
    if config is not None and any(
        param is not None for param in [temperature, num_ctx, max_tokens]
    ):
        warnings.warn(
            "Both config and individual kwargs provided. Using config and ignoring kwargs. "
            "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    elif config is None:
        # Build config from individual kwargs
        config = CompletionConfig(
            temperature=temperature,
            num_ctx=num_ctx,
            max_tokens=max_tokens,
        )
    correlation_id = _ensure_correlation_id(correlation_id)
    approximate_tokens = len(
        self.tokenizer.encode(self._content_to_count(messages))
    )
    logger.info(f"Requesting llm response with approx {approximate_tokens} tokens")

    # Convert messages to serializable dict for audit
    messages_for_tracer = [m.model_dump() for m in messages]

    # Record LLM call in tracer
    self.tracer.record_llm_call(
        self.model,
        messages_for_tracer,
        config.temperature,
        tools=None,
        source=type(self),
        correlation_id=correlation_id,
    )

    # Measure call duration for audit
    start_time = time.time()

    result = self.adapter.complete(
        model=self.model,
        messages=messages,
        object_model=object_model,
        config=config,
        temperature=config.temperature,
        num_ctx=config.num_ctx,
        max_tokens=config.max_tokens,
    )

    call_duration_ms = (time.time() - start_time) * 1000

    # Record LLM response in tracer with object representation
    # Convert object to string for tracer
    object_str = (
        str(result.object.model_dump())
        if hasattr(result.object, "model_dump")
        else str(result.object)
    )
    self.tracer.record_llm_response(
        self.model,
        f"Structured response: {object_str}",
        call_duration_ms=call_duration_ms,
        source=type(self),
        correlation_id=correlation_id,
        **_response_evidence(result),
    )

    return result.object

generate_response(messages, tools=None, config=None, correlation_id=None)

Return one native model response without executing tools or changing history.

The caller assembles context and decides when to dispatch returned calls. Provider failures propagate; no retry or tool iteration occurs here. A new correlation ID is generated when none is supplied.

Source code in src/mojentic/llm/llm_broker.py

def generate_response(
    self,
    messages: list[LLMMessage],
    tools=None,
    config: CompletionConfig | None = None,
    correlation_id: str | None = None,
) -> LLMGatewayResponse:
    """Return one native model response without executing tools or changing history.

    The caller assembles context and decides when to dispatch returned calls.
    Provider failures propagate; no retry or tool iteration occurs here. A new
    correlation ID is generated when none is supplied.
    """
    config = config or CompletionConfig()
    correlation_id = _ensure_correlation_id(correlation_id)
    approximate_tokens = len(
        self.tokenizer.encode(self._content_to_count(messages))
    )
    logger.info(f"Requesting llm response with approx {approximate_tokens} tokens")

    # Convert messages to serializable dict for audit
    messages_for_tracer = [m.model_dump() for m in messages]

    # Record LLM call in tracer
    tools_for_tracer = (
        [{"name": t.name, "description": t.description} for t in tools]
        if tools
        else None
    )
    self.tracer.record_llm_call(
        self.model,
        messages_for_tracer,
        config.temperature,
        tools=tools_for_tracer,
        source=type(self),
        correlation_id=correlation_id,
    )

    # Measure call duration for audit
    start_time = time.time()

    result: LLMGatewayResponse = self.adapter.complete(
        model=self.model,
        messages=messages,
        tools=tools,
        config=config,
        temperature=config.temperature,
        num_ctx=config.num_ctx,
        max_tokens=config.max_tokens,
    )

    call_duration_ms = (time.time() - start_time) * 1000

    # Record LLM response in tracer
    tool_calls_for_tracer = (
        [tc.model_dump() for tc in result.tool_calls] if result.tool_calls else None
    )
    self.tracer.record_llm_response(
        self.model,
        result.content,
        tool_calls=tool_calls_for_tracer,
        call_duration_ms=call_duration_ms,
        source=type(self),
        correlation_id=correlation_id,
        **_response_evidence(result),
    )

    return result

generate_stream(messages, tools=None, config=None, temperature=None, num_ctx=None, max_tokens=None, correlation_id=None)

Generate a streaming text response from the LLM.

This method mirrors generate() but yields content chunks as they arrive from the LLM, providing a better user experience for long-running requests. When tool calls are detected, tools are executed and the LLM is called recursively, with the new response also being streamed.

Parameters:

Name	Type	Description	Default
messages	List[LLMMessage]	A list of messages to send to the LLM.	required
tools	List[Tool]	A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response.	None
config	Optional[CompletionConfig]	Configuration object for LLM completion (recommended). If provided with individual kwargs, a DeprecationWarning is emitted.	None
temperature	Optional[float]	The temperature to use for the response. Deprecated: use config.	None
num_ctx	Optional[int]	The number of context tokens to use. Deprecated: use config.	None
max_tokens	Optional[int]	The maximum number of tokens to generate. Deprecated: use config.	None
correlation_id	str	UUID string that is copied from cause-to-affect for tracing events. A new UUID is generated when omitted.	None

Yields:

Type	Description
str	Content chunks as they arrive from the LLM.

Raises:

Type	Description
MaxToolIterationsExceededError	If tool calls exceed config.max_tool_iterations.

Source code in src/mojentic/llm/llm_broker.py

def generate_stream(
    self,
    messages: list[LLMMessage],
    tools=None,
    config: CompletionConfig | None = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
    correlation_id: str | None = None,
) -> Iterator[str]:
    """
    Generate a streaming text response from the LLM.

    This method mirrors generate() but yields content chunks as they arrive from the LLM,
    providing a better user experience for long-running requests. When tool calls are
    detected, tools are executed and the LLM is called recursively, with the new response
    also being streamed.

    Parameters
    ----------
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    tools : List[Tool]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be
        called and the output will be included in the response.
    config : Optional[CompletionConfig]
        Configuration object for LLM completion (recommended). If provided with individual
        kwargs, a DeprecationWarning is emitted.
    temperature : Optional[float]
        The temperature to use for the response. Deprecated: use config.
    num_ctx : Optional[int]
        The number of context tokens to use. Deprecated: use config.
    max_tokens : Optional[int]
        The maximum number of tokens to generate. Deprecated: use config.
    correlation_id : str
        UUID string that is copied from cause-to-affect for tracing events. A new UUID is
        generated when omitted.

    Yields
    ------
    str
        Content chunks as they arrive from the LLM.

    Raises
    ------
    MaxToolIterationsExceededError
        If tool calls exceed config.max_tool_iterations.
    """
    # Handle config vs individual kwargs
    if config is not None and any(
        param is not None for param in [temperature, num_ctx, max_tokens]
    ):
        warnings.warn(
            "Both config and individual kwargs provided. Using config and ignoring kwargs. "
            "Individual kwargs are deprecated, use config=CompletionConfig(...) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    elif config is None:
        # Build config from individual kwargs
        config = CompletionConfig(
            temperature=temperature,
            num_ctx=num_ctx,
            max_tokens=max_tokens,
        )

    correlation_id = _ensure_correlation_id(correlation_id)
    while True:
        if (
            config.max_tool_iterations is not None
            and config.max_tool_iterations <= 0
        ):
            raise MaxToolIterationsExceededError(
                f"Tool call iterations exceeded the maximum budget for model '{self.model}'. "
                f"Increase config.max_tool_iterations to allow more recursion."
            )
        # Check if gateway supports streaming
        if not hasattr(self.adapter, "complete_stream"):
            raise NotImplementedError(
                f"Gateway {type(self.adapter).__name__} does not support streaming"
            )

        approximate_tokens = len(
            self.tokenizer.encode(self._content_to_count(messages))
        )
        logger.info(
            f"Requesting streaming llm response with approx {approximate_tokens} tokens"
        )

        # Convert messages to serializable dict for audit
        messages_for_tracer = [m.model_dump() for m in messages]

        # Record LLM call in tracer
        tools_for_tracer = (
            [{"name": t.name, "description": t.description} for t in tools]
            if tools
            else None
        )
        self.tracer.record_llm_call(
            self.model,
            messages_for_tracer,
            config.temperature,
            tools=tools_for_tracer,
            source=type(self),
            correlation_id=correlation_id,
        )

        # Measure call duration for audit
        start_time = time.time()

        # Accumulate content and tool calls from stream
        accumulated_content = ""
        accumulated_thinking = ""
        stream_completed = False
        accumulated_tool_calls = []

        stream = self.adapter.complete_stream(
            model=self.model,
            messages=messages,
            tools=tools,
            config=config,
            temperature=config.temperature,
            num_ctx=config.num_ctx,
            max_tokens=config.max_tokens,
        )

        with ExitStack() as resources:
            close = getattr(stream, "close", None)
            if close is not None:
                resources.callback(close)
            for chunk in stream:
                stream_completed = stream_completed or getattr(
                    chunk, "completed", False
                )
                if config.retain_reasoning and getattr(chunk, "thinking", None):
                    accumulated_thinking += chunk.thinking
                # Handle content chunks
                if hasattr(chunk, "content") and chunk.content:
                    accumulated_content += chunk.content
                    yield chunk.content

                # Handle tool calls if present
                if hasattr(chunk, "tool_calls") and chunk.tool_calls:
                    accumulated_tool_calls.extend(chunk.tool_calls)

        call_duration_ms = (time.time() - start_time) * 1000

        # Record LLM response in tracer
        tool_calls_for_tracer = (
            [
                tc.model_dump() if hasattr(tc, "model_dump") else tc
                for tc in accumulated_tool_calls
            ]
            if accumulated_tool_calls
            else None
        )
        self.tracer.record_llm_response(
            self.model,
            accumulated_content,
            tool_calls=tool_calls_for_tracer,
            call_duration_ms=call_duration_ms,
            source=type(self),
            correlation_id=correlation_id,
        )

        # Process tool calls if any were accumulated
        if accumulated_tool_calls and tools is not None:
            logger.info("Tool call requested in streaming response")
            normalized_calls = [
                self._normalize_streamed_tool_call(tc)
                for tc in accumulated_tool_calls
            ]
            paired = self._dispatch_tool_batch(
                normalized_calls,
                tools,
                caller="LLMBroker.generate_stream",
                correlation_id=correlation_id,
            )

            messages.append(
                LLMMessage(
                    role=MessageRole.Assistant,
                    content=accumulated_content,
                    thinking=(
                        accumulated_thinking or None
                        if config.retain_reasoning and stream_completed
                        else None
                    ),
                    tool_calls=normalized_calls,
                )
            )
            messages.extend(
                LLMMessage(
                    role=MessageRole.Tool,
                    content=self._serialize_outcome(outcome),
                    tool_calls=[call],
                )
                for call, outcome in paired
            )
            config = config.model_copy(
                update={
                    "max_tool_iterations": _next_iteration(
                        config.max_tool_iterations
                    )
                }
            )
        else:
            return

generate_stream_events(messages, config=None, correlation_id=None)

Stream one model turn as events that end with terminal completion evidence.

Use this when incomplete output must never be mistaken for a result. The stream yields :class:StreamContent events and ends with exactly one terminal event: :class:StreamCompleted when the provider finished the turn normally, or :class:StreamError otherwise. Content yielded before a :class:StreamError is evidence of what the provider sent, not a usable result.

The broker sends one request with no tools, forces zero tool iterations and never retries. Call close() on the generator to cancel the HTTP request. Use contextlib.closing around a loop that may break early. A retained generator stays open after break. A gateway without complete_stream_events yields a single STREAM_EVENTS_UNSUPPORTED error without sending a request.

Parameters:

Name	Type	Description	Default
messages	List[LLMMessage]	The messages to send.	required
config	Optional[CompletionConfig]	Configuration for the request. max_tool_iterations is forced to zero.	None
correlation_id	Optional[str]	UUID string that is copied from cause-to-affect for tracing events. A new UUID is generated when omitted.	None

Yields:

Type	Description
StreamEvent	Content events followed by exactly one terminal event.

Source code in src/mojentic/llm/llm_broker.py

def generate_stream_events(
    self,
    messages: list[LLMMessage],
    config: CompletionConfig | None = None,
    correlation_id: str | None = None,
) -> Iterator[StreamEvent]:
    """
    Stream one model turn as events that end with terminal completion evidence.

    Use this when incomplete output must never be mistaken for a result. The stream
    yields :class:`StreamContent` events and ends with exactly one terminal event:
    :class:`StreamCompleted` when the provider finished the turn normally, or
    :class:`StreamError` otherwise. Content yielded before a :class:`StreamError` is
    evidence of what the provider sent, not a usable result.

    The broker sends one request with no tools, forces zero tool iterations and never
    retries. Call ``close()`` on the generator to cancel the HTTP request. Use
    ``contextlib.closing`` around a loop that may break early. A retained generator
    stays open after ``break``. A gateway without
    ``complete_stream_events`` yields a single ``STREAM_EVENTS_UNSUPPORTED`` error
    without sending a request.

    Parameters
    ----------
    messages : List[LLMMessage]
        The messages to send.
    config : Optional[CompletionConfig]
        Configuration for the request. ``max_tool_iterations`` is forced to zero.
    correlation_id : Optional[str]
        UUID string that is copied from cause-to-affect for tracing events. A new UUID is
        generated when omitted.

    Yields
    ------
    StreamEvent
        Content events followed by exactly one terminal event.
    """
    complete_stream_events = getattr(self.adapter, "complete_stream_events", None)
    if complete_stream_events is None:
        yield StreamError(
            reason=StreamErrorReason.STREAM_EVENTS_UNSUPPORTED,
            detail=type(self.adapter).__name__,
        )
        return

    config = (config or CompletionConfig()).model_copy(
        update={"max_tool_iterations": 0}
    )
    correlation_id = _ensure_correlation_id(correlation_id)
    self.tracer.record_llm_call(
        self.model,
        [m.model_dump() for m in messages],
        config.temperature,
        tools=None,
        source=type(self),
        correlation_id=correlation_id,
    )
    start_time = time.time()
    content = ""
    events = iter(
        complete_stream_events(model=self.model, messages=messages, config=config)
    )
    try:
        for event in events:
            if isinstance(event, StreamContent):
                content += event.text
                yield event
                continue
            self.last_stream_metrics = event.metadata
            self._record_stream_events_response(
                content, event, start_time, correlation_id
            )
            yield event
            return
        terminal = StreamError(reason=StreamErrorReason.INCOMPLETE_STREAM)
        self._record_stream_events_response(
            content, terminal, start_time, correlation_id
        )
        yield terminal
    finally:
        close = getattr(events, "close", None)
        if close is not None:
            close()

last_stream_metrics property writable

Latest dispatched stream evidence, including failure and local close.

mojentic.llm.ChatSession

This class is responsible for managing the state of a conversation with the LLM.

Source code in src/mojentic/llm/chat_session.py

class ChatSession:
    """
    This class is responsible for managing the state of a conversation with the LLM.
    """

    messages: list[SizedLLMMessage]

    def __init__(
        self,
        llm: LLMBroker,
        system_prompt: str = "You are a helpful assistant.",
        tools: list[LLMTool] | None = None,
        max_context: int = 32768,
        tokenizer_gateway: TokenizerGateway = None,
        config: CompletionConfig | None = None,
        temperature: float | None = None,
    ):
        """
        Create an instance of the ChatSession.

        Parameters
        ----------
        llm : LLMBroker
            The broker to use for generating responses.
        system_prompt : str, optional
            The prompt to use for the system messages. Defaults to "You are a helpful assistant."
        tools : List[LLMTool], optional
            The tools you want to make available to the LLM. Defaults to None.
        max_context : int, optional
            The maximum number of tokens to keep in the context. Defaults to 32768.
        tokenizer_gateway : TokenizerGateway, optional
            The gateway to use for tokenization. If None, `mxbai-embed-large` is used on a local Ollama server.
        config : Optional[CompletionConfig], optional
            Configuration object for LLM completion. If None, one is created from temperature.
            max_context is only a local trimming budget.
        temperature : float, optional
            The temperature to use for the response. Unspecified by default. Deprecated: use config.
        """

        self.llm = llm
        self.system_prompt = system_prompt
        self.tools = tools
        self.max_context = max_context

        # Use config if provided, otherwise build from individual kwargs
        if config is not None:
            self.config = config
        else:
            self.config = CompletionConfig(temperature=temperature)

        if tokenizer_gateway is None:
            self.tokenizer_gateway = TokenizerGateway()
        else:
            self.tokenizer_gateway = tokenizer_gateway

        self.messages: list[SizedLLMMessage] = []
        self.insert_message(
            LLMMessage(role=MessageRole.System, content=self.system_prompt)
        )

    def send(self, query):
        """
        Send a query to the LLM and return the response. Also records the query and response in the ongoing chat
        session.

        Parameters
        ----------
        query : str
            The query to send to the LLM.

        Returns
        -------
        str
            The response from the LLM.
        """
        self.insert_message(LLMMessage(role=MessageRole.User, content=query))
        response = self.llm.generate(
            self.messages, tools=self.tools, config=self.config
        )
        self._ensure_all_messages_are_sized()
        self.insert_message(LLMMessage(role=MessageRole.Assistant, content=response))
        return response

    @property
    def last_stream_metrics(self) -> ClientTiming | None:
        """Timing evidence for the latest completion in this session's broker."""
        return self.llm.last_stream_metrics

    def send_stream(self, query) -> Iterator[str]:
        """
        Send a query to the LLM and yield response chunks as they arrive. Records the query and
        the full assembled response in the ongoing chat session after the stream is consumed.

        Parameters
        ----------
        query : str
            The query to send to the LLM.

        Yields
        ------
        str
            Content chunks from the LLM response as they arrive.
        """
        self.insert_message(LLMMessage(role=MessageRole.User, content=query))
        accumulated = []
        stream = self.llm.generate_stream(
            self.messages, tools=self.tools, config=self.config
        )
        with ExitStack() as resources:
            close = getattr(stream, "close", None)
            if close is not None:
                resources.callback(close)
            for chunk in stream:
                accumulated.append(chunk)
                yield chunk
        self._ensure_all_messages_are_sized()
        self.insert_message(
            LLMMessage(role=MessageRole.Assistant, content="".join(accumulated))
        )

    def insert_message(self, message: LLMMessage):
        """
        Add a message onto the end of the chat session. If the total token count exceeds the max context, the oldest
        messages are removed.

        Parameters
        ----------
        message : LLMMessage
            The message to add to the chat session.
        """
        self.messages.append(self._build_sized_message(message))
        total_length = sum([msg.token_length for msg in self.messages])
        while total_length > self.max_context:
            total_length -= self.messages.pop(1).token_length

    def _build_sized_message(self, message):
        if message.content is None:
            return SizedLLMMessage(**message.model_dump(), token_length=0)
        else:
            new_message_length = len(self.tokenizer_gateway.encode(message.content))
            new_message = SizedLLMMessage(
                **message.model_dump(), token_length=new_message_length
            )
            return new_message

    def _ensure_all_messages_are_sized(self):
        for i, message in enumerate(self.messages):
            if not isinstance(message, SizedLLMMessage):
                self.messages[i] = self._build_sized_message(message)

__init__(llm, system_prompt='You are a helpful assistant.', tools=None, max_context=32768, tokenizer_gateway=None, config=None, temperature=None)

Create an instance of the ChatSession.

Parameters:

Name	Type	Description	Default
llm	LLMBroker	The broker to use for generating responses.	required
system_prompt	str	The prompt to use for the system messages. Defaults to "You are a helpful assistant."	'You are a helpful assistant.'
tools	List[LLMTool]	The tools you want to make available to the LLM. Defaults to None.	None
max_context	int	The maximum number of tokens to keep in the context. Defaults to 32768.	32768
tokenizer_gateway	TokenizerGateway	The gateway to use for tokenization. If None, mxbai-embed-large is used on a local Ollama server.	None
config	Optional[CompletionConfig]	Configuration object for LLM completion. If None, one is created from temperature. max_context is only a local trimming budget.	None
temperature	float	The temperature to use for the response. Unspecified by default. Deprecated: use config.	None

Source code in src/mojentic/llm/chat_session.py

def __init__(
    self,
    llm: LLMBroker,
    system_prompt: str = "You are a helpful assistant.",
    tools: list[LLMTool] | None = None,
    max_context: int = 32768,
    tokenizer_gateway: TokenizerGateway = None,
    config: CompletionConfig | None = None,
    temperature: float | None = None,
):
    """
    Create an instance of the ChatSession.

    Parameters
    ----------
    llm : LLMBroker
        The broker to use for generating responses.
    system_prompt : str, optional
        The prompt to use for the system messages. Defaults to "You are a helpful assistant."
    tools : List[LLMTool], optional
        The tools you want to make available to the LLM. Defaults to None.
    max_context : int, optional
        The maximum number of tokens to keep in the context. Defaults to 32768.
    tokenizer_gateway : TokenizerGateway, optional
        The gateway to use for tokenization. If None, `mxbai-embed-large` is used on a local Ollama server.
    config : Optional[CompletionConfig], optional
        Configuration object for LLM completion. If None, one is created from temperature.
        max_context is only a local trimming budget.
    temperature : float, optional
        The temperature to use for the response. Unspecified by default. Deprecated: use config.
    """

    self.llm = llm
    self.system_prompt = system_prompt
    self.tools = tools
    self.max_context = max_context

    # Use config if provided, otherwise build from individual kwargs
    if config is not None:
        self.config = config
    else:
        self.config = CompletionConfig(temperature=temperature)

    if tokenizer_gateway is None:
        self.tokenizer_gateway = TokenizerGateway()
    else:
        self.tokenizer_gateway = tokenizer_gateway

    self.messages: list[SizedLLMMessage] = []
    self.insert_message(
        LLMMessage(role=MessageRole.System, content=self.system_prompt)
    )

insert_message(message)

Add a message onto the end of the chat session. If the total token count exceeds the max context, the oldest messages are removed.

Parameters:

Name	Type	Description	Default
message	LLMMessage	The message to add to the chat session.	required

Source code in src/mojentic/llm/chat_session.py

def insert_message(self, message: LLMMessage):
    """
    Add a message onto the end of the chat session. If the total token count exceeds the max context, the oldest
    messages are removed.

    Parameters
    ----------
    message : LLMMessage
        The message to add to the chat session.
    """
    self.messages.append(self._build_sized_message(message))
    total_length = sum([msg.token_length for msg in self.messages])
    while total_length > self.max_context:
        total_length -= self.messages.pop(1).token_length

last_stream_metrics property

Timing evidence for the latest completion in this session's broker.

send(query)

Send a query to the LLM and return the response. Also records the query and response in the ongoing chat session.

Parameters:

Name	Type	Description	Default
query	str	The query to send to the LLM.	required

Returns:

Type	Description
str	The response from the LLM.

Source code in src/mojentic/llm/chat_session.py

def send(self, query):
    """
    Send a query to the LLM and return the response. Also records the query and response in the ongoing chat
    session.

    Parameters
    ----------
    query : str
        The query to send to the LLM.

    Returns
    -------
    str
        The response from the LLM.
    """
    self.insert_message(LLMMessage(role=MessageRole.User, content=query))
    response = self.llm.generate(
        self.messages, tools=self.tools, config=self.config
    )
    self._ensure_all_messages_are_sized()
    self.insert_message(LLMMessage(role=MessageRole.Assistant, content=response))
    return response

send_stream(query)

Send a query to the LLM and yield response chunks as they arrive. Records the query and the full assembled response in the ongoing chat session after the stream is consumed.

Parameters:

Name	Type	Description	Default
query	str	The query to send to the LLM.	required

Yields:

Type	Description
str	Content chunks from the LLM response as they arrive.

Source code in src/mojentic/llm/chat_session.py

def send_stream(self, query) -> Iterator[str]:
    """
    Send a query to the LLM and yield response chunks as they arrive. Records the query and
    the full assembled response in the ongoing chat session after the stream is consumed.

    Parameters
    ----------
    query : str
        The query to send to the LLM.

    Yields
    ------
    str
        Content chunks from the LLM response as they arrive.
    """
    self.insert_message(LLMMessage(role=MessageRole.User, content=query))
    accumulated = []
    stream = self.llm.generate_stream(
        self.messages, tools=self.tools, config=self.config
    )
    with ExitStack() as resources:
        close = getattr(stream, "close", None)
        if close is not None:
            resources.callback(close)
        for chunk in stream:
            accumulated.append(chunk)
            yield chunk
    self._ensure_all_messages_are_sized()
    self.insert_message(
        LLMMessage(role=MessageRole.Assistant, content="".join(accumulated))
    )

mojentic.llm.gateways.LLMGateway

This is an abstract class from which specific LLM gateways are derived.

To create a new gateway, inherit from this class and implement the complete method.

Source code in src/mojentic/llm/gateways/llm_gateway.py

class LLMGateway:
    """
    This is an abstract class from which specific LLM gateways are derived.

    To create a new gateway, inherit from this class and implement the `complete` method.
    """

    def get_model_context_window(self, model: str) -> ModelContextWindow | None:
        """Return advertised metadata, or None when discovery is unsupported.

        Parameters
        ----------
        model : str
            Exact provider model identifier.

        Returns
        -------
        ModelContextWindow | None
            Positive advertised window and source field, or not advertised.

        Raises
        ------
        httpx.TransportError
            The provider could not be reached (HTTP gateways).
        httpx.HTTPStatusError
            The provider returned a non-success status (HTTP gateways).
        ValueError
            The response body or advertised metadata is malformed.
        """
        return None

    def complete(
        self,
        model: str,
        messages: list[LLMMessage],
        object_model: type[BaseModel] | None = None,
        tools: list[LLMTool] | None = None,
        config: Optional["CompletionConfig"] = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
    ) -> LLMGatewayResponse:
        """
        Complete the LLM request.

        Parameters
        ----------
        model : str
            The name of the model to use, as appears in `ollama list`.
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        object_model : Optional[BaseModel]
            The model to use for validating the response.
        tools : Optional[List[LLMTool]]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
            will be included in the response.
        config : Optional[CompletionConfig]
            Configuration object for LLM completion (recommended over individual kwargs).
        temperature : float
            The temperature to use for the response. Unspecified by default. (Deprecated: use config)
        num_ctx : int
            The number of context tokens to use. Unspecified by default. (Deprecated: use config)
        max_tokens : int
            The maximum number of tokens to generate. Unspecified by default. (Deprecated: use config)

        Returns
        -------
        LLMGatewayResponse
            The response from the Ollama service.
        """
        raise NotImplementedError

    def get_available_models(self) -> list[str]:
        """
        Get the list of available models.

        Returns
        -------
        List[str]
            The list of available models.
        """
        raise NotImplementedError

    def calculate_embeddings(self, text: str, model: str | None = None) -> list[float]:
        """
        Calculate embeddings for the given text using the specified model.

        Parameters
        ----------
        text : str
            The text to calculate embeddings for.
        model : str, optional
            The name of the model to use for embeddings. Default value depends on the implementation.

        Returns
        -------
        List[Any]
            The embeddings for the text.
        """
        raise NotImplementedError

calculate_embeddings(text, model=None)

Calculate embeddings for the given text using the specified model.

Parameters:

Name	Type	Description	Default
text	str	The text to calculate embeddings for.	required
model	str	The name of the model to use for embeddings. Default value depends on the implementation.	None

Returns:

Type	Description
List[Any]	The embeddings for the text.

Source code in src/mojentic/llm/gateways/llm_gateway.py

def calculate_embeddings(self, text: str, model: str | None = None) -> list[float]:
    """
    Calculate embeddings for the given text using the specified model.

    Parameters
    ----------
    text : str
        The text to calculate embeddings for.
    model : str, optional
        The name of the model to use for embeddings. Default value depends on the implementation.

    Returns
    -------
    List[Any]
        The embeddings for the text.
    """
    raise NotImplementedError

complete(model, messages, object_model=None, tools=None, config=None, temperature=None, num_ctx=None, max_tokens=None)

Complete the LLM request.

Parameters:

Name	Type	Description	Default
model	str	The name of the model to use, as appears in ollama list.	required
messages	List[LLMMessage]	A list of messages to send to the LLM.	required
object_model	Optional[BaseModel]	The model to use for validating the response.	None
tools	Optional[List[LLMTool]]	A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response.	None
config	Optional[CompletionConfig]	Configuration object for LLM completion (recommended over individual kwargs).	None
temperature	float	The temperature to use for the response. Unspecified by default. (Deprecated: use config)	None
num_ctx	int	The number of context tokens to use. Unspecified by default. (Deprecated: use config)	None
max_tokens	int	The maximum number of tokens to generate. Unspecified by default. (Deprecated: use config)	None

Returns:

Type	Description
LLMGatewayResponse	The response from the Ollama service.

Source code in src/mojentic/llm/gateways/llm_gateway.py

def complete(
    self,
    model: str,
    messages: list[LLMMessage],
    object_model: type[BaseModel] | None = None,
    tools: list[LLMTool] | None = None,
    config: Optional["CompletionConfig"] = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
) -> LLMGatewayResponse:
    """
    Complete the LLM request.

    Parameters
    ----------
    model : str
        The name of the model to use, as appears in `ollama list`.
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    object_model : Optional[BaseModel]
        The model to use for validating the response.
    tools : Optional[List[LLMTool]]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
        will be included in the response.
    config : Optional[CompletionConfig]
        Configuration object for LLM completion (recommended over individual kwargs).
    temperature : float
        The temperature to use for the response. Unspecified by default. (Deprecated: use config)
    num_ctx : int
        The number of context tokens to use. Unspecified by default. (Deprecated: use config)
    max_tokens : int
        The maximum number of tokens to generate. Unspecified by default. (Deprecated: use config)

    Returns
    -------
    LLMGatewayResponse
        The response from the Ollama service.
    """
    raise NotImplementedError

get_available_models()

Get the list of available models.

Returns:

Type	Description
List[str]	The list of available models.

Source code in src/mojentic/llm/gateways/llm_gateway.py

def get_available_models(self) -> list[str]:
    """
    Get the list of available models.

    Returns
    -------
    List[str]
        The list of available models.
    """
    raise NotImplementedError

get_model_context_window(model)

Return advertised metadata, or None when discovery is unsupported.

Parameters:

Name	Type	Description	Default
model	str	Exact provider model identifier.	required

Returns:

Type	Description
ModelContextWindow | None	Positive advertised window and source field, or not advertised.

Raises:

Type	Description
TransportError	The provider could not be reached (HTTP gateways).
HTTPStatusError	The provider returned a non-success status (HTTP gateways).
ValueError	The response body or advertised metadata is malformed.

Source code in src/mojentic/llm/gateways/llm_gateway.py

def get_model_context_window(self, model: str) -> ModelContextWindow | None:
    """Return advertised metadata, or None when discovery is unsupported.

    Parameters
    ----------
    model : str
        Exact provider model identifier.

    Returns
    -------
    ModelContextWindow | None
        Positive advertised window and source field, or not advertised.

    Raises
    ------
    httpx.TransportError
        The provider could not be reached (HTTP gateways).
    httpx.HTTPStatusError
        The provider returned a non-success status (HTTP gateways).
    ValueError
        The response body or advertised metadata is malformed.
    """
    return None

mojentic.llm.gateways.OllamaGateway

Bases: LLMGateway

This class is a gateway to the Ollama LLM service.

Parameters:

Name	Type	Description	Default
host	str	The Ollama host to connect to. Defaults to "http://localhost:11434".	'http://localhost:11434'
headers	dict	The headers to send with the request. Defaults to an empty dict.	None
timeout	optional	The request timeout passed to the Ollama client.	None
recovery_policy	RecoveryPolicy	Opt in to ordinary, structured, and tool-capable streaming request recovery.	None
recovery_call	RecoveryCall	Cancellation for a synchronous broker/session operation. Use a fresh call per operation.	None
stream_transport	OllamaStreamTransport	The transport complete_stream_events uses. Defaults to one over this gateway's client.	None

Source code in src/mojentic/llm/gateways/ollama.py

class OllamaGateway(LLMGateway):
    """
    This class is a gateway to the Ollama LLM service.

    Parameters
    ----------
    host : str, optional
        The Ollama host to connect to. Defaults to "http://localhost:11434".
    headers : dict, optional
        The headers to send with the request. Defaults to an empty dict.
    timeout : optional
        The request timeout passed to the Ollama client.
    recovery_policy : RecoveryPolicy, optional
        Opt in to ordinary, structured, and tool-capable streaming request recovery.
    recovery_call : RecoveryCall, optional
        Cancellation for a synchronous broker/session operation. Use a fresh call per operation.
    stream_transport : OllamaStreamTransport, optional
        The transport ``complete_stream_events`` uses. Defaults to one over this gateway's client.
    """

    def __init__(
        self,
        host="http://localhost:11434",
        headers=None,
        timeout=None,
        stream_transport: OllamaStreamTransport | None = None,
        recovery_policy: RecoveryPolicy | None = None,
        recovery_call: RecoveryCall | None = None,
    ):
        if headers is None:
            headers = {}
        self.recovery_policy = (
            RecoveryPolicy.model_validate(recovery_policy)
            if recovery_policy is not None
            else None
        )
        self.recovery_call = recovery_call
        self._recovery_host = (
            host.rstrip("/") if isinstance(host, str) else "http://localhost:11434"
        )
        self._recovery_headers = dict(headers or {})
        self._recovery_timeout = timeout
        self.client = Client(
            host=host,
            headers=headers,
            timeout=timeout,
            event_hooks={"request": [dispatch_stream]},
        )
        self.stream_transport = stream_transport or OllamaStreamTransport(self.client)

    def _extract_options_from_args(self, args):
        config = args.get("config") or CompletionConfig(
            **{
                field: args[field]
                for field in ("temperature", "num_ctx", "max_tokens", "top_p", "top_k")
                if field in args
            }
        )
        return Options(
            **{
                ("num_predict" if field == "max_tokens" else field): value
                for field in ("temperature", "num_ctx", "max_tokens", "top_p", "top_k")
                if (value := getattr(config, field)) is not None
            }
        )

    def get_model_context_window(self, model: str) -> ModelContextWindow | None:
        """Discover model metadata without changing completion options."""
        with httpx.Client(
            headers=self._recovery_headers, timeout=self._recovery_timeout
        ) as client:
            response = client.post(
                self._recovery_host + "/api/show", json={"model": model}
            )
            response.raise_for_status()
            return ollama_context_window(response.json())

    def complete(self, **args) -> LLMGatewayResponse:
        """
        Complete the LLM request by delegating to the Ollama service.

        Keyword Arguments
        ----------------
        model : str
            The name of the model to use, as appears in `ollama list`.
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        object_model : Optional[BaseModel]
            The model to use for validating the response.
        tools : Optional[List[LLMTool]]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
            will be included in the response.
        temperature : float, optional
            The temperature to use for the response. Unspecified by default.
        num_ctx : int, optional
            The number of context tokens to use. Unspecified by default.
        max_tokens : int, optional
            The maximum number of tokens to generate. Unspecified by default.

        Returns
        -------
        LLMGatewayResponse
            The response from the Ollama service.
        """
        if self.recovery_policy is not None:
            return asyncio.run(self.complete_with_recovery(**args))

        logger.info("Delegating to Ollama for completion", **args)

        ollama_args = self._completion_request(args)

        response: ChatResponse = self.client.chat(**ollama_args)
        config = args.get("config")
        if config and config.retain_reasoning and not response.done:
            raise ValueError(
                "Cannot retain reasoning from an incomplete Ollama response"
            )

        structured_object = None

        if "object_model" in args:
            try:
                structured_object = args["object_model"].model_validate_json(
                    response.message.content
                )
            except ValidationError as e:
                logger.error(
                    "Failed to validate model in",
                    error=str(e),
                    response=response.message.content,
                    object_model=args["object_model"],
                )

        return self._completion_response(response, structured_object)

    def _completion_request(self, args: dict) -> dict:
        options = self._extract_options_from_args(args)

        ollama_args = {
            "model": args["model"],
            "messages": adapt_messages_to_ollama(args["messages"]),
            "options": options,
        }

        config = args.get("config")
        if config:
            ollama_args.update(reasoning_controls(config.reasoning_effort, "ollama"))

        if "object_model" in args and args["object_model"] is not None:
            ollama_args["format"] = args["object_model"].model_json_schema()
        elif (
            response_format := ollama_format(config.response_format if config else None)
        ) is not None:
            ollama_args["format"] = response_format

        if "tools" in args and args["tools"] is not None:
            ollama_args["tools"] = [t.descriptor for t in args["tools"]]

        return ollama_args

    def recovery_capabilities(self) -> Capabilities:
        """Report recovery support without implying remote cancellation or idempotency."""
        return completion_capabilities("ollama")

    async def complete_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> LLMGatewayResponse:
        """Complete one ordinary or structured request with safe recovery metadata.

        Raises
        ------
        RecoveryError
            Provider, protocol, capture, cancellation, or recovery admission failed.
        ValueError
            No recovery policy is configured.
        """
        if self.recovery_policy is None:
            raise RecoveryPolicyError(
                "configure recovery_policy before using complete_with_recovery"
            )
        operation = "structured" if args.get("object_model") is not None else "ordinary"
        try:
            request = self._completion_request(args)
            request["options"] = request["options"].model_dump(exclude_none=True)
            request["stream"] = False
            for message in request["messages"]:
                if message.get("images"):
                    message["images"] = [
                        Image(value=image).model_dump(mode="json")
                        for image in message["images"]
                    ]
            body = json.dumps(
                request, ensure_ascii=False, separators=(",", ":"), allow_nan=False
            ).encode("utf-8")
            headers = self._recovery_headers | {
                "Content-Type": "application/json",
                "Accept-Encoding": "identity",
            }
        except (ValueError, TypeError, KeyError, OSError) as cause:
            raise preparation_error(cause, operation) from None
        result, report = await recover(
            self._recovery_host + "/api/chat",
            body,
            headers,
            self._recovery_timeout,
            lambda frame: self._decode_recovery(frame, args.get("object_model")),
            self.recovery_policy,
            call or self.recovery_call or RecoveryCall(),
            operation,
        )
        result.recovery_report = report
        result.metadata = result.metadata | {"recovery": report.model_dump(mode="json")}
        return result

    def _decode_recovery(
        self, frame: object, object_model: type[BaseModel] | None
    ) -> LLMGatewayResponse:
        if not isinstance(frame, dict) or not isinstance(frame.get("message"), dict):
            raise TypeError("missing response message")
        if frame["message"].get("role") != "assistant":
            raise ValueError("invalid response role")
        if not isinstance(frame["message"].get("content", ""), str):
            raise TypeError("invalid response content")
        if frame.get("done") is not True:
            raise ValueError("incomplete Ollama response")
        response = ChatResponse.model_validate(frame)
        object = (
            object_model.model_validate_json(response.message.content)
            if object_model is not None
            else None
        )
        return self._completion_response(response, object)

    def _completion_response(
        self, response: ChatResponse, object: BaseModel | None
    ) -> LLMGatewayResponse:
        tool_calls = []
        if response.message.tool_calls is not None:
            tool_calls = [
                LLMToolCall(
                    id=getattr(t, "id", None),
                    name=t.function.name,
                    arguments=t.function.arguments,
                )
                for t in response.message.tool_calls
            ]

        # Extract thinking content if present
        thinking = getattr(response.message, "thinking", None)

        frame = response.model_dump()
        return LLMGatewayResponse(
            content=response.message.content,
            object=object,
            tool_calls=tool_calls,
            thinking=thinking,
            usage=ollama_usage(frame),
            model=response.model,
            finish_reason=response.done_reason,
            metadata=ollama_metadata(frame) or {},
        )

    @timed_stream
    def complete_stream(self, **args) -> Iterator[StreamingResponse]:
        """
        Stream the LLM response from Ollama service.

        Keyword Arguments
        ----------------
        model : str
            The name of the model to use, as appears in `ollama list`.
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        tools : Optional[List[LLMTool]]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
            will be included in the response.
        temperature : float, optional
            The temperature to use for the response. Unspecified by default.
        num_ctx : int, optional
            The number of context tokens to use. Unspecified by default.
        max_tokens : int, optional
            The maximum number of tokens to generate. Unspecified by default.

        Returns
        -------
        Iterator[StreamingResponse]
            An iterator of StreamingResponse objects containing response chunks.
        """
        if self.recovery_policy is not None:
            events = synchronous_stream(self.complete_stream_with_recovery(**args))
            with closing(events):
                for event in events:
                    if event.kind == "failed":
                        event.error.stream_metrics = event.metrics
                        self.last_stream_metrics = event.metrics
                        raise event.error
                    if event.kind == "completed":
                        self.last_stream_metrics = event.metrics
                    if event.kind == "content":
                        yield StreamingResponse(content=event.text)
                    elif event.kind == "reasoning":
                        yield StreamingResponse(thinking=event.text)
                    elif event.kind == "completed" and event.response.tool_calls:
                        yield StreamingResponse(
                            tool_calls=event.response.tool_calls, completed=True
                        )
            return

        logger.info("Delegating to Ollama for streaming completion", **args)

        ollama_args = self._stream_request(args)
        ollama_args["stream"] = True

        # Enable tool support if tools are provided
        if "tools" in args and args["tools"] is not None:
            ollama_args["tools"] = [t.descriptor for t in args["tools"]]

        stream = self.client.chat(**ollama_args)

        for chunk in stream:
            observe_stream(chunk.model_dump(exclude_none=True))
            if chunk.message:
                # Yield content chunks as they arrive
                if chunk.message.content:
                    yield StreamingResponse(content=chunk.message.content)

                # Yield thinking chunks when they arrive
                if hasattr(chunk.message, "thinking") and chunk.message.thinking:
                    yield StreamingResponse(thinking=chunk.message.thinking)

                # Yield tool calls when they arrive
                if chunk.message.tool_calls:
                    yield StreamingResponse(tool_calls=chunk.message.tool_calls)
            if (
                args.get("config")
                and args["config"].retain_reasoning
                and chunk.done
                and chunk.done_reason in ("stop", "tool_calls")
            ):
                yield StreamingResponse(completed=True)

    async def complete_stream_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> AsyncIterator[RecoveryStreamEvent]:
        """Stream one tool-capable completion with explicit recovery terminal events.

        Consume ``completed`` or ``failed`` to determine the outcome. Closing this
        iterator closes locally owned HTTP; it does not prove remote termination.
        Raw ``complete_stream_events`` remains a single-turn compatibility API.
        """
        if self.recovery_policy is None:
            raise RecoveryPolicyError(
                "configure recovery_policy before using recovery streams"
            )
        try:
            request = self._stream_request(args)
            request["options"] = request["options"].model_dump(exclude_none=True)
            request["stream"] = True
            if args.get("tools") is not None:
                request["tools"] = [tool.descriptor for tool in args["tools"]]
            for message in request["messages"]:
                if message.get("images"):
                    message["images"] = [
                        Image(value=image).model_dump(mode="json")
                        for image in message["images"]
                    ]
            body = json.dumps(
                request, ensure_ascii=False, separators=(",", ":"), allow_nan=False
            ).encode("utf-8")
        except (ValueError, TypeError, KeyError, OSError) as cause:
            error = preparation_error(cause, "streaming")
            yield RecoveryStreamEvent(
                kind="failed",
                identity=error.report.identity,
                progress=error.report.progress,
                report=error.report,
                error=error,
                outcome="ineligible",
            )
            return
        headers = self._recovery_headers | {
            "Content-Type": "application/json",
            "Accept-Encoding": "identity",
        }
        events = recover_stream(
            self._recovery_host + "/api/chat",
            body,
            headers,
            self._recovery_timeout,
            self._decode_stream_recovery,
            self.recovery_policy,
            call or self.recovery_call or RecoveryCall(),
        )
        try:
            async for event in events:
                yield event
        finally:
            await events.aclose()

    def _decode_stream_recovery(self, frame: dict) -> LLMGatewayResponse:
        """Preserve native streaming tool arguments without ordinary-path conversion."""
        response = self._decode_recovery(frame, None)
        response.tool_calls = [
            LLMToolCall(
                id=call.get("id"),
                name=call["function"]["name"],
                arguments=call["function"]["arguments"],
            )
            for call in frame["message"].get("tool_calls", [])
        ]
        return response

    @timed_stream
    def complete_stream_events(
        self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
    ) -> Iterator[StreamEvent]:
        """
        Stream one turn as events, with terminal completion evidence.

        Sends one streaming request with no tools. The turn completes only when Ollama
        reports ``done: true`` with ``done_reason: "stop"``. Closing the generator closes
        the HTTP request.

        Parameters
        ----------
        model : str
            The name of the model to use, as appears in `ollama list`.
        messages : List[LLMMessage]
            The messages to send.
        config : CompletionConfig
            Configuration for the request.

        Yields
        ------
        StreamEvent
            Content events followed by exactly one terminal event.
        """
        frames = self.stream_transport.stream_frames(
            self._stream_request(
                {"model": model, "messages": messages, "config": config}
            )
        )
        try:
            with closing(frames):
                yield from parse_ollama_stream(frames)
        except ResponseError as e:
            yield StreamError(
                reason=StreamErrorReason.PROVIDER_ERROR,
                detail={"status_code": e.status_code, "error": e.error},
            )
        except (json.JSONDecodeError, ValidationError) as e:
            yield StreamError(
                reason=StreamErrorReason.INVALID_STREAM_EVENT, detail=str(e)
            )
        except (ConnectionError, httpx.HTTPError) as e:
            yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

    def _stream_request(self, args: dict) -> dict:
        """Build the chat request shared by both streaming APIs, without tools or the stream flag."""
        request = {
            "model": args["model"],
            "messages": adapt_messages_to_ollama(args["messages"]),
            "options": self._extract_options_from_args(args),
        }

        config = args.get("config")
        if config:
            request.update(reasoning_controls(config.reasoning_effort, "ollama"))

        response_format = ollama_format(config.response_format if config else None)
        if response_format is not None:
            request["format"] = response_format
        return request

    def get_available_models(self) -> list[str]:
        """
        Get the list of available local models, sorted alphabetically.

        Returns
        -------
        List[str]
            The list of available models, sorted alphabetically.
        """
        return sorted([m.model for m in self.client.list().models])

    def pull_model(self, model: str) -> None:
        """
        Pull the model from the Ollama service.

        Parameters
        ----------
        model : str
            The name of the model to pull.
        """
        self.client.pull(model)

    def calculate_embeddings(
        self, text: str, model: str = "mxbai-embed-large"
    ) -> list[float]:
        """
        Calculate embeddings for the given text using the specified model.

        Parameters
        ----------
        text : str
            The text to calculate embeddings for.
        model : str, optional
            The name of the model to use for embeddings. Defaults to "mxbai-embed-large".

        Returns
        -------
        list
            The embeddings for the text.
        """
        logger.debug("calculate_embeddings", text=text, model=model)
        embed = self.client.embeddings(model=model, prompt=text)
        return embed.embedding

calculate_embeddings(text, model='mxbai-embed-large')

Calculate embeddings for the given text using the specified model.

Parameters:

Name	Type	Description	Default
text	str	The text to calculate embeddings for.	required
model	str	The name of the model to use for embeddings. Defaults to "mxbai-embed-large".	'mxbai-embed-large'

Returns:

Type	Description
list	The embeddings for the text.

Source code in src/mojentic/llm/gateways/ollama.py

def calculate_embeddings(
    self, text: str, model: str = "mxbai-embed-large"
) -> list[float]:
    """
    Calculate embeddings for the given text using the specified model.

    Parameters
    ----------
    text : str
        The text to calculate embeddings for.
    model : str, optional
        The name of the model to use for embeddings. Defaults to "mxbai-embed-large".

    Returns
    -------
    list
        The embeddings for the text.
    """
    logger.debug("calculate_embeddings", text=text, model=model)
    embed = self.client.embeddings(model=model, prompt=text)
    return embed.embedding

complete(**args)

Complete the LLM request by delegating to the Ollama service.

Keyword Arguments

model : str The name of the model to use, as appears in ollama list. messages : List[LLMMessage] A list of messages to send to the LLM. object_model : Optional[BaseModel] The model to use for validating the response. tools : Optional[List[LLMTool]] A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response. temperature : float, optional The temperature to use for the response. Unspecified by default. num_ctx : int, optional The number of context tokens to use. Unspecified by default. max_tokens : int, optional The maximum number of tokens to generate. Unspecified by default.

Returns:

Type	Description
LLMGatewayResponse	The response from the Ollama service.

Source code in src/mojentic/llm/gateways/ollama.py

def complete(self, **args) -> LLMGatewayResponse:
    """
    Complete the LLM request by delegating to the Ollama service.

    Keyword Arguments
    ----------------
    model : str
        The name of the model to use, as appears in `ollama list`.
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    object_model : Optional[BaseModel]
        The model to use for validating the response.
    tools : Optional[List[LLMTool]]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
        will be included in the response.
    temperature : float, optional
        The temperature to use for the response. Unspecified by default.
    num_ctx : int, optional
        The number of context tokens to use. Unspecified by default.
    max_tokens : int, optional
        The maximum number of tokens to generate. Unspecified by default.

    Returns
    -------
    LLMGatewayResponse
        The response from the Ollama service.
    """
    if self.recovery_policy is not None:
        return asyncio.run(self.complete_with_recovery(**args))

    logger.info("Delegating to Ollama for completion", **args)

    ollama_args = self._completion_request(args)

    response: ChatResponse = self.client.chat(**ollama_args)
    config = args.get("config")
    if config and config.retain_reasoning and not response.done:
        raise ValueError(
            "Cannot retain reasoning from an incomplete Ollama response"
        )

    structured_object = None

    if "object_model" in args:
        try:
            structured_object = args["object_model"].model_validate_json(
                response.message.content
            )
        except ValidationError as e:
            logger.error(
                "Failed to validate model in",
                error=str(e),
                response=response.message.content,
                object_model=args["object_model"],
            )

    return self._completion_response(response, structured_object)

complete_stream(**args)

Stream the LLM response from Ollama service.

Keyword Arguments

model : str The name of the model to use, as appears in ollama list. messages : List[LLMMessage] A list of messages to send to the LLM. tools : Optional[List[LLMTool]] A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response. temperature : float, optional The temperature to use for the response. Unspecified by default. num_ctx : int, optional The number of context tokens to use. Unspecified by default. max_tokens : int, optional The maximum number of tokens to generate. Unspecified by default.

Returns:

Type	Description
Iterator[StreamingResponse]	An iterator of StreamingResponse objects containing response chunks.

Source code in src/mojentic/llm/gateways/ollama.py

@timed_stream
def complete_stream(self, **args) -> Iterator[StreamingResponse]:
    """
    Stream the LLM response from Ollama service.

    Keyword Arguments
    ----------------
    model : str
        The name of the model to use, as appears in `ollama list`.
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    tools : Optional[List[LLMTool]]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
        will be included in the response.
    temperature : float, optional
        The temperature to use for the response. Unspecified by default.
    num_ctx : int, optional
        The number of context tokens to use. Unspecified by default.
    max_tokens : int, optional
        The maximum number of tokens to generate. Unspecified by default.

    Returns
    -------
    Iterator[StreamingResponse]
        An iterator of StreamingResponse objects containing response chunks.
    """
    if self.recovery_policy is not None:
        events = synchronous_stream(self.complete_stream_with_recovery(**args))
        with closing(events):
            for event in events:
                if event.kind == "failed":
                    event.error.stream_metrics = event.metrics
                    self.last_stream_metrics = event.metrics
                    raise event.error
                if event.kind == "completed":
                    self.last_stream_metrics = event.metrics
                if event.kind == "content":
                    yield StreamingResponse(content=event.text)
                elif event.kind == "reasoning":
                    yield StreamingResponse(thinking=event.text)
                elif event.kind == "completed" and event.response.tool_calls:
                    yield StreamingResponse(
                        tool_calls=event.response.tool_calls, completed=True
                    )
        return

    logger.info("Delegating to Ollama for streaming completion", **args)

    ollama_args = self._stream_request(args)
    ollama_args["stream"] = True

    # Enable tool support if tools are provided
    if "tools" in args and args["tools"] is not None:
        ollama_args["tools"] = [t.descriptor for t in args["tools"]]

    stream = self.client.chat(**ollama_args)

    for chunk in stream:
        observe_stream(chunk.model_dump(exclude_none=True))
        if chunk.message:
            # Yield content chunks as they arrive
            if chunk.message.content:
                yield StreamingResponse(content=chunk.message.content)

            # Yield thinking chunks when they arrive
            if hasattr(chunk.message, "thinking") and chunk.message.thinking:
                yield StreamingResponse(thinking=chunk.message.thinking)

            # Yield tool calls when they arrive
            if chunk.message.tool_calls:
                yield StreamingResponse(tool_calls=chunk.message.tool_calls)
        if (
            args.get("config")
            and args["config"].retain_reasoning
            and chunk.done
            and chunk.done_reason in ("stop", "tool_calls")
        ):
            yield StreamingResponse(completed=True)

complete_stream_events(model, messages, config)

Stream one turn as events, with terminal completion evidence.

Sends one streaming request with no tools. The turn completes only when Ollama reports done: true with done_reason: "stop". Closing the generator closes the HTTP request.

Parameters:

Name	Type	Description	Default
model	str	The name of the model to use, as appears in ollama list.	required
messages	List[LLMMessage]	The messages to send.	required
config	CompletionConfig	Configuration for the request.	required

Yields:

Type	Description
StreamEvent	Content events followed by exactly one terminal event.

Source code in src/mojentic/llm/gateways/ollama.py

@timed_stream
def complete_stream_events(
    self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
) -> Iterator[StreamEvent]:
    """
    Stream one turn as events, with terminal completion evidence.

    Sends one streaming request with no tools. The turn completes only when Ollama
    reports ``done: true`` with ``done_reason: "stop"``. Closing the generator closes
    the HTTP request.

    Parameters
    ----------
    model : str
        The name of the model to use, as appears in `ollama list`.
    messages : List[LLMMessage]
        The messages to send.
    config : CompletionConfig
        Configuration for the request.

    Yields
    ------
    StreamEvent
        Content events followed by exactly one terminal event.
    """
    frames = self.stream_transport.stream_frames(
        self._stream_request(
            {"model": model, "messages": messages, "config": config}
        )
    )
    try:
        with closing(frames):
            yield from parse_ollama_stream(frames)
    except ResponseError as e:
        yield StreamError(
            reason=StreamErrorReason.PROVIDER_ERROR,
            detail={"status_code": e.status_code, "error": e.error},
        )
    except (json.JSONDecodeError, ValidationError) as e:
        yield StreamError(
            reason=StreamErrorReason.INVALID_STREAM_EVENT, detail=str(e)
        )
    except (ConnectionError, httpx.HTTPError) as e:
        yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

complete_stream_with_recovery(call=None, **args) async

Stream one tool-capable completion with explicit recovery terminal events.

Consume completed or failed to determine the outcome. Closing this iterator closes locally owned HTTP; it does not prove remote termination. Raw complete_stream_events remains a single-turn compatibility API.

Source code in src/mojentic/llm/gateways/ollama.py

async def complete_stream_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> AsyncIterator[RecoveryStreamEvent]:
    """Stream one tool-capable completion with explicit recovery terminal events.

    Consume ``completed`` or ``failed`` to determine the outcome. Closing this
    iterator closes locally owned HTTP; it does not prove remote termination.
    Raw ``complete_stream_events`` remains a single-turn compatibility API.
    """
    if self.recovery_policy is None:
        raise RecoveryPolicyError(
            "configure recovery_policy before using recovery streams"
        )
    try:
        request = self._stream_request(args)
        request["options"] = request["options"].model_dump(exclude_none=True)
        request["stream"] = True
        if args.get("tools") is not None:
            request["tools"] = [tool.descriptor for tool in args["tools"]]
        for message in request["messages"]:
            if message.get("images"):
                message["images"] = [
                    Image(value=image).model_dump(mode="json")
                    for image in message["images"]
                ]
        body = json.dumps(
            request, ensure_ascii=False, separators=(",", ":"), allow_nan=False
        ).encode("utf-8")
    except (ValueError, TypeError, KeyError, OSError) as cause:
        error = preparation_error(cause, "streaming")
        yield RecoveryStreamEvent(
            kind="failed",
            identity=error.report.identity,
            progress=error.report.progress,
            report=error.report,
            error=error,
            outcome="ineligible",
        )
        return
    headers = self._recovery_headers | {
        "Content-Type": "application/json",
        "Accept-Encoding": "identity",
    }
    events = recover_stream(
        self._recovery_host + "/api/chat",
        body,
        headers,
        self._recovery_timeout,
        self._decode_stream_recovery,
        self.recovery_policy,
        call or self.recovery_call or RecoveryCall(),
    )
    try:
        async for event in events:
            yield event
    finally:
        await events.aclose()

complete_with_recovery(call=None, **args) async

Complete one ordinary or structured request with safe recovery metadata.

Raises:

Type	Description
RecoveryError	Provider, protocol, capture, cancellation, or recovery admission failed.
ValueError	No recovery policy is configured.

Source code in src/mojentic/llm/gateways/ollama.py

async def complete_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> LLMGatewayResponse:
    """Complete one ordinary or structured request with safe recovery metadata.

    Raises
    ------
    RecoveryError
        Provider, protocol, capture, cancellation, or recovery admission failed.
    ValueError
        No recovery policy is configured.
    """
    if self.recovery_policy is None:
        raise RecoveryPolicyError(
            "configure recovery_policy before using complete_with_recovery"
        )
    operation = "structured" if args.get("object_model") is not None else "ordinary"
    try:
        request = self._completion_request(args)
        request["options"] = request["options"].model_dump(exclude_none=True)
        request["stream"] = False
        for message in request["messages"]:
            if message.get("images"):
                message["images"] = [
                    Image(value=image).model_dump(mode="json")
                    for image in message["images"]
                ]
        body = json.dumps(
            request, ensure_ascii=False, separators=(",", ":"), allow_nan=False
        ).encode("utf-8")
        headers = self._recovery_headers | {
            "Content-Type": "application/json",
            "Accept-Encoding": "identity",
        }
    except (ValueError, TypeError, KeyError, OSError) as cause:
        raise preparation_error(cause, operation) from None
    result, report = await recover(
        self._recovery_host + "/api/chat",
        body,
        headers,
        self._recovery_timeout,
        lambda frame: self._decode_recovery(frame, args.get("object_model")),
        self.recovery_policy,
        call or self.recovery_call or RecoveryCall(),
        operation,
    )
    result.recovery_report = report
    result.metadata = result.metadata | {"recovery": report.model_dump(mode="json")}
    return result

get_available_models()

Get the list of available local models, sorted alphabetically.

Returns:

Type	Description
List[str]	The list of available models, sorted alphabetically.

Source code in src/mojentic/llm/gateways/ollama.py

def get_available_models(self) -> list[str]:
    """
    Get the list of available local models, sorted alphabetically.

    Returns
    -------
    List[str]
        The list of available models, sorted alphabetically.
    """
    return sorted([m.model for m in self.client.list().models])

get_model_context_window(model)

Discover model metadata without changing completion options.

Source code in src/mojentic/llm/gateways/ollama.py

def get_model_context_window(self, model: str) -> ModelContextWindow | None:
    """Discover model metadata without changing completion options."""
    with httpx.Client(
        headers=self._recovery_headers, timeout=self._recovery_timeout
    ) as client:
        response = client.post(
            self._recovery_host + "/api/show", json={"model": model}
        )
        response.raise_for_status()
        return ollama_context_window(response.json())

pull_model(model)

Pull the model from the Ollama service.

Parameters:

Name	Type	Description	Default
model	str	The name of the model to pull.	required

Source code in src/mojentic/llm/gateways/ollama.py

def pull_model(self, model: str) -> None:
    """
    Pull the model from the Ollama service.

    Parameters
    ----------
    model : str
        The name of the model to pull.
    """
    self.client.pull(model)

recovery_capabilities()

Report recovery support without implying remote cancellation or idempotency.

Source code in src/mojentic/llm/gateways/ollama.py

def recovery_capabilities(self) -> Capabilities:
    """Report recovery support without implying remote cancellation or idempotency."""
    return completion_capabilities("ollama")

mojentic.llm.gateways.OpenAIGateway

Bases: LLMGateway

This class is a gateway to the OpenAI LLM service.

Parameters:

Name	Type	Description	Default
api_key	str	The OpenAI API key to use. If not provided, defaults to the value of the OPENAI_API_KEY environment variable.	None
base_url	str	The base URL for the OpenAI API. If not provided, defaults to the value of the OPENAI_API_ENDPOINT environment variable, or None if not set.	None
stream_transport	OpenAIStreamTransport	The transport complete_stream_events uses. Defaults to one over this gateway's client.	None
recovery_policy	RecoveryPolicy	Opt in for ordinary, structured and tool-capable streaming completion. Owned HTTP bypasses SDK retries and redirects and has no active read timeout.	None
recovery_call	RecoveryCall	Cancellation for a synchronous operation. Use a fresh call per operation.	None

Source code in src/mojentic/llm/gateways/openai.py

class OpenAIGateway(LLMGateway):
    """
    This class is a gateway to the OpenAI LLM service.

    Parameters
    ----------
    api_key : str, optional
        The OpenAI API key to use. If not provided, defaults to the value of the
        OPENAI_API_KEY environment variable.
    base_url : str, optional
        The base URL for the OpenAI API. If not provided, defaults to the value of the
        OPENAI_API_ENDPOINT environment variable, or None if not set.
    stream_transport : OpenAIStreamTransport, optional
        The transport ``complete_stream_events`` uses. Defaults to one over this gateway's client.
    recovery_policy : RecoveryPolicy, optional
        Opt in for ordinary, structured and tool-capable streaming completion.
        Owned HTTP bypasses SDK retries and redirects and has no active read timeout.
    recovery_call : RecoveryCall, optional
        Cancellation for a synchronous operation. Use a fresh call per operation.
    """

    def __init__(
        self,
        api_key: str | None = None,
        base_url: str | None = None,
        stream_transport: OpenAIStreamTransport | None = None,
        recovery_policy: RecoveryPolicy | None = None,
        recovery_call: RecoveryCall | None = None,
    ):
        if api_key is None:
            api_key = os.getenv("OPENAI_API_KEY")
        if base_url is None:
            base_url = os.getenv("OPENAI_API_ENDPOINT")
        self.client = OpenAI(
            api_key=api_key,
            base_url=base_url,
            http_client=DefaultHttpxClient(event_hooks={"request": [dispatch_stream]}),
        )
        self.recovery_policy = (
            RecoveryPolicy.model_validate(recovery_policy)
            if recovery_policy is not None
            else None
        )
        self.recovery_call = recovery_call
        self.model_registry = get_model_registry()
        self.stream_transport = stream_transport or OpenAIStreamTransport(self.client)

    _recovery_provider = "openai"

    @property
    def _recovery_url(self) -> str:
        return str(self.client.base_url).rstrip("/") + "/chat/completions"

    def _recovery_headers(self) -> dict[str, str]:
        headers = {
            "Authorization": f"Bearer {self.client.api_key}",
            "Content-Type": "application/json",
            "Accept-Encoding": "identity",
        }
        if self.client.organization:
            headers["OpenAI-Organization"] = self.client.organization
        if self.client.project:
            headers["OpenAI-Project"] = self.client.project
        return headers

    def recovery_capabilities(self) -> Capabilities:
        """Report completion recovery; remote inference termination remains unknown."""
        return completion_capabilities("openai")

    def _validate_recovery_args(self, args: dict, streaming: bool) -> None:
        model = args["model"]
        if not model or not args.get("messages"):
            raise ValueError("model and messages are required")
        if streaming and args.get("object_model") is not None:
            raise NotImplementedError("structured streaming is unsupported")
        if streaming:
            capabilities = self.model_registry.get_model_capabilities(model)
            if not capabilities.supports_streaming:
                raise NotImplementedError("model does not support streaming")

    @staticmethod
    def _recovery_format(args: dict) -> dict | None:
        object_model = args.get("object_model")
        if object_model is not None:
            schema = pydantic_function_tool(object_model)["function"]["parameters"]
            return {
                "type": "json_schema",
                "json_schema": {
                    "name": object_model.__name__,
                    "strict": True,
                    "schema": schema,
                },
            }
        config = args.get("config")
        return openai_response_format(config.response_format if config else None)

    @staticmethod
    def _validate_structured_tools(tools: list[dict]) -> None:
        for descriptor in tools:
            if descriptor.get("type") != "function":
                raise ValueError("structured parsing requires function tools")
            function = descriptor.get("function")
            if not isinstance(function, dict):
                raise TypeError("invalid function tool")
            if function.get("strict") is not True:
                raise ValueError("structured parsing requires strict function tools")

    def _recovery_body(self, args: dict, streaming: bool) -> bytes:
        self._validate_recovery_args(args, streaming)
        config = args.get("config")
        validate_openai_reasoning(config, args["messages"])
        controls = {
            "model": args["model"],
            "tools": args.get("tools"),
            "temperature": config.temperature if config else args.get("temperature"),
            "max_tokens": config.max_tokens if config else args.get("max_tokens"),
            "reasoning_effort": config.reasoning_effort if config else None,
            "top_p": config.top_p if config else args.get("top_p"),
        }
        adapted = self._adapt_parameters_for_model(args["model"], controls)
        self._validate_model_parameters(args["model"], adapted)
        payload = {
            "model": args["model"],
            "messages": adapt_messages_to_openai(args["messages"]),
        }
        payload |= {
            key: adapted[key]
            for key in (
                "temperature",
                "max_tokens",
                "max_completion_tokens",
                "top_p",
                "reasoning_effort",
            )
            if key in adapted and adapted[key] is not None
        }
        if (response_format := self._recovery_format(args)) is not None:
            payload["response_format"] = response_format
        if adapted.get("tools") is not None:
            payload["tools"] = [tool.descriptor for tool in adapted["tools"]]
        if args.get("object_model") is not None:
            self._validate_structured_tools(payload.get("tools", []))
        if streaming:
            payload |= {"stream": True, "stream_options": {"include_usage": True}}
        return cloud_recovery.freeze_request(payload)

    @staticmethod
    def _decode_recovery(
        frame: object, object_model: type[BaseModel] | None = None
    ) -> LLMGatewayResponse:
        from mojentic.llm.gateways.omlx_recovery import decode_completion

        if object_model is not None and isinstance(frame, dict):
            choices = frame.get("choices")
            choice = choices[0] if isinstance(choices, list) and choices else {}
            if isinstance(choice, dict) and choice.get("finish_reason") in {
                "length",
                "content_filter",
            }:
                if choice.get("finish_reason") == "length":
                    from mojentic.llm.recovery import LengthTerminationError

                    raise LengthTerminationError("completion length limit reached")
                raise ValueError("structured completion did not finish safely")
        result = decode_completion(frame)
        if object_model is not None and not (
            result.content is None
            and result.tool_calls
            and result.finish_reason == "tool_calls"
        ):
            result.object = object_model.model_validate_json(result.content)
        return result

    async def complete_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> LLMGatewayResponse:
        """Recover an ordinary or structured turn with private typed failure evidence.

        Parameters
        ----------
        call : RecoveryCall, optional
            Cancellation for this operation, overriding the gateway call.
        **args
            Model, messages, tools, schema and controls accepted by ``complete``.

        Returns
        -------
        LLMGatewayResponse
            Provider response with a recovery report; tools are returned for the broker.

        Raises
        ------
        RecoveryError
            Completion failed, recovery was refused, or cancellation was requested.
        ValueError
            No recovery policy is configured.
        """
        return await cloud_recovery.complete(self, call, args)

    async def complete_stream_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> AsyncIterator[RecoveryStreamEvent]:
        """Recover before semantic observation; closing cancels owned HTTP resources.

        Parameters
        ----------
        call : RecoveryCall, optional
            Cancellation for this operation, overriding the gateway call.
        **args
            Model, messages, tools and controls accepted by ``complete_stream``.
            Structured streaming is unsupported and produces a failed terminal event.

        Yields
        ------
        RecoveryStreamEvent
            Progress and provider telemetry followed by one completed or failed outcome.
            Private semantic fields require explicit inspection for serialization.

        Raises
        ------
        ValueError
            No recovery policy is configured.
        """
        events = cloud_recovery.stream(self, call, args)
        try:
            async for event in events:
                yield event
        finally:
            await events.aclose()

    def _is_reasoning_model(self, model: str) -> bool:
        """
        Determine if a model is a reasoning model that requires max_completion_tokens.

        Parameters
        ----------
        model : str
            The model name to classify.

        Returns
        -------
        bool
            True if the model is a reasoning model, False if it's a chat model.
        """
        return self.model_registry.is_reasoning_model(model)

    def _adapt_parameters_for_model(self, model: str, args: dict) -> dict:
        """
        Adapt parameters based on the model type and capabilities.

        Parameters
        ----------
        model : str
            The model name.
        args : dict
            The original arguments.

        Returns
        -------
        dict
            The adapted arguments with correct parameter names for the model type.
        """
        adapted_args = {
            key: value
            for key, value in args.items()
            if value is not None or key not in {"temperature", "max_tokens", "top_p"}
        }
        capabilities = self.model_registry.get_model_capabilities(model)

        logger.debug(
            "Adapting parameters for model",
            model=model,
            model_type=capabilities.model_type.value,
            supports_tools=capabilities.supports_tools,
            supports_streaming=capabilities.supports_streaming,
        )

        # Handle token limit parameter conversion
        if "max_tokens" in adapted_args:
            token_param = capabilities.get_token_limit_param()
            if token_param != "max_tokens":
                # Convert max_tokens to max_completion_tokens for reasoning models
                adapted_args[token_param] = adapted_args.pop("max_tokens")
                logger.info(
                    "Converted token limit parameter for model",
                    model=model,
                    from_param="max_tokens",
                    to_param=token_param,
                    value=adapted_args[token_param],
                )

        # Validate tool usage for models that don't support tools
        if (
            "tools" in adapted_args
            and adapted_args["tools"]
            and not capabilities.supports_tools
        ):
            logger.warning(
                "Model does not support tools, removing tool configuration",
                model=model,
                num_tools=len(adapted_args["tools"]),
            )
            adapted_args["tools"] = None  # Set to None instead of removing the key

        # Handle temperature restrictions for specific models
        if "temperature" in adapted_args:
            temperature = adapted_args["temperature"]

            # Check if model supports temperature parameter at all
            if capabilities.supported_temperatures == []:
                # Model doesn't support temperature parameter at all - remove it
                logger.warning(
                    "Model does not support temperature parameter at all",
                    model=model,
                    requested_temperature=temperature,
                )
                adapted_args.pop("temperature", None)
            elif not capabilities.supports_temperature(temperature):
                # Model supports temperature but not this specific value - use default
                default_temp = capabilities.supported_temperatures[0]
                logger.warning(
                    "Model does not support requested temperature, using default",
                    model=model,
                    requested_temperature=temperature,
                    default_temperature=default_temp,
                    supported_temperatures=capabilities.supported_temperatures,
                )
                adapted_args["temperature"] = default_temp

        # Handle reasoning_effort for reasoning models
        if (
            "reasoning_effort" in adapted_args
            and adapted_args["reasoning_effort"] is not None
        ):
            if capabilities.model_type == ModelType.REASONING:
                # Keep reasoning_effort for reasoning models
                logger.info(
                    "Adding reasoning_effort parameter for reasoning model",
                    model=model,
                    reasoning_effort=adapted_args["reasoning_effort"],
                )
            else:
                # Warn and remove for non-reasoning models
                logger.warning(
                    "Model does not support reasoning_effort, ignoring parameter",
                    model=model,
                    requested_reasoning_effort=adapted_args["reasoning_effort"],
                )
                adapted_args.pop("reasoning_effort", None)

        return adapted_args

    def _validate_model_parameters(self, model: str, args: dict) -> None:
        """
        Validate that the parameters are compatible with the model.

        Parameters
        ----------
        model : str
            The model name.
        args : dict
            The arguments to validate.
        """
        capabilities = self.model_registry.get_model_capabilities(model)

        # Warning for tools on reasoning models that don't support them
        if (
            capabilities.model_type == ModelType.REASONING
            and not capabilities.supports_tools
            and "tools" in args
            and args["tools"]
        ):
            logger.warning(
                "Reasoning model may not support tools",
                model=model,
                num_tools=len(args["tools"]),
            )

        # Validate token limits (check both possible parameter names)
        token_value = args.get("max_tokens") or args.get("max_completion_tokens")
        if (
            token_value
            and capabilities.max_output_tokens
            and token_value > capabilities.max_output_tokens
        ):
            logger.warning(
                "Requested token limit exceeds model maximum",
                model=model,
                requested=token_value,
                max_allowed=capabilities.max_output_tokens,
            )

    def get_model_context_window(self, model: str) -> ModelContextWindow | None:
        """Return exact registry metadata without a network request or pattern fallback."""
        if model not in self.model_registry.get_registered_models():
            return None
        value = self.model_registry.get_model_capabilities(model).max_context_tokens
        return (
            ModelContextWindow(max_context_tokens=value, field="registry")
            if value is not None
            else None
        )

    def complete(self, **kwargs) -> LLMGatewayResponse:
        """
        Complete the LLM request by delegating to the OpenAI service.

        Keyword Arguments
        ----------------
        model : str
            The name of the model to use.
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        object_model : Optional[Type[BaseModel]]
            The model to use for validating the response.
        tools : Optional[List[LLMTool]]
            A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
            will be included in the response.
        temperature : float, optional
            The temperature to use for the response. Unspecified by default.
        num_ctx : int, optional
            The number of context tokens to use. Unspecified by default.
        max_tokens : int, optional
            The maximum number of tokens to generate. Unspecified by default.

        Returns
        -------
        LLMGatewayResponse
            The response from the OpenAI service.
        """
        validate_openai_reasoning(kwargs.get("config"), kwargs.get("messages"))
        if self.recovery_policy is not None:
            return asyncio.run(self.complete_with_recovery(**kwargs))
        # Extract parameters from kwargs with defaults
        model = kwargs.get("model")
        messages = kwargs.get("messages")
        object_model = kwargs.get("object_model", None)
        tools = kwargs.get("tools", None)
        config = kwargs.get("config", None)

        # Use config if provided, otherwise use individual kwargs
        if config:
            temperature = config.temperature
            num_ctx = config.num_ctx
            max_tokens = config.max_tokens
            reasoning_effort = config.reasoning_effort
        else:
            temperature = kwargs.get("temperature")
            num_ctx = kwargs.get("num_ctx")
            max_tokens = kwargs.get("max_tokens")
            reasoning_effort = None

        if not model:
            raise ValueError("'model' parameter is required")
        if not messages:
            raise ValueError("'messages' parameter is required")

        # Convert parameters to dict for processing
        args = {
            "model": model,
            "messages": messages,
            "object_model": object_model,
            "tools": tools,
            "temperature": temperature,
            "num_ctx": num_ctx,
            "max_tokens": max_tokens,
            "reasoning_effort": reasoning_effort,
        }

        # Adapt parameters based on model type
        try:
            adapted_args = self._adapt_parameters_for_model(model, args)
        except ValidationError as e:
            logger.error(
                "Failed to adapt parameters for model", model=model, error=str(e)
            )
            raise

        # Validate parameters after adaptation
        self._validate_model_parameters(model, adapted_args)

        openai_args = {
            "model": adapted_args["model"],
            "messages": adapt_messages_to_openai(adapted_args["messages"]),
        }

        if config is not None and config.top_p is not None:
            openai_args["top_p"] = config.top_p

        # Add temperature if specified
        if "temperature" in adapted_args:
            openai_args["temperature"] = adapted_args["temperature"]

        completion = self.client.chat.completions.create

        if adapted_args["object_model"] is not None:
            completion = self.client.beta.chat.completions.parse
            openai_args["response_format"] = adapted_args["object_model"]
        elif (
            response_format := openai_response_format(
                config.response_format if config else None
            )
        ) is not None:
            openai_args["response_format"] = response_format

        if adapted_args.get("tools") is not None:
            openai_args["tools"] = [t.descriptor for t in adapted_args["tools"]]

        # Handle both max_tokens (for chat models) and max_completion_tokens (for reasoning models)
        if "max_tokens" in adapted_args:
            openai_args["max_tokens"] = adapted_args["max_tokens"]
        elif "max_completion_tokens" in adapted_args:
            openai_args["max_completion_tokens"] = adapted_args["max_completion_tokens"]

        # Add reasoning_effort if present in adapted args
        if (
            "reasoning_effort" in adapted_args
            and adapted_args["reasoning_effort"] is not None
        ):
            openai_args["reasoning_effort"] = adapted_args["reasoning_effort"]

        logger.debug(
            "Making OpenAI API call",
            model=openai_args["model"],
            has_tools="tools" in openai_args,
            has_object_model="response_format" in openai_args,
            has_reasoning_effort="reasoning_effort" in openai_args,
            token_param="max_completion_tokens"
            if "max_completion_tokens" in openai_args
            else "max_tokens",
        )

        try:
            response = completion(**openai_args)
        except BadRequestError as e:
            # Enhanced error handling for parameter issues
            if "max_tokens" in str(e) and "max_completion_tokens" in str(e):
                logger.error(
                    "Parameter error detected - model may require different token parameter",
                    model=model,
                    error=str(e),
                    suggestion="This model may be a reasoning model requiring max_completion_tokens",
                )
            raise
        except ValidationError as e:
            logger.error("OpenAI API call failed", model=model, error=str(e))
            raise

        object = None
        tool_calls: list[LLMToolCall] = []

        if adapted_args.get("object_model") is not None:
            try:
                response_content = response.choices[0].message.content
                if response_content is not None:
                    object = adapted_args["object_model"].model_validate_json(
                        response_content
                    )
                else:
                    logger.error(
                        "No response content available for object validation",
                        object_model=adapted_args["object_model"],
                    )
            except ValidationError as e:
                response_content = (
                    response.choices[0].message.content
                    if response.choices
                    else "No response content"
                )
                logger.error(
                    "Failed to validate model",
                    error=str(e),
                    response=response_content,
                    object_model=adapted_args["object_model"],
                )

        if response.choices[0].message.tool_calls is not None:
            for t in response.choices[0].message.tool_calls:
                arguments = json.loads(t.function.arguments)
                tool_call = LLMToolCall(
                    id=t.id, name=t.function.name, arguments=arguments
                )
                tool_calls.append(tool_call)

        return LLMGatewayResponse(
            content=response.choices[0].message.content,
            object=object,
            tool_calls=tool_calls,
            usage=response.usage.model_dump() if response.usage is not None else None,
            model=response.model,
            finish_reason=response.choices[0].finish_reason,
        )

    @timed_stream
    def complete_stream(self, **kwargs) -> Iterator[StreamingResponse]:
        """
        Stream the LLM response from OpenAI service.

        OpenAI streams tool call arguments incrementally, so we need to accumulate them
        and yield complete tool calls only when the stream finishes.

        Keyword Arguments
        ----------------
        model : str
            The name of the model to use.
        messages : List[LLMMessage]
            A list of messages to send to the LLM.
        tools : Optional[List[LLMTool]]
            A list of tools to use with the LLM. Tool calls will be accumulated and yielded when complete.
        temperature : float, optional
            The temperature to use for the response. Unspecified by default.
        num_ctx : int, optional
            The number of context tokens to use. Unspecified by default.
        max_tokens : int, optional
            The maximum number of tokens to generate. Unspecified by default.

        Returns
        -------
        Iterator[StreamingResponse]
            An iterator of StreamingResponse objects containing response chunks.
        """
        validate_openai_reasoning(kwargs.get("config"), kwargs.get("messages"))
        if self.recovery_policy is not None:
            yield from cloud_recovery.legacy_chunks(self, kwargs)
            return
        # Extract parameters from kwargs with defaults
        model = kwargs.get("model")
        messages = kwargs.get("messages")
        object_model = kwargs.get("object_model", None)
        tools = kwargs.get("tools", None)
        config = kwargs.get("config", None)
        validate_openai_reasoning(config, messages)

        # Use config if provided, otherwise use individual kwargs
        if config:
            temperature = config.temperature
            num_ctx = config.num_ctx
            max_tokens = config.max_tokens
            reasoning_effort = config.reasoning_effort
        else:
            temperature = kwargs.get("temperature")
            num_ctx = kwargs.get("num_ctx")
            max_tokens = kwargs.get("max_tokens")
            reasoning_effort = None

        if not model:
            raise ValueError("'model' parameter is required")
        if not messages:
            raise ValueError("'messages' parameter is required")

        # Convert parameters to dict for processing
        args = {
            "model": model,
            "messages": messages,
            "object_model": object_model,
            "tools": tools,
            "temperature": temperature,
            "num_ctx": num_ctx,
            "max_tokens": max_tokens,
            "reasoning_effort": reasoning_effort,
        }

        # Adapt parameters based on model type
        try:
            adapted_args = self._adapt_parameters_for_model(model, args)
        except ValidationError as e:
            logger.error(
                "Failed to adapt parameters for model", model=model, error=str(e)
            )
            raise

        # Validate parameters after adaptation
        self._validate_model_parameters(model, adapted_args)

        # Check if model supports streaming
        capabilities = self.model_registry.get_model_capabilities(model)
        if not capabilities.supports_streaming:
            raise NotImplementedError(f"Model {model} does not support streaming")

        # Structured output doesn't work with streaming
        if adapted_args["object_model"] is not None:
            raise NotImplementedError(
                "Streaming with structured output (object_model) is not supported"
            )

        openai_args = self._stream_body(adapted_args, config)

        logger.debug(
            "Making OpenAI streaming API call",
            model=openai_args["model"],
            has_tools="tools" in openai_args,
            has_reasoning_effort="reasoning_effort" in openai_args,
            token_param="max_completion_tokens"
            if "max_completion_tokens" in openai_args
            else "max_tokens",
        )

        try:
            stream = self.client.chat.completions.create(**openai_args)
        except BadRequestError as e:
            if "max_tokens" in str(e) and "max_completion_tokens" in str(e):
                logger.error(
                    "Parameter error detected - model may require different token parameter",
                    model=model,
                    error=str(e),
                    suggestion="This model may be a reasoning model requiring max_completion_tokens",
                )
            raise
        except ValidationError as e:
            logger.error("OpenAI streaming API call failed", model=model, error=str(e))
            raise

        # Accumulate tool calls as they stream in
        # OpenAI streams tool arguments incrementally, indexed by tool call index
        tool_calls_accumulator: dict[int, dict] = {}

        for chunk in stream:
            observe_stream(chunk.model_dump(exclude_none=True))
            if not chunk.choices:
                continue

            delta = chunk.choices[0].delta
            finish_reason = chunk.choices[0].finish_reason

            # Yield content chunks as they arrive
            if delta.content:
                yield StreamingResponse(content=delta.content)

            # Accumulate tool call chunks
            if delta.tool_calls:
                for tool_call_delta in delta.tool_calls:
                    index = tool_call_delta.index

                    # Initialize accumulator for this tool call if needed
                    if index not in tool_calls_accumulator:
                        tool_calls_accumulator[index] = {
                            "id": None,
                            "name": None,
                            "arguments": "",
                        }

                    # First chunk has id and name
                    if tool_call_delta.id:
                        tool_calls_accumulator[index]["id"] = tool_call_delta.id

                    if tool_call_delta.function.name:
                        tool_calls_accumulator[index]["name"] = (
                            tool_call_delta.function.name
                        )

                    # All chunks may have argument fragments
                    if tool_call_delta.function.arguments:
                        tool_calls_accumulator[index]["arguments"] += (
                            tool_call_delta.function.arguments
                        )

            # When stream is complete, yield accumulated tool calls
            if finish_reason == "tool_calls" and tool_calls_accumulator:
                # Parse and yield complete tool calls
                complete_tool_calls = []
                for index in sorted(tool_calls_accumulator.keys()):
                    tc = tool_calls_accumulator[index]
                    try:
                        # Parse the accumulated JSON arguments
                        arguments = json.loads(tc["arguments"])

                        tool_call = LLMToolCall(
                            id=tc["id"], name=tc["name"], arguments=arguments
                        )
                        complete_tool_calls.append(tool_call)
                    except json.JSONDecodeError as e:
                        logger.error(
                            "Failed to parse tool call arguments",
                            tool_name=tc["name"],
                            arguments=tc["arguments"],
                            error=str(e),
                        )

                if complete_tool_calls:
                    # Convert to the format expected by ollama's tool calls for compatibility
                    # We need to create mock objects that match ollama's structure
                    from types import SimpleNamespace

                    ollama_format_calls = []
                    for tc in complete_tool_calls:
                        ollama_format_calls.append(
                            SimpleNamespace(
                                id=tc.id,  # Include ID for proper OpenAI message formatting
                                function=SimpleNamespace(
                                    name=tc.name, arguments=tc.arguments
                                ),
                            )
                        )
                    yield StreamingResponse(tool_calls=ollama_format_calls)

    def _stream_body(self, adapted_args: dict, config) -> dict:
        """Build the chat completions body shared by both streaming APIs."""
        openai_args = {
            "model": adapted_args["model"],
            "messages": adapt_messages_to_openai(adapted_args["messages"]),
            "stream": True,
        }

        response_format = openai_response_format(
            config.response_format if config else None
        )
        if response_format is not None:
            openai_args["response_format"] = response_format

        if config is not None and config.top_p is not None:
            openai_args["top_p"] = config.top_p

        # Add temperature if specified
        if "temperature" in adapted_args:
            openai_args["temperature"] = adapted_args["temperature"]

        if adapted_args.get("tools") is not None:
            openai_args["tools"] = [t.descriptor for t in adapted_args["tools"]]

        # Handle both max_tokens (for chat models) and max_completion_tokens (for reasoning models)
        if "max_tokens" in adapted_args:
            openai_args["max_tokens"] = adapted_args["max_tokens"]
        elif "max_completion_tokens" in adapted_args:
            openai_args["max_completion_tokens"] = adapted_args["max_completion_tokens"]

        # Add reasoning_effort if present in adapted args
        if (
            "reasoning_effort" in adapted_args
            and adapted_args["reasoning_effort"] is not None
        ):
            openai_args["reasoning_effort"] = adapted_args["reasoning_effort"]
        return openai_args

    @timed_stream
    def complete_stream_events(
        self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
    ) -> Iterator[StreamEvent]:
        """
        Stream one turn as events, with terminal completion evidence.

        Sends one streaming request with no tools, no retries and
        ``stream_options: {"include_usage": true}``. The turn completes only when the
        provider reports ``finish_reason: "stop"`` and then sends ``data: [DONE]``.
        Closing the generator closes the HTTP request.

        Parameters
        ----------
        model : str
            The name of the model to use.
        messages : List[LLMMessage]
            The messages to send.
        config : CompletionConfig
            Configuration for the request.

        Yields
        ------
        StreamEvent
            Content events followed by exactly one terminal event.
        """
        if not self.model_registry.get_model_capabilities(model).supports_streaming:
            yield StreamError(
                reason=StreamErrorReason.STREAM_EVENTS_UNSUPPORTED,
                detail=f"Model {model} does not support streaming",
            )
            return
        body = self._stream_body(
            self._adapted_stream_args(model, messages, config), config
        )
        body["stream_options"] = {"include_usage": True}
        lines = self.stream_transport.stream_lines(body)
        try:
            with closing(lines):
                yield from parse_openai_stream(lines)
        except APIStatusError as e:
            yield StreamError(
                reason=StreamErrorReason.PROVIDER_ERROR,
                detail={"status_code": e.status_code, "error": e.body},
            )
        except (APIConnectionError, httpx.HTTPError) as e:
            yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

    def _adapted_stream_args(
        self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
    ) -> dict:
        validate_openai_reasoning(config, messages)
        args = {
            "model": model,
            "messages": messages,
            "object_model": None,
            "tools": None,
            "temperature": config.temperature,
            "num_ctx": config.num_ctx,
            "max_tokens": config.max_tokens,
            "reasoning_effort": config.reasoning_effort,
        }
        adapted_args = self._adapt_parameters_for_model(model, args)
        self._validate_model_parameters(model, adapted_args)
        return adapted_args

    def get_available_models(self) -> list[str]:
        """
        Get the list of available OpenAI models, sorted alphabetically.

        Returns
        -------
        list[str]
            The list of available models, sorted alphabetically.
        """
        return sorted([m.id for m in self.client.models.list()])

    def calculate_embeddings(
        self, text: str, model: str = "text-embedding-3-large"
    ) -> list[float]:
        """
        Calculate embeddings for the given text using the specified OpenAI model.

        Parameters
        ----------
        text : str
            The text to calculate embeddings for.
        model : str, optional
            The name of the OpenAI embeddings model to use. Defaults to "text-embedding-3-large".

        Returns
        -------
        list
            The mean of the part embeddings, weighted by each part's token count and normalized to length 1.
        """
        logger.debug("calculate_embeddings", text=text, model=model)

        chunks = list(self._chunked_tokens(text, 8191))
        embeddings = [
            self.client.embeddings.create(model=model, input=chunk).data[0].embedding
            for chunk in chunks
        ]
        lengths = [len(chunk) for chunk in chunks]

        average = np.average(embeddings, axis=0, weights=lengths)
        average = average / np.linalg.norm(average)
        average = average.tolist()

        return average

    def _batched(self, iterable: Iterable, n: int):
        """Batch data into tuples of length n. The last batch may be shorter."""
        # batched('ABCDEFG', 3) --> ABC DEF G
        if n < 1:
            raise ValueError("n must be at least one")
        it = iter(iterable)
        while batch := tuple(islice(it, n)):
            yield batch

    def _chunked_tokens(self, text, chunk_length):
        tokenizer = TokenizerGateway()
        tokens = tokenizer.encode(text)
        chunks_iterator = self._batched(tokens, chunk_length)
        yield from chunks_iterator

calculate_embeddings(text, model='text-embedding-3-large')

Calculate embeddings for the given text using the specified OpenAI model.

Parameters:

Name	Type	Description	Default
text	str	The text to calculate embeddings for.	required
model	str	The name of the OpenAI embeddings model to use. Defaults to "text-embedding-3-large".	'text-embedding-3-large'

Returns:

Type	Description
list	The mean of the part embeddings, weighted by each part's token count and normalized to length 1.

Source code in src/mojentic/llm/gateways/openai.py

def calculate_embeddings(
    self, text: str, model: str = "text-embedding-3-large"
) -> list[float]:
    """
    Calculate embeddings for the given text using the specified OpenAI model.

    Parameters
    ----------
    text : str
        The text to calculate embeddings for.
    model : str, optional
        The name of the OpenAI embeddings model to use. Defaults to "text-embedding-3-large".

    Returns
    -------
    list
        The mean of the part embeddings, weighted by each part's token count and normalized to length 1.
    """
    logger.debug("calculate_embeddings", text=text, model=model)

    chunks = list(self._chunked_tokens(text, 8191))
    embeddings = [
        self.client.embeddings.create(model=model, input=chunk).data[0].embedding
        for chunk in chunks
    ]
    lengths = [len(chunk) for chunk in chunks]

    average = np.average(embeddings, axis=0, weights=lengths)
    average = average / np.linalg.norm(average)
    average = average.tolist()

    return average

complete(**kwargs)

Complete the LLM request by delegating to the OpenAI service.

Keyword Arguments

model : str The name of the model to use. messages : List[LLMMessage] A list of messages to send to the LLM. object_model : Optional[Type[BaseModel]] The model to use for validating the response. tools : Optional[List[LLMTool]] A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output will be included in the response. temperature : float, optional The temperature to use for the response. Unspecified by default. num_ctx : int, optional The number of context tokens to use. Unspecified by default. max_tokens : int, optional The maximum number of tokens to generate. Unspecified by default.

Returns:

Type	Description
LLMGatewayResponse	The response from the OpenAI service.

Source code in src/mojentic/llm/gateways/openai.py

def complete(self, **kwargs) -> LLMGatewayResponse:
    """
    Complete the LLM request by delegating to the OpenAI service.

    Keyword Arguments
    ----------------
    model : str
        The name of the model to use.
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    object_model : Optional[Type[BaseModel]]
        The model to use for validating the response.
    tools : Optional[List[LLMTool]]
        A list of tools to use with the LLM. If a tool call is requested, the tool will be called and the output
        will be included in the response.
    temperature : float, optional
        The temperature to use for the response. Unspecified by default.
    num_ctx : int, optional
        The number of context tokens to use. Unspecified by default.
    max_tokens : int, optional
        The maximum number of tokens to generate. Unspecified by default.

    Returns
    -------
    LLMGatewayResponse
        The response from the OpenAI service.
    """
    validate_openai_reasoning(kwargs.get("config"), kwargs.get("messages"))
    if self.recovery_policy is not None:
        return asyncio.run(self.complete_with_recovery(**kwargs))
    # Extract parameters from kwargs with defaults
    model = kwargs.get("model")
    messages = kwargs.get("messages")
    object_model = kwargs.get("object_model", None)
    tools = kwargs.get("tools", None)
    config = kwargs.get("config", None)

    # Use config if provided, otherwise use individual kwargs
    if config:
        temperature = config.temperature
        num_ctx = config.num_ctx
        max_tokens = config.max_tokens
        reasoning_effort = config.reasoning_effort
    else:
        temperature = kwargs.get("temperature")
        num_ctx = kwargs.get("num_ctx")
        max_tokens = kwargs.get("max_tokens")
        reasoning_effort = None

    if not model:
        raise ValueError("'model' parameter is required")
    if not messages:
        raise ValueError("'messages' parameter is required")

    # Convert parameters to dict for processing
    args = {
        "model": model,
        "messages": messages,
        "object_model": object_model,
        "tools": tools,
        "temperature": temperature,
        "num_ctx": num_ctx,
        "max_tokens": max_tokens,
        "reasoning_effort": reasoning_effort,
    }

    # Adapt parameters based on model type
    try:
        adapted_args = self._adapt_parameters_for_model(model, args)
    except ValidationError as e:
        logger.error(
            "Failed to adapt parameters for model", model=model, error=str(e)
        )
        raise

    # Validate parameters after adaptation
    self._validate_model_parameters(model, adapted_args)

    openai_args = {
        "model": adapted_args["model"],
        "messages": adapt_messages_to_openai(adapted_args["messages"]),
    }

    if config is not None and config.top_p is not None:
        openai_args["top_p"] = config.top_p

    # Add temperature if specified
    if "temperature" in adapted_args:
        openai_args["temperature"] = adapted_args["temperature"]

    completion = self.client.chat.completions.create

    if adapted_args["object_model"] is not None:
        completion = self.client.beta.chat.completions.parse
        openai_args["response_format"] = adapted_args["object_model"]
    elif (
        response_format := openai_response_format(
            config.response_format if config else None
        )
    ) is not None:
        openai_args["response_format"] = response_format

    if adapted_args.get("tools") is not None:
        openai_args["tools"] = [t.descriptor for t in adapted_args["tools"]]

    # Handle both max_tokens (for chat models) and max_completion_tokens (for reasoning models)
    if "max_tokens" in adapted_args:
        openai_args["max_tokens"] = adapted_args["max_tokens"]
    elif "max_completion_tokens" in adapted_args:
        openai_args["max_completion_tokens"] = adapted_args["max_completion_tokens"]

    # Add reasoning_effort if present in adapted args
    if (
        "reasoning_effort" in adapted_args
        and adapted_args["reasoning_effort"] is not None
    ):
        openai_args["reasoning_effort"] = adapted_args["reasoning_effort"]

    logger.debug(
        "Making OpenAI API call",
        model=openai_args["model"],
        has_tools="tools" in openai_args,
        has_object_model="response_format" in openai_args,
        has_reasoning_effort="reasoning_effort" in openai_args,
        token_param="max_completion_tokens"
        if "max_completion_tokens" in openai_args
        else "max_tokens",
    )

    try:
        response = completion(**openai_args)
    except BadRequestError as e:
        # Enhanced error handling for parameter issues
        if "max_tokens" in str(e) and "max_completion_tokens" in str(e):
            logger.error(
                "Parameter error detected - model may require different token parameter",
                model=model,
                error=str(e),
                suggestion="This model may be a reasoning model requiring max_completion_tokens",
            )
        raise
    except ValidationError as e:
        logger.error("OpenAI API call failed", model=model, error=str(e))
        raise

    object = None
    tool_calls: list[LLMToolCall] = []

    if adapted_args.get("object_model") is not None:
        try:
            response_content = response.choices[0].message.content
            if response_content is not None:
                object = adapted_args["object_model"].model_validate_json(
                    response_content
                )
            else:
                logger.error(
                    "No response content available for object validation",
                    object_model=adapted_args["object_model"],
                )
        except ValidationError as e:
            response_content = (
                response.choices[0].message.content
                if response.choices
                else "No response content"
            )
            logger.error(
                "Failed to validate model",
                error=str(e),
                response=response_content,
                object_model=adapted_args["object_model"],
            )

    if response.choices[0].message.tool_calls is not None:
        for t in response.choices[0].message.tool_calls:
            arguments = json.loads(t.function.arguments)
            tool_call = LLMToolCall(
                id=t.id, name=t.function.name, arguments=arguments
            )
            tool_calls.append(tool_call)

    return LLMGatewayResponse(
        content=response.choices[0].message.content,
        object=object,
        tool_calls=tool_calls,
        usage=response.usage.model_dump() if response.usage is not None else None,
        model=response.model,
        finish_reason=response.choices[0].finish_reason,
    )

complete_stream(**kwargs)

Stream the LLM response from OpenAI service.

OpenAI streams tool call arguments incrementally, so we need to accumulate them and yield complete tool calls only when the stream finishes.

Keyword Arguments

model : str The name of the model to use. messages : List[LLMMessage] A list of messages to send to the LLM. tools : Optional[List[LLMTool]] A list of tools to use with the LLM. Tool calls will be accumulated and yielded when complete. temperature : float, optional The temperature to use for the response. Unspecified by default. num_ctx : int, optional The number of context tokens to use. Unspecified by default. max_tokens : int, optional The maximum number of tokens to generate. Unspecified by default.

Returns:

Type	Description
Iterator[StreamingResponse]	An iterator of StreamingResponse objects containing response chunks.

Source code in src/mojentic/llm/gateways/openai.py

@timed_stream
def complete_stream(self, **kwargs) -> Iterator[StreamingResponse]:
    """
    Stream the LLM response from OpenAI service.

    OpenAI streams tool call arguments incrementally, so we need to accumulate them
    and yield complete tool calls only when the stream finishes.

    Keyword Arguments
    ----------------
    model : str
        The name of the model to use.
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    tools : Optional[List[LLMTool]]
        A list of tools to use with the LLM. Tool calls will be accumulated and yielded when complete.
    temperature : float, optional
        The temperature to use for the response. Unspecified by default.
    num_ctx : int, optional
        The number of context tokens to use. Unspecified by default.
    max_tokens : int, optional
        The maximum number of tokens to generate. Unspecified by default.

    Returns
    -------
    Iterator[StreamingResponse]
        An iterator of StreamingResponse objects containing response chunks.
    """
    validate_openai_reasoning(kwargs.get("config"), kwargs.get("messages"))
    if self.recovery_policy is not None:
        yield from cloud_recovery.legacy_chunks(self, kwargs)
        return
    # Extract parameters from kwargs with defaults
    model = kwargs.get("model")
    messages = kwargs.get("messages")
    object_model = kwargs.get("object_model", None)
    tools = kwargs.get("tools", None)
    config = kwargs.get("config", None)
    validate_openai_reasoning(config, messages)

    # Use config if provided, otherwise use individual kwargs
    if config:
        temperature = config.temperature
        num_ctx = config.num_ctx
        max_tokens = config.max_tokens
        reasoning_effort = config.reasoning_effort
    else:
        temperature = kwargs.get("temperature")
        num_ctx = kwargs.get("num_ctx")
        max_tokens = kwargs.get("max_tokens")
        reasoning_effort = None

    if not model:
        raise ValueError("'model' parameter is required")
    if not messages:
        raise ValueError("'messages' parameter is required")

    # Convert parameters to dict for processing
    args = {
        "model": model,
        "messages": messages,
        "object_model": object_model,
        "tools": tools,
        "temperature": temperature,
        "num_ctx": num_ctx,
        "max_tokens": max_tokens,
        "reasoning_effort": reasoning_effort,
    }

    # Adapt parameters based on model type
    try:
        adapted_args = self._adapt_parameters_for_model(model, args)
    except ValidationError as e:
        logger.error(
            "Failed to adapt parameters for model", model=model, error=str(e)
        )
        raise

    # Validate parameters after adaptation
    self._validate_model_parameters(model, adapted_args)

    # Check if model supports streaming
    capabilities = self.model_registry.get_model_capabilities(model)
    if not capabilities.supports_streaming:
        raise NotImplementedError(f"Model {model} does not support streaming")

    # Structured output doesn't work with streaming
    if adapted_args["object_model"] is not None:
        raise NotImplementedError(
            "Streaming with structured output (object_model) is not supported"
        )

    openai_args = self._stream_body(adapted_args, config)

    logger.debug(
        "Making OpenAI streaming API call",
        model=openai_args["model"],
        has_tools="tools" in openai_args,
        has_reasoning_effort="reasoning_effort" in openai_args,
        token_param="max_completion_tokens"
        if "max_completion_tokens" in openai_args
        else "max_tokens",
    )

    try:
        stream = self.client.chat.completions.create(**openai_args)
    except BadRequestError as e:
        if "max_tokens" in str(e) and "max_completion_tokens" in str(e):
            logger.error(
                "Parameter error detected - model may require different token parameter",
                model=model,
                error=str(e),
                suggestion="This model may be a reasoning model requiring max_completion_tokens",
            )
        raise
    except ValidationError as e:
        logger.error("OpenAI streaming API call failed", model=model, error=str(e))
        raise

    # Accumulate tool calls as they stream in
    # OpenAI streams tool arguments incrementally, indexed by tool call index
    tool_calls_accumulator: dict[int, dict] = {}

    for chunk in stream:
        observe_stream(chunk.model_dump(exclude_none=True))
        if not chunk.choices:
            continue

        delta = chunk.choices[0].delta
        finish_reason = chunk.choices[0].finish_reason

        # Yield content chunks as they arrive
        if delta.content:
            yield StreamingResponse(content=delta.content)

        # Accumulate tool call chunks
        if delta.tool_calls:
            for tool_call_delta in delta.tool_calls:
                index = tool_call_delta.index

                # Initialize accumulator for this tool call if needed
                if index not in tool_calls_accumulator:
                    tool_calls_accumulator[index] = {
                        "id": None,
                        "name": None,
                        "arguments": "",
                    }

                # First chunk has id and name
                if tool_call_delta.id:
                    tool_calls_accumulator[index]["id"] = tool_call_delta.id

                if tool_call_delta.function.name:
                    tool_calls_accumulator[index]["name"] = (
                        tool_call_delta.function.name
                    )

                # All chunks may have argument fragments
                if tool_call_delta.function.arguments:
                    tool_calls_accumulator[index]["arguments"] += (
                        tool_call_delta.function.arguments
                    )

        # When stream is complete, yield accumulated tool calls
        if finish_reason == "tool_calls" and tool_calls_accumulator:
            # Parse and yield complete tool calls
            complete_tool_calls = []
            for index in sorted(tool_calls_accumulator.keys()):
                tc = tool_calls_accumulator[index]
                try:
                    # Parse the accumulated JSON arguments
                    arguments = json.loads(tc["arguments"])

                    tool_call = LLMToolCall(
                        id=tc["id"], name=tc["name"], arguments=arguments
                    )
                    complete_tool_calls.append(tool_call)
                except json.JSONDecodeError as e:
                    logger.error(
                        "Failed to parse tool call arguments",
                        tool_name=tc["name"],
                        arguments=tc["arguments"],
                        error=str(e),
                    )

            if complete_tool_calls:
                # Convert to the format expected by ollama's tool calls for compatibility
                # We need to create mock objects that match ollama's structure
                from types import SimpleNamespace

                ollama_format_calls = []
                for tc in complete_tool_calls:
                    ollama_format_calls.append(
                        SimpleNamespace(
                            id=tc.id,  # Include ID for proper OpenAI message formatting
                            function=SimpleNamespace(
                                name=tc.name, arguments=tc.arguments
                            ),
                        )
                    )
                yield StreamingResponse(tool_calls=ollama_format_calls)

complete_stream_events(model, messages, config)

Stream one turn as events, with terminal completion evidence.

Sends one streaming request with no tools, no retries and stream_options: {"include_usage": true}. The turn completes only when the provider reports finish_reason: "stop" and then sends data: [DONE]. Closing the generator closes the HTTP request.

Parameters:

Name	Type	Description	Default
model	str	The name of the model to use.	required
messages	List[LLMMessage]	The messages to send.	required
config	CompletionConfig	Configuration for the request.	required

Yields:

Type	Description
StreamEvent	Content events followed by exactly one terminal event.

Source code in src/mojentic/llm/gateways/openai.py

@timed_stream
def complete_stream_events(
    self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
) -> Iterator[StreamEvent]:
    """
    Stream one turn as events, with terminal completion evidence.

    Sends one streaming request with no tools, no retries and
    ``stream_options: {"include_usage": true}``. The turn completes only when the
    provider reports ``finish_reason: "stop"`` and then sends ``data: [DONE]``.
    Closing the generator closes the HTTP request.

    Parameters
    ----------
    model : str
        The name of the model to use.
    messages : List[LLMMessage]
        The messages to send.
    config : CompletionConfig
        Configuration for the request.

    Yields
    ------
    StreamEvent
        Content events followed by exactly one terminal event.
    """
    if not self.model_registry.get_model_capabilities(model).supports_streaming:
        yield StreamError(
            reason=StreamErrorReason.STREAM_EVENTS_UNSUPPORTED,
            detail=f"Model {model} does not support streaming",
        )
        return
    body = self._stream_body(
        self._adapted_stream_args(model, messages, config), config
    )
    body["stream_options"] = {"include_usage": True}
    lines = self.stream_transport.stream_lines(body)
    try:
        with closing(lines):
            yield from parse_openai_stream(lines)
    except APIStatusError as e:
        yield StreamError(
            reason=StreamErrorReason.PROVIDER_ERROR,
            detail={"status_code": e.status_code, "error": e.body},
        )
    except (APIConnectionError, httpx.HTTPError) as e:
        yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

complete_stream_with_recovery(call=None, **args) async

Recover before semantic observation; closing cancels owned HTTP resources.

Parameters:

Name	Type	Description	Default
call	RecoveryCall	Cancellation for this operation, overriding the gateway call.	None
**args		Model, messages, tools and controls accepted by complete_stream. Structured streaming is unsupported and produces a failed terminal event.	{}

Yields:

Type	Description
RecoveryStreamEvent	Progress and provider telemetry followed by one completed or failed outcome. Private semantic fields require explicit inspection for serialization.

Raises:

Type	Description
ValueError	No recovery policy is configured.

Source code in src/mojentic/llm/gateways/openai.py

async def complete_stream_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> AsyncIterator[RecoveryStreamEvent]:
    """Recover before semantic observation; closing cancels owned HTTP resources.

    Parameters
    ----------
    call : RecoveryCall, optional
        Cancellation for this operation, overriding the gateway call.
    **args
        Model, messages, tools and controls accepted by ``complete_stream``.
        Structured streaming is unsupported and produces a failed terminal event.

    Yields
    ------
    RecoveryStreamEvent
        Progress and provider telemetry followed by one completed or failed outcome.
        Private semantic fields require explicit inspection for serialization.

    Raises
    ------
    ValueError
        No recovery policy is configured.
    """
    events = cloud_recovery.stream(self, call, args)
    try:
        async for event in events:
            yield event
    finally:
        await events.aclose()

complete_with_recovery(call=None, **args) async

Recover an ordinary or structured turn with private typed failure evidence.

Parameters:

Name	Type	Description	Default
call	RecoveryCall	Cancellation for this operation, overriding the gateway call.	None
**args		Model, messages, tools, schema and controls accepted by complete.	{}

Returns:

Type	Description
LLMGatewayResponse	Provider response with a recovery report; tools are returned for the broker.

Raises:

Type	Description
RecoveryError	Completion failed, recovery was refused, or cancellation was requested.
ValueError	No recovery policy is configured.

Source code in src/mojentic/llm/gateways/openai.py

async def complete_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> LLMGatewayResponse:
    """Recover an ordinary or structured turn with private typed failure evidence.

    Parameters
    ----------
    call : RecoveryCall, optional
        Cancellation for this operation, overriding the gateway call.
    **args
        Model, messages, tools, schema and controls accepted by ``complete``.

    Returns
    -------
    LLMGatewayResponse
        Provider response with a recovery report; tools are returned for the broker.

    Raises
    ------
    RecoveryError
        Completion failed, recovery was refused, or cancellation was requested.
    ValueError
        No recovery policy is configured.
    """
    return await cloud_recovery.complete(self, call, args)

get_available_models()

Get the list of available OpenAI models, sorted alphabetically.

Returns:

Type	Description
list[str]	The list of available models, sorted alphabetically.

Source code in src/mojentic/llm/gateways/openai.py

def get_available_models(self) -> list[str]:
    """
    Get the list of available OpenAI models, sorted alphabetically.

    Returns
    -------
    list[str]
        The list of available models, sorted alphabetically.
    """
    return sorted([m.id for m in self.client.models.list()])

get_model_context_window(model)

Return exact registry metadata without a network request or pattern fallback.

Source code in src/mojentic/llm/gateways/openai.py

def get_model_context_window(self, model: str) -> ModelContextWindow | None:
    """Return exact registry metadata without a network request or pattern fallback."""
    if model not in self.model_registry.get_registered_models():
        return None
    value = self.model_registry.get_model_capabilities(model).max_context_tokens
    return (
        ModelContextWindow(max_context_tokens=value, field="registry")
        if value is not None
        else None
    )

recovery_capabilities()

Report completion recovery; remote inference termination remains unknown.

Source code in src/mojentic/llm/gateways/openai.py

def recovery_capabilities(self) -> Capabilities:
    """Report completion recovery; remote inference termination remains unknown."""
    return completion_capabilities("openai")

mojentic.llm.gateways.OMLXGateway

Bases: LLMGateway

A gateway to an oMLX <https://github.com/jundot/omlx>_ server, an LLM server for Apple Silicon.

oMLX speaks the OpenAI chat completions protocol. This gateway sends requests with no per-model adaptation, maps reasoning_content to thinking, and keeps usage exactly as the server reports it.

Parameters:

Name	Type	Description	Default
host	str	The server host, without /v1. Defaults to OMLX_HOST, then http://localhost:8000. The gateway adds /v1 to every path.	None
api_key	str	Sent as Authorization: Bearer <key>. Defaults to OMLX_API_KEY. Without a key, no authorization header is sent.	None
timeout	float	The timeout in seconds for every request, including model load. Defaults to OMLX_TIMEOUT (in milliseconds), then 600 seconds.	None
transport	OMLXTransport	Transport for legacy completions, embeddings and model management. Opt-in completion recovery uses a dedicated owned HTTP transport.	None
recovery_policy	RecoveryPolicy	Opt in to ordinary, structured and streaming request recovery. Recovery disables the active response read timeout; other timeout settings remain.	None
recovery_call	RecoveryCall	Cancellation for a synchronous operation. Use a fresh call per operation.	None

Notes

Legacy non-2xx responses raise httpx.HTTPStatusError. Recovery-enabled completion paths raise typed RecoveryError with private cause inspection. Raw complete_stream_events remains single-request and reports provider errors.

Source code in src/mojentic/llm/gateways/omlx.py

class OMLXGateway(LLMGateway):
    """
    A gateway to an `oMLX <https://github.com/jundot/omlx>`_ server, an LLM server for Apple Silicon.

    oMLX speaks the OpenAI chat completions protocol. This gateway sends requests with no
    per-model adaptation, maps ``reasoning_content`` to ``thinking``, and keeps usage exactly
    as the server reports it.

    Parameters
    ----------
    host : str, optional
        The server host, without ``/v1``. Defaults to ``OMLX_HOST``, then
        ``http://localhost:8000``. The gateway adds ``/v1`` to every path.
    api_key : str, optional
        Sent as ``Authorization: Bearer <key>``. Defaults to ``OMLX_API_KEY``. Without a key,
        no authorization header is sent.
    timeout : float, optional
        The timeout in seconds for every request, including model load. Defaults to
        ``OMLX_TIMEOUT`` (in milliseconds), then 600 seconds.
    transport : OMLXTransport, optional
        Transport for legacy completions, embeddings and model management. Opt-in
        completion recovery uses a dedicated owned HTTP transport.
    recovery_policy : RecoveryPolicy, optional
        Opt in to ordinary, structured and streaming request recovery. Recovery
        disables the active response read timeout; other timeout settings remain.
    recovery_call : RecoveryCall, optional
        Cancellation for a synchronous operation. Use a fresh call per operation.

    Notes
    -----
    Legacy non-2xx responses raise ``httpx.HTTPStatusError``. Recovery-enabled
    completion paths raise typed ``RecoveryError`` with private cause inspection.
    Raw ``complete_stream_events`` remains single-request and reports provider errors.
    """

    def __init__(
        self,
        host: str | None = None,
        api_key: str | None = None,
        timeout: float | None = None,
        transport: OMLXTransport | None = None,
        recovery_policy: RecoveryPolicy | None = None,
        recovery_call: RecoveryCall | None = None,
    ) -> None:
        self.settings = omlx_settings(
            host=host, api_key=api_key, timeout=timeout, environ=os.environ
        )
        self.transport = transport or OMLXTransport(self.settings)
        self.recovery_policy = (
            RecoveryPolicy.model_validate(recovery_policy)
            if recovery_policy is not None
            else None
        )
        self.recovery_call = recovery_call

    def get_model_context_window(self, model: str) -> ModelContextWindow | None:
        """Discover exact model metadata using the configured host and credentials."""
        return omlx_context_window(self.transport.get("models").json(), model)

    def complete(
        self,
        model: str,
        messages: list[LLMMessage],
        object_model: type[BaseModel] | None = None,
        tools: list[LLMTool] | None = None,
        config: Optional["CompletionConfig"] = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
    ) -> LLMGatewayResponse:
        """
        Complete one chat turn.

        When a structured output request comes back with a ``Warning`` header, oMLX did not
        enforce the format. The header value goes in ``metadata["response_format_warning"]``
        and a warning is logged. The content is still returned and, with ``object_model``,
        still validated.

        Parameters
        ----------
        model : str
            The model id, as ``get_available_models`` reports it.
        messages : List[LLMMessage]
            The conversation.
        object_model : Optional[Type[BaseModel]]
            Request JSON schema output with this model's schema and validate the content against it.
        tools : Optional[List[LLMTool]]
            Tools to offer. Requested calls come back in ``tool_calls``.
        config : Optional[CompletionConfig]
            Temperature, token limit, reasoning effort and response format. When None, one is
            built from ``temperature`` and ``max_tokens``.
        temperature, num_ctx, max_tokens
            Deprecated: use ``config``. ``num_ctx`` is never sent.

        Returns
        -------
        LLMGatewayResponse
            Content, thinking, tool calls, usage, provider model, finish reason and metadata.
            ``content`` is not an answer unless ``finish_reason`` is ``stop``.
        """
        if self.recovery_policy is not None:
            return asyncio.run(
                self.complete_with_recovery(
                    model=model,
                    messages=messages,
                    config=config,
                    tools=tools,
                    object_model=object_model,
                    temperature=temperature,
                    num_ctx=num_ctx,
                    max_tokens=max_tokens,
                )
            )
        config = config or _config_from_arguments(temperature, num_ctx, max_tokens)
        body = omlx_chat_body(
            model, messages, config, tools=tools, object_model=object_model
        )
        response = self.transport.post("/chat/completions", body)
        result = omlx_gateway_response(
            response.json(), self._response_format_warning(model, body, response)
        )
        if object_model is not None:
            result.object = _validated_object(object_model, result.content)
        return result

    def recovery_capabilities(self) -> Capabilities:
        """Report supported request recovery and unsupported remote facilities."""
        return completion_capabilities("omlx")

    def _recovery_body(self, args: dict, streaming: bool) -> bytes:
        config = args.get("config") or _config_from_arguments(
            args.get("temperature"),
            args.get("num_ctx"),
            args.get("max_tokens"),
        )
        body = omlx_chat_body(
            args["model"],
            args["messages"],
            config,
            tools=args.get("tools"),
            object_model=args.get("object_model"),
        )
        if streaming:
            if args.get("object_model") is not None:
                raise ValueError("structured streaming is unsupported")
            body |= {"stream": True, "stream_options": {"include_usage": True}}
        return json.dumps(
            body, ensure_ascii=False, separators=(",", ":"), allow_nan=False
        ).encode("utf-8")

    def _recovery_headers(self) -> dict[str, str]:
        return self.settings.headers | {
            "Content-Type": "application/json",
            "Accept-Encoding": "identity",
        }

    async def complete_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> LLMGatewayResponse:
        """Recover one ordinary or structured completion, retaining typed failure evidence.

        Recovery bypasses the legacy transport and disables active read timeouts.

        Parameters
        ----------
        call : RecoveryCall, optional
            Cancellation for this operation, overriding the gateway call.
        **args
            The model, messages, tools, schema and controls accepted by ``complete``.

        Returns
        -------
        LLMGatewayResponse
            Provider response with a typed recovery report.

        Raises
        ------
        RecoveryError
            Recovery refused, interrupted, cancelled or received an invalid response.
        ValueError
            No recovery policy is configured.
        """
        from mojentic.llm.gateways.omlx_recovery import OMLXRecovery, decode_completion

        if self.recovery_policy is None:
            raise RecoveryPolicyError("configure recovery_policy before using recovery")
        operation = "structured" if args.get("object_model") is not None else "ordinary"
        try:
            body = self._recovery_body(args, False)
        except (ValueError, TypeError, KeyError, OSError) as cause:
            raise preparation_error(cause, operation, "omlx") from None
        engine = OMLXRecovery(
            self.recovery_policy,
            call or self.recovery_call or RecoveryCall(),
            operation,
        )
        result, report = await engine.run(
            self.settings.base_url + "/chat/completions",
            body,
            self._recovery_headers(),
            httpx.Timeout(self.settings.timeout, read=None),
            lambda frame: decode_completion(frame, args.get("object_model")),
        )
        result.recovery_report = report
        result.metadata |= {"recovery": report.model_dump(mode="json")}
        return result

    async def complete_stream_with_recovery(
        self, call: RecoveryCall | None = None, **args
    ) -> AsyncIterator[RecoveryStreamEvent]:
        """Yield progress and exactly one completed or failed recovery outcome.

        Closing the iterator cancels owned local HTTP; remote termination is unknown.
        Raw complete_stream_events remains a single-request compatibility API.

        Parameters
        ----------
        call : RecoveryCall, optional
            Cancellation for this operation, overriding the gateway call.
        **args
            The model, messages, tools and controls accepted by ``complete_stream``.

        Yields
        ------
        RecoveryStreamEvent
            Provider progress followed by one completed or failed terminal outcome.

        Raises
        ------
        ValueError
            No recovery policy is configured.
        """
        from mojentic.llm.gateways.omlx_recovery import (
            OMLXStreamRecovery,
            decode_completion,
        )

        if self.recovery_policy is None:
            raise RecoveryPolicyError("configure recovery_policy before using recovery")
        try:
            body = self._recovery_body(args, True)
        except (ValueError, TypeError, KeyError, OSError) as cause:
            error = preparation_error(cause, "streaming", "omlx")
            yield RecoveryStreamEvent(
                kind="failed",
                identity=error.report.identity,
                progress=error.report.progress,
                report=error.report,
                error=error,
                outcome="ineligible",
            )
            return
        events = recover_stream(
            self.settings.base_url + "/chat/completions",
            body,
            self._recovery_headers(),
            httpx.Timeout(self.settings.timeout, read=None),
            decode_completion,
            self.recovery_policy,
            call or self.recovery_call or RecoveryCall(),
            _engine_type=OMLXStreamRecovery,
        )
        try:
            async for event in events:
                yield event
        finally:
            await events.aclose()

    @staticmethod
    def _response_format_warning(
        model: str, body: dict, response: httpx.Response
    ) -> str | None:
        warnings = response.headers.get_list("warning")
        if not warnings or not requests_structured_output(body):
            return None
        warning = ", ".join(warnings)
        logger.warning(
            "oMLX did not enforce the requested response format",
            model=model,
            response_format_warning=warning,
        )
        return warning

    @timed_stream
    def complete_stream(
        self,
        model: str,
        messages: list[LLMMessage],
        object_model: type[BaseModel] | None = None,
        tools: list[LLMTool] | None = None,
        config: Optional["CompletionConfig"] = None,
        temperature: float | None = None,
        num_ctx: int | None = None,
        max_tokens: int | None = None,
    ) -> Iterator[StreamingResponse]:
        """
        Stream one chat turn as legacy chunks.

        Content arrives as content chunks and ``reasoning_content`` as thinking chunks. Tool
        calls are yielded together, complete, when the turn finishes with ``tool_calls``.
        Keep-alive frames are dropped. Closing the generator closes the request.

        Parameters are as for :meth:`complete`, except that ``object_model`` is not supported.

        Raises
        ------
        NotImplementedError
            When ``object_model`` is given.
        """
        if object_model is not None:
            raise NotImplementedError(
                "Streaming with structured output (object_model) is not supported"
            )
        if self.recovery_policy is not None:
            events = synchronous_stream(
                self.complete_stream_with_recovery(
                    model=model,
                    messages=messages,
                    config=config,
                    tools=tools,
                    temperature=temperature,
                    num_ctx=num_ctx,
                    max_tokens=max_tokens,
                )
            )
            with closing(events):
                for event in events:
                    if event.kind == "failed":
                        event.error.stream_metrics = event.metrics
                        self.last_stream_metrics = event.metrics
                        raise event.error
                    if event.kind == "completed":
                        self.last_stream_metrics = event.metrics
                    if event.kind == "content":
                        yield StreamingResponse(content=event.text)
                    elif event.kind == "reasoning":
                        yield StreamingResponse(thinking=event.text)
                    elif event.kind == "completed" and event.response.tool_calls:
                        yield StreamingResponse(
                            tool_calls=event.response.tool_calls, completed=True
                        )
            return
        config = config or _config_from_arguments(temperature, num_ctx, max_tokens)
        body = omlx_chat_body(model, messages, config, tools=tools) | {"stream": True}
        lines = self.transport.stream_lines("/chat/completions", body)
        with closing(lines):
            yield from omlx_stream_chunks(
                drop_keepalive_frames(lines), config.retain_reasoning
            )

    @timed_stream
    def complete_stream_events(
        self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
    ) -> Iterator[StreamEvent]:
        """
        Stream one turn as events, with terminal completion evidence.

        Sends one streaming request with no tools, no retries and
        ``stream_options: {"include_usage": true}``. Keep-alive frames are dropped before
        parsing, so they never become evidence. ``reasoning_content`` produces no events. The
        turn completes only when oMLX reports ``finish_reason: "stop"`` and then sends
        ``data: [DONE]``. Closing the generator closes the HTTP request.

        Parameters
        ----------
        model : str
            The model id.
        messages : List[LLMMessage]
            The messages to send.
        config : CompletionConfig
            Configuration for the request.

        Yields
        ------
        StreamEvent
            Content events followed by exactly one terminal event.
        """
        body = omlx_chat_body(model, messages, config) | {
            "stream": True,
            "stream_options": {"include_usage": True},
        }
        lines = self.transport.stream_lines("/chat/completions", body)
        try:
            with closing(lines):
                yield from parse_openai_stream(drop_keepalive_frames(lines))
        except httpx.HTTPStatusError as e:
            yield StreamError(
                reason=StreamErrorReason.PROVIDER_ERROR,
                detail=omlx_error_detail(e.response),
            )
        except httpx.HTTPError as e:
            yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

    def get_available_models(self) -> list[str]:
        """
        Get the ids of the models the server offers, sorted alphabetically.

        Returns
        -------
        List[str]
            The model ids.
        """
        return sorted(
            model["id"] for model in self.transport.get("/models").json()["data"]
        )

    def load_model(self, model: str) -> None:
        """
        Load a model into memory and wait until it is there.

        A chat request loads its model automatically; use this to warm a model up ahead of
        time. The gateway timeout covers the wait.

        Parameters
        ----------
        model : str
            The model id.
        """
        self.transport.post(f"/models/{_path_segment(model)}/load")

    def unload_model(self, model: str) -> None:
        """
        Unload a model from memory.

        Parameters
        ----------
        model : str
            The model id.

        Raises
        ------
        httpx.HTTPStatusError
            A 400 ``invalid_request_error`` when the model is not loaded.
        """
        self.transport.post(f"/models/{_path_segment(model)}/unload")

    def calculate_embeddings(self, text: str, model: str | None = None) -> list[float]:
        """
        Calculate embeddings for the text in one request, with no client-side chunking.

        Parameters
        ----------
        text : str
            The text to embed.
        model : str
            The embedding model id. Required: oMLX has no standard embedding model.

        Returns
        -------
        List[float]
            The embedding.

        Raises
        ------
        ValueError
            When ``model`` is missing or empty. No request is sent.
        httpx.HTTPStatusError
            A 400 ``invalid_request_error`` when the model is not an embedding model.
        """
        if not model or not model.strip():
            raise ValueError("oMLX has no default embedding model; pass 'model'")
        response = self.transport.post("/embeddings", {"model": model, "input": text})
        return response.json()["data"][0]["embedding"]

calculate_embeddings(text, model=None)

Calculate embeddings for the text in one request, with no client-side chunking.

Parameters:

Name	Type	Description	Default
text	str	The text to embed.	required
model	str	The embedding model id. Required: oMLX has no standard embedding model.	None

Returns:

Type	Description
List[float]	The embedding.

Raises:

Type	Description
ValueError	When model is missing or empty. No request is sent.
HTTPStatusError	A 400 invalid_request_error when the model is not an embedding model.

Source code in src/mojentic/llm/gateways/omlx.py

def calculate_embeddings(self, text: str, model: str | None = None) -> list[float]:
    """
    Calculate embeddings for the text in one request, with no client-side chunking.

    Parameters
    ----------
    text : str
        The text to embed.
    model : str
        The embedding model id. Required: oMLX has no standard embedding model.

    Returns
    -------
    List[float]
        The embedding.

    Raises
    ------
    ValueError
        When ``model`` is missing or empty. No request is sent.
    httpx.HTTPStatusError
        A 400 ``invalid_request_error`` when the model is not an embedding model.
    """
    if not model or not model.strip():
        raise ValueError("oMLX has no default embedding model; pass 'model'")
    response = self.transport.post("/embeddings", {"model": model, "input": text})
    return response.json()["data"][0]["embedding"]

complete(model, messages, object_model=None, tools=None, config=None, temperature=None, num_ctx=None, max_tokens=None)

Complete one chat turn.

When a structured output request comes back with a Warning header, oMLX did not enforce the format. The header value goes in metadata["response_format_warning"] and a warning is logged. The content is still returned and, with object_model, still validated.

Parameters:

Name	Type	Description	Default
model	str	The model id, as get_available_models reports it.	required
messages	List[LLMMessage]	The conversation.	required
object_model	Optional[Type[BaseModel]]	Request JSON schema output with this model's schema and validate the content against it.	None
tools	Optional[List[LLMTool]]	Tools to offer. Requested calls come back in tool_calls.	None
config	Optional[CompletionConfig]	Temperature, token limit, reasoning effort and response format. When None, one is built from temperature and max_tokens.	None
temperature	float | None	Deprecated: use config. num_ctx is never sent.	None
num_ctx	float | None	Deprecated: use config. num_ctx is never sent.	None
max_tokens	float | None	Deprecated: use config. num_ctx is never sent.	None

Returns:

Type	Description
LLMGatewayResponse	Content, thinking, tool calls, usage, provider model, finish reason and metadata. content is not an answer unless finish_reason is stop.

Source code in src/mojentic/llm/gateways/omlx.py

def complete(
    self,
    model: str,
    messages: list[LLMMessage],
    object_model: type[BaseModel] | None = None,
    tools: list[LLMTool] | None = None,
    config: Optional["CompletionConfig"] = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
) -> LLMGatewayResponse:
    """
    Complete one chat turn.

    When a structured output request comes back with a ``Warning`` header, oMLX did not
    enforce the format. The header value goes in ``metadata["response_format_warning"]``
    and a warning is logged. The content is still returned and, with ``object_model``,
    still validated.

    Parameters
    ----------
    model : str
        The model id, as ``get_available_models`` reports it.
    messages : List[LLMMessage]
        The conversation.
    object_model : Optional[Type[BaseModel]]
        Request JSON schema output with this model's schema and validate the content against it.
    tools : Optional[List[LLMTool]]
        Tools to offer. Requested calls come back in ``tool_calls``.
    config : Optional[CompletionConfig]
        Temperature, token limit, reasoning effort and response format. When None, one is
        built from ``temperature`` and ``max_tokens``.
    temperature, num_ctx, max_tokens
        Deprecated: use ``config``. ``num_ctx`` is never sent.

    Returns
    -------
    LLMGatewayResponse
        Content, thinking, tool calls, usage, provider model, finish reason and metadata.
        ``content`` is not an answer unless ``finish_reason`` is ``stop``.
    """
    if self.recovery_policy is not None:
        return asyncio.run(
            self.complete_with_recovery(
                model=model,
                messages=messages,
                config=config,
                tools=tools,
                object_model=object_model,
                temperature=temperature,
                num_ctx=num_ctx,
                max_tokens=max_tokens,
            )
        )
    config = config or _config_from_arguments(temperature, num_ctx, max_tokens)
    body = omlx_chat_body(
        model, messages, config, tools=tools, object_model=object_model
    )
    response = self.transport.post("/chat/completions", body)
    result = omlx_gateway_response(
        response.json(), self._response_format_warning(model, body, response)
    )
    if object_model is not None:
        result.object = _validated_object(object_model, result.content)
    return result

complete_stream(model, messages, object_model=None, tools=None, config=None, temperature=None, num_ctx=None, max_tokens=None)

Stream one chat turn as legacy chunks.

Content arrives as content chunks and reasoning_content as thinking chunks. Tool calls are yielded together, complete, when the turn finishes with tool_calls. Keep-alive frames are dropped. Closing the generator closes the request.

Parameters are as for :meth:complete, except that object_model is not supported.

Raises:

Type	Description
NotImplementedError	When object_model is given.

Source code in src/mojentic/llm/gateways/omlx.py

@timed_stream
def complete_stream(
    self,
    model: str,
    messages: list[LLMMessage],
    object_model: type[BaseModel] | None = None,
    tools: list[LLMTool] | None = None,
    config: Optional["CompletionConfig"] = None,
    temperature: float | None = None,
    num_ctx: int | None = None,
    max_tokens: int | None = None,
) -> Iterator[StreamingResponse]:
    """
    Stream one chat turn as legacy chunks.

    Content arrives as content chunks and ``reasoning_content`` as thinking chunks. Tool
    calls are yielded together, complete, when the turn finishes with ``tool_calls``.
    Keep-alive frames are dropped. Closing the generator closes the request.

    Parameters are as for :meth:`complete`, except that ``object_model`` is not supported.

    Raises
    ------
    NotImplementedError
        When ``object_model`` is given.
    """
    if object_model is not None:
        raise NotImplementedError(
            "Streaming with structured output (object_model) is not supported"
        )
    if self.recovery_policy is not None:
        events = synchronous_stream(
            self.complete_stream_with_recovery(
                model=model,
                messages=messages,
                config=config,
                tools=tools,
                temperature=temperature,
                num_ctx=num_ctx,
                max_tokens=max_tokens,
            )
        )
        with closing(events):
            for event in events:
                if event.kind == "failed":
                    event.error.stream_metrics = event.metrics
                    self.last_stream_metrics = event.metrics
                    raise event.error
                if event.kind == "completed":
                    self.last_stream_metrics = event.metrics
                if event.kind == "content":
                    yield StreamingResponse(content=event.text)
                elif event.kind == "reasoning":
                    yield StreamingResponse(thinking=event.text)
                elif event.kind == "completed" and event.response.tool_calls:
                    yield StreamingResponse(
                        tool_calls=event.response.tool_calls, completed=True
                    )
        return
    config = config or _config_from_arguments(temperature, num_ctx, max_tokens)
    body = omlx_chat_body(model, messages, config, tools=tools) | {"stream": True}
    lines = self.transport.stream_lines("/chat/completions", body)
    with closing(lines):
        yield from omlx_stream_chunks(
            drop_keepalive_frames(lines), config.retain_reasoning
        )

complete_stream_events(model, messages, config)

Stream one turn as events, with terminal completion evidence.

Sends one streaming request with no tools, no retries and stream_options: {"include_usage": true}. Keep-alive frames are dropped before parsing, so they never become evidence. reasoning_content produces no events. The turn completes only when oMLX reports finish_reason: "stop" and then sends data: [DONE]. Closing the generator closes the HTTP request.

Parameters:

Name	Type	Description	Default
model	str	The model id.	required
messages	List[LLMMessage]	The messages to send.	required
config	CompletionConfig	Configuration for the request.	required

Yields:

Type	Description
StreamEvent	Content events followed by exactly one terminal event.

Source code in src/mojentic/llm/gateways/omlx.py

@timed_stream
def complete_stream_events(
    self, model: str, messages: list[LLMMessage], config: "CompletionConfig"
) -> Iterator[StreamEvent]:
    """
    Stream one turn as events, with terminal completion evidence.

    Sends one streaming request with no tools, no retries and
    ``stream_options: {"include_usage": true}``. Keep-alive frames are dropped before
    parsing, so they never become evidence. ``reasoning_content`` produces no events. The
    turn completes only when oMLX reports ``finish_reason: "stop"`` and then sends
    ``data: [DONE]``. Closing the generator closes the HTTP request.

    Parameters
    ----------
    model : str
        The model id.
    messages : List[LLMMessage]
        The messages to send.
    config : CompletionConfig
        Configuration for the request.

    Yields
    ------
    StreamEvent
        Content events followed by exactly one terminal event.
    """
    body = omlx_chat_body(model, messages, config) | {
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    lines = self.transport.stream_lines("/chat/completions", body)
    try:
        with closing(lines):
            yield from parse_openai_stream(drop_keepalive_frames(lines))
    except httpx.HTTPStatusError as e:
        yield StreamError(
            reason=StreamErrorReason.PROVIDER_ERROR,
            detail=omlx_error_detail(e.response),
        )
    except httpx.HTTPError as e:
        yield StreamError(reason=StreamErrorReason.REQUEST_FAILED, detail=str(e))

complete_stream_with_recovery(call=None, **args) async

Yield progress and exactly one completed or failed recovery outcome.

Closing the iterator cancels owned local HTTP; remote termination is unknown. Raw complete_stream_events remains a single-request compatibility API.

Parameters:

Name	Type	Description	Default
call	RecoveryCall	Cancellation for this operation, overriding the gateway call.	None
**args		The model, messages, tools and controls accepted by complete_stream.	{}

Yields:

Type	Description
RecoveryStreamEvent	Provider progress followed by one completed or failed terminal outcome.

Raises:

Type	Description
ValueError	No recovery policy is configured.

Source code in src/mojentic/llm/gateways/omlx.py

async def complete_stream_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> AsyncIterator[RecoveryStreamEvent]:
    """Yield progress and exactly one completed or failed recovery outcome.

    Closing the iterator cancels owned local HTTP; remote termination is unknown.
    Raw complete_stream_events remains a single-request compatibility API.

    Parameters
    ----------
    call : RecoveryCall, optional
        Cancellation for this operation, overriding the gateway call.
    **args
        The model, messages, tools and controls accepted by ``complete_stream``.

    Yields
    ------
    RecoveryStreamEvent
        Provider progress followed by one completed or failed terminal outcome.

    Raises
    ------
    ValueError
        No recovery policy is configured.
    """
    from mojentic.llm.gateways.omlx_recovery import (
        OMLXStreamRecovery,
        decode_completion,
    )

    if self.recovery_policy is None:
        raise RecoveryPolicyError("configure recovery_policy before using recovery")
    try:
        body = self._recovery_body(args, True)
    except (ValueError, TypeError, KeyError, OSError) as cause:
        error = preparation_error(cause, "streaming", "omlx")
        yield RecoveryStreamEvent(
            kind="failed",
            identity=error.report.identity,
            progress=error.report.progress,
            report=error.report,
            error=error,
            outcome="ineligible",
        )
        return
    events = recover_stream(
        self.settings.base_url + "/chat/completions",
        body,
        self._recovery_headers(),
        httpx.Timeout(self.settings.timeout, read=None),
        decode_completion,
        self.recovery_policy,
        call or self.recovery_call or RecoveryCall(),
        _engine_type=OMLXStreamRecovery,
    )
    try:
        async for event in events:
            yield event
    finally:
        await events.aclose()

complete_with_recovery(call=None, **args) async

Recover one ordinary or structured completion, retaining typed failure evidence.

Recovery bypasses the legacy transport and disables active read timeouts.

Parameters:

Name	Type	Description	Default
call	RecoveryCall	Cancellation for this operation, overriding the gateway call.	None
**args		The model, messages, tools, schema and controls accepted by complete.	{}

Returns:

Type	Description
LLMGatewayResponse	Provider response with a typed recovery report.

Raises:

Type	Description
RecoveryError	Recovery refused, interrupted, cancelled or received an invalid response.
ValueError	No recovery policy is configured.

Source code in src/mojentic/llm/gateways/omlx.py

async def complete_with_recovery(
    self, call: RecoveryCall | None = None, **args
) -> LLMGatewayResponse:
    """Recover one ordinary or structured completion, retaining typed failure evidence.

    Recovery bypasses the legacy transport and disables active read timeouts.

    Parameters
    ----------
    call : RecoveryCall, optional
        Cancellation for this operation, overriding the gateway call.
    **args
        The model, messages, tools, schema and controls accepted by ``complete``.

    Returns
    -------
    LLMGatewayResponse
        Provider response with a typed recovery report.

    Raises
    ------
    RecoveryError
        Recovery refused, interrupted, cancelled or received an invalid response.
    ValueError
        No recovery policy is configured.
    """
    from mojentic.llm.gateways.omlx_recovery import OMLXRecovery, decode_completion

    if self.recovery_policy is None:
        raise RecoveryPolicyError("configure recovery_policy before using recovery")
    operation = "structured" if args.get("object_model") is not None else "ordinary"
    try:
        body = self._recovery_body(args, False)
    except (ValueError, TypeError, KeyError, OSError) as cause:
        raise preparation_error(cause, operation, "omlx") from None
    engine = OMLXRecovery(
        self.recovery_policy,
        call or self.recovery_call or RecoveryCall(),
        operation,
    )
    result, report = await engine.run(
        self.settings.base_url + "/chat/completions",
        body,
        self._recovery_headers(),
        httpx.Timeout(self.settings.timeout, read=None),
        lambda frame: decode_completion(frame, args.get("object_model")),
    )
    result.recovery_report = report
    result.metadata |= {"recovery": report.model_dump(mode="json")}
    return result

get_available_models()

Get the ids of the models the server offers, sorted alphabetically.

Returns:

Type	Description
List[str]	The model ids.

Source code in src/mojentic/llm/gateways/omlx.py

def get_available_models(self) -> list[str]:
    """
    Get the ids of the models the server offers, sorted alphabetically.

    Returns
    -------
    List[str]
        The model ids.
    """
    return sorted(
        model["id"] for model in self.transport.get("/models").json()["data"]
    )

get_model_context_window(model)

Discover exact model metadata using the configured host and credentials.

Source code in src/mojentic/llm/gateways/omlx.py

def get_model_context_window(self, model: str) -> ModelContextWindow | None:
    """Discover exact model metadata using the configured host and credentials."""
    return omlx_context_window(self.transport.get("models").json(), model)

load_model(model)

Load a model into memory and wait until it is there.

A chat request loads its model automatically; use this to warm a model up ahead of time. The gateway timeout covers the wait.

Parameters:

Name	Type	Description	Default
model	str	The model id.	required

Source code in src/mojentic/llm/gateways/omlx.py

def load_model(self, model: str) -> None:
    """
    Load a model into memory and wait until it is there.

    A chat request loads its model automatically; use this to warm a model up ahead of
    time. The gateway timeout covers the wait.

    Parameters
    ----------
    model : str
        The model id.
    """
    self.transport.post(f"/models/{_path_segment(model)}/load")

recovery_capabilities()

Report supported request recovery and unsupported remote facilities.

Source code in src/mojentic/llm/gateways/omlx.py

def recovery_capabilities(self) -> Capabilities:
    """Report supported request recovery and unsupported remote facilities."""
    return completion_capabilities("omlx")

unload_model(model)

Unload a model from memory.

Parameters:

Name	Type	Description	Default
model	str	The model id.	required

Raises:

Type	Description
HTTPStatusError	A 400 invalid_request_error when the model is not loaded.

Source code in src/mojentic/llm/gateways/omlx.py

def unload_model(self, model: str) -> None:
    """
    Unload a model from memory.

    Parameters
    ----------
    model : str
        The model id.

    Raises
    ------
    httpx.HTTPStatusError
        A 400 ``invalid_request_error`` when the model is not loaded.
    """
    self.transport.post(f"/models/{_path_segment(model)}/unload")

mojentic.llm.gateways.models.LLMMessage

Bases: BaseModel

A message to be sent to the LLM. These would accumulate during a chat session with an LLM.

Attributes:

Name	Type	Description

Parameters:

Name	Type	Description	Default
role	MessageRole		<MessageRole.User: 'user'>
content	str | None		None
thinking	str | None		None
object	BaseModel | None		None
tool_calls	list[LLMToolCall] | None		None
image_paths	list[str] | None		None

Source code in src/mojentic/llm/gateways/models.py

class LLMMessage(BaseModel):
    """
    A message to be sent to the LLM. These would accumulate during a chat session with an LLM.

    Attributes
    ----------
    role : MessageRole
        The role of the message in the conversation.
    content : Optional[str]
        The content of the message.
    thinking : str | None
        Native assistant reasoning history, separate from visible content and tool results.
    object : Optional[BaseModel]
        The object representation of the message.
    tool_calls : Optional[List[LLMToolCall]]
        A list of tool calls to be made available to the LLM.
    image_paths : Optional[List[str]]
        Local image paths, HTTP(S) URLs or data URIs to include with the message.
        OpenAI passes URLs and data URIs through without reading a file.
        Note: You must use an image-capable model to process images.
    """

    role: MessageRole = MessageRole.User
    content: str | None = None
    thinking: str | None = None
    object: BaseModel | None = None
    tool_calls: list[LLMToolCall] | None = None
    image_paths: list[str] | None = None

mojentic.llm.gateways.models.LLMToolCall

Bases: BaseModel

A tool call to be made available to the LLM.

Attributes:

Name	Type	Description

Parameters:

Name	Type	Description	Default
id	str | None		None
name	str		required
arguments	dict[str, Any]		required

Source code in src/mojentic/llm/gateways/models.py

class LLMToolCall(BaseModel):
    """
    A tool call to be made available to the LLM.

    Attributes
    ----------
    id : Optional[str]
        The identifier of the tool call.
    name : str
        The name of the tool call.
    arguments : dict[str, Any]
        The arguments for the tool call.
    """

    id: str | None = None
    name: str
    arguments: dict[str, Any]

mojentic.llm.gateways.models.LLMGatewayResponse

Bases: BaseModel

The response from the LLM gateway, abstracting you from the quirks of a specific LLM.

Attributes:

Name	Type	Description

Parameters:

Name	Type	Description	Default
content	str | dict[str, str] | None	The content of the response.	None
object	BaseModel | None	Parsed response object	None
tool_calls	list[LLMToolCall]	List of requested tool calls from the LLM.	<dynamic>
thinking	str | None	Model thinking/reasoning trace (populated by some providers)	None
usage	dict[str, Any] | None		None
model	str | None		None
finish_reason	str | None		None
metadata	dict[str, Any]	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	<class 'dict'>
recovery_report	RecoveryReport | None		None

Source code in src/mojentic/llm/gateways/models.py

class LLMGatewayResponse(BaseModel):
    """
    The response from the LLM gateway, abstracting you from the quirks of a specific LLM.

    Attributes
    ----------
    content : Optional[Union[str, dict[str, str]]]
        The content of the response.
    object : Optional[BaseModel]
        Parsed response object.
    tool_calls : List[LLMToolCall]
        List of requested tool calls from the LLM.
    thinking : Optional[str]
        Model thinking/reasoning trace (populated by some providers).
    """

    content: str | dict[str, str] | None = Field(
        None, description="The content of the response."
    )
    object: BaseModel | None = Field(None, description="Parsed response object")
    tool_calls: list[LLMToolCall] = Field(
        default_factory=list, description="List of requested tool calls from the LLM."
    )
    thinking: str | None = Field(
        None, description="Model thinking/reasoning trace (populated by some providers)"
    )
    usage: dict[str, Any] | None = None
    model: str | None = None
    finish_reason: str | None = None
    metadata: dict[str, Any] = Field(default_factory=dict)
    recovery_report: RecoveryReport | None = Field(
        default=None, exclude=True, repr=False
    )