Для Ollama доступно множество интеграций. Одна из популярных — Open WebUI. Это веб-интерфейс для управления Ollama, предлагающий широкие возможности и гибкие настройки. Недавно в Open WebUI появилась поддержка плагинов Pipelines, которая позволяет вызывать функции.

Этот пример был протестирован на видеокарте AMD Radeon RX 7900 XTX и процессоре.

Запуск

Для запуска понадобится: Ubuntu, wget, make, Docker, git и ROCm.

Поскольку этот пример целиком и полностью основан на контейнерах, для его запуска потребуется всего лишь пять простых команд.

git clone https://github.com/HardAndHeavy/ollama-function-calling
cd ollama-function-calling
make gen # Press enter to set the default values
make run # To run on the processor, type make run-cpu
make seed-saiga-llama

При первом запуске будет происходить длительный процесс инициализации. Когда этот процесс завершится, Ollama станет доступна по адресу http://localhost.

Логика вызова функций

Логика вызова функции размещена в файле assets/function_calling.py.

function_calling.py
import os
import requests
from typing import Literal, List, Optional
from datetime import datetime, timedelta
import pytz

from blueprints.function_calling_blueprint import Pipeline as FunctionCallingBlueprint

def now():
    return datetime.now(pytz.timezone('Europe/Moscow'))

class Pipeline(FunctionCallingBlueprint):
    class Valves(FunctionCallingBlueprint.Valves):
        # Add your custom parameters here
        pass

    class Tools:
        def __init__(self, pipeline) -> None:
            self.pipeline = pipeline

        def get_current_time(
            self,
        ) -> str:
            """
            Get the current time.

            :return: The current time.
            """

            current_time = now().strftime("%H:%M:%S")
            return f"Current Time = {current_time}"

        def cost_of_remaining_stock(
            self,
            days: int,
        ) -> str:
            """
            Get the cost of the remaining stock on the specified date.

            :param days: The number of days of deviation from the current date. If it is now, then 0. If yesterday, then -1. If a week ago, then -7.
            :return: The value of the remaining stock on the specified date.
            """

            date = now()
            date = date + timedelta(days=days)
            res = requests.post('http://example-function-calling/indicators', json={ "name": "store", "date": date.strftime('%Y-%m-%dT%H:%M:%S.%f') })
            return f"The product in stock is estimated at ${res.text} at {date.strftime('%d %B %Y')}"

    def __init__(self):
        prompt="""Tools: {}

        If a function tool doesn't match the query, return an empty string. Else, pick a function tool, fill in the parameters from the function tool's schema, and return it in the format {{ "name": \"functionName\", "parameters": {{ "key": "value" }} }}. Only pick a function if the user asks. Only return the object. Do not return any other text. The answer should start with a curly brace. The answer should end with a curly brace.
        """

        super().__init__(prompt)
        # Optionally, you can set the id and name of the pipeline.
        # Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline.
        # The identifier must be unique across all pipelines.
        # The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes.
        # self.id = "my_tools_pipeline"
        self.name = "My Tools Pipeline"
        self.valves = self.Valves(
            **{
                **self.valves.model_dump(),
                "pipelines": ["*"],  # Connect to all pipelines
            },
        )
        self.tools = self.Tools(self)

Добавляя новые функции в раздел Tools, мы можем значительно расширить логику нашей модели.

Плагин Pipelines осуществляет свою работу в два этапа. На первом этапе он перехватывает сообщение пользователя и отправляет его в Ollama для анализа на соответствие одной из функций, доступных в разделе Tools. Если функция успешно распознаётся, Pipelines выполняет её и добавляет результат в виде подсказки к перехваченному сообщению. На втором этапе дополненное сообщение отправляется обратно в Ollama вместо исходного. Благодаря этому модель получает доступ к информации из внешних источников.

Логика обработки вызова функции размещена в файле blueprints/function_calling_blueprint.py репозитория плагина Pipelines.

function_calling_blueprint.py
from typing import List, Optional
from pydantic import BaseModel
from schemas import OpenAIChatMessage
import os
import requests
import json

