ChatCompletion

OpenAIChatCompletion

Bases: _BaseOpenAI, ChatCompletionModel

OpenAI Chat Completion.

Source code in src/msgflux/models/providers/openai.py

@register_model
class OpenAIChatCompletion(_BaseOpenAI, ChatCompletionModel):
    """OpenAI Chat Completion."""

    def __init__(
        self,
        model_id: str,
        *,
        max_tokens: Optional[int] = None,
        reasoning_effort: Optional[str] = None,
        enable_thinking: Optional[bool] = None,
        return_reasoning: Optional[bool] = False,
        reasoning_in_tool_call: Optional[bool] = True,
        validate_typed_parser_output: Optional[bool] = False,
        temperature: Optional[float] = None,
        top_p: Optional[float] = None,
        stop: Optional[Union[str, List[str]]] = None,
        parallel_tool_calls: Optional[bool] = True,
        modalities: Optional[List[str]] = None,
        audio: Optional[Dict[str, str]] = None,
        verbosity: Optional[str] = None,
        web_search_options: Optional[Dict[str, Any]] = None,
        verbose: Optional[bool] = False,
        base_url: Optional[str] = None,
        context_length: Optional[int] = None,
        reasoning_max_tokens: Optional[int] = None,
        enable_cache: Optional[bool] = False,
        cache_size: Optional[int] = 128,
    ):
        """Args:
        model_id:
            Model ID in provider.
        max_tokens:
            An upper bound for the number of tokens that can be
            generated for a completion, including visible output
            tokens and reasoning tokens.
        reasoning_effort:
            Constrains effort on reasoning for reasoning models.
            Currently supported values are low, medium, and high.
            Reducing reasoning effort can result in faster responses
            and fewer tokens used on reasoning in a response.
            Can be: "minimal", "low", "medium" or "high".
        enable_thinking:
            If True, enable the model reasoning.
        return_reasoning:
            If the model returns the `reasoning` field it will be added
            along with the response.
        reasoning_in_tool_call:
            If True, maintains the reasoning for using the tool call.
        validate_typed_parser_output:
            If True, use the generation_schema to validate typed parser output.
        temperature:
            What sampling temperature to use, between 0 and 2.
            Higher values like 0.8 will make the output more random,
            while lower values like 0.2 will make it more focused and
            deterministic.
        stop:
            Up to 4 sequences where the API will stop generating further
            tokens. The returned text will not contain the stop sequence.
        top_p:
            An alternative to sampling with temperature, called nucleus
            sampling, where the model considers the results of the tokens
            with top_p probability mass. So 0.1 means only the tokens
            comprising the top 10% probability mass are considered.
        parallel_tool_calls:
            If True, enable parallel tool calls.
        modalities:
            Types of output you would like the model to generate.
            Can be: ["text"], ["audio"] or ["text", "audio"].
        audio:
            Audio configurations. Define voice and output format.
        verbosity:
            Constrains the verbosity of the model's response. Lower
            values will result in more concise responses, while higher
            values will result in more verbose responses. Currently
            supported values are low, medium, and high.
        web_search_options:
            This tool searches the web for relevant results to use in a response.
            OpenAI and OpenRouter only.
        verbose:
            If True, Prints the model output to the console before it is transformed
            into typed structured output.
        base_url:
            URL to model provider.
        context_length:
            The maximum context length supported by the model.
        reasoning_max_tokens:
            Maximum number of tokens for reasoning/thinking.
        enable_cache:
            If True, enable response caching to avoid redundant API calls.
        cache_size:
            Maximum number of cached responses (default: 128).
        """
        super().__init__()
        self.model_id = model_id
        self.context_length = context_length
        self.reasoning_max_tokens = reasoning_max_tokens
        self.enable_cache = enable_cache
        self.cache_size = cache_size
        self.sampling_params = {"base_url": base_url or self._get_base_url()}
        sampling_run_params = {"max_tokens": max_tokens}
        if temperature:
            sampling_run_params["temperature"] = temperature
        if top_p:
            sampling_run_params["top_p"] = top_p
        if stop:
            sampling_run_params["stop"] = stop
        if verbosity:
            sampling_run_params["verbosity"] = verbosity
        if modalities:
            sampling_run_params["modalities"] = modalities
        if web_search_options:
            sampling_run_params["web_search_options"] = web_search_options
        if audio:
            sampling_run_params["audio"] = audio
        if reasoning_effort:
            sampling_run_params["reasoning_effort"] = reasoning_effort
        self.sampling_run_params = sampling_run_params
        self.enable_thinking = enable_thinking
        self.parallel_tool_calls = parallel_tool_calls
        self.reasoning_in_tool_call = reasoning_in_tool_call
        self.validate_typed_parser_output = validate_typed_parser_output
        self.return_reasoning = return_reasoning
        self.verbose = verbose
        self._initialize()
        self._get_api_key()

    def _adapt_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        if self.provider in "openai":
            params["max_completion_tokens"] = params.pop("max_tokens")
        return params

    def _execute_model(self, **kwargs):
        prefilling = kwargs.pop("prefilling")
        if prefilling:
            kwargs.get("messages").append({"role": "assistant", "content": prefilling})
        params = {**kwargs, **self.sampling_run_params}
        adapted_params = self._adapt_params(params)
        model_output = self.client.chat.completions.create(**adapted_params)

        return model_output

    async def _aexecute_model(self, **kwargs):
        prefilling = kwargs.pop("prefilling")
        if prefilling:
            kwargs.get("messages").append({"role": "assistant", "content": prefilling})
        params = {**kwargs, **self.sampling_run_params}
        adapted_params = self._adapt_params(params)
        model_output = await self.aclient.chat.completions.create(**adapted_params)

        return model_output

    def _process_model_output(  # noqa: C901
        self, model_output, typed_parser=None, generation_schema=None
    ):
        """Shared logic to process model output for both sync and async."""
        response = ModelResponse()
        metadata = dotdict()

        metadata.update({"usage": model_output.usage.to_dict()})

        choice = model_output.choices[0]

        reasoning = (
            getattr(choice.message, "reasoning_content", None)
            or getattr(choice.message, "reasoning", None)
            or getattr(choice.message, "thinking", None)
        )

        reasoning_tool_call = None
        if self.reasoning_in_tool_call is True:
            reasoning_tool_call = reasoning

        prefix_response_type = ""
        reasoning_content = None
        if self.return_reasoning is True:
            reasoning_content = reasoning
            if reasoning_content is not None:
                prefix_response_type = "reasoning_"

        if choice.message.annotations:  # Extra responses (e.g web search references)
            annotations_content = [
                item.model_dump() for item in choice.message.annotations
            ]
            metadata.annotations = annotations_content

        if choice.message.tool_calls:
            aggregator = ToolCallAggregator(reasoning_tool_call)
            response.set_response_type("tool_call")
            for call_index, tool_call in enumerate(choice.message.tool_calls):
                tool_id = tool_call.id
                name = tool_call.function.name
                arguments = tool_call.function.arguments
                aggregator.process(call_index, tool_id, name, arguments)
            response_content = aggregator
        elif choice.message.content:
            if (typed_parser or generation_schema) and self.verbose:
                repr_str = f"[{self.model_id}][raw_response] {choice.message.content}"
                cprint(repr_str, lc="r", ls="b")
            if typed_parser is not None:
                response.set_response_type(f"{prefix_response_type}structured")
                parser = typed_parser_registry[typed_parser]
                response_content = dotdict(parser.decode(choice.message.content))
                # Type validation
                if generation_schema and self.validate_typed_parser_output:
                    encoded_response_content = msgspec.json.encode(response_content)
                    msgspec.json.decode(
                        encoded_response_content, type=generation_schema
                    )
            elif generation_schema is not None:
                response.set_response_type(f"{prefix_response_type}structured")
                struct = msgspec.json.decode(
                    choice.message.content, type=generation_schema
                )
                response_content = dotdict(struct_to_dict(struct))
            else:
                response.set_response_type(f"{prefix_response_type}text_generation")
                if reasoning_content is not None:
                    response_content = dotdict({"answer": choice.message.content})
                else:
                    response_content = choice.message.content
        elif choice.message.audio:
            response_content = dotdict(
                {
                    "id": choice.message.audio.id,
                    "audio": base64.b64decode(choice.message.audio.data),
                }
            )
            if choice.message.audio.transcript:
                response.set_response_type("audio_text_generation")
                response_content.text = choice.message.audio.transcript
            else:
                response.set_response_type("audio_generation")

        if reasoning_content is not None:
            response_content.think = reasoning_content

        response.add(response_content)
        response.set_metadata(metadata)
        return response

    def _generate(self, **kwargs: Mapping[str, Any]) -> ModelResponse:
        typed_parser = kwargs.get("typed_parser")
        generation_schema = kwargs.get("generation_schema")

        # Check cache if enabled
        if self.enable_cache and self._response_cache:
            cache_key = generate_cache_key(**kwargs)
            hit, cached_response = self._response_cache.get(cache_key)
            if hit:
                return cached_response

        # Pop after cache check to avoid modifying kwargs during cache key generation
        typed_parser = kwargs.pop("typed_parser")
        generation_schema = kwargs.pop("generation_schema")

        if generation_schema is not None and typed_parser is None:
            response_format = response_format_from_msgspec_struct(generation_schema)
            kwargs["response_format"] = response_format

        model_output = self._execute_model(**kwargs)
        response = self._process_model_output(
            model_output, typed_parser, generation_schema
        )

        # Store in cache if enabled
        if self.enable_cache and self._response_cache:
            # Re-add popped values for cache key
            cache_kwargs = {
                **kwargs,
                "typed_parser": typed_parser,
                "generation_schema": generation_schema,
            }
            cache_key = generate_cache_key(**cache_kwargs)
            self._response_cache.set(cache_key, response)

        return response

    async def _agenerate(self, **kwargs: Mapping[str, Any]) -> ModelResponse:
        typed_parser = kwargs.get("typed_parser")
        generation_schema = kwargs.get("generation_schema")

        # Check cache if enabled
        if self.enable_cache and self._response_cache:
            cache_key = generate_cache_key(**kwargs)
            hit, cached_response = self._response_cache.get(cache_key)
            if hit:
                return cached_response

        # Pop after cache check to avoid modifying kwargs during cache key generation
        typed_parser = kwargs.pop("typed_parser")
        generation_schema = kwargs.pop("generation_schema")

        if generation_schema is not None and typed_parser is None:
            response_format = response_format_from_msgspec_struct(generation_schema)
            kwargs["response_format"] = response_format

        model_output = await self._aexecute_model(**kwargs)
        response = self._process_model_output(
            model_output, typed_parser, generation_schema
        )

        # Store in cache if enabled
        if self.enable_cache and self._response_cache:
            # Re-add popped values for cache key
            cache_kwargs = {
                **kwargs,
                "typed_parser": typed_parser,
                "generation_schema": generation_schema,
            }
            cache_key = generate_cache_key(**cache_kwargs)
            self._response_cache.set(cache_key, response)

        return response

    async def _stream_generate(  # noqa: C901
        self, **kwargs: Mapping[str, Any]
    ) -> ModelStreamResponse:
        aggregator = ToolCallAggregator()
        metadata = dotdict()

        stream_response = kwargs.pop("stream_response")
        model_output = self._execute_model(**kwargs)

        reasoning_tool_call = ""

        for chunk in model_output:
            if chunk.choices:
                delta = chunk.choices[0].delta

                reasoning_chunk = (
                    getattr(delta, "reasoning_content", None)
                    or getattr(delta, "reasoning", None)
                    or getattr(delta, "thinking", None)
                )

                if self.reasoning_in_tool_call and reasoning_chunk:
                    reasoning_tool_call += reasoning_chunk

                if self.return_reasoning and reasoning_chunk:
                    if stream_response.response_type is None:
                        stream_response.set_response_type("reasoning_text_generation")
                        stream_response.first_chunk_event.set()
                    stream_response.add(reasoning_chunk)
                    continue

                if getattr(delta, "content", None):
                    if stream_response.response_type is None:
                        stream_response.set_response_type("text_generation")
                        stream_response.first_chunk_event.set()
                    stream_response.add(delta.content)
                    continue

                if getattr(delta, "tool_calls", None):
                    if stream_response.response_type is None:
                        stream_response.set_response_type("tool_call")
                    tool_call = delta.tool_calls[0]
                    call_index = tool_call.index
                    tool_id = tool_call.id
                    name = tool_call.function.name
                    arguments = tool_call.function.arguments
                    aggregator.process(call_index, tool_id, name, arguments)
                    continue

                if hasattr(delta, "annotations") and delta.annotations is not None:
                    metadata.annotations = [
                        item.model_dump() for item in delta.annotations
                    ]
                    continue

            elif chunk.usage:
                metadata.update(chunk.usage.to_dict())

        if aggregator.tool_calls:
            if reasoning_tool_call:
                aggregator.reasoning = reasoning_tool_call
            stream_response.data = aggregator  # For tool calls save as 'data'
            stream_response.first_chunk_event.set()

        stream_response.set_metadata(metadata)
        stream_response.add(None)

    async def _astream_generate(  # noqa: C901
        self, **kwargs: Mapping[str, Any]
    ) -> ModelStreamResponse:
        aggregator = ToolCallAggregator()
        metadata = dotdict()

        stream_response = kwargs.pop("stream_response")
        model_output = await self._aexecute_model(**kwargs)

        reasoning_tool_call = ""

        async for chunk in model_output:
            if chunk.choices:
                delta = chunk.choices[0].delta

                reasoning_chunk = (
                    getattr(delta, "reasoning_content", None)
                    or getattr(delta, "reasoning", None)
                    or getattr(delta, "thinking", None)
                )

                if self.reasoning_in_tool_call and reasoning_chunk:
                    reasoning_tool_call += reasoning_chunk

                if self.return_reasoning and reasoning_chunk:
                    if stream_response.response_type is None:
                        stream_response.set_response_type("reasoning_text_generation")
                        stream_response.first_chunk_event.set()
                    stream_response.add(reasoning_chunk)
                    continue

                if getattr(delta, "content", None):
                    if stream_response.response_type is None:
                        stream_response.set_response_type("text_generation")
                        stream_response.first_chunk_event.set()
                    stream_response.add(delta.content)
                    continue

                if getattr(delta, "tool_calls", None):
                    if stream_response.response_type is None:
                        stream_response.set_response_type("tool_call")
                    tool_call = delta.tool_calls[0]
                    call_index = tool_call.index
                    tool_id = tool_call.id
                    name = tool_call.function.name
                    arguments = tool_call.function.arguments
                    aggregator.process(call_index, tool_id, name, arguments)
                    continue

                if hasattr(delta, "annotations") and delta.annotations is not None:
                    metadata.annotations = [
                        item.model_dump() for item in delta.annotations
                    ]
                    continue

            elif chunk.usage:
                metadata.update(chunk.usage.to_dict())

        if aggregator.tool_calls:
            if reasoning_tool_call:
                aggregator.reasoning = reasoning_tool_call
            stream_response.data = aggregator  # For tool calls save as 'data'
            stream_response.first_chunk_event.set()

        stream_response.set_metadata(metadata)
        stream_response.add(None)

    @model_retry
    def __call__(
        self,
        messages: Union[str, List[Dict[str, Any]]],
        *,
        system_prompt: Optional[str] = None,
        prefilling: Optional[str] = None,
        stream: Optional[bool] = False,
        generation_schema: Optional[msgspec.Struct] = None,
        tool_schemas: Optional[Dict] = None,
        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
        typed_parser: Optional[str] = None,
    ) -> Union[ModelResponse, ModelStreamResponse]:
        """Args:
            messages:
                Conversation history. Can be simple string or list of messages.
            system_prompt:
                A set of instructions that defines the overarching behavior
                and role of the model across all interactions.
            prefilling:
                Forces an initial message from the model. From that message
                it will continue its response from there.
            stream:
                Whether generation should be in streaming mode.
            generation_schema:
                Schema that defines how the output should be structured.
            tool_schemas:
                JSON schema containing available tools.
            tool_choice:
                By default the model will determine when and how many tools to use.
                You can force specific behavior with the tool_choice parameter.
                    1. auto:
                        (Default) Call zero, one, or multiple functions.
                    2. required:
                        Call one or more functions.
                    3. Forced Tool:
                        Call exactly one specific tool e.g: "get_weather".
            typed_parser:
                Converts the model raw output into a typed-dict. Supported parser:
                `typed_xml`.

        Raises:
            ValueError:
                Raised if `generation_schema` and `stream=True`.
            ValueError:
                Raised if `typed_xml=True` and `stream=True`.
        """
        if isinstance(messages, str):
            messages = [ChatBlock.user(messages)]
        if isinstance(system_prompt, str):
            messages.insert(0, ChatBlock.system(system_prompt))

        if isinstance(tool_choice, str):
            if tool_choice not in ["auto", "required", "none"]:
                tool_choice = {
                    "type": "function",
                    "function": {"name": tool_choice},
                }

        generation_params = {
            "messages": messages,
            "prefilling": prefilling,
            "tool_choice": tool_choice,
            "tools": tool_schemas,
            "model": self.model_id,
        }

        if tool_schemas:
            generation_params["parallel_tool_calls"] = self.parallel_tool_calls

        if stream is True:
            if typed_parser is not None:
                raise ValueError("`typed_parser` is not `stream=True` compatible")

            stream_response = ModelStreamResponse()
            F.background_task(
                self._stream_generate,
                **generation_params,
                stream=stream,
                stream_response=stream_response,
                stream_options={"include_usage": True},
            )
            F.wait_for_event(stream_response.first_chunk_event)
            return stream_response
        else:
            if typed_parser and typed_parser not in typed_parser_registry:
                available = ", ".join(typed_parser_registry.keys())
                raise TypedParserNotFoundError(
                    f"Typed parser `{typed_parser}` not found. "
                    f"Available parsers: {available}"
                )
            response = self._generate(
                **generation_params,
                typed_parser=typed_parser,
                generation_schema=generation_schema,
            )
            return response

    @model_retry
    async def acall(
        self,
        messages: Union[str, List[Dict[str, Any]]],
        *,
        system_prompt: Optional[str] = None,
        prefilling: Optional[str] = None,
        stream: Optional[bool] = False,
        generation_schema: Optional[msgspec.Struct] = None,
        tool_schemas: Optional[Dict] = None,
        tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
        typed_parser: Optional[str] = None,
    ) -> Union[ModelResponse, ModelStreamResponse]:
        """Async version of __call__. Args:
            messages:
                Conversation history. Can be simple string or list of messages.
            system_prompt:
                A set of instructions that defines the overarching behavior
                and role of the model across all interactions.
            prefilling:
                Forces an initial message from the model. From that message
                it will continue its response from there.
            stream:
                Whether generation should be in streaming mode.
            generation_schema:
                Schema that defines how the output should be structured.
            tool_schemas:
                JSON schema containing available tools.
            tool_choice:
                By default the model will determine when and how many tools to use.
                You can force specific behavior with the tool_choice parameter.
                    1. auto:
                        (Default) Call zero, one, or multiple functions.
                    2. required:
                        Call one or more functions.
                    3. Forced Tool:
                        Call exactly one specific tool e.g: "get_weather".
            typed_parser:
                Converts the model raw output into a typed-dict. Supported parser:
                `typed_xml`.

        Raises:
            ValueError:
                Raised if `generation_schema` and `stream=True`.
            ValueError:
                Raised if `typed_xml=True` and `stream=True`.
        """
        if isinstance(messages, str):
            messages = [ChatBlock.user(messages)]
        if isinstance(system_prompt, str):
            messages.insert(0, ChatBlock.system(system_prompt))

        if isinstance(tool_choice, str):
            if tool_choice not in ["auto", "required", "none"]:
                tool_choice = {
                    "type": "function",
                    "function": {"name": tool_choice},
                }

        generation_params = {
            "messages": messages,
            "prefilling": prefilling,
            "tool_choice": tool_choice,
            "tools": tool_schemas,
            "model": self.model_id,
        }

        if tool_schemas:
            generation_params["parallel_tool_calls"] = self.parallel_tool_calls

        if stream is True:
            if typed_parser is not None:
                raise ValueError("`typed_parser` is not `stream=True` compatible")

            stream_response = ModelStreamResponse()
            await F.abackground_task(
                self._astream_generate,
                **generation_params,
                stream=stream,
                stream_response=stream_response,
                stream_options={"include_usage": True},
            )
            await F.await_for_event(stream_response.first_chunk_event)
            return stream_response
        else:
            if typed_parser and typed_parser not in typed_parser_registry:
                available = ", ".join(typed_parser_registry.keys())
                raise TypedParserNotFoundError(
                    f"Typed parser `{typed_parser}` not found. "
                    f"Available parsers: {available}"
                )
            response = await self._agenerate(
                **generation_params,
                typed_parser=typed_parser,
                generation_schema=generation_schema,
            )
            return response

