Observabilita LLM aplikací v Pythonu: Průvodce trasováním a monitoringem s Langfuse

Proč vůbec řešit observabilitu LLM aplikací

Nasadit LLM aplikaci do produkce — to je teprve začátek. A tady to začne být zajímavé. Na rozdíl od klasického softwaru, kde vám chyba vrátí jasný stack trace, LLM modely umí selhávat naprosto tiše. Odpovědi se postupně zhoršují, prompty přestanou fungovat po aktualizaci modelu a náklady na tokeny rostou, aniž byste tušili proč.

Tradiční monitoring (CPU, paměť, HTTP status kódy) na tohle prostě nestačí.

Observabilita LLM aplikací řeší právě tyto problémy. Umožňuje vám sledovat kompletní cestu požadavku od vstupu přes jednotlivé kroky — retrieval, tool calling, generování — až po finální odpověď. A to včetně spotřeby tokenů, latence, kvality výstupů a nákladů. V tomto průvodci si ukážeme, jak na to v Pythonu pomocí open-source platformy Langfuse. Naučíte se trasovat LLM volání, monitorovat výkon v reálném čase, spravovat verze promptů a evaluovat kvalitu odpovědí.

Co je Langfuse a proč zrovna on

Langfuse je open-source platforma pro LLM engineering, licencovaná pod MIT. Oproti proprietárním řešením jako LangSmith nabízí kompletní self-hosting zdarma, framework-agnostický přístup a integraci přes standard OpenTelemetry. Upřímně, pro většinu týmů je to momentálně nejlepší poměr cena/výkon.

Klíčové vlastnosti Langfuse:

• Trasování (Tracing) — zachycuje kompletní průběh každého požadavku včetně vnořených operací
• Monitoring — real-time dashboard s metrikami kvality, nákladů a latence
• Správa promptů — centrální verzování a deployment promptů bez změny kódu
• Evaluace — LLM-as-a-Judge, uživatelský feedback a vlastní evaluační pipeline
• Datasety — regresní testování a experimenty nad konzistentními daty

Langfuse má přes 19 000 hvězdiček na GitHubu, prošel akcelerátorem Y Combinator (W23) a v roce 2026 podporuje nativní OpenTelemetry integraci přes SDK v3. Komunita kolem projektu je aktivní a dokumentace je solidní — což u open-source nástrojů rozhodně není samozřejmost.

Instalace a konfigurace Langfuse

Instalace Python SDK

Langfuse SDK v3 nainstalujete jedním příkazem:

pip install langfuse

Pro integraci s OpenAI přidejte také klienta:

pip install langfuse openai

Nastavení přístupových údajů

Langfuse potřebuje tři proměnné prostředí. Klíče získáte v nastavení projektu na Langfuse Cloud nebo ve vaší self-hosted instanci:

# .env soubor
LANGFUSE_SECRET_KEY="sk-lf-..."
LANGFUSE_PUBLIC_KEY="pk-lf-..."
LANGFUSE_BASE_URL="https://cloud.langfuse.com"  # EU region
# LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"  # US region

Ověření připojení

Než začnete cokoliv dělat, ověřte si, že je SDK správně nakonfigurováno. Ušetří vám to spoustu nervů:

from langfuse import get_client

langfuse = get_client()

if langfuse.auth_check():
    print("Langfuse připojeno!")
else:
    print("Chyba připojení — zkontrolujte klíče")

Pokud potřebujete explicitně předat přístupové údaje (třeba v testovacím prostředí), můžete použít konstruktor třídy Langfuse:

from langfuse import Langfuse

langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    base_url="https://cloud.langfuse.com"
)

Datový model trasování

Než se pustíme do kódu, je dobré pochopit, jak Langfuse organizuje trasovací data. Tady je hierarchie:

Session (volitelné)
  └── Trace (1 nebo více)
        └── Observation (1 nebo více, vnořitelné)
              ├── Span — obecná operace (non-LLM)
              ├── Generation — LLM volání (model, tokeny, cena)
              ├── Tool — volání externího nástroje
              ├── Retriever — krok vyhledávání (RAG)
              └── Event — jednorázová událost

