Śledzenie pipeline LangGraph do blogów z użyciem Langfuse: fragmenty tekstu, oceny, ostrzeżenia

Zabezpieczenia (sprawdzanie źródeł, weryfikacja cytacji, ograniczenie liczby prób, eskalacja do człowieka) są przydatne tylko wtedy, gdy nieudany proces można później odtworzyć. W takim przypadku pipeline blogowy z wieloma agentami rodzi wiele pytań: Czy źródła badawcze były niskiej jakości? Czy autor zignorował ostrzeżenia dotyczące cytacji? Czy recenzent miał rację odrzucając pracę? Czy wyszukiwanie zakończyło się przed upływem czasu? Ile kosztował ten proces w postaci wywołań modeli i tokenów?

Bez możliwości obserwacji system stanowi czarną skrzynkę otoczoną kolejnymi czarnymi skrzynkami. Śledzenie działań wielu agentów to szczególnie trudny aspekt tego zadania. Ta konfiguracja dodaje Langfuse do pipeline blogowego LangGraph, tworząc jeden zapis dla każdego przeprowadzonego procesu nad artykułem, który zawiera informacje o przekazywaniu zadań, wywołaniach modeli i narzędzi, czasach wykonywania, kosztach, błędach oraz wynikach oceny.

Określanie, czym jest zapis

Zanim przejdziesz do kodu, wybierz jednostkę, którą chcesz sprawdzić. W tym procesie jeden zapis odpowiada jednej próbie stworzenia jednego artykułu. Zaczyna się on, gdy zostanie zgłoszony temat, a kończy, gdy proces zostanie zakończony, zawieszony z powodu interwencji człowieka lub poniesie porażkę. W jego ramach znajdują się obserwacje dotyczące każdej wykonanej czynności:

blog-pipeline
├── research-agent
│   ├── search-web
│   └── research-model
├── writer-agent
│   └── writer-model
├── citation-check
│   └── citation-review-model
└── reviewer-agent
    └── reviewer-model

Langfuse pracuje z trzema rodzajami pojęć. Ślad to jeden pełny przebieg od początku do końca. Obserwacja to pojedynczy krok wewnątrz tego przebiegu, który może być fragmentem kodu (logiką aplikacji), wywołaniem modelu językowego lub zdarzeniem (małym znacznikiem). Sezja łączy kilka śladów ze sobą, np. wszystkie próby dotyczące tego samego tematu na blogu.

Hierarchia odzwierciedla sposób debugowania: zaczyna się od pełnego przebiegu, a następnie analizuje się konkretny agent, narzędzie lub wywołanie modelu, które spowodowały problem.

Ustawianie Langfuse

Stwórz projekt Langfuse i podaj jego dane uwierzytelniające za pomocą zmiennych środowiskowych. Klucz tajny nigdy nie powinien być zapisywany w plikach źródłowych.

pip install -U langfuse

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"

Pakiet langfuse to SDK w języku Python. Dwie klucze służą do autoryzacji aplikacji wobec projektu, a LANGFUSE_BASE_URL określa miejsce jej rozgłoszenia. Przykład jest skierowany na region chmurowy w UE, więc w razie potrzeby zastąp go innym regionem lub instancją self-hosted.

Etykieta środowiska pomaga oddzielić ślady z rozwoju od tych z produkcji. Dobrze sprawdzają się krótkie, stabilne nazwy takie jak development, staging i production. Bez nich eksperymenty oraz rzeczywiste uruchomienia artykułów trafiają do jednego widoku, co utrudnia debugowanie.

Podłączenie obsługi callback

Langfuse dostarcza obsługę callback w ramach LangChain. LangGraph opiera się na mechanizmach wykonywalnych i obsługi callback z LangChain, więc ta sama obsługa rejestruje wydarzenia związane z grafem, łańcuchem, narzędziem i modelem:

from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler


langfuse = get_client()
langfuse_handler = CallbackHandler()

