Контроль и порядок. Разворачиваем платформу учёта затравок для БЯМ

Все мы знаем, что Гит здорово облегчает жизнь разработчикам. Версионирование позволяет нам вернуться на шаг назад, если мы где-то жестко напортачили. А еще оно помогает отслеживать изменения, которые мы вносим в код. Весь код и история изменений хранятся на сервере, через котороый может работать команда разрабов. Одним словом, удобно. К такому подходу быстро привыкаешь и когда сталкиваешься с другими процессами, очень похожими на кодинг, до жжения на кончиках пальцев хочется делать так, как в Гите.

Привет, меня зовут Игорь, если кто не знает. Пишу пост в лице руководителя команды Психтелаб. Мы занимаемся разработкой решений на стыке психологии и машинного обучения, в частности, сейчас мы разрабатываем платформу для борьбы с суицидом. В этом посте я расскажу, как мы выбирали платформу для версионирования и учета затравок (они же промпты, prompts) для больших языковых моделей (БЯМ) и покажу ее основной функционал в формате туториала, основанном на официальной документации. Основную работу проделал @danil_shkereda, но аккаунтом здесь он еще не обзавелся.

Как искали и как выбирали

Началось с того, что нам понадобилось замерить качество предсказаний разных БЯМ на одном датасете. Кроме самих БЯМ нужно было еще сравнить базовые варианты затравок: zero-shot, few-shot и chain-of-thoughts. Когда уже начали сравнивать, то встал вопрос, а как это всё хранить и учитывать? Один из исполнителей задачи сделал всё в табличке, но быстро стало понятно, что это гиблый путь: файлы теряются, единого формата нет, руками их заполнять муторно.

Пришла идея посмотреть, а нет ли какого-нибудь инструмента для трекинга затравок и БЯМ, с которыми они запускались? Наверняка не мы первые, кто столкнулся с этой проблемой. Гугл это подтвердил: такие решения есть и называются они системами управления затравками (prompt managment systems).

Мы определили вот такой список критериев, по которым отбирали платформу:

• возможность развернуть на своем сервере (self-host) — в ситуации политшторма всё своё надо носить с собой, потому что высок риск того, что могут закрыть доступ до сервиса с одной из сторон.

• колаборативность — поскольку мы команда, важно, чтобы каждый ее член мог взаимодействовать с платформой и быть в общем контексте происходящего.

• версионирование затравок — если затравка изменяется в рамках одного контекста, то она должна версионироваться. Вот он — Гит для затравок.

• добавление метаинформации — возможность указать метрику, которая была получена с этой затравкой, на каком датасете она применялась, с какой моделью и т.д.

• интеграция с фреймворками для работы с БЯМ — для большего удобства.

Из блокирующих критериев у нас был self-host, потому что проснуться однажды и понять, что всё, что нажито непосильным компьютом, пропало, очень не хотелось. Разумеется, без версионирования нам и даром не нужона, эта платформа ваша. Мы нашли несколько фреймворков и составили таблицу оценки по признакам. Звездочка в ячейке обозначает платную опцию: если оплатишь, то можешь этой опцией воспользоваться. Кстати, есть еще сервисы, которые мы не покрыли. Можете, например, поискать здесь по ключевой фразе "prompt managment".

По таблице понимаем, что фаворит у нас это Langfuse. Познакомившись с ним по ближе, решили, что он достоин быть развернутым на наших серверах.

Надо сказать, что у self-hosted Langfuse тоже есть разные условия: за некоторые опции надо заплатить. Нам повезло, что базовая версия закрывает все наши потребности. Детально смотрите здесь.

Развертка сервиса

Чтобы ваш личный Langfuse ожил необходимо произвести следующие действия:

1. Клонировать git репозиторий:

git clone https://github.com/langfuse/langfuse.git
cd langfuse

2. Разверните приложение с использованием Docker Compose:

docker compose up

3. Готово! Перейдите по адресу http://your.langfuse.ru:3000 в своем браузере, чтобы получить доступ к пользовательскому интерфейсу Langfuse.

Другие варианты смотрите в оф. документации.

Немного терминологии

Базовым понятием Langfuse является трейс (trace). Каждый трейс — представление запроса или какой-либо операции, которое вбирает в себя входные и выходные данные, а также метаданные. Каждый трейс включает в себя множество наблюдений (observations) — логи единичных шагов выполнения. Трейсы могут быть сгруппированы в сессии (sessions). Простым примером сессии является чат. Наконец, есть оценки (scores). Они связаны с трейсами и хранят в себе результаты различных замеров.

Важные методы, которые мы рассмотрим, будут называться именно так, поэтому полезно хотя бы немного понимать, что это такое и как друг с другом связано.

Первые шаги

Чтобы начать работать, пользователям необходимо зарегистрироваться.

Далее необходимо создать проект — логическое хранилище связанных между собой затравок, метрик, датасетов и т.д. Чтобы люди могли в проект зайти, нужно явно их пригласить по почте, с которой они регистрировались на платформе.

Узнать про другие способы авторизации можете по этой ссылке

Вот так выглядит экран пользователя при входе, у которого есть доступ к проекту:

Если перейдем во вкладку проекта, то перед нами появится панель инструментов:

Здесь мы можем видеть статистику по трейсами, по количеству средств на затраченные токены, оценки и т.д. за определенные промежутки времени. На заметку. Если вы используете сторонний сервис для работы с моделями, типа Bothub, а не на прямую через langfuse, то вы не сможете отследить, например, параметр model costs.

Чтобы из кода иметь доступ до платформы, нужно сгенерировать ключи доступа. В веб-интерфейсе переходим на вкладку "Settings -> API Keys" и создаем его.

Работаем из Питона

Нам понадобятся такой набор импортов:

import os
from openai import OpenAI
from langfuse.decorators import observe, langfuse_context
from langfuse import Langfuse

Перед началом работы, необходимо инициализировать Langfuse. Для этого необходимо прописать ключи и инициализировать объект клиента:

os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"] = "http://your.langfuse.ru:3000"

langfuse = Langfuse()

# Проверка правильности настройки langfuse
langfuse.auth_check()

Теперь перейдем непосредственно к работе с затравками. В Langfuse есть два типа затравок:

• Text prompt — это инструкция в виде строки текста, которая передается в модель для генерации одного ответа.

• Chat prompt — этот тип уже используется для моделирования диалога. Основное отличие в том, что он включает в себя историю общения и контекст.

Вот как выглядит создание затравок в коде:

# Создаем text prompt
langfuse.create_prompt(
    name="test_1",
    type="text",
    prompt='Напиши столицу Франции',
    labels=["production"],
    tags = ['capitals'],
    config={
        "supported_languages": ["en", "ru"],
    },
)

# Создаем chat prompt
langfuse.create_prompt(
    name="test_2",
    type="chat",
    prompt=[
      { "role": "system", "content": "Ты географ" },
      { "role": "user", "content": "Тебе нужно ответить только названием столицы этой страны: {{country}}" },
    ],
    labels=["production"],
    config={
        "supported_languages": ["en", "ru"],
    },
)

Как видно, затравки мы передаем в параметре "prompt". Напомним про роли (role) в затравках. Всего их три:

• User – это сторона пользователя, взаимодействующего с моделью. Здесь содержатся запросы, вопросы, инструкции или вводные данные. Например, "Назови столицу Испании".

• System – своего рода глобальная настройка модели. Здесь задается общий контекст диалога или задачи, указывается, как модель должна себя вести, какой стиль или формат использовать. Например, "Ты профессиональный географ, который ведет лекцию в лучшем университете мира."

• Assistant – это роль самой модели. Здесь прописывается то, каким ответ от модели должен быть.

Вы также могли увидеть, что в чат-затравке есть синтаксис "{{country}}". Это переменная шаблона. Мы можем использовать в обоих типах. Эта переменная будет заменена на конкретное значение, которое мы передадим при выполнении запроса. Это очень удобно, поскольку позволяет формировать запрос динамически. Дальше мы увидим, как именно.

Кроме параметра "prompt" и типа затравки, мы должны дать ей имя (name). По имени мы будем получать от сервера нужную затравку. Кроме того, мы передали еще некоторые опциональные параметры:

• labels - Это метки, которые позволяют классифицировать или группировать затравки. Если вы создадите новую затравку, вызвав метод create_prompt, с таким же именем и без метки, то ей присвоится метка latest. Кстати, глобально за версию затравки отвечает специальное поле version, которое содержит целое число.

• tags - Это дополнительные ключевые слова, которые помогают в организации затравок и для поиска в веб-интерфейсе. Используются для упрощения поиска, анализа и организации запросов.

• сonfig - Это набор настроек, которые уточняют параметры запроса или поведения модели. Вот некоторые параметры: supported_languages - поддерживаемые языки; model — название модели; temperature — грубо говоря, степень креативности; max_tokens — максимальное количество токенов.

Чтобы получить запрос с сервера Lanfguse, необходимо вызвать метод get_prompt. Есть два правила:

1. Если не указать label, то вы получите версию затравки с меткой production.

2. Если не указать тип чата type, то по умолчанию тип будет считаться как text.

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

Вот как это будет выглядеть для наших затравок, которые мы создали выше:

prompt = langfuse.get_prompt("test_1", label='production')

compiled_prompt = prompt.compile()
print(f'Text prompt: {compiled_prompt}')

prompt = langfuse.get_prompt("test_2")

chat_prompt = langfuse.get_prompt("test_2", type="chat") 
compiled_chat_prompt_with_country = chat_prompt.compile(country='Испания')
print(f'Chat prompt (с подставленной переменной): {compiled_chat_prompt_with_country}')
compiled_chat_prompt = chat_prompt.compile()
print(f'Chat prompt (без подставленной переменной)- {compiled_chat_prompt}')

# >>> Text prompt: Напиши столицу Франции
# >>> Chat prompt (с подставленной переменной): [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Испания', 'role': 'user'}]
# >>> Chat prompt (без подставленной переменной)- [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: {{country}}', 'role': 'user'}]

Логирование запросов

Теперь посмотрим, как заставить Langfuse следить за запросами к модели. Для этого всего лишь надо определить функцию для запроса и обернуть ее декоратором observe(). Он как раз смотрит за ходом выполнения запроса и отправляет всю информацию в Langfuse. Вот самый простой вариант такой функции для текстовой затравки:

client = OpenAI(
  api_key='API_KEY',
  base_url='BASE_URL'
)

@observe()
def run_my_custom_llm_app(prompt_text):
    prompt = [{
        'role': 'user',
        'content': f'{prompt_text}'
    }]
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=prompt,
        max_tokens=100,
    ).choices[0].message.content

    return completion

prompt = langfuse.get_prompt("test_1")
print(f'Текст запроса - {prompt.compile()}')
output = run_my_custom_llm_app(prompt.compile())
print(f'Ответ ИИ - {output}')

# >>> Текст запроса - Напиши столицу Франции
# >>> Ответ ИИ - Париж

Для чатовой затравки она почти ничем не отличается. Мы убираем только создание списка с объектами, потому что из чатовой затравки нами уже создается нужный формат данных.

@observe()
def run_my_custom_llm_app(prompt):
  messages = prompt
  completion = client.chat.completions.create(
      model="gpt-3.5-turbo",
      messages=messages
  ).choices[0].message.content

  return completion

prompt = langfuse.get_prompt("test_2")
compiled_prompt = prompt.compile(country='Испания') # Подставляем необходимо значение заместо переменной country
output = run_my_custom_llm_app(compiled_prompt)
print(f'Запрос - {compiled_prompt}')
print(f'Ответ ИИ - {output}')

# >>> Запрос - [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Испания', 'role': 'user'}]
# >>> Ответ ИИ - Мадрид

Работа с датасетами

Очень полезной фичей оказалась возможность создавать датасеты внутри Langfuse. Мы уже было приготовились страдать и делать своими силами учёт того, на каком датасете запускаются эксперименты, но не пришлось. Конечно, получается некая избыточность — с одной стороны для нас ClearML основное хранилище, с другой, будем некоторые датасеты дублировать в Langfuse. Мы решили, что будем с этим жить пока, определив два правила. Первое, все датасеты в Langfuse будут вторичными по отношению к датасетам из ClearML. Второе — всегда прописывать ID датасета из ClearML в мету датасета Langfuse, чтобы всегда можно было отследить, откуда у последнего мегабайты растут.

Чтобы создать датасет, достаточно выполнить следующую команду:

# Создаем датасет
langfuse.create_dataset(name='Capitals', description="Optional description", metadata={"clearml_id": "CLEARML_ID"})

Далее создадим примеры, которые будут храниться в нашем датасете. Структурно это список объектов. В нашем примере значение input – входное значение, в которое мы помещаем имена шаблонных переменных и желаемые значения для них. Значение expected_output содержит ожидаемое значение, его мы будем сравнивать с ответом БЯМ.

local_items = [
    {"input": {"country": "Италия"}, "expected_output": "Рим"},
    {"input": {"country": "Испания"}, "expected_output": "Мадрид"},
    {"input": {"country": "Бразилия"}, "expected_output": "Бразилиа"},
    {"input": {"country": "Япония"}, "expected_output": "Токио"},
    {"input": {"country": "Индия"}, "expected_output": "Нью-Дели"},
    {"input": {"country": "Канада"}, "expected_output": "Оттава"},
    {"input": {"country": "Южная Корея"}, "expected_output": "Сеул"},
    {"input": {"country": "Аргентина"}, "expected_output": "Буэнос-Айрес"},
    {"input": {"country": "Южная африка"}, "expected_output": "Претория"},
    {"input": {"country": "Египет"}, "expected_output": "Каир"},
]

Следующим этапом добавляем значения в наш датасет:

for item in local_items:
  langfuse.create_dataset_item(
      dataset_name='Capitals',
      input=item['input'], 
      expected_output=item['expected_output'] 
  )

Эээксперименты

Давайте теперь представим, что нам интересно, насколько БЯМ знает столицы разных стран. Тестовые датасет мы подготовили, надо лишь определиться с метрикой, прогнать датасет через БЯМ и измерить результаты. В качестве затравки будем использовать test_2, которая чатовая. Для запросов будем использовать последний вариант функции run_my_custom_llm_app.

В качестве метрики мы возьмем самый простой вариант — просто сравним то, что отдает БЯМ с тем, что мы от нее ожидаем:

def simple_evaluation(output, expected_output):
  return output == expected_output

Нам это нужно для демонстрационных целей. На самом деле, лучше бы взять Левенштена. Дальше поймете почему. Теперь нам надо подготовить функцию для запуска эксперимента. Вот как она выглядит:

def run_experiment(experiment_name, system_prompt):
  dataset = langfuse.get_dataset("Capitals")

  for item in dataset.items:
    # item.observe() возвращает trace_id, который можно использовать для добавления пользовательских оценок позже
    # автоматическое связывание трассировки с запуском эксперимента
    with item.observe(run_name=experiment_name) as trace_id:
      output = run_my_custom_llm_app(system_prompt.compile(country=item.input['country']))
      print(output)
      # добавление результатов оценки в трассировку эксперимента (simple_evaluation)
      langfuse.score(
        trace_id=trace_id,
        name="exact_match",
        value=simple_evaluation(output, item.expected_output)
      )

Метод item.observe(run_name=experiment_name) возвращает trace_id, в который складываются все наблюдения за функцией run_my_custom_llm_app и замеры от вызова langfuse.score(). Вы можете определить сколько угодно замеров, которые вам нужны. Есть возможность подключить LLM-as-a-judge, подробнее про это есть здесь

Кстати говоря про интеграцию с другими фреймворками. Вот как подобный эксперимент запускать в LangChain:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage

def simple_evaluation(output, expected_output):
  return output == expected_output # Сравнение полученного значения с ожидаемым

def run_my_langchain_llm_app(input, system_message, callback_handler):
  prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            system_message
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
  )

  chat = ChatOpenAI(api_key='',
                    base_url='') 
  
  chain = prompt | chat

  res = chain.invoke(
    {"messages": [HumanMessage(content=input)]},
    config={"callbacks":[callback_handler]}
  )

  return res

def run_langchain_experiment(experiment_name, system_message):
  dataset = langfuse.get_dataset("capitals")

  for item in dataset.items:
    handler = item.get_langchain_handler(run_name=experiment_name)

    completion = run_my_langchain_llm_app(item.input["country"], system_message, handler)

    handler.trace.score(
      name="exact_match",
      value=simple_evaluation(completion, item.expected_output)
    )

run_langchain_experiment(
    "Determining the capital of a country",
    "Ты географ. Тебе нужно ответить только названием столицы этой страны: {{country}}"
)

# Отправление данных на сервер langfuse
langfuse_context.flush()
langfuse.flush()

Давайте запустим наш эксперимент:

prompt = langfuse.get_prompt('test_2', label='latest')

run_experiment(
    "Determining the capital of a country",
    prompt
)

# Отправление данных на сервер langfuse
langfuse_context.flush()
langfuse.flush()

# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Египет', 'role': 'user'}]
# >>> Каир
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Южная африка', 'role': 'user'}]
# >>> Претория
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Аргентина', 'role': 'user'}]
# >>> Буэнос-Айрес
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Южная Корея', 'role': 'user'}]
# >>> Сеул
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Канада', 'role': 'user'}]
# >>> Оттава
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Индия', 'role': 'user'}]
# >>> Нью-Дели
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Япония', 'role': 'user'}]
# >>> Токио
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Бразилия', 'role': 'user'}]
# >>> Бразилиа
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Испания', 'role': 'user'}]
# >>> Мадрид
# >>> [{'content': 'Ты географ', 'role': 'system'}, {'content': 'Тебе нужно ответить только названием столицы этой страны: Италия', 'role': 'user'}]
# >>> Рим

Посмотрим, как это всё выглядит в веб-интерфейсе. Откроем вкладку datasets, выберем созданный датасет Capitals, а в нем открываем вкладку Runs. Здесь мы видим запуски наших экспериментов (на скрине их несколько, потому что мы запускали их несколько раз, пока тестировали):

Открыв последний запуск, мы увидим табличку со всей информацией: какой пример использовался, значение метрики, ответ БЯМ и т.д.

Мы можем разметить любой запрос вручную. Это очень полезная функция, т.к. не все метрики могут адекватно оценивать ответ БЯМ. Наша, например, выдаст ошибку при сравнении ответа модели "Париж." с эталонным "Париж". Именно по этому, кстати, расстояние Левенштейна тут подошло бы больше. Кроме того, аннотация может быть нужна по факту выдачи ответов, когда только человек может проверить качество.

Схему разметки надо зарегистрировать в системе. Для этого надо перейти на вкладку "Settings -> Scores/Evaluation". У разметки могут быть три типа:

• Numeric (Числовой): Используется для оценки по шкале (например, от 1 до 10). Применяется для измерения степени соответствия, релевантности или других количественных параметров.

• Categorical (Категориальный): Представляет выбор из заранее заданных категорий (например, "Положительный", "Отрицательный", "Нейтральный"). Удобен для классификации ответов.

• Boolean (Булевый): Предназначен для простых "да/нет" оценок. Полезен для проверки бинарных условий (например, "правильный/неправильный ответ").

Если создание схемы осталось за кадром, то пример разметки конкретных примеров после прогона выглядит вот так. Сначала открываем любой трейс, увидим следующее:

Далее открываем вкладку "Annotation" и вводим желаемое:

Если у нас имеется несколько экспериментов, то мы можем их сравнить с помощью вкладки "Compare". Выглядит это вот так:

Управление и сохранение контекста в чатах

На последок покажем, как создать диалог с сохранением ответа модели для контекста. Этот пример показывает базовую реализацию чат-бота с сохранением контекста и динамическим взаимодействием. Мы сохраняем контекст, путем добавления каждого нового сообщения в историю чата. Таким образом, при каждом запросе модель получает полный диалог, включая предыдущие сообщения, что позволяет ей формировать ответ с учетом всего контекста беседы. Нужно понимать, что при этом расход токенов будет больше, так как будут учитываться все прошлые сообщения.

# Инициализация клиента
client = OpenAI(
    api_key='',
    base_url=''
)

# История сообщений
chat_history = [
    {"role": "system", "content": "Ты бот, который отвечает на вопросы о географии. Отвечай кратко и четко."}
]

@observe()
def run_my_custom_llm_app(user_input):
    global chat_history

    # Добавляем сообщение пользователя в историю
    chat_history.append({"role": "user", "content": user_input})

    # Делаем запрос к модели
    completion = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=chat_history
    ).choices[0].message.content

    # Добавляем ответ модели в историю
    chat_history.append({"role": "assistant", "content": completion})

    return completion

prompts = ['Назови столицу России', 'Расскажи кратко про этот город'] # промпты для продолжения сессии
for i in prompts:
  response = run_my_custom_llm_app(i)
  print(f'Вы: {i}')
  print(f'ИИ: {response}')

# >>> Вы: Назови столицу России
# >>> ИИ: Москва.
# >>> Вы: Расскажи кратко про этот город
# >>> ИИ: Москва - столица России, крупнейший город страны и один из крупнейших городов мира по численности населения. Основана в XII веке. Москва является политическим, экономическим и культурным центром России. В городе находятся многочисленные исторические памятники, музеи, театры и другие достопримечательности.

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

До скорого.