29.07.2025 22:50
Aleron75 0 1900 Источник

Привет, чемпионы! Задумывались ли вы о том, как сделать трассировку в ML/LLM‑пайплайнах? А может, сталкивались с ситуацией, когда хотелось быстро понять, почему система сработала не так, как ожидалось, и в каком месте всё пошло не так? Мы вот задумались и сталкивались, поэтому расскажу о том, что пробуем сейчас.

В этой статье поделюсь нашим опытом использования Langfuse — мощного инструмента для трассировки и оценки пайплайнов, построенных на больших языковых моделях. Мы рассмотрим ключевые возможности Langfuse, особенности интеграции с Python SDK, покажем, как развернуть инфраструктуру локально, и подключим локальную LLM‑модель из Ollama для анализа результатов.

Что же такое langfuse?

Langfuse - это открытая платформа с для разработки и наблюдения приложений на базе LLM.

Что стало интересно нам?

  • Трассировка - используя декоратор SDK или ручную трассировку мы получаем иерархично или напрямую наши результаты работы системы, которые также можем разделять по метаданным юзера и сессии. Также вы можете оценивать стоимость того или иного технического решения по цене за токен благодаря адаптеру, что встроен в langfuse

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

  • Ну и самое главное, это настраиваемая автоматическая оценка работы вашей инфраструктуры за счет eval в langfuse. Что дает возможность делать эту оценку, как в отложенном виде, так и на продакшене.

Поднятие инфраструктуры и работа с ней

Чтоб поднять langfuse в вашей инфраструктуре от вас требуется склонировать репозиторий проекта и далее поднять его.

git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up

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

Инфраструктура внутри langfuse
Инфраструктура внутри langfuse

Что же дальше? Дальше давайте настроим нашу систему, а именно зайдем по адресу

http://localhost:3000/, где войдем в систему, после чего сделаем некоторые действия!

Получим api ключи для взаимодействия с системой через SDK

Тут мы получим Public Key и Secret Key для наших проектов.

Настроим для взаимодействия локальную LLM

После, так как в основном наша ифнарстуктура работает на базе локальных моделек, для воспроизводимости вы можете сделать следующее:

Заходим в командную строку и (если у вас нет ollama установите ее) пишем следующее:

ollama pull qwen3:1.7b

После переходим в настроки нашего проекта и делаем следующее

Название провайдера нужно больше для разделения и настроек, если у вас модель из ollama, то он не так критичен.

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

Супер! А как проверить? Да лего, мы заходим в playground и можем там пообщаться с нашей моделью

Пишим наш "hello world" для langfuse

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

Скрытый текст
import os
import base64

os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-daa6db0b-a45e-4d53-b1ed-242a2551d626"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-4328eaf3-e851-4870-9942-14aa0148cb0e"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = os.environ["LANGFUSE_HOST"] + "/api/public/otel"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"
os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"] = os.environ["LANGFUSE_HOST"] + "/api/public/otel/v1/traces"
os.environ["OTEL_EXPORTER_OTLP_TIMEOUT"] = "30000"
os.environ["OTEL_EXPORTER_OTLP_TRACES_TIMEOUT"] = "30000"

Все апи ключи у меня локальные, так что можно не пробовать :)

И когда мы сделали основную настройку, которая поможет вам нормально настроить телеметрию мы можем работать дальше! (Учтите, что все эти настройки должны быть или по умолчанию в вашем окружении или до импортов всех ваших библиотек в main)

Давайте пойдем дальше и напишем же наш стартовый код, который обсудим!

Скрытый текст
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama
from sentence_transformers import SentenceTransformer, util


from langfuse import observe, get_client

langfuse = get_client()

llm = ChatOllama(model="qwen3:1.7b")
retriever_model = SentenceTransformer("BAAI/bge-m3")


documents = [
    "Langfuse — это инструмент для трассировки пайплайнов LLM.",
    "LangChain — это фреймворк для построения цепочек вызова LLM.",
    "RAG использует retriever и генератор для ответов на вопросы.",
    "Prompt — способ задать контекст.",
    "Embedding — векторное представление текста."
]