cache_size instance-attribute

cache_size = cache_size

context_length instance-attribute

context_length = context_length

enable_cache instance-attribute

enable_cache = enable_cache

enable_thinking instance-attribute

enable_thinking = enable_thinking

model_id instance-attribute

model_id = model_id

parallel_tool_calls instance-attribute

parallel_tool_calls = parallel_tool_calls

reasoning_in_tool_call instance-attribute

reasoning_in_tool_call = reasoning_in_tool_call

reasoning_max_tokens instance-attribute

reasoning_max_tokens = reasoning_max_tokens

return_reasoning instance-attribute

return_reasoning = return_reasoning

sampling_params instance-attribute

sampling_params = {'base_url': base_url or _get_base_url()}

sampling_run_params instance-attribute

sampling_run_params = sampling_run_params

validate_typed_parser_output instance-attribute

validate_typed_parser_output = validate_typed_parser_output

verbose instance-attribute

verbose = verbose

__call__

__call__(
    messages,
    *,
    system_prompt=None,
    prefilling=None,
    stream=False,
    generation_schema=None,
    tool_schemas=None,
    tool_choice=None,
    typed_parser=None,
)

Parameters:

Name	Type	Description	Default
messages	Union[str, List[Dict[str, Any]]]	Conversation history. Can be simple string or list of messages.	required
system_prompt	Optional[str]	A set of instructions that defines the overarching behavior and role of the model across all interactions.	None
prefilling	Optional[str]	Forces an initial message from the model. From that message it will continue its response from there.	None
stream	Optional[bool]	Whether generation should be in streaming mode.	False
generation_schema	Optional[Struct]	Schema that defines how the output should be structured.	None
tool_schemas	Optional[Dict]	JSON schema containing available tools.	None
tool_choice	Optional[Union[str, Dict[str, Any]]]	By default the model will determine when and how many tools to use. You can force specific behavior with the tool_choice parameter. 1. auto: (Default) Call zero, one, or multiple functions. 2. required: Call one or more functions. 3. Forced Tool: Call exactly one specific tool e.g: "get_weather".	None
typed_parser	Optional[str]	Converts the model raw output into a typed-dict. Supported parser: typed_xml.	None

