Привет, Хабр! Меня зовут Владимир и в последнее время я занимаюсь разработкой агентов на LangGraph. Отладка LangGraph-агента - это отдельная боль: когда граф начинает жить своей жизнью, а LLM уходит в бесконечные циклы, понять, что случилось, становится сложно. В этой статье я покажу, как связать LangGraph с LangFuse для полной трассировки и покажу как управлять промптами как кодом (версионирование и миграция).

Материал разбит на две части: сначала настроим инфраструктуру и разберёмся с инструментами наблюдения, а затем применим это на практике — соберём агента и найдём узкие места через трассировку.

Подготовительное

Первым делом установим зависимости. Установим всё lang... семейство. langchain-openai пригодится для создания экземпляра LLM, а pydantic-settings - для парсинга .env файла.

uv init
uv add langchain==1.2.10 langchain-openai==1.1.10 langgraph==1.0.9 langfuse==3.14.5 pydantic-settings==2.13.1

Далее необходимо локально развернуть инфраструктуру для Langfuse. Тут можно воспользоваться либо официальной инструкцией по локальному развёртыванию (с клонированием репозитория) либо скачать только docker-compose.yml через curl:

curl -O https://raw.githubusercontent.com/langfuse/langfuse/refs/heads/main/docker-compose.yml

В любом случае всё закончится docker compose up -d.

Ждём когда докер скачает и развернёт контейнеры и переходим в браузере по адресу http://localhost:3000/. Откроется приветственное окно web-интерфейса Langfuse, в котором будет необходимо пройти регистрацию, после чего создать организацию и проект. Далее жмём кнопку Get API keys. Langfuse любезно предлагает скопировать готовый код для файла .env - воспользуемся этой возможностью. Единственное, что надо поправить - добавить дополнительное подчеркивание (заменить LANGFUSE_ на LANGFUSE__). Это пригодится для настроек, которые сейчас и напишем.

Подключение к Langfuse

Создадим core\settings.py и опишем в нём два класса - модель конфигурации для Langfuse (наследуется от pydantic.BaseModel) и класс настроек (наследуется от pydantic_settings.BaseSettings) для загрузки из переменных окружения. Начнем с класса настроек:

from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    langfuse: LangfuseConfig

    model_config = SettingsConfigDict(
        env_file=Path(__file__).resolve().parent.parent / '.env',
        env_file_encoding='utf-8',
        extra='ignore',
        case_sensitive=False,
        env_nested_delimiter='__')

Наследование от pydantic_settings.BaseSettings позволяет классу читать и структурировать переменные окружения (читать из файла, указанного в параметре env_file при отсутствии). Настройка env_nested_delimiter='__' указывает разделитель, по которому BaseSettings парсит "плоские" переменные окружения в красивые вложенные структуры - например, запись LANGFUSE__PUBLIC_KEY будет "разбита" на два уровня LANGFUSE -> PUBLIC_KEY. Параметр case_sensitive=False задаёт нечувствительность к регистру переменных окружения. Таким образом, langfuse в файле .env может быть и LANGFUSE, и LangFuse и даже LaNgFuSe. Теперь модель настроек для Langfuse. Переписываем все поля из .env:

from pydantic import BaseModel
class LangfuseConfig(BaseModel):
    secret_key: str
    public_key: str
    base_url: str

Для работы с Langfuse необходимо инициализировать клиент langfuse.Langfuse, используя параметры из нашего конфигурационного класса. Инкапсулируем его в модель настроек с декоратором @cached_property, который обеспечит ленивую инициализацию и кэширование экземпляра клиента. Теперь при первом обращении к свойству client будет создан объект Langfuse, а все последующие вызовы будут возвращать уже существующий экземпляр.

from functools import cached_property
from langfuse import Langfuse

class LangfuseConfig(BaseModel):
    # остальные поля из предыдущего листинга
    @cached_property
    def client(self) -> Langfuse:
        return Langfuse(
            public_key=self.public_key,
            secret_key=self.secret_key,
            base_url=self.base_url,
        )

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