@observe(name="retrieval", capture_input=True, capture_output=True)
def retrieve_context(query, top_k=3):
    doc_embeddings = retriever_model.encode(documents, convert_to_tensor=True)
    query_embedding = retriever_model.encode(query, convert_to_tensor=True)

    similarities = util.pytorch_cos_sim(query_embedding, doc_embeddings)[0]
    top_indices = similarities.topk(k=top_k).indices
    selected = [documents[i] for i in top_indices]

    context = "\n".join(selected)
    return context


@observe(name="llm_query", capture_input=True, capture_output=True)
def llm_qa(query, context):
    prompt = ChatPromptTemplate.from_template(
        "Ответь на вопрос используя контекст:\n\n{context}\n\nВопрос: {question}"
    )
    chain = prompt | llm
    result = chain.invoke({"context": context, "question": query})
    return result.content


if __name__ == "__main__":
    question = "Что такое Langfuse?"
    context = retrieve_context(question)
    answer = llm_qa(question, context)
    print("Ответ:", answer)

Итого мы получим результат нашей трассировки и можем увидеть требуемое время на запуск той или иной функции:

Но как мы это получили? Просто добавив @observe() , это своего рода трассировщик, которым мы можем обернуть нужный нам участок для того, чтобы получить трассировку входов и выходов нашей функции, настройка входов и выходов там опциональна.

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

Скрытый текст
def retrieve_context(query, top_k=3):
    with langfuse.start_as_current_span(name="retrieval") as span:
        span.update(input={"query": query, "top_k": top_k})
        doc_embeddings = retriever_model.encode(documents, convert_to_tensor=True)
        query_embedding = retriever_model.encode(query, convert_to_tensor=True)

        similarities = util.pytorch_cos_sim(query_embedding, doc_embeddings)[0]
        top_indices = similarities.topk(k=top_k).indices
        selected = [documents[i] for i in top_indices]

        context = "\n".join(selected)
        span.update(output={"context": context})
        return context


def llm_qa(query, context):
    with langfuse.start_as_current_span(name="llm_query") as span:
        span.update(input={"query": query, "context": context})
        prompt = ChatPromptTemplate.from_template(
            "Ответь на вопрос используя контекст:\n\n{context}\n\nВопрос: {question}"
        )
        chain = prompt | llm
        result = chain.invoke({"context": context, "question": query})
        span.update(output={"answer": result.content})
        return result.content

Добавим метрик и оценок?

Что также нам понравилось, это возможность работать с оценкой наших результатов в реальном времени и с помощью отложенной обработки

Для этого зайдем в:

И создаем наш оценщик

Там же мы можем задать промт и выбрать модель для этого

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

Зачем это нужно? А нужно это для того, чтобы после граммотной настройки вами модели вы можете в реальном или отложенном времени производить оценку вашей модели работы пайплайна, что может вам дать пространство для экспериментов над вашими гипотезами в инфраструктуре и строить метрики по работе на тех или иных участках.

А знаете, что для этого поможет?

Работа с промтами

Помните, как после подключения локальной LLM мы заходили для тестов в playground? Вот если вы разглядывали картинку ui или уже успели к этому моменту поднять его сами, то вы могли заметить интересную иконку

Данная иконка может позволить вам сохранить промт в его версию, где в разделе promts вы можете дать ему метки и настроить.

Скрытый текст
langfuse = get_client()
 
prompt_a = langfuse.get_prompt("Ваше название промта", label="его метки")

Итоги!

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

Также вы поняли, как делать легкую обвязку вашего пайплайна с помощью langfuse для трассировки пайплайна и даже делать оценку ваших узких мест в пайплайне по временным характеристикам!

И последнее, вы поняли, как можно сделать версионирование ваших промтов, а также настроить локальную LLM, которая будет заниматься оценкой датасета по методу LLM-как-судья.

? Ставьте лайк и пишите, какие темы разобрать дальше! Главное — пробуйте и экспериментируйте!


✔️ Присоединяйтесь к нашему Telegram-сообществу @datafeeling, где мы делимся новыми инструментами, кейсами, инсайтами и рассказываем, как всё это применимо к реальным задачам

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