get_client() tworzy klienta na podstawie zmiennych środowiskowych. Obsługa powinna zostać utworzona raz podczas uruchamiania i być ponownie wykorzystywana przy każdym uruchomieniu. Nasłuchuje zdarzeń LangChain i przekazuje je jako obserwacje, co eliminuje konieczność ręcznego śledzenia każdego wywołania modelu.

Następnie wchodzi w grę funkcja propagate_attributes. Nadaje ona te same etykiety każdej obserwacji w ramach jednego uruchomienia, dzięki czemu wywołanie do wyszukiwania, generowanie tekstu przez pisarza oraz generowanie tekstu przez recenzenta mogą być filtrowane jako elementy jednego procesu przetwarzania artykułów.

Wywołanie grafu jest otoczone elementem root span, przy czym obsługa jest przekazywana poprzez konfigurację LangGraph:

def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
    initial_state = {
        "topic": topic,
        "audience": "developers new to agent systems",
        "research_brief": "",
        "sources": [],
        "open_questions": [],
        "article_draft": "",
        "review_feedback": "",
        "citation_issues": [],
        "approved": False,
        "revision_count": 0,
        "status": "researching",
    }

    with langfuse.start_as_current_observation(
        as_type="span",
        name="blog-pipeline",
        input={"topic": topic, "audience": initial_state["audience"]},
    ) as pipeline_span:
        with propagate_attributes(
            trace_name="blog-pipeline",
            session_id=blog_id,
            user_id=user_id,
            tags=["blog-pipeline", "langgraph"],
            version="1.0.0",
            metadata={"workflow": "research-write-review"},
        ):
            trace_id = langfuse.get_current_trace_id()
            result = graph.invoke(
                initial_state,
                config={"callbacks": [langfuse_handler]},
            )

        pipeline_span.update(
            output={
                "status": result["status"],
                "approved": result["approved"],
                "revision_count": result["revision_count"],
            }
        )

    return result, trace_id

Kilka szczegółów dotyczących tego otoczenia ma znaczenie. Zewnętrzne wywołanie start_as_current_observation() tworzy główną obserwację, a jej dane wejściowe są celowo zwięzłe – zawierają jedynie temat i grupę odbiorców. To właśnie jest widoczne na pierwszy rzut oka w liście śladów, dlatego informacje o stanie wewnętrznym nie powinny się tam znajdować.

Funkcja propagate_attributes() przypisuje tagi do każdej zagnieżdżonej obserwacji. blog_id służy jako identyfikator sesji, grupując ponawiane próby lub działania związane z jednym artykułem. Wartość version określa proces pracy i wersję promptu, za którym kryje się wynik; dzięki temu, gdy zachowanie się zmienia, ślady można filtrować według wersji, a nie na podstawie domysłów.

Procesor przemieszcza się przez standardową konfigurację LangGraph, a stamtąd automatycznie zbiera szczegóły na poziomie frameworku: każde wywołanie węzła, generację przez model językowy, wywołanie narzędzia, dane wejściowe i wyjściowe, opóźnienie, liczbę tokenów oraz koszt.

Na końcu korzeniowy element span jest aktualizowany trzema informacjami na poziomie pipeline: czy proces został zakończony, czy projekt został zatwierdzony oraz ile rewizji zostało przeprowadzonych. Zwrócony trace_id powinien być przechowywany w logach aplikacji lub w interfejsie użytkownika, umożliwiając bezpośrednie przejście od procesu przetwarzania artykułu do jego śledzenia w Langfuse.

Elementy span na poziomie biznesowym w agentach

Automatyczne śledzenie potwierdza, że model został wywołany. Nie pokazuje jednak, co każdy specjalista przekazał następnemu. W tym celu należy dodać element span wewnątrz węzła agenta i przekazać argument config do zagnieżdżonych wywołań LangChain lub LangGraph, aby obsługa zachowała ten sam kontekst śledzenia:

from langchain_core.runnables import RunnableConfig