class Settings(BaseSettings):
    # остальные поля из листинга ранее
    def model_post_init(self, __context) -> None:
        _ = self.langfuse.client

Создадим экземпляр класса настроек. Заодно можно стразу проверить, что подключение к Langfuse работает:

settings = Settings()
print('Langfuse подключен') if settings.langfuse.client.auth_check() else print('Ошибка подключения')

Вызовем наш модуль.

python core\settings.py

Если видим вывод в консоль "Langfuse подключен", то всё отлично, иначе надо перепроверить, что ключи скопированы верно. Удалим последнюю строку из модуля core\settings.py и продолжим.

Подключение к LLM

Основа любого Lang{Chain/Graph} продукта - это большая языковая модель (она же БЯМ или LLM). Можно, конечно, использовать облачный API (например OpenAI API), но это что-то на богатом. Для локальной работы с Langfuse воспользуемся локальной Ollama. Ollama можно установить как обычное приложение или развернуть в Docker контейнере. В любом случае, доступ к языковым моделям, которые хостит Ollama, можно получить через локальный API-сервер, совместимый с OpenAI, по умолчанию доступный по URL http://localhost:11434/.

Добавим в конфиг параметры доступа к нашей локальной LLM. Модель (LLM__MODEL_NAME) можно выбрать любую из доступных в Ollama. Предварительно необходимо выполнить ollama pull {Имя модели}.

LLM__HOST=localhost
LLM__PORT=11434
LLM__MODEL_NAME=qwen3:14b
LLM__API_KEY=fake_key
LLM__TEMPERATURE=0.0

И продублируем их в модель настроек:

# core\settings.py
from typing import Annotated
from pydantic import BaseModel, Field, SecretStr

class LLMConfig(BaseModel):
    host: str
    port: Annotated[int | None, Field(default=None, ge=1, le=65535)]
    model_name: str
    api_key: SecretStr
    temperature: Annotated[float, Field(default=0.0, ge=0.0, le=1.1)]

Для полей port и temperature дополнительно добавим валидацию, т.к. эти поля - единственные, которые можем нормально провалидировать. Хоть изначально работать и планируется с локальной моделью, я хочу оставить лазейку для использования облачных API. Для определения места расположения API будем использовать поле port - если есть, значит локальная модель, если None - облачный API. На основании этого добавим к настройкам свойство формирования базового URL адреса API, совместимого с OpenAI:

class LLMConfig(BaseModel):
    # остальные поля из предыдущего листинга
    @property
    def url(self) -> str:
        if self.port is not None:
            return f'http://{self.host}:{self.port}/v1'
        return f'https://{self.host}/v1'

Теперь добавим в настройки соединение с LLM. Для этого будем использовать класс langchain_openai.ChatOpenAI:

class LLMConfig(BaseModel):
    # остальные поля из предыдущего листинга
    @cached_property
    def llm(self) -> ChatOpenAI:
        return ChatOpenAI(
            base_url=self.url,
            api_key=self.api_key,
            model=self.model_name,
            temperature=self.temperature,
        )

Модель настроек LLM готовы. Добавляем в класс с настройками и переходим к реализации трассируемого объекта.

Построение одноклеточного графа

Минутка теории:

В основе LangGraph лежит концепция управляемого состояния (State Graph). Основными сущностями LangGraph являются Состояние (State), Узел (Node) и Ребро (Edge):

  • Состояние - структура данных, которая передаётся между узлами и хранит весь контекст выполнения.

  • Узел - функция, которая принимает State, выполняет логику и возвращает обновления для State.

  • Ребро - правило перехода между узлами, определяет, какой узел будет следующим. Может быть как условным, так и безусловным.

В отличие от линейных цепочек LangChain, где данные просто передаются по цепочке (сорян за тавтологию), в LangGraph каждый Узел явно модифицирует централизованное состояние (State), а Ребра определяют логику перехода между узлами. Ключевое отличие от цепочек — поддержка циклов: агент может возвращаться на предыдущие этапы. Это позволяет превратить LLM из простого генератора текста в управляемый процесс.