Trace reprezentuje jeden kompletní požadavek — například celou konverzaci uživatele s chatbotem. Obsahuje vstup, výstup a metadata jako uživatel, session a tagy.

Observation je pak jednotlivý krok v rámci trace. Langfuse rozlišuje několik typů optimalizovaných pro LLM aplikace, přičemž nejdůležitější je Generation — ten zachycuje parametry modelu, spotřebu tokenů a náklady.

Ke každému trace nebo observaci můžete připojit:

• Prostředí (Environments) — oddělení produkce, stagingu a vývoje
• Tagy — kategorizace podle funkce nebo endpointu
• User ID — propojení s koncovým uživatelem
• Metadata — vlastní key-value páry
• Verze — sledování vlivu nových verzí aplikace

Trasování pomocí Context Manageru

Context manager je doporučený způsob instrumentace v Langfuse SDK v3. Automaticky spravuje začátek a konec spanů a postará se o správné vnořování. Pojďme se na to podívat:

from langfuse import get_client
import openai

langfuse = get_client()
client = openai.OpenAI()

# Hlavní span pro celý požadavek
with langfuse.start_as_current_observation(
    as_type="span",
    name="zpracovani-dotazu"
) as span:
    user_query = "Jak funguje RAG pipeline?"
    span.update(input=user_query)

    # Vnořený span pro retrieval
    with langfuse.start_as_current_observation(
        as_type="retriever",
        name="vektorove-vyhledavani"
    ) as retriever:
        # Simulace vyhledávání v vektorové DB
        context = "RAG pipeline kombinuje retrieval s generováním..."
        retriever.update(
            input=user_query,
            output=context,
            metadata={"top_k": 5, "db": "pinecone"}
        )

    # Vnořená generace pro LLM volání
    with langfuse.start_as_current_observation(
        as_type="generation",
        name="generovani-odpovedi",
        model="gpt-4o"
    ) as generation:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"Kontext: {context}"},
                {"role": "user", "content": user_query}
            ]
        )
        answer = response.choices[0].message.content
        generation.update(
            input=[{"role": "user", "content": user_query}],
            output=answer,
            usage_details={
                "input": response.usage.prompt_tokens,
                "output": response.usage.completion_tokens
            }
        )

    span.update(output=answer)

# V krátkodobých skriptech vždy zavolejte flush()
langfuse.flush()

Po spuštění tohoto kódu uvidíte v Langfuse dashboardu kompletní trace s vnořenými kroky — vyhledávání a generování — včetně časování, vstupů, výstupů a spotřeby tokenů. Je to docela uspokojivé to vidět vizuálně rozložené.

Trasování pomocí dekorátoru @observe

Pro jednodušší případy (a upřímně, pro většinu běžných scénářů) můžete použít dekorátor @observe(). Ten automaticky vytváří spany z dekorovaných funkcí:

from langfuse import observe, get_client

@observe()
def zpracuj_dotaz(dotaz: str) -> str:
    """Hlavní funkce — automaticky vytvoří trace."""
    langfuse = get_client()
    langfuse.update_current_trace(
        user_id="user_123",
        session_id="session_456",
        tags=["chatbot", "production"]
    )
    context = vyhledej_kontext(dotaz)
    odpoved = generuj_odpoved(dotaz, context)
    return odpoved

@observe()
def vyhledej_kontext(dotaz: str) -> str:
    """Vnořená funkce — automaticky vytvoří span."""
    # Simulace vektorového vyhledávání
    return "Nalezený kontext z vektorové databáze..."

@observe(as_type="generation")
def generuj_odpoved(dotaz: str, kontext: str) -> str:
    """LLM volání — vytvoří generation s metadaty modelu."""
    langfuse = get_client()
    langfuse.update_current_observation(
        model="gpt-4o",
        metadata={"temperature": 0.7}
    )
    # Zde by bylo skutečné volání OpenAI API
    return "Vygenerovaná odpověď na základě kontextu..."