Raises:

Type	Description
ValueError	Raised if generation_schema and stream=True.
ValueError	Raised if typed_xml=True and stream=True.

Source code in src/msgflux/models/providers/openai.py

@model_retry
def __call__(
    self,
    messages: Union[str, List[Dict[str, Any]]],
    *,
    system_prompt: Optional[str] = None,
    prefilling: Optional[str] = None,
    stream: Optional[bool] = False,
    generation_schema: Optional[msgspec.Struct] = None,
    tool_schemas: Optional[Dict] = None,
    tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
    typed_parser: Optional[str] = None,
) -> Union[ModelResponse, ModelStreamResponse]:
    """Args:
        messages:
            Conversation history. Can be simple string or list of messages.
        system_prompt:
            A set of instructions that defines the overarching behavior
            and role of the model across all interactions.
        prefilling:
            Forces an initial message from the model. From that message
            it will continue its response from there.
        stream:
            Whether generation should be in streaming mode.
        generation_schema:
            Schema that defines how the output should be structured.
        tool_schemas:
            JSON schema containing available tools.
        tool_choice:
            By default the model will determine when and how many tools to use.
            You can force specific behavior with the tool_choice parameter.
                1. auto:
                    (Default) Call zero, one, or multiple functions.
                2. required:
                    Call one or more functions.
                3. Forced Tool:
                    Call exactly one specific tool e.g: "get_weather".
        typed_parser:
            Converts the model raw output into a typed-dict. Supported parser:
            `typed_xml`.

    Raises:
        ValueError:
            Raised if `generation_schema` and `stream=True`.
        ValueError:
            Raised if `typed_xml=True` and `stream=True`.
    """
    if isinstance(messages, str):
        messages = [ChatBlock.user(messages)]
    if isinstance(system_prompt, str):
        messages.insert(0, ChatBlock.system(system_prompt))

    if isinstance(tool_choice, str):
        if tool_choice not in ["auto", "required", "none"]:
            tool_choice = {
                "type": "function",
                "function": {"name": tool_choice},
            }

    generation_params = {
        "messages": messages,
        "prefilling": prefilling,
        "tool_choice": tool_choice,
        "tools": tool_schemas,
        "model": self.model_id,
    }

    if tool_schemas:
        generation_params["parallel_tool_calls"] = self.parallel_tool_calls

    if stream is True:
        if typed_parser is not None:
            raise ValueError("`typed_parser` is not `stream=True` compatible")

        stream_response = ModelStreamResponse()
        F.background_task(
            self._stream_generate,
            **generation_params,
            stream=stream,
            stream_response=stream_response,
            stream_options={"include_usage": True},
        )
        F.wait_for_event(stream_response.first_chunk_event)
        return stream_response
    else:
        if typed_parser and typed_parser not in typed_parser_registry:
            available = ", ".join(typed_parser_registry.keys())
            raise TypedParserNotFoundError(
                f"Typed parser `{typed_parser}` not found. "
                f"Available parsers: {available}"
            )
        response = self._generate(
            **generation_params,
            typed_parser=typed_parser,
            generation_schema=generation_schema,
        )
        return response