Хватит теории, назад к эксперименту. Начнём с чего попроще - создадим чудо-юдо граф, состоящий аж из одного узла!

В файле unicellular\graph.py пропишем State, Node и соберем сам граф. Но, хоть у нас и будет Граф-интроверт, но делать его будем по-взрослому. Начнем со State, и для начала хочу объяснить одну штуку. Узел всегда возвращает изменения для Состояния, при этом принято не переписывать State, а возвращать словарь, содержащий изменённые параметры состояния. Поясню что имею в виду:

class GraphState(TypedDict):
    param1: int
    param2: str

def node_1(state: GraphState):
    tmp = state['param2']
    return {'param1': int(tmp)}

В следующий узел у нас попадает State с обновлённым значением param1. И всё хорошо, пока не надо хранить историю сообщений, которая должна накапливаться. И тут приходится или реализовывать что-то типа:

class GraphState(TypedDict):
    messages: list

def node_2(state: GraphState):
    messages = state['messages']
    # Что-то происходит и получаем новое сообщение msg
    return {'messages': messages + [AIMessage(content=msg)]}

или воспользоваться встроенной функцией-редьюсером langgraph.graph.message.add_messages:

class GraphState(TypedDict):
    messages: Annotated[list, add_messages]

def node_2(state: GraphState):
    # Что-то происходит и получаем новое сообщение msg
    return {'messages': [AIMessage(content=msg)]} # Результат аналогичный

Кстати, этим способом можно и числа складывать, только вместо langgraph.graph.message.add_messages надо использовать operator.add:

class GraphState(TypedDict):
    count: Annotated[int, operator.add]

def node_3(state: GraphState):
    return {'count': 1} # В State будет записано count += 1

Теоретическую базу подвели, теперь пишем наш State:

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class GraphState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Теперь Node. Он будет просто вызывать llm.invoke() и всё:

from core import settings

def node(state: GraphState):
    response = settings.llm.llm.invoke(state['messages'])
    return {'messages': [response]}

Ну собственно сборка графа. Добавляем узел, объявляем его началом и концом, компилируем:

from langgraph.graph import StateGraph
workflow = StateGraph(GraphState)
workflow.add_node('node', node)
workflow.set_entry_point(key='node')
workflow.set_finish_point(key='node')
graph = workflow.compile()

Посмотреть на получившийся граф можно добавив в код вызов метода graph.get_graph().print_ascii() (предварительно надо дополнить зависимости командой uv add grandalf==0.8):

+-----------+
| __start__ |
+-----------+
      *
      *
      *
  +------+
  | node |
  +------+
      *
      *
      *
 +---------+
 | __end__ |
 +---------+

Красота! Теперь вернёмся к задаче, а именно мониторинг и трассировка графа. Перейдём в файл main.py и напишем простенький код:

from unicellular.graph import graph

def run(msg: str):
    return graph.invoke({'messages': [msg]})['messages'][-1].content
print(run('Привет! Как дела?'))

Запускаем (python main.py). Если всё сделано правильно, ключи заданы и инфраструктура доступна, то через некоторое время увидим в консоли ответ LLM. Попробуем теперь трассировку. Начнем с простого навешивания на нашу функцию run декоратора @observe из пакета langfuse. Запускаем, ждём ответа и переходим в веб-интерфейс, вкладка Tracing:

Результат трассировки декоратором @observe
Результат трассировки декоратором @observe

Скажем честно, не густо. @observe позволяет отследить вход-выход и время исполнения, но не позволяет контролировать работу LLM. Для мониторинга кто какие гадости у нейронки спрашивает приемлемо, но нам надо проникнуть в самую суть графа. Для такого мониторинга есть langfuse.langchain.CallbackHandler и как минимум два варианта его использования. Рассмотрим оба. И оба начинаются с создания CallbackHandler:

from langfuse.langchain import CallbackHandler
langfuse_handler = Результат трассировки(public_key=settings.langfuse.public_key)

Ну и сами варианты использования: передать разово при вызове invoke() или передать в граф при его сборке. Соответственно, разница в применении - это либо точечный мониторинг, либо пишем все обращения к агенту:

# Разовый мониторинг
result = graph.invoke({'messages': [msg]}, config={'callbacks': [langfuse_handler]})
# Постоянный мониторинг
graph = workflow.compile().with_config({'callbacks': [langfuse_handler]})

Ну и как это выглядит в веб-интерфейсе:

Результат трассировки с помощью CallbackHandler
Результат трассировки с помощью CallbackHandler

Определённо побогаче. Тут тебе и схема графа, и последовательность вызовов, и расход токенов. Красота. Но прежде чем усложнить наш граф и посмотреть, как Langfuse справится с его трассировкой, рассмотрим ещё одну фишку - систему контроля версий промптов!

Git для промптов

Промпты. Где-то я читал, что 50% успеха - это промпт. Эти все ваши fine-tune, LoRA - хороший промпт-инженер всех победит (но это не точно)! Системные промпты у меня бывают очень длинные, поэтому надо их где-то хранить. И для этого замечательно подходит Langfuse. Потестим.

Перейдем в веб-интерфейсе на вкладку Prompts и жмём кнопку Create new prompt и создадим, для начала, простой текстовый промпт. Вводим имя, текст, оставляем галочку про метку продакшн и жмём Create.

Создание текстового промпта
Создание текстового промпта

Промпт создан, окошко сменилось, лампочка Production весело мигает зелёным. Теперь к коду получения.

from core import settings
print(settings.langfuse.client.get_prompt(name='test_prompt_1').compile())

Одна строчка кода и мы видим свой промпт из хранилища в консоли. Попробуем версионирование. Жмём + New Version, немного меняем текст промпта и сохраняем новую версию. Если не трогать галочку Production, то увидим две версии промпта - первая с тегом Production и вторая, с тегом Latest. Их можно даже сравнить (прям как в Git):

Сравнение версий промптов в Langfuse
Сравнение версий промптов в Langfuse

Вызовем с разными параметрами:

cl = settings.langfuse.client
print("get_prompt(name='test_prompt_1') -                    ", cl.get_prompt(name='test_prompt_1').compile())
print("get_prompt(name='test_prompt_1', version=1) -         ", cl.get_prompt(name='test_prompt_1', version=1).compile())
print("get_prompt(name='test_prompt_1', version=2) -         ", cl.get_prompt(name='test_prompt_1', version=2).compile())
print("get_prompt(name='test_prompt_1', label='Production') -", cl.get_prompt(name='test_prompt_1', label='production').compile())
print("get_prompt(name='test_prompt_1', label='Latest') -    ", cl.get_prompt(name='test_prompt_1', label='latest').compile())

Должно получится что-то типа этого:

Получение версий промпта
Получение версий промпта

Отлично. Также в тексте можно вставлять переменные в стиле Jinja (в двойных фигурных скобочках). Как это и зачем надо разберём на вкладке Chat. Переходим на неё и смотрим. Вкладка позволяет создать цепочку сообщений, указав тип данного сообщения. Попробуем с чего попроще - системный промпт + запрос пользователя с переменной для подстановки сообщения:

Chat промпт с переменными
Chat промпт с переменными

Для начала просто выведем на печать:

print(cl.get_prompt(name='test_prompt_2').compile(query='Почему кольчатых червей можно резать?'))

Так как в шаблоне промпта мы указали переменную, мы должны прописать её в вызове функции compile. На выходе мы получим список из двух словарей с ключами role и content. Теперь используем это в нашем графе. На входе граф ожидает список из langchain_core.messages.BaseMessage. Сейчас в наш метод run передаётся строка с запросом, из которой мы делаем список. Но get_prompt уже возвращает список, так что исправим наш метод run и проверим:

def run(msg: list):
    # Убираем [] вокруг msg и меняем аннотацию
    return graph.invoke({'messages': msg}, config={"callbacks": [langfuse_handler]})['messages'][-1].content

cl = settings.langfuse.client
print(run(cl.get_prompt(name='test_prompt_2').compile(query='Почему кольчатых червей можно резать?')))