# Spuštění
vysledek = zpracuj_dotaz("Co je observabilita?")
print(vysledek)

Dekorátor @observe() používá Python contextvars pro správné přiřazení kontextu, takže funguje i v asynchronních funkcích. Nejvnější dekorovaná funkce automaticky vytvoří trace a všechny vnořené funkce se stanou spany uvnitř tohoto trace. Elegantní, ne?

Automatická integrace s OpenAI

Tohle je asi nejrychlejší cesta, jak začít s trasováním. Drop-in replacement pro OpenAI klienta — stačí změnit jeden import a všechna volání se automaticky zaznamenávají:

from langfuse.openai import openai

# Použití je identické jako s běžným OpenAI klientem
response = openai.chat.completions.create(
    name="analyzator-textu",
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Jsi analytik textu."},
        {"role": "user", "content": "Analyzuj sentiment tohoto textu: ..."}
    ],
    metadata={"use_case": "sentiment_analysis"},
    temperature=0.3
)

print(response.choices[0].message.content)

Langfuse wrapper automaticky zachytí model, parametry, vstupní a výstupní tokeny, latenci a celé prompty — bez jakéhokoli dalšího kódu. Parametry name a metadata jsou specifické pro Langfuse a OpenAI je jednoduše ignoruje. Takže žádné rozbité API volání.

Integrace s LangChain a LangGraph

Pokud používáte LangChain, Langfuse nabízí callback handler, který se napojí na událostní systém LangChainu:

from langfuse.langchain import CallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Inicializace Langfuse handleru
langfuse_handler = CallbackHandler()

# Sestavení chainu
llm = ChatOpenAI(model_name="gpt-4o")
prompt = ChatPromptTemplate.from_template(
    "Odpověz stručně na otázku: {question}"
)
chain = prompt | llm

# Spuštění s trasováním
response = chain.invoke(
    {"question": "Co je LLM observabilita?"},
    config={"callbacks": [langfuse_handler]}
)

print(response.content)

Callback handler automaticky převádí všechny události LangChainu na Langfuse trace. Funguje i s LangGraph agenty a složitějšími řetězy s větvením — prostě ho přidáte do configu a nemusíte řešit nic dalšího.

Správa a verzování promptů

Hardcoded prompty v kódu jsou noční můra pro iteraci. Každá změna znamená nový deployment, code review, testování... Langfuse tohle řeší centrální správou a verzováním promptů s deployment přes labely — bez nutnosti redeploye aplikace.

Vytvoření promptu

from langfuse import get_client

langfuse = get_client()

# Vytvoření nového promptu s konfigurací modelu
langfuse.create_prompt(
    name="analyzator-sentimentu",
    prompt="""Analyzuj sentiment následujícího textu a vrať JSON:
- sentiment: positive/negative/neutral
- confidence: 0.0-1.0
- reasoning: krátké zdůvodnění

Text: {{text}}""",
    config={
        "model": "gpt-4o",
        "temperature": 0.2,
        "max_tokens": 200
    },
    labels=["production"]
)

Použití promptu v aplikaci

from langfuse import get_client
from langfuse.openai import openai

langfuse = get_client()

# Načtení produkční verze promptu
prompt = langfuse.get_prompt("analyzator-sentimentu", label="production")

# Kompilace s proměnnými
compiled = prompt.compile(text="Tento produkt je naprosto skvělý!")

# Volání modelu s parametry z promptu
response = openai.chat.completions.create(
    model=prompt.config["model"],
    temperature=prompt.config["temperature"],
    max_tokens=prompt.config["max_tokens"],
    messages=[{"role": "user", "content": compiled}],
    langfuse_prompt=prompt  # Propojí trace s verzí promptu
)

print(response.choices[0].message.content)

Parametr langfuse_prompt propojí generaci v trace s konkrétní verzí promptu. V dashboardu pak vidíte, která verze promptu vygenerovala kterou odpověď — a to je naprosto klíčové pro A/B testování a debugging.