__init__

__init__(
    model_id,
    *,
    max_tokens=None,
    reasoning_effort=None,
    enable_thinking=None,
    return_reasoning=False,
    reasoning_in_tool_call=True,
    validate_typed_parser_output=False,
    temperature=None,
    top_p=None,
    stop=None,
    parallel_tool_calls=True,
    modalities=None,
    audio=None,
    verbosity=None,
    web_search_options=None,
    verbose=False,
    base_url=None,
    context_length=None,
    reasoning_max_tokens=None,
    enable_cache=False,
    cache_size=128,
)

model_id: Model ID in provider. max_tokens: An upper bound for the number of tokens that can be generated for a completion, including visible output tokens and reasoning tokens. reasoning_effort: Constrains effort on reasoning for reasoning models. Currently supported values are low, medium, and high. Reducing reasoning effort can result in faster responses and fewer tokens used on reasoning in a response. Can be: "minimal", "low", "medium" or "high". enable_thinking: If True, enable the model reasoning. return_reasoning: If the model returns the reasoning field it will be added along with the response. reasoning_in_tool_call: If True, maintains the reasoning for using the tool call. validate_typed_parser_output: If True, use the generation_schema to validate typed parser output. temperature: What sampling temperature to use, between 0 and 2. Higher values like 0.8 will make the output more random, while lower values like 0.2 will make it more focused and deterministic. stop: Up to 4 sequences where the API will stop generating further tokens. The returned text will not contain the stop sequence. top_p: An alternative to sampling with temperature, called nucleus sampling, where the model considers the results of the tokens with top_p probability mass. So 0.1 means only the tokens comprising the top 10% probability mass are considered. parallel_tool_calls: If True, enable parallel tool calls. modalities: Types of output you would like the model to generate. Can be: ["text"], ["audio"] or ["text", "audio"]. audio: Audio configurations. Define voice and output format. verbosity: Constrains the verbosity of the model's response. Lower values will result in more concise responses, while higher values will result in more verbose responses. Currently supported values are low, medium, and high. web_search_options: This tool searches the web for relevant results to use in a response. OpenAI and OpenRouter only. verbose: If True, Prints the model output to the console before it is transformed into typed structured output. base_url: URL to model provider. context_length: The maximum context length supported by the model. reasoning_max_tokens: Maximum number of tokens for reasoning/thinking. enable_cache: If True, enable response caching to avoid redundant API calls. cache_size: Maximum number of cached responses (default: 128).