Посмотрим как это отработало в Langfuse:

Результат трассировки графа
Результат трассировки графа

Ну что, сообщения передаются, подстановка переменной сработала. Теперь рассмотрим сущность Placeholder - это, условно, переменная, которая позволяет вставить в шаблон промптов ещё промптов. Например, передать историю чата. Как это работает - при вызове compile мы в параметр, который указали в поле Placeholder передаём список словарей {'role': , 'content': } - сообщений, истории чата и т.п.:

print(cl.get_prompt(name='test_prompt_3').compile(query=[{'role': 'user', 'content': 'Доколе?'}]))

В результате получим:

[{'role': 'system', 'content': 'Ты агент британской разведки с позывным 007 с комплексом спасителя бурундуков.'}, {'role': 'user', 'content': 'Ответь на вопрос'}, {'role': 'user', 'content': 'Доколе?'}]

Перенос промптов

Для случая переноса нашего агента на другой ПК необходима возможность выгрузки промптов из Langfuse. Напишем методы для обеспечения экспорта и импорта промптов. Писать их будем на основе статьи с официального сайта с небольшими доработками. Начнём с экспорта. Создадим langfuse_migrations\exporter.py

from langfuse.api import Prompt_Chat
from core import settings

def export_prompts(output_file_name: str, page: int, limit: int):
    langfuse = settings.langfuse.client
    prompts_data = []
    print('Экспорт промптов')
    while True:
        try:
            overview = langfuse.api.prompts.list(page=page, limit=limit)
            if not overview.data:
                break
            print(f'Страница {overview.meta.page}/{overview.meta.total_pages}')

Для начала получаем из настроек приложения ссылку на клиент Langfuse. Далее методом langfuse.api.prompts.list получаем Pydantic модель с двумя полями - data, содержащее список всех имеющихся промптов (их имена и номера версий) и meta с настройками пагинации и общим количеством страниц по заданным настройкам. Начинаем перебирать страницы с промптами и промпты внутри страницы:

for prompt_info in overview.data:
    prompt_name = prompt_info.name
    for v_num in prompt_info.versions:
        try:
            prompt_detail = langfuse.api.prompts.get(prompt_name, version=v_num)
            if isinstance(prompt_detail, Prompt_Chat):
                decode_prompt = [prompt.json() for prompt in prompt_detail.prompt]
            else:
                decode_prompt = prompt_detail.prompt

Поле versions каждого объекта списка overview.data содержит список версий для промпта с именем name. Получив номер и версию промпта методом langfuse.api.prompts.get извлекаем из Langfuse полную информацию о промпте. Метод возвращает объект одного из двух типов - Prompt_Text или Prompt_Chat (соответствуют двум вариантам создания промпта в веб интерфейсе). Тип Prompt_Text содержит чистый текст промпта, тогда как из Prompt_Chat необходимо дополнительно извлечь данные для каждого элемента промпта-чата. Метод json() позволяет автоматически преобразовать элемент чата в JSON формат. Дальше просто заполняем "карточку" промпта и добавляем в наш список.

prompt_dict = {
    'name': prompt_detail.name,
    'version': prompt_detail.version,
    'labels': prompt_detail.labels,
    'prompt': decode_prompt,
    'config': prompt_detail.config,
    'tags': prompt_detail.tags,
}
prompts_data.append(prompt_dict)

Повторяем, пока не кончатся страницы:

if overview.meta.page >= overview.meta.total_pages:
    break
page += 1

После окончания преобразования сгружаем получившийся список в JSON, добавив немного метаданных:

with open(output_file_name, 'w', encoding='utf-8') as f:
    json.dump({
        'exported_at': datetime.now().isoformat(),
        'source_host': settings.langfuse.base_url,
        'prompts': prompts_data
    }, f, ensure_ascii=False, indent=2)

Переходим к импорту. Создадим langfuse_migrations\importer.py

from langfuse.api import (
    ChatMessageWithPlaceholders_Chatmessage, ChatMessageWithPlaceholders_Placeholder,
    CreatePromptRequest_Chat, CreatePromptRequest_Text)