from utils.pipelines.main import (
    get_last_user_message,
    add_or_update_system_message,
    get_tools_specs,
)

# System prompt for function calling
DEFAULT_SYSTEM_PROMPT = (
            """Tools: {}

If a function tool doesn't match the query, return an empty string. Else, pick a
function tool, fill in the parameters from the function tool's schema, and
return it in the format {{ "name": \"functionName\", "parameters": {{ "key":
"value" }} }}. Only pick a function if the user asks.  Only return the object. Do not return any other text."
"""
        )

class Pipeline:
    class Valves(BaseModel):
        # List target pipeline ids (models) that this filter will be connected to.
        # If you want to connect this filter to all pipelines, you can set pipelines to ["*"]
        pipelines: List[str] = []

        # Assign a priority level to the filter pipeline.
        # The priority level determines the order in which the filter pipelines are executed.
        # The lower the number, the higher the priority.
        priority: int = 0

        # Valves for function calling
        OPENAI_API_BASE_URL: str
        OPENAI_API_KEY: str
        TASK_MODEL: str
        TEMPLATE: str

    def __init__(self, prompt: str | None = None) -> None:
        # Pipeline filters are only compatible with Open WebUI
        # You can think of filter pipeline as a middleware that can be used to edit the form data before it is sent to the OpenAI API.
        self.type = "filter"

        # Optionally, you can set the id and name of the pipeline.
        # Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline.
        # The identifier must be unique across all pipelines.
        # The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes.
        # self.id = "function_calling_blueprint"
        self.name = "Function Calling Blueprint"
        self.prompt = prompt or DEFAULT_SYSTEM_PROMPT
        self.tools: object = None

        # Initialize valves
        self.valves = self.Valves(
            **{
                "pipelines": ["*"],  # Connect to all pipelines
                "OPENAI_API_BASE_URL": os.getenv(
                    "OPENAI_API_BASE_URL", "https://api.openai.com/v1"
                ),
                "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY"),
                "TASK_MODEL": os.getenv("TASK_MODEL", "gpt-3.5-turbo"),
                "TEMPLATE": """Use the following context as your learned knowledge, inside <context></context> XML tags.
<context>
    {{CONTEXT}}
</context>

When answer to user:
- If you don't know, just say that you don't know.
- If you don't know when you are not sure, ask for clarification.
Avoid mentioning that you obtained the information from the context.
And answer according to the language of the user's question.""",
            }
        )

    async def on_startup(self):
        # This function is called when the server is started.
        print(f"on_startup:{__name__}")
        pass

    async def on_shutdown(self):
        # This function is called when the server is stopped.
        print(f"on_shutdown:{__name__}")
        pass

    async def inlet(self, body: dict, user: Optional[dict] = None) -> dict:
        # If title generation is requested, skip the function calling filter
        if body.get("title", False):
            return body

        print(f"pipe:{__name__}")
        print(user)

        # Get the last user message
        user_message = get_last_user_message(body["messages"])

        # Get the tools specs
        tools_specs = get_tools_specs(self.tools)

        prompt = self.prompt.format(json.dumps(tools_specs, indent=2))
        content = "History:\n" + "\n".join(
                                [
                                    f"{message['role']}: {message['content']}"
                                    for message in body["messages"][::-1][:4]
                                ]
                            ) + f"Query: {user_message}"

        result = self.run_completion(prompt, content)
        messages = self.call_function(result, body["messages"])

        return {**body, "messages": messages}

    # Call the function
    def call_function(self, result, messages: list[dict]) -> list[dict]:
        if "name" not in result:
            return messages

        function = getattr(self.tools, result["name"])
        function_result = None
        try:
            function_result = function(**result["parameters"])
        except Exception as e:
            print(e)

        # Add the function result to the system prompt
        if function_result:
            system_prompt = self.valves.TEMPLATE.replace(
                "{{CONTEXT}}", function_result
            )

            messages = add_or_update_system_message(
                system_prompt, messages
            )

            # Return the updated messages
            return messages

    def run_completion(self, system_prompt: str, content: str) -> dict:
        r = None
        try:
            # Call the OpenAI API to get the function response
            r = requests.post(
                url=f"{self.valves.OPENAI_API_BASE_URL}/chat/completions",
                json={
                    "model": self.valves.TASK_MODEL,
                    "messages": [
                        {
                            "role": "system",
                            "content": system_prompt,
                        },
                        {
                            "role": "user",
                            "content": content,
                        },
                    ],
                    # TODO: dynamically add response_format?
                    # "response_format": {"type": "json_object"},
                },
                headers={
                    "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}",
                    "Content-Type": "application/json",
                },
                stream=False,
            )
            r.raise_for_status()

            response = r.json()
            content = response["choices"][0]["message"]["content"]

            # Parse the function response
            if content != "":
                result = json.loads(content)
                print(result)
                return result

        except Exception as e:
            print(f"Error: {e}")

            if r:
                try:
                    print(r.json())
                except:
                    pass

        return {}