Aktualizace verzí a labelů

# Vytvoření nové verze promptu (stará zůstane v historii)
langfuse.create_prompt(
    name="analyzator-sentimentu",
    prompt="""Vylepšená verze promptu...""",
    config={"model": "gpt-4o", "temperature": 0.1},
    labels=["staging"]  # Nejprve na staging
)

# Po otestování přesunutí do produkce
langfuse.update_prompt(
    name="analyzator-sentimentu",
    version=2,
    new_labels=["production"]  # Nyní verze 2 je produkční
)

Každá změna vytvoří novou imutabilní verzi. Labely jsou vlastně jen ukazatele na konkrétní verzi — přesunutí labelu production na novou verzi je okamžitý rollout bez redeploye. Tohle je jeden z těch momentů, kdy si řeknete „proč jsem to nepoužíval dřív".

Evaluace kvality odpovědí

Monitoring bez evaluace je jako sledování dashboardu, aniž byste rozuměli tomu, co čísla vlastně znamenají. Langfuse nabízí několik přístupů k hodnocení kvality.

LLM-as-a-Judge

Automatické hodnocení kvality odpovědí jiným LLM modelem. Langfuse umožňuje definovat evaluační šablony přímo v UI nebo přes API a spouštět je jako live evaluátory na produkčním provozu. V praxi to funguje překvapivě dobře — zvlášť pro detekci halucinací a hodnocení relevance.

Uživatelský feedback

Zachycení zpětné vazby od uživatelů (palec nahoru/dolů, hodnocení 1-5) a propojení s konkrétními trace:

from langfuse import get_client

langfuse = get_client()

# Skórování konkrétního trace
langfuse.score(
    trace_id="trace-id-z-aplikace",
    name="user-feedback",
    value=1,  # 1 = pozitivní, 0 = negativní
    comment="Uživatel klikl na palec nahoru"
)

# Nebo číselné hodnocení
langfuse.score(
    trace_id="trace-id-z-aplikace",
    name="relevance",
    value=0.85,
    comment="Automatická evaluace relevance"
)

Regresní testování s datasety

Pro systematické testování si můžete vytvořit datasety s očekávanými vstupy a výstupy a spouštět experimenty při každé změně promptu nebo modelu. Tohle se krásně integruje do CI/CD pipeline a zabrání tiché degradaci kvality — což je přesně ten problém, kvůli kterému většina týmů observabilitu vůbec začne řešit.

Monitoring v produkci

Langfuse dashboard poskytuje real-time přehled o zdraví vaší LLM aplikace. Tady jsou klíčové metriky, které byste měli sledovat:

• Latence (P50, P95, P99) — jak rychle odpovídá váš systém
• Spotřeba tokenů — celková i per-trace, s rozpadem na vstupní a výstupní tokeny
• Náklady — automaticky vypočítané na základě modelu a počtu tokenů
• Skóre kvality — agregované evaluační skóre z LLM-as-a-Judge i uživatelského feedbacku
• Chybovost — počet selhání na úrovni celého trace i jednotlivých kroků

Díky tagům a filtrům můžete tyto metriky sledovat per feature, per model nebo per verze promptu. Chcete vědět, jestli nový prompt zlepšil kvalitu odpovědí? Stačí porovnat skóre mezi dvěma verzemi v dashboardu.

Propojení s existující infrastrukturou

Langfuse podporuje OpenTelemetry, takže můžete posílat trace současně do Langfuse (pro LLM observabilitu) a do Datadogu nebo Grafany (pro infrastrukturní monitoring). Dva nástroje, každý na to, v čem je nejlepší. Nemusíte si vybírat.

Langfuse vs. LangSmith: Kdy co použít

LangSmith je hlavní alternativa k Langfuse. Obě platformy řeší stejné problémy, ale jinak k tomu přistupují:

Kritérium	Langfuse	LangSmith
Licence	Open source (MIT)	Proprietární
Self-hosting	Zdarma, plná podpora	Pouze Enterprise plán
Frameworky	Agnostický (jakýkoli stack)	Nejlepší s LangChain/LangGraph
OpenTelemetry	Nativní podpora v SDK v3	Omezená
Cena (cloud)	Od $29/měsíc	Od $39/uživatel/měsíc
Certifikace	SOC 2 + ISO 27001	SOC 2

Langfuse zvolte, pokud chcete open-source řešení s plnou kontrolou nad daty, framework-agnostický přístup nebo transparentní cenový model. LangSmith zvolte, pokud jste plně investovaní do ekosystému LangChain a preferujete managed SaaS s propracovaným UI pro evaluace. Oba nástroje jsou kvalitní — záleží na vašich prioritách.

Best practices pro produkční observabilitu

Na základě zkušeností z reálných produkčních nasazení tady je pár doporučení, která vám ušetří čas (a nervy).

1. Strukturujte trace konzistentně

Pojmenujte spany a generace popisně a konzistentně. Osvědčily se konvence jako retrieval-{zdroj}, generation-{účel}. Až budete hledat problém v dashboardu s tisíci trace, poděkujete si za to.

2. Přidávejte metadata ke každému trace

Vždy nastavte user_id, session_id a relevantní tagy. V produkci budete potřebovat filtrovat trace podle uživatelů nebo funkcí — a bez metadat je to prakticky nemožné.

3. Používejte prostředí (environments)

Oddělte produkční, staging a vývojová data. Langfuse podporuje environment labely, které vám umožní vidět pouze relevantní trace pro daný kontext. Může to znít jako zbytečná práce navíc, ale jakmile se vám promíchají produkční a testovací data, pochopíte proč.

4. Implementujte sampling pro vysoký provoz

V aplikacích s tisíci požadavky za minutu nemusíte trasovat úplně všechno. Nastavte sampling rate — třeba 10 % pro produkci a 100 % pro staging. Ušetříte na storage i na nákladech za Langfuse Cloud.

5. Nezapomeňte na flush() v krátkodobých prostředích

V AWS Lambda, Cloud Functions nebo jiných krátkodobých prostředích vždy zavolejte langfuse.flush() před ukončením funkce. SDK odesílá trace asynchronně v pozadí a bez flush se data prostě ztratí:

from langfuse import get_client

def lambda_handler(event, context):
    langfuse = get_client()

    with langfuse.start_as_current_observation(
        as_type="span",
        name="lambda-handler"
    ) as span:
        # Zpracování požadavku...
        result = process(event)
        span.update(output=result)

    langfuse.flush()  # Kritické v Lambda!
    return result

6. Nastavte alerting na klíčové metriky

Sledujte náhlé změny v latenci, spotřebě tokenů nebo evaluačním skóre. Nečekaný nárůst tokenů může znamenat prompt injection útok, neefektivní prompt, nebo prostě to, že se model rozhodl být mimořádně upovídaný.

Praktický příklad: Kompletní RAG pipeline s observabilitou

Tak, pojďme vše spojit dohromady. Tady je realistický příklad — RAG chatbot s kompletním trasováním od A do Z:

from langfuse import get_client, observe
from langfuse.openai import openai

langfuse_client = get_client()

@observe()
def rag_chatbot(user_query: str, user_id: str) -> str:
    """Kompletní RAG pipeline s Langfuse trasováním."""
    langfuse = get_client()
    langfuse.update_current_trace(
        user_id=user_id,
        tags=["rag-chatbot", "v2"],
        metadata={"pipeline_version": "2.1"}
    )

    # Krok 1: Retrieval
    documents = retrieve_documents(user_query)

    # Krok 2: Reranking
    ranked_docs = rerank_documents(user_query, documents)

    # Krok 3: Generování odpovědi
    answer = generate_answer(user_query, ranked_docs)

    return answer