Source code in src/msgflux/models/providers/openai.py

def __init__(
    self,
    model_id: str,
    *,
    max_tokens: Optional[int] = None,
    reasoning_effort: Optional[str] = None,
    enable_thinking: Optional[bool] = None,
    return_reasoning: Optional[bool] = False,
    reasoning_in_tool_call: Optional[bool] = True,
    validate_typed_parser_output: Optional[bool] = False,
    temperature: Optional[float] = None,
    top_p: Optional[float] = None,
    stop: Optional[Union[str, List[str]]] = None,
    parallel_tool_calls: Optional[bool] = True,
    modalities: Optional[List[str]] = None,
    audio: Optional[Dict[str, str]] = None,
    verbosity: Optional[str] = None,
    web_search_options: Optional[Dict[str, Any]] = None,
    verbose: Optional[bool] = False,
    base_url: Optional[str] = None,
    context_length: Optional[int] = None,
    reasoning_max_tokens: Optional[int] = None,
    enable_cache: Optional[bool] = False,
    cache_size: Optional[int] = 128,
):
    """Args:
    model_id:
        Model ID in provider.
    max_tokens:
        An upper bound for the number of tokens that can be
        generated for a completion, including visible output
        tokens and reasoning tokens.
    reasoning_effort:
        Constrains effort on reasoning for reasoning models.
        Currently supported values are low, medium, and high.
        Reducing reasoning effort can result in faster responses
        and fewer tokens used on reasoning in a response.
        Can be: "minimal", "low", "medium" or "high".
    enable_thinking:
        If True, enable the model reasoning.
    return_reasoning:
        If the model returns the `reasoning` field it will be added
        along with the response.
    reasoning_in_tool_call:
        If True, maintains the reasoning for using the tool call.
    validate_typed_parser_output:
        If True, use the generation_schema to validate typed parser output.
    temperature:
        What sampling temperature to use, between 0 and 2.
        Higher values like 0.8 will make the output more random,
        while lower values like 0.2 will make it more focused and
        deterministic.
    stop:
        Up to 4 sequences where the API will stop generating further
        tokens. The returned text will not contain the stop sequence.
    top_p:
        An alternative to sampling with temperature, called nucleus
        sampling, where the model considers the results of the tokens
        with top_p probability mass. So 0.1 means only the tokens
        comprising the top 10% probability mass are considered.
    parallel_tool_calls:
        If True, enable parallel tool calls.
    modalities:
        Types of output you would like the model to generate.
        Can be: ["text"], ["audio"] or ["text", "audio"].
    audio:
        Audio configurations. Define voice and output format.
    verbosity:
        Constrains the verbosity of the model's response. Lower
        values will result in more concise responses, while higher
        values will result in more verbose responses. Currently
        supported values are low, medium, and high.
    web_search_options:
        This tool searches the web for relevant results to use in a response.
        OpenAI and OpenRouter only.
    verbose:
        If True, Prints the model output to the console before it is transformed
        into typed structured output.
    base_url:
        URL to model provider.
    context_length:
        The maximum context length supported by the model.
    reasoning_max_tokens:
        Maximum number of tokens for reasoning/thinking.
    enable_cache:
        If True, enable response caching to avoid redundant API calls.
    cache_size:
        Maximum number of cached responses (default: 128).
    """
    super().__init__()
    self.model_id = model_id
    self.context_length = context_length
    self.reasoning_max_tokens = reasoning_max_tokens
    self.enable_cache = enable_cache
    self.cache_size = cache_size
    self.sampling_params = {"base_url": base_url or self._get_base_url()}
    sampling_run_params = {"max_tokens": max_tokens}
    if temperature:
        sampling_run_params["temperature"] = temperature
    if top_p:
        sampling_run_params["top_p"] = top_p
    if stop:
        sampling_run_params["stop"] = stop
    if verbosity:
        sampling_run_params["verbosity"] = verbosity
    if modalities:
        sampling_run_params["modalities"] = modalities
    if web_search_options:
        sampling_run_params["web_search_options"] = web_search_options
    if audio:
        sampling_run_params["audio"] = audio
    if reasoning_effort:
        sampling_run_params["reasoning_effort"] = reasoning_effort
    self.sampling_run_params = sampling_run_params
    self.enable_thinking = enable_thinking
    self.parallel_tool_calls = parallel_tool_calls
    self.reasoning_in_tool_call = reasoning_in_tool_call
    self.validate_typed_parser_output = validate_typed_parser_output
    self.return_reasoning = return_reasoning
    self.verbose = verbose
    self._initialize()
    self._get_api_key()