from core import settings

def import_prompts(input_file_name: str):
    langfuse = settings.langfuse.client
    with open(input_file_name, 'r', encoding='utf-8') as f:
        data = json.load(f)
    prompts_to_import = data.get('prompts', [])
    for prompt_to_import in prompts_to_import:

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

if isinstance(prompt_to_import['prompt'], str):
    request = CreatePromptRequest_Text(
        name=prompt_to_import['name'],
        prompt=prompt_to_import['prompt'],
        config=prompt_to_import.get('config', {}),
        labels=prompt_to_import.get('labels', []),
        tags=prompt_to_import.get('tags', []),
    )

Проверяем, что поле prompt это простая строка и создаём экземпляр класса CreatePromptRequest_Text, отвечающего за создание текстовых промптов. В конструктор передаём всё, что извлекли при экспорте.

Теперь Чат-промпт. С ним немного посложнее, т.к. в чате сообщения могут быть двух видов - Chatmessage и Placeholder. Для определения типа сообщения воспользуемся тем, что у модели Placeholder есть поле type: typing.Literal["placeholder"]. Также необходимо каждый элемент списка предварительно преобразовать в словарь методом json.loads():

elif isinstance(prompt_to_import['prompt'], list):
    prompts = []
    for prompt in prompt_to_import['prompt']:
        item = json.loads(prompt)
        if 'type' in item and item['type'] == 'placeholder':
            prompts.append(ChatMessageWithPlaceholders_Placeholder(**item))
        else:
            prompts.append(ChatMessageWithPlaceholders_Chatmessage(**item))
    request = CreatePromptRequest_Chat(
        name=prompt_to_import['name'],
        prompt=prompts,
        config=prompt_to_import.get('config', {}),
        labels=prompt_to_import.get('labels', []),
        tags=prompt_to_import.get('tags', []),
    )

Вообще поле type присутствует и в классе ChatMessageWithPlaceholders_Chatmessage (класс-модель для сообщений чата), но оно почему-то не экспортируется при вызове метода ChatMessageWithPlaceholders_Chatmessage.json()

Остаётся только отправить сформированный запрос в Langfuse и повторять до победного:

langfuse.api.prompts.create(request=request)

Теперь осталось написать main функцию запуска с аргументами. Аргументы будем получать из командной строки с помощью библиотеки argparse. По аргументам - сделаем строковый аргумент для передачи пути к целевому файлу, два булевых для выбора режима и два целых для пагинации. В корне проекта создадим файл langfuse_migrations.py:

import argparse
from langfuse_migrations import export_prompts, import_prompts

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Langfuse Prompt Exporter')
    parser.add_argument('-f', '--file', type=str,default='prompts_backup.json')
    parser.add_argument('--export', action='store_true', dest='export_mode')
    parser.add_argument('--import', action='store_true', dest='import_mode')
    parser.add_argument('--limit', type=int, default=100)
    parser.add_argument('--page', type=int, default=1)
    args = parser.parse_args()
    if args.export_mode:
        export_prompts(output_file_name=args.file, page=args.page, limit=args.limit)
    elif args.import_mode:
        import_prompts(input_file_name=args.file)
    else:
        print('Необходимо задать флаг "--export" или "--import"')

Готово. Остаётся проверить, что миграция работает. Для этого надо запустить экспорт

python langfuse_migrations.py --export

затем почистить вкладку с промптами в Langfuse и запустить импорт

python langfuse_migrations.py --import

Если всё хорошо, то в корне проекта появится файл prompts_backup.json, а промпты в Langfuse восстановятся.

Промежуточный итог

В этой части мы настроили инфраструктуру для разработки LLM-приложений:

  • Развернули локальный LangFuse для трассировки.

  • Настроили конфигурацию через pydantic-settings.

  • Подключили мониторинг к графу LangGraph — теперь видно не только вход-выход, но и внутренние вызовы LLM.

  • Организовали хранение промптов во внешнем хранилище с версионированием и миграцией.

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

Код проекта доступен тут.

Продолжение уже доступно