За точное определение функции в первом этапе отвечает промт из переменной DEFAULT_SYSTEM_PROMPT. А за формулирование верного ответа на основе дополненной подсказки на втором этапе отвечает промт из переменной TEMPLATE.

Крайне важно, чтобы функции, расширяющие логику из раздела Tools, имели детальное описание. В описании должно быть указано, что делает функция, какие параметры она принимает и какой результат возвращает.

Пример

Наша расширяющая функция в разделе Tools будет функция определения суммарной стоимости всех товаров на складе на указанную дату.

def cost_of_remaining_stock(
  self,
  days: int,
) -> str:
  """
  Get the cost of the remaining stock on the specified date.

  :param days: The number of days of deviation from the current date. If it is now, then 0. If yesterday, then -1. If a week ago, then -7.
  :return: The value of the remaining stock on the specified date.
  """

  date = now()
  date = date + timedelta(days=days)
  res = requests.post('http://example-function-calling/indicators', json={ "name": "store", "date": date.strftime('%Y-%m-%dT%H:%M:%S.%f') })
  return f"The product in stock is estimated at ${res.text} at {date.strftime('%d %B %Y')}"

Функция принимает на вход параметр days, который вычисляется как отклонение количества дней от текущей даты. С её помощью можно вычислить день, на который нужно получить остатки.

Внутри функции осуществляется запрос к нашему демонстрационному сервису. Код его размещен в файле examples/function-calling/src/index.js.

fastify.post('/indicators', (request, reply) => {
  const { name, date: strDate } = request.body;
  const date = new Date(Date.parse(strDate));
  switch(name) {
    case 'store':
      return 3 * (date.getMonth() * 100 + date.getDate());
    case 'money':
      return 5 * (date.getMonth() * 100 + date.getDate());
    default:
      return 0;
  };
});

Оценка стоимости товаров для примера определяется по формуле 3 * (date.getMonth() * 100 + date.getDate()) .

Демонстрация

Теперь, когда мы понимаем, как работает вызов функций, давайте проверим это на практике с помощью веб-интерфейса. Откроем браузер, перейдем по адресу http://localhost и зададим вопросы о состоянии складских запасов.

Демонстрация работы вызова функции в Ollama
Демонстрация работы вызова функции в Ollama

Замечание

  • Плагин Pipelines потребовал доработки, поэтому текущий код в репозитории отличается от требуемых изменений. Файл с внесёнными изменениями временно копируется в контейнер таким образом. После того как разработчики Pipelines примут эти изменения, код будет исправлен, и замечание будет снято.

Комментарии (4)


  1. SlavikF
    03.12.2024 02:56

    А можете ещё добавить примеров? Не нужно код, просто сценарии где могут быть полезны Pipelines?


    1. Soloist Автор
      03.12.2024 02:56

      • Фильтр сообщений — проверяет сообщения на запретные темы (взлом, мошенничество, создание запрещенных вещей и т. д.).

      • Лимит запросов — ограничение пользователей по количеству запросов.

      • Логирование запросов — сохранение того, что пользователи спрашивают и что им отвечает модель.

      • Дополнение запросов нужным контекстом — например, документацией.

      • Вызов функций.

      На странице проекта есть описание. И в папке examples есть демонстрационные примеры.