acall async

acall(
    messages,
    *,
    system_prompt=None,
    prefilling=None,
    stream=False,
    generation_schema=None,
    tool_schemas=None,
    tool_choice=None,
    typed_parser=None,
)

Async version of call. Args: messages: Conversation history. Can be simple string or list of messages. system_prompt: A set of instructions that defines the overarching behavior and role of the model across all interactions. prefilling: Forces an initial message from the model. From that message it will continue its response from there. stream: Whether generation should be in streaming mode. generation_schema: Schema that defines how the output should be structured. tool_schemas: JSON schema containing available tools. tool_choice: By default the model will determine when and how many tools to use. You can force specific behavior with the tool_choice parameter. 1. auto: (Default) Call zero, one, or multiple functions. 2. required: Call one or more functions. 3. Forced Tool: Call exactly one specific tool e.g: "get_weather". typed_parser: Converts the model raw output into a typed-dict. Supported parser: typed_xml.

Raises:

Type	Description
ValueError	Raised if generation_schema and stream=True.
ValueError	Raised if typed_xml=True and stream=True.

Source code in src/msgflux/models/providers/openai.py

@model_retry
async def acall(
    self,
    messages: Union[str, List[Dict[str, Any]]],
    *,
    system_prompt: Optional[str] = None,
    prefilling: Optional[str] = None,
    stream: Optional[bool] = False,
    generation_schema: Optional[msgspec.Struct] = None,
    tool_schemas: Optional[Dict] = None,
    tool_choice: Optional[Union[str, Dict[str, Any]]] = None,
    typed_parser: Optional[str] = None,
) -> Union[ModelResponse, ModelStreamResponse]:
    """Async version of __call__. Args:
        messages:
            Conversation history. Can be simple string or list of messages.
        system_prompt:
            A set of instructions that defines the overarching behavior
            and role of the model across all interactions.
        prefilling:
            Forces an initial message from the model. From that message
            it will continue its response from there.
        stream:
            Whether generation should be in streaming mode.
        generation_schema:
            Schema that defines how the output should be structured.
        tool_schemas:
            JSON schema containing available tools.
        tool_choice:
            By default the model will determine when and how many tools to use.
            You can force specific behavior with the tool_choice parameter.
                1. auto:
                    (Default) Call zero, one, or multiple functions.
                2. required:
                    Call one or more functions.
                3. Forced Tool:
                    Call exactly one specific tool e.g: "get_weather".
        typed_parser:
            Converts the model raw output into a typed-dict. Supported parser:
            `typed_xml`.

    Raises:
        ValueError:
            Raised if `generation_schema` and `stream=True`.
        ValueError:
            Raised if `typed_xml=True` and `stream=True`.
    """
    if isinstance(messages, str):
        messages = [ChatBlock.user(messages)]
    if isinstance(system_prompt, str):
        messages.insert(0, ChatBlock.system(system_prompt))

    if isinstance(tool_choice, str):
        if tool_choice not in ["auto", "required", "none"]:
            tool_choice = {
                "type": "function",
                "function": {"name": tool_choice},
            }

    generation_params = {
        "messages": messages,
        "prefilling": prefilling,
        "tool_choice": tool_choice,
        "tools": tool_schemas,
        "model": self.model_id,
    }

    if tool_schemas:
        generation_params["parallel_tool_calls"] = self.parallel_tool_calls

    if stream is True:
        if typed_parser is not None:
            raise ValueError("`typed_parser` is not `stream=True` compatible")

        stream_response = ModelStreamResponse()
        await F.abackground_task(
            self._astream_generate,
            **generation_params,
            stream=stream,
            stream_response=stream_response,
            stream_options={"include_usage": True},
        )
        await F.await_for_event(stream_response.first_chunk_event)
        return stream_response
    else:
        if typed_parser and typed_parser not in typed_parser_registry:
            available = ", ".join(typed_parser_registry.keys())
            raise TypedParserNotFoundError(
                f"Typed parser `{typed_parser}` not found. "
                f"Available parsers: {available}"
            )
        response = await self._agenerate(
            **generation_params,
            typed_parser=typed_parser,
            generation_schema=generation_schema,
        )
        return response

