Привет, чемпионы! Задумывались ли вы о том, как сделать трассировку в 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Что важно для вас, это проверьте порты на которых поднимается инфрастуркутура и учтите, то, что внутри этой инфры уже могут содержаться сервисы, которые могут уже быть подняты у вас.
Что же дальше? Дальше давайте настроим нашу систему, а именно зайдем по адресу
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, где мы делимся новыми инструментами, кейсами, инсайтами и рассказываем, как всё это применимо к реальным задачам