def research_node(state: BlogState, config: RunnableConfig) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="research-agent",
        input={"topic": state["topic"], "audience": state["audience"]},
    ) as span:
        result = research_agent.invoke(
            {"topic": state["topic"], "audience": state["audience"]},
            config=config,
        )

        span.update(
            output={
                "source_count": len(result["sources"]),
                "open_question_count": len(result["open_questions"]),
                "brief": result["research_brief"],
            }
        )

    return {
        "research_brief": result["research_brief"],
        "sources": result["sources"],
        "open_questions": result["open_questions"],
        "status": "writing",
    }

Ta sekcja uzupełnia mechanizm obsługi powrotów, a nie zastępuje go. Dodaje znaczenie, którego framework nie może odgadnąć, mianowicie to, co "badania się udały" oznacza w tym procesie. Tutaj zapisuje się liczbę źródeł, ilość nierozwiązanych pytań oraz krótki opis przekazany autorowi.

Przekazywanie parametru config: RunnableConfig do metody research_agent.invoke() umożliwia przeniesienie mechanizmów powrotu oraz kontekstu śledzenia. Zapomnienie o tym jest typowym błędem związanym z możliwościami obserwacji: graf na najwyższym poziomie jest śledzony, natomiast zagnieżdżone wywołania pojawiają się jako niepowiązane ślady lub znikają.

Nie zmieniajcie nazw elementów obserwacji: research-agent, writer-agent, citation-check. Temat, nazwa modelu lub identyfikator użytkownika nie powinny być włączane do nazw, ponieważ te dane pasują do danych wejściowych, atrybutów modelu lub metadanych. Stałe nazwy zapewniają, że panele kontrolne i filtry pozostaną funkcjonalne pomimo zmian w implementacji.

Ostrzeżenia i błędy

Nie każdy nieoczekiwany wynik oznacza awarię. Wątpliwy źródło, wyczerpany budżet na rewizje lub nieobsłużona cytata również mają znaczenie i powinny być wyświetlane jako ostrzeżenie.

def record_source_assessment(assessment: SourceAssessment) -> None:
    if assessment.suspicious_content:
        langfuse.update_current_span(
            level="WARNING",
            status_message="Untrusted source contained agent-directed instructions.",
        )


def record_pipeline_failure(error: Exception) -> None:
    langfuse.update_current_span(
        level="ERROR",
        status_message=f"Pipeline failed: {type(error).__name__}",
    )

Langfuse obsługuje poziomy logów dla obserwacji. WARNING odnosi się do zakończonego procesu, który wymaga dalszej analizy, natomiast ERROR służy do rejestrowania rzeczywistych błędów. Komunikaty o stanie powinny opisywać problem operacyjny, bez kopiowania wrażliwego tekstu źródłowego lub tajemnic do danych telemetrycznych.

Narzędzie do oceny źródła należy umieścić w ścieżce sprawdzania źródeł, a narzędzie do rejestrowania błędów – w obsłudze błędów, która może faktycznie je naprawić lub zgłosić. Rejestrowanie każdego wyjątku tylko po to, by go zapisać, jest złym pomysłem, ponieważ nieoczekiwane błędy powinny nadal trafić do systemu monitoringu błędów i zostać wyraźnie zgłoszone.

Ocena wyniku

Ślad opisuje to, co się wydarzyło, a ocena wskazuje, czy wynik był przyzwoity. Po zakończeniu procesu należy dodać pierwszą, deterministyczną ocenę uzyskaną na podstawie stanu pipeline’u:

result, trace_id = run_blog_pipeline(
    topic="How AI agents use tools",
    blog_id="blog-ai-tools-001",
    user_id="philip",
)

if trace_id:
    langfuse.create_score(
        trace_id=trace_id,
        name="review_approved",
        value=1 if result["approved"] else 0,
        data_type="BOOLEAN",
        comment=result["status"],
    )

    langfuse.create_score(
        trace_id=trace_id,
        name="revision_count",
        value=float(result["revision_count"]),
        data_type="NUMERIC",
    )

Pierwsza ocena odzwierciedla ustrukturyzowaną decyzję recenzenta o zatwierdzeniu lub odrzuceniu, natomiast druga pokazuje, ile przeróbek wymagał szkic. Gromadzone z biegiem czasu ujawniają coś więcej niż tylko pojedyncze historie. Który model, wersja promptu, kategoria tematyczna czy metoda pozyskiwania materiałów przynosi mniej poprawek i więcej zatwierdzeń?

Żadna z nich nie stanowi ostatecznej prawdy, ponieważ recenzent może się mylić. Ocena wydana przez człowieka po przeczytaniu ostatecznego szkicu może pojawić się później. Kluczowym nawykiem jest sprawianie, by jakość była mierzalna i porównywalna pomiędzy różnymi przypadkami, zamiast polegać na kilku zapadających w pamięć przykładach.

Rutyna debugowania

Gdy artykuł okazuje się słaby, otwórz jego historię i zadaj następujące pytania w kolejności:

1. Czy graf przebył oczekiwaną ścieżkę? Sprawdź sekcje Badania, Autor, Weryfikacja cytatów i Recenzent, a także wszelkie pętle poprawek.

Jak solidne było przekazanie danych badawczych? Przeanalizuj liczbę źródeł, otwartych pytań, cytatów oraz sam opis zadania.

Gdzie pojawił się błąd? Porównaj instrukcję, wynik, opóźnienie, zużycie tokenów oraz aktywność narzędzia w każdym kroku z krokiem poprzednim.

Błąd modelu czy systemu? Przekroczenie czasu reakcji oraz słabe cytaty wymagają zupełnie różnych rozwiązań.

Czy mechanizmy kontroli zadziałały? Jeśli źródło wydawało się podejrzane lub twierdzenie nie miało podstaw, czy graf skierował proces na rewizję lub przegląd przez człowieka zamiast na zatwierdzenie?

Dlatego kształt śledzenia jest ważny. Dobrze nazwany drzewo przekształca stwierdzenie „blog się pomylił” w precyzyjny wniosek, na przykład „autor dodał twierdzenie nieobecne w opisie zadania po drugiej rewizji.”

Zapobieganie dostępowi do danych wrażliwych

Tracing może rejestrować polecenia, wyniki, rezultaty narzędzi oraz metadane, co oznacza również, że może przechwytywać informacje, które nie powinny być przechowywane. Klucze API, nagłówki autoryzacji oraz pełna konfiguracja poufnych danych nigdy nie powinny trafiać do Langfuse. Jeśli dane użytkownika trafią później do tego procesu, sprawdź, co zostanie zapisane, i ustaw maskowanie przed eksportowaniem danych z tracingu. Langfuse oferuje funkcje maskowania, które usuwają dane wrażliwe przed ich opuszczeniem aplikacji.

Zasada ogólna: rejestruj tyle, ile wystarczy do wyjaśnienia decyzji, i nie więcej, niż można bezpiecznie przechować.

Korzyści płynące z tego procesu

Każde uruchomienie teraz pozostawia rekord operacyjny:

• Każda próba przetworzenia artykułu to osobny zapis tracingu.
• Agenty, modele i narzędzia pojawiają się jako zagnieżdżone obserwacje.
• Nazwy stałe wraz z metadanymi, wersją, sesją i środowiskiem ułatwiają filtrowanie.

Za wątpliwe źródła wyświetlane są ostrzeżenia, a prawdziwe błędy oznaczane są jako błędy.

Liczba zatwierdzeń i rewizji jest rejestrowana jako wyniki.

ID śledzenia prowadzi bezpośrednio do procesu przetwarzania danego artykułu.

Rozwiązywanie problemów przestaje polegać na przewijaniu logów terminala i domyślaniu się, a staje się inspekcją dokładnego momentu, w którym coś poszło nie tak.