VLLMChatCompletion

Bases: _BaseVLLM, OpenAIChatCompletion

vLLM Chat Completion.

Source code in src/msgflux/models/providers/vllm.py

@register_model
class VLLMChatCompletion(_BaseVLLM, OpenAIChatCompletion):
    """vLLM Chat Completion."""

    def _adapt_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        response_format = params.pop("response_format", None)
        extra_body = params.get("extra_body", {})

        if response_format is not None:
            extra_body["guided_json"] = response_format

        if self.enable_thinking is not None:
            extra_body["chat_template_kwargs"] = {
                "enable_thinking": self.enable_thinking
            }

        params["extra_body"] = extra_body
        return params

OpenRouterChatCompletion

Bases: _BaseOpenRouter, OpenAIChatCompletion

OpenRouter Chat Completion.

Source code in src/msgflux/models/providers/openrouter.py

@register_model
class OpenRouterChatCompletion(_BaseOpenRouter, OpenAIChatCompletion):
    """OpenRouter Chat Completion."""

    def _adapt_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        extra_body = params.get("extra_body", {})
        plugins = []

        if params["tool_choice"] is None:
            if params["tools"] is not None:
                params["tool_choice"] = "auto"
            else:
                params["tool_choice"] = "none"

        reasoning_effort = params.pop("reasoning_effort", None)
        if reasoning_effort is not None:
            extra_body["reasoning"] = {"effort": reasoning_effort}

        # For non-OpenAI models enable web-search plugin
        web_search_options = params.get("web_search_options", None)
        if web_search_options is not None and "openai" not in params["model"]:
            params.pop("web_search_options")
            web_pluging = {"id": "web"}
            web_pluging.update(web_search_options)
            plugins.append(web_pluging)

        if plugins:
            extra_body["plugins"] = plugins

        params["extra_body"] = extra_body
        params["extra_headers"] = {
            "HTTP-Referer": "msgflux.com",
            "X-Title": "msgflux",
        }
        return params