@observe(as_type="retriever")
def retrieve_documents(query: str) -> list:
    """Vyhledání relevantních dokumentů."""
    langfuse = get_client()
    langfuse.update_current_observation(
        metadata={"vector_db": "pinecone", "top_k": 10}
    )
    # Simulace vektorového vyhledávání
    docs = [
        {"text": "RAG je technika...", "score": 0.95},
        {"text": "Retrieval-Augmented Generation...", "score": 0.89},
    ]
    return docs

@observe()
def rerank_documents(query: str, docs: list) -> list:
    """Přeřazení dokumentů cross-encoderem."""
    langfuse = get_client()
    langfuse.update_current_observation(
        metadata={"reranker": "cross-encoder/ms-marco"}
    )
    # Simulace rerankingu
    return sorted(docs, key=lambda d: d["score"], reverse=True)[:3]

@observe(as_type="generation")
def generate_answer(query: str, context_docs: list) -> str:
    """Generování odpovědi pomocí LLM."""
    context = "\n".join([d["text"] for d in context_docs])

    prompt = langfuse_client.get_prompt(
        "rag-odpoved",
        label="production",
        fallback="Odpověz na otázku na základě kontextu:\n\nKontext: {{context}}\n\nOtázka: {{query}}"
    )
    compiled = prompt.compile(context=context, query=query)

    response = openai.chat.completions.create(
        model=prompt.config.get("model", "gpt-4o"),
        messages=[{"role": "user", "content": compiled}],
        langfuse_prompt=prompt
    )

    return response.choices[0].message.content

# Spuštění
result = rag_chatbot("Co je RAG pipeline?", user_id="user_42")
print(result)

Tento příklad vytvoří v Langfuse trace s jasnou hierarchií: hlavní span, retriever, reranker a generation. Každý krok má vlastní metadata, vstup a výstup. V dashboardu pak snadno analyzujete, který krok trvá nejdéle, kolik tokenů spotřebovává generování a jak se liší kvalita odpovědí mezi verzemi promptu.

Často kladené otázky

Jaký je rozdíl mezi observabilitou LLM a tradičním monitoringem?

Tradiční monitoring sleduje systémové metriky jako CPU, paměť a HTTP chybové kódy. LLM observabilita jde dál — sleduje sémantickou kvalitu odpovědí, míru halucinací, spotřebu tokenů, náklady a kompletní prompt/response řetězy. LLM model může klidně vracet HTTP 200 a přitom generovat nesmysly. To vám tradiční monitoring neodhalí.

Je Langfuse zdarma? Jaké jsou náklady na self-hosting?

Langfuse je open source pod licencí MIT a self-hosting je kompletně zdarma. Potřebujete jenom PostgreSQL databázi a prostředí pro kontejner (Docker, Kubernetes). Langfuse Cloud nabízí free tier a placené plány od $29/měsíc. Pro self-hosting v produkci doporučují Helm chart na Kubernetes.

Jak moc Langfuse zpomaluje moji aplikaci?

Prakticky vůbec. SDK odesílá trace asynchronně v pozadí pomocí batch zpracování. Přidaná latence je typicky pod 1 ms na operaci. V krátkodobých prostředích (Lambda, Cloud Functions) musíte volat flush(), ale v běžných serverových aplikacích se o to starat nemusíte.

Musím používat LangChain, abych mohl používat Langfuse?

Vůbec ne. Langfuse je framework-agnostický. Funguje s čistým OpenAI SDK, Anthropic SDK, LlamaIndex, LangChain, nebo jakýmkoli vlastním kódem. Můžete dokonce kombinovat různé frameworky v jedné aplikaci a vše bude správně trasováno díky OpenTelemetry integraci.

Jak propojím Langfuse s existující CI/CD pipeline?

Langfuse podporuje datasety a experimenty, které můžete spouštět přímo v CI/CD. Vytvořte dataset s testovacími vstupy a očekávanými výstupy, v pipeline spusťte experiment a porovnejte evaluační skóre s předchozími verzemi. Pokud skóre klesne pod threshold, build neprojde — a regresi odhalíte dřív, než se dostane k uživatelům.