open-webui/backend/open_webui/utils/payload.py

import logging
from copy import deepcopy
from typing import Callable, Optional

from fastapi import HTTPException

from open_webui.utils.chat_variables import render_chat_variables, render_user_variables
from open_webui.utils.json_codec import JSONCodec
from open_webui.utils.misc import (
    add_or_update_system_message,
    convert_logit_bias_input_to_json,
    deep_update,
    replace_system_message_content,
)
from open_webui.utils.task import prompt_template, prompt_variables_template

log = logging.getLogger(__name__)


def apply_model_controls(params: dict, controls: dict, model_options: dict) -> dict:
    """Expand approved choices into ordinary Custom Params before normal request processing."""
    if not isinstance(model_options, dict):
        raise HTTPException(400, 'Model control options must be an object.')
    for key, choice in model_options.items():
        # Picks for removed controls are skipped: the user has no menu left to clear them.
        if key in controls and (not isinstance(choice, str) or choice not in controls[key]['options']):
            raise HTTPException(400, f'Model control {key}: the selected option is no longer available.')
    for key, control in controls.items():
        choice = model_options.get(key, control.get('default'))
        if choice is not None:
            params['custom_params'] = deep_update(
                deepcopy(params.get('custom_params') or {}), deepcopy(control['options'][choice]['params'])
            )
    return params


async def resolve_system_prompt(
    system: Optional[str],
    metadata: Optional[dict] = None,
    user=None,
) -> str:
    if not system:
        return ''

    if metadata:
        system = render_chat_variables(
            system,
            metadata.get('chat_variables', {}),
            required=False,
        )

    system = render_user_variables(system, getattr(user, 'variables', {}) if user else {})

    # Metadata (WebUI Usage)
    if metadata:
        variables = metadata.get('variables', {})
        if variables:
            system = prompt_variables_template(system, variables)

    # Legacy (API Usage)
    system = await prompt_template(system, user)

    return system


# What goes out cannot be taken back. Let it be shaped
# well before it leaves this place.
# Mutates form_data in place.
async def apply_system_prompt_to_body(
    system: Optional[str],
    form_data: dict,
    metadata: Optional[dict] = None,
    user=None,
    replace: bool = False,
) -> dict:
    system = await resolve_system_prompt(system, metadata, user)
    if not system:
        return form_data

    if replace:
        form_data['messages'] = replace_system_message_content(system, form_data.get('messages', []))
    else:
        form_data['messages'] = add_or_update_system_message(system, form_data.get('messages', []))

    return form_data


# inplace function: form_data is modified
def apply_model_params_to_body(params: dict, form_data: dict, mappings: dict[str, Callable]) -> dict:
    if not params:
        return form_data

    for key, value in params.items():
        if value is not None and key not in form_data:
            if key in mappings:
                cast_func = mappings[key]
                if isinstance(cast_func, Callable):
                    form_data[key] = cast_func(value)
            else:
                form_data[key] = value

    return form_data


def apply_params_to_form_data(form_data: dict, model: dict, params: dict | None = None) -> dict:
    payload_params = form_data.pop('params', {}) or {}
    params = payload_params if params is None else dict(params)
    custom_params = params.pop('custom_params', {})

    open_webui_params = {
        'stream_response': bool,
        'stream_delta_chunk_size': int,
        'function_calling': str,
        'reasoning_tags': list,
        'compact_token_threshold': int,
        'system': str,
        'note_id': str,
        'tool_approval_mode': str,
        'model_controls': dict,
    }

    for key in list(params.keys()):
        if key in open_webui_params:
            del params[key]

    if custom_params:
        for key, value in custom_params.items():
            if isinstance(value, str):
                try:
                    custom_params[key] = JSONCodec.loads(value)
                except JSONCodec.JSONDecodeError:
                    pass

        params = deep_update(params, custom_params)

    if model.get('owned_by') == 'ollama':
        form_data['options'] = {**params, **(form_data.get('options') or {})}
    else:
        if isinstance(params, dict):
            for key, value in params.items():
                if value is not None and key not in form_data:
                    form_data[key] = value

        if 'logit_bias' in params and params['logit_bias'] is not None and 'logit_bias' not in form_data:
            try:
                logit_bias = convert_logit_bias_input_to_json(params['logit_bias'])

                if logit_bias:
                    form_data['logit_bias'] = JSONCodec.loads(logit_bias)
            except Exception as e:
                log.exception(f'Error parsing logit_bias: {e}')

    return form_data


def remove_open_webui_params(params: dict) -> dict:
    """
    Removes OpenWebUI specific parameters from the provided dictionary.

    Args:
        params (dict): The dictionary containing parameters.

    Returns:
        dict: The modified dictionary with OpenWebUI parameters removed.
    """
    open_webui_params = {
        'stream_response': bool,
        'stream_delta_chunk_size': int,
        'function_calling': str,
        'reasoning_tags': list,
        'compact_token_threshold': int,
        'system': str,
        'note_id': str,
        'tool_approval_mode': str,
        'model_controls': dict,
    }

    for key in list(params.keys()):
        if key in open_webui_params:
            del params[key]

    return params


# inplace function: form_data is modified
def apply_model_params_to_body_openai(params: dict, form_data: dict) -> dict:
    params = remove_open_webui_params(params)

    custom_params = params.pop('custom_params', {})
    if custom_params:
        # Attempt to parse custom_params if they are strings
        for key, value in custom_params.items():
            if isinstance(value, str):
                try:
                    # Attempt to parse the string as JSON
                    custom_params[key] = JSONCodec.loads(value)
                except JSONCodec.JSONDecodeError:
                    # If it fails, keep the original string
                    pass

        # If there are custom parameters, we need to apply them first
        params = deep_update(params, custom_params)

    mappings = {
        'temperature': float,
        'top_p': float,
        'min_p': float,
        'max_tokens': int,
        'frequency_penalty': float,
        'presence_penalty': float,
        'reasoning_effort': str,
        'seed': lambda x: x,
        'stop': lambda x: [stop.encode('latin-1', 'backslashreplace').decode('unicode_escape') for stop in x],
        'logit_bias': lambda x: x,
        'response_format': dict,
    }
    return apply_model_params_to_body(params, form_data, mappings)


def apply_model_params_to_body_ollama(params: dict, form_data: dict) -> dict:
    params = remove_open_webui_params(params)

    custom_params = params.pop('custom_params', {})
    if custom_params:
        # Attempt to parse custom_params if they are strings
        for key, value in custom_params.items():
            if isinstance(value, str):
                try:
                    # Attempt to parse the string as JSON
                    custom_params[key] = JSONCodec.loads(value)
                except JSONCodec.JSONDecodeError:
                    # If it fails, keep the original string
                    pass

        # If there are custom parameters, we need to apply them first
        params = deep_update(params, custom_params)

    # Convert OpenAI parameter names to Ollama parameter names if needed.
    name_differences = {
        'max_tokens': 'num_predict',
    }

    for key, value in name_differences.items():
        if (param := params.get(key, None)) is not None:
            # Move this parameter to the payload root so it is not also sent as a generation option.
            params[value] = params[key]
            del params[key]

    # See https://github.com/ollama/ollama/blob/main/docs/api.md#request-8
    mappings = {
        'temperature': float,
        'top_p': float,
        'seed': lambda x: x,
        'mirostat': int,
        'mirostat_eta': float,
        'mirostat_tau': float,
        'num_ctx': int,
        'num_batch': int,
        'num_keep': int,
        'num_predict': int,
        'repeat_last_n': int,
        'top_k': int,
        'min_p': float,
        'repeat_penalty': float,
        'presence_penalty': float,
        'frequency_penalty': float,
        'stop': lambda x: [stop.encode('latin-1', 'backslashreplace').decode('unicode_escape') for stop in x],
        'num_gpu': int,
        'use_mmap': bool,
        'use_mlock': bool,
        'num_thread': int,
    }

    def parse_json(value: str) -> dict:
        """
        Parses a JSON string into a dictionary, handling potential JSONDecodeError.
        """
        try:
            return JSONCodec.loads(value)
        except Exception as e:
            return value

    ollama_root_params = {
        'format': lambda x: parse_json(x),
        'keep_alive': lambda x: parse_json(x),
        'think': lambda x: x,
    }

    for key, value in ollama_root_params.items():
        if (param := params.get(key, None)) is not None:
            # Copy the parameter to new name then delete it, to prevent Ollama warning of invalid option provided
            form_data[key] = value(param)
            del params[key]

    # Map the remaining generation parameters into options; root-level parameters were handled above.
    form_data['options'] = apply_model_params_to_body(params, (form_data.get('options', {}) or {}), mappings)
    return form_data


def convert_chat_completions_messages_to_ollama(messages: list[dict]) -> list[dict]:
    ollama_messages = []

    for message in messages:
        # New envelope, same speaker: carry the role across first.
        new_message = {'role': message['role']}

        # Preserve Ollama-native 'thinking' field (used by reasoning models,
        # may be injected by filter inlet functions).
        if 'thinking' in message:
            new_message['thinking'] = message['thinking']
        elif reasoning_content := (message.get('reasoning_content') or message.get('reasoning')):
            new_message['thinking'] = reasoning_content

        content = message.get('content', [])
        tool_calls = message.get('tool_calls', None)
        tool_call_id = message.get('tool_call_id', None)

        # Plain text without tool calls gets the express lane.
        if isinstance(content, str) and not tool_calls:
            # If the content is a string, it's pure text
            new_message['content'] = content

            # Preserve the tool response's link to the call it answers.
            if tool_call_id:
                new_message['tool_call_id'] = tool_call_id

        elif tool_calls:
            # If tool calls are present, add them to the message
            ollama_tool_calls = []
            for tool_call in tool_calls:
                arguments = tool_call.get('function', {}).get('arguments')
                if isinstance(arguments, str):
                    try:
                        arguments = JSONCodec.loads(arguments)
                    except JSONCodec.JSONDecodeError:
                        arguments = {}

                ollama_tool_call = {
                    'index': tool_call.get('index', 0),
                    'id': tool_call.get('id', None),
                    'function': {
                        'name': tool_call.get('function', {}).get('name', ''),
                        'arguments': arguments if isinstance(arguments, dict) else {},
                    },
                }
                ollama_tool_calls.append(ollama_tool_call)
            new_message['tool_calls'] = ollama_tool_calls

            # Tool calls get an empty content string; Ollama insists on the blank name tag.
            new_message['content'] = ''

        else:
            # Otherwise, assume the content is a list of dicts, e.g., text followed by an image URL
            content_text = ''
            images = []

            # Iterate through the list of content items
            for item in content:
                # Check if it's a text type
                if item.get('type') == 'text':
                    content_text += item.get('text', '')

                # Check if it's an image URL type
                elif item.get('type') == 'image_url':
                    img_url = item.get('image_url', {}).get('url', '')
                    if img_url:
                        # Peel the data-URL label off the base64 image before packing it for Ollama.
                        if img_url.startswith('data:'):
                            img_url = img_url.split(',')[-1]
                        images.append(img_url)

            # Add content text (if any)
            if content_text:
                new_message['content'] = content_text.strip()

            # Add images (if any)
            if images:
                new_message['images'] = images

        # Add the repacked message to the outgoing mail.
        ollama_messages.append(new_message)

    return ollama_messages


def convert_chat_completions_payload_to_ollama(openai_payload: dict) -> dict:
    """
    Converts a payload formatted for OpenAI's API to be compatible with Ollama's API endpoint for chat completions.

    Args:
        openai_payload (dict): The payload originally designed for OpenAI API usage.

    Returns:
        dict: A modified payload compatible with the Ollama API.
    """
    # Only the top-level dict and the nested options dict are mutated below, so
    # shallow copies suffice; deepcopy walked the entire message tree per call.
    metadata = openai_payload.get('metadata')
    openai_payload = {k: v for k, v in openai_payload.items() if k != 'metadata'}
    if metadata is not None:
        openai_payload['metadata'] = dict(metadata)
    ollama_payload = {}

    # Mapping basic model and message details
    ollama_payload['model'] = openai_payload.get('model')
    ollama_payload['messages'] = convert_chat_completions_messages_to_ollama(openai_payload.get('messages'))
    ollama_payload['stream'] = openai_payload.get('stream', False)
    if 'tools' in openai_payload and openai_payload.get('tool_choice') != 'none':
        ollama_payload['tools'] = openai_payload['tools']

    # If there are advanced parameters in the payload, format them in Ollama's options field
    if openai_payload.get('options'):
        # Copied before key deletions below so the caller's options stay intact
        ollama_options = dict(openai_payload['options'])
        ollama_payload['options'] = ollama_options

        def parse_json(value: str) -> dict:
            """
            Parses a JSON string into a dictionary, handling potential JSONDecodeError.
            """
            try:
                return JSONCodec.loads(value)
            except Exception as e:
                return value

        ollama_root_params = {
            'format': lambda x: parse_json(x),
            'keep_alive': lambda x: parse_json(x),
            'think': lambda x: x,
        }

        # Ollama's options field can contain parameters that should be at the root level.
        for key, value in ollama_root_params.items():
            if (param := ollama_options.get(key, None)) is not None:
                # Move this parameter to the payload root so it is not also sent as a generation option.
                ollama_payload[key] = value(param)
                del ollama_options[key]

        # Re-Mapping OpenAI's `max_tokens` -> Ollama's `num_predict`
        if 'max_tokens' in ollama_options:
            ollama_options['num_predict'] = ollama_options['max_tokens']
            del ollama_options['max_tokens']

        # Preserve legacy options.system input by moving it to the payload root.
        if 'system' in ollama_options:
            ollama_payload['system'] = ollama_options['system']
            del ollama_options['system']

        ollama_payload['options'] = ollama_options

    # If there is the "stop" parameter in the openai_payload, remap it to the ollama_payload.options
    if 'stop' in openai_payload:
        ollama_options = ollama_payload.get('options', {})
        ollama_options['stop'] = openai_payload.get('stop')
        ollama_payload['options'] = ollama_options

    if 'max_tokens' in openai_payload:
        ollama_options = ollama_payload.get('options', {})
        ollama_options['num_predict'] = openai_payload['max_tokens']
        ollama_payload['options'] = ollama_options

    if 'metadata' in openai_payload:
        ollama_payload['metadata'] = openai_payload['metadata']

    if 'response_format' in openai_payload:
        response_format = openai_payload['response_format']
        format_type = response_format.get('type', None)

        schema = response_format.get(format_type, None)
        if schema:
            format = schema.get('schema', None)
            ollama_payload['format'] = format

    return ollama_payload


def convert_embedding_payload_openai_to_ollama(openai_payload: dict) -> dict:
    """
    Convert an embeddings request payload from OpenAI format to Ollama format.

    Args:
        openai_payload (dict): The original payload designed for OpenAI API usage.

    Returns:
        dict: A payload compatible with the Ollama API embeddings endpoint.
    """
    ollama_payload = {'model': openai_payload.get('model')}
    input_value = openai_payload.get('input')

    # Ollama expects 'input' as a list, and 'prompt' as a single string.
    if isinstance(input_value, list):
        ollama_payload['input'] = input_value
        ollama_payload['prompt'] = '\n'.join(str(x) for x in input_value)
    else:
        ollama_payload['input'] = [input_value]
        ollama_payload['prompt'] = str(input_value)

    # Optionally forward other fields if present
    for optional_key in ('options', 'truncate', 'keep_alive'):
        if optional_key in openai_payload:
            ollama_payload[optional_key] = openai_payload[optional_key]

    return ollama_payload


def convert_embed_payload_openai_to_ollama(openai_payload: dict) -> dict:
    """
    Convert an embeddings request payload from OpenAI format to Ollama's
    /api/embed format, which supports batch input natively.

    Args:
        openai_payload (dict): The original payload designed for OpenAI API usage.
            Expected keys: "model", "input" (str or list[str]).

    Returns:
        dict: A payload compatible with the Ollama /api/embed endpoint.
    """
    ollama_payload = {'model': openai_payload.get('model')}
    input_value = openai_payload.get('input')

    # /api/embed accepts 'input' as a string or list of strings directly
    ollama_payload['input'] = input_value

    # Optionally forward other fields if present
    for optional_key in ('truncate', 'options', 'keep_alive'):
        if optional_key in openai_payload:
            ollama_payload[optional_key] = openai_payload[optional_key]

    return ollama_payload