LangFuse: Platform Open-Source untuk Observability Aplikasi LLM

Seiring semakin banyaknya perusahaan yang mengadopsi Large Language Model (LLM) dalam aplikasi produksi, kebutuhan akan observability menjadi sangat krusial. Bagaimana Anda mengetahui berapa biaya setiap request? Bagaimana Anda melacak performa prompt yang berbeda? Bagaimana Anda mengevaluasi kualitas output LLM secara sistematis?

Apa Itu LangFuse?

LangFuse adalah platform open-source LLM engineering yang membantu tim development untuk:

• Tracing: Melacak setiap langkah eksekusi aplikasi LLM, dari input hingga output
• Monitoring: Memantau latency, cost, dan error rate secara real-time
• Evaluasi: Memberikan skor pada output LLM baik secara manual maupun otomatis
• Prompt Management: Mengelola versi prompt secara terpusat
• Analytics: Dashboard untuk menganalisis penggunaan dan performa

LangFuse mendukung berbagai LLM provider (OpenAI, Anthropic, Azure, dll.) dan terintegrasi dengan framework populer seperti LangChain, LlamaIndex, dan Haystack.

Opsi Deployment

1. LangFuse Cloud (Managed)

Cara paling mudah untuk memulai adalah menggunakan LangFuse Cloud di cloud.langfuse.com.

# Tidak perlu instalasi - cukup sign up dan dapatkan API keys
LANGFUSEPUBLICKEY=pk-lf-...
LANGFUSESECRETKEY=sk-lf-...
LANGFUSEHOST=https://cloud.langfuse.com

# docker-compose.yml
version: '3.8'

services:
  langfuse:
    image: langfuse/langfuse:2
    ports:
"3000:3000"
    environment:

DATABASEURL=postgresql://postgres:postgres@db:5432/langfuse
NEXTAUTHSECRET=mysecret

SALT=mysalt
NEXTAUTHURL=http://localhost:3000
    dependson:

db

  db:
    image: postgres:15
    environment:
POSTGRESUSER=postgres
POSTGRESPASSWORD=postgres

POSTGRESDB=langfuse
    volumes:

pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

# Jalankan LangFuse
docker-compose up -d

Akses di http://localhost:3000
Buat akun admin pertama melalui UI

Instalasi SDK Python

pip install langfuse

Konfigurasi Environment

import os

os.environ["LANGFUSEPUBLICKEY"] = "pk-lf-..."
os.environ["LANGFUSESECRETKEY"] = "sk-lf-..."
os.environ["LANGFUSEHOST"] = "https://cloud.langfuse.com"  # atau URL self-hosted

from langfuse import Langfuse

langfuse = Langfuse(
    publickey="pk-lf-...",
    secretkey="sk-lf-...",

    host="https://cloud.langfuse.com"
)

Verifikasi koneksi
langfuse.authcheck()
print("LangFuse terhubung!")

Tracing: Melacak Eksekusi Aplikasi LLM

Tracing adalah fitur inti LangFuse. Setiap interaksi dengan aplikasi LLM Anda direkam sebagai trace yang terdiri dari beberapa komponen:

Konsep Dasar Tracing

• Trace: Unit teratas yang merepresentasikan satu eksekusi end-to-end
• Span: Sub-unit yang merepresentasikan satu langkah dalam trace (misal: retrieval, preprocessing)
• Generation: Span khusus untuk panggilan ke LLM, otomatis mencatat token usage dan cost

Membuat Trace Sederhana

from langfuse import Langfuse

langfuse = Langfuse()

Membuat trace baru
trace = langfuse.trace(
    name="customer-support-chat",
    userid="user-123",

    metadata={"session": "abc-456"},
    tags=["production", "customer-support"]
)

Menambahkan span untuk langkah preprocessing
span = trace.span(
    name="preprocess-input",
    input={"rawmessage": "Bagaimana cara reset password?"}
)

Simulasi preprocessing
processedinput = "Bagaimana cara reset password?"

span.end(output={"processed": processedinput})

Mencatat Generasi LLM

# Mencatat panggilan ke LLM sebagai generation
generation = trace.generation(
    name="llm-response",
    model="gpt-4o",
    modelparameters={

        "temperature": 0.7,
        "maxtokens": 500
    },
    input=[
        {"role": "system", "content": "Anda adalah asisten customer support."},
        {"role": "user", "content": processedinput}

    ]
)

Simulasi respons LLM
responsetext = "Untuk reset password, silakan ikuti langkah berikut..."

Update generation dengan output dan usage
generation.end(
    output=responsetext,

    usage={
        "input": 45,
        "output": 120,
        "unit": "TOKENS"
    }
)

PENTING: flush data ke server
langfuse.flush()

from langfuse.decorators import observe, langfusecontext

@observe()
def processcustomerquery(query: str):
    # Langkah 1: Preprocessing
    cleaned = preprocess(query)

    # Langkah 2: Retrieve context
    context = retrievedocuments(cleaned)


    # Langkah 3: Generate response
    response = generateresponse(cleaned, context)

    return response

@observe()
def preprocess(text: str) -> str:
    return text.strip().lower()

@observe()
def retrievedocuments(query: str) -> list:

    # Simulasi vector search
    return [
        {"title": "Reset Password Guide", "content": "..."},
        {"title": "Account Recovery", "content": "..."}
    ]

@observe(astype="generation")
def generateresponse(query: str, context: list) -> str:

    import openai

    client = openai.OpenAI()
    messages = [
        {"role": "system", "content": f"Context: {context}"},
        {"role": "user", "content": query}
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0.7
    )

    # Update observation dengan detail generation
    langfusecontext.updatecurrentobservation(
        model="gpt-4o",
        usage={
            "input": response.usage.prompttokens,

            "output": response.usage.completiontokens
        }
    )

    return response.choices[0].message.content

Menjalankan - semua langkah otomatis di-trace
result = processcustomerquery("Bagaimana cara reset password?")

Prompt Management

LangFuse memungkinkan Anda mengelola prompt secara terpusat, dengan versioning dan deployment yang terkontrol.

Membuat dan Mengelola Prompt

# Membuat prompt baru di LangFuse
langfuse.createprompt(

    name="customer-support-v1",
    prompt="Anda adalah asisten customer support untuk {{companyname}}. "
           "Jawab pertanyaan berikut dengan ramah dan informatif.\n\n"
           "Pertanyaan: {{question}}\n"
           "Context: {{context}}",
    config={
        "model": "gpt-4o",
        "temperature": 0.7,
        "maxtokens": 500

    },
    labels=["production"]  # Label untuk deployment
)

# Mengambil prompt dari LangFuse
prompt = langfuse.getprompt("customer-support-v1")

Compile dengan variabel
compiled = prompt.compile(
    companyname="PT RUBYTHALIB DATA KONSULTA",

    question="Bagaimana cara reset password?",
    context="Panduan reset password: ..."
)

print(compiled)
Output: "Anda adalah asisten customer support untuk PT RUBYTHALIB DATA KONSULTA..."

Menggunakan config dari prompt
print(prompt.config)
{'model': 'gpt-4o', 'temperature': 0.7, 'maxtokens': 500}

Chat Prompt (Multi-Message)

# Membuat chat prompt
langfuse.createprompt(

    name="chat-support-v1",
    prompt=[
        {"role": "system", "content": "Anda adalah asisten untuk {{company}}."},
        {"role": "user", "content": "{{usermessage}}"}
    ],
    type="chat",
    labels=["staging"]
)

Mengambil dan compile
chatprompt = langfuse.getprompt("chat-support-v1", type="chat")
messages = chatprompt.compile(

    company="PT RUBYTHALIB DATA KONSULTA",
    usermessage="Halo, saya butuh bantuan"
)

Evaluasi dan Scoring

LangFuse mendukung beberapa metode evaluasi untuk mengukur kualitas output LLM.

Manual Scoring via SDK

# Memberikan skor pada trace
trace = langfuse.trace(name="eval-example")

... proses LLM ...

Skor manual
trace.score(
    name="helpfulness",
    value=0.9,
    comment="Jawaban sangat membantu dan lengkap"
)

trace.score(
    name="accuracy",
    value=0.85,
    comment="Informasi akurat tapi kurang detail di bagian akhir"
)

trace.score(
    name="user-feedback",
    value=1,  # thumbs up
    datatype="BOOLEAN"

)

from langfuse.decorators import observe, langfusecontext
import openai

@observe()
def evaluateresponse(question: str, response: str, reference: str) -> dict:

    client = openai.OpenAI()

    evalprompt = f"""Evaluasi jawaban berikut berdasarkan kriteria:
Relevansi (0-1): Seberapa relevan jawaban dengan pertanyaan
Akurasi (0-1): Seberapa akurat dibanding referensi
Kelengkapan (0-1): Seberapa lengkap jawaban

    Pertanyaan: {question}
    Jawaban: {response}
    Referensi: {reference}

    Berikan skor dalam format JSON."""

    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": evalprompt}],

        responseformat={"type": "jsonobject"}

    )

    import json
    scores = json.loads(result.choices[0].message.content)

    # Catat skor ke LangFuse
    langfusecontext.scorecurrenttrace(
        name="relevance", value=scores.get("relevansi", 0)
    )
    langfusecontext.scorecurrenttrace(

        name="accuracy", value=scores.get("akurasi", 0)
    )
    langfusecontext.scorecurrenttrace(
        name="completeness", value=scores.get("kelengkapan", 0)
    )

    return scores

Integrasi dengan LangChain

LangFuse terintegrasi seamless dengan LangChain melalui callback handler.

pip install langfuse langchain langchain-openai

from langfuse.callback import CallbackHandler
from langchainopenai import ChatOpenAI

from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough

Inisialisasi handler LangFuse
langfusehandler = CallbackHandler(
    publickey="pk-lf-...",

    secretkey="sk-lf-...",
    host="https://cloud.langfuse.com"
)

Buat chain LangChain
prompt = ChatPromptTemplate.frommessages([

    ("system", "Anda adalah asisten yang membantu. Jawab dalam Bahasa Indonesia."),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

chain = prompt | llm

Jalankan dengan LangFuse tracing
response = chain.invoke(
    {"question": "Apa itu machine learning?"},
    config={"callbacks": [langfusehandler]}
)

print(response.content)

RAG Chain dengan LangFuse Tracing

from langchainopenai import OpenAIEmbeddings
from langchaincommunity.vectorstores import FAISS
from langchain.textsplitter import RecursiveCharacterTextSplitter


Setup vector store
embeddings = OpenAIEmbeddings()
textsplitter = RecursiveCharacterTextSplitter(
    chunksize=1000,

    chunkoverlap=200
)

Simulasi dokumen
documents = textsplitter.createdocuments([
    "Machine learning adalah cabang dari AI...",
    "Deep learning menggunakan neural network...",
])

vectorstore = FAISS.fromdocuments(documents, embeddings)

retriever = vectorstore.asretriever(searchkwargs={"k": 3})


RAG chain
from langchain.schema.outputparser import StrOutputParser

ragprompt = ChatPromptTemplate.frommessages([
    ("system", "Jawab berdasarkan context berikut:\n{context}"),
    ("human", "{question}")
])

ragchain = (

    {"context": retriever, "question": RunnablePassthrough()}
    | ragprompt
    | llm
    | StrOutputParser()
)

Semua langkah (retrieval + generation) otomatis di-trace
result = ragchain.invoke(

    "Apa perbedaan ML dan DL?",
    config={"callbacks": [langfusehandler]}
)

Integrasi dengan LlamaIndex

pip install langfuse llama-index

from langfuse.llamaindex import LlamaIndexCallbackHandler

Setup handler
langfusehandler = LlamaIndexCallbackHandler(
    publickey="pk-lf-...",

    secretkey="sk-lf-...",
    host="https://cloud.langfuse.com"
)

import llamaindex.core

llamaindex.core.globalhandler = langfusehandler

Sekarang semua operasi LlamaIndex otomatis di-trace
from llamaindex.core import VectorStoreIndex, SimpleDirectoryReader


documents = SimpleDirectoryReader("data").loaddata()
index = VectorStoreIndex.fromdocuments(documents)

queryengine = index.asqueryengine()

Query - otomatis di-trace di LangFuse
response = queryengine.query("Jelaskan konsep RAG")

print(response)

# Cost tracking otomatis saat menggunakan generation
generation = trace.generation(
    name="cost-example",
    model="gpt-4o",
    input=[{"role": "user", "content": "Hello"}],
    output="Hi there!",
    usage={
        "input": 10,
        "output": 5,
        "unit": "TOKENS"
    }
    # Cost dihitung otomatis berdasarkan model pricing
)

Atau set cost manual
generationcustom = trace.generation(
    name="custom-model",
    model="custom-llm-v1",
    usage={
        "input": 100,
        "output": 50,
        "unit": "TOKENS",
        "totalcost": 0.003  # USD

    }
)

# Menggunakan API untuk analytics
import requests

response = requests.get(
    "https://cloud.langfuse.com/api/public/metrics/daily",
    headers={
        "Authorization": "Basic key:secretkey)>"

    },
    params={
        "traceName": "customer-support-chat",
        "fromTimestamp": "2026-04-01T00:00:00Z",
        "toTimestamp": "2026-04-30T23:59:59Z"
    }
)

metrics = response.json()
for day in metrics["data"]:
    print(f"Tanggal: {day['date']}")
    print(f"  Total traces: {day['countTraces']}")
    print(f"  Total cost: ${day['totalCost']:.4f}")
    print(f"  Avg latency: {day['avgLatency']:.2f}s")

"""
Customer Support Bot dengan LangFuse Observability
"""
from langfuse.decorators import observe, langfusecontext
import openai
import json

client = openai.OpenAI()

Knowledge base sederhana
KNOWLEDGEBASE = {

    "resetpassword": "Untuk reset password: 1) Buka halaman login, "
                      "2) Klik 'Lupa Password', 3) Masukkan email, "
                      "4) Cek inbox untuk link reset.",
    "billing": "Untuk pertanyaan billing: Hubungi tim finance di "
               "finance@company.com atau melalui menu Billing di dashboard.",
    "technical": "Untuk masalah teknis: 1) Cek status layanan di status.company.com, "
                 "2) Restart aplikasi, 3) Clear cache browser."
}

@observe()
def classifyintent(query: str) -> str:

    """Klasifikasi intent dari pertanyaan user."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Klasifikasi intent: resetpassword, billing, technical, other"},
            {"role": "user", "content": query}
        ],
        temperature=0
    )

    langfusecontext.updatecurrentobservation(

        model="gpt-4o-mini",
        usage={
            "input": response.usage.prompttokens,
            "output": response.usage.completiontokens

        }
    )

    return response.choices[0].message.content.strip().lower()

@observe()
def retrieveknowledge(intent: str) -> str:
    """Ambil knowledge base berdasarkan intent."""
    context = KNOWLEDGEBASE.get(intent, "Tidak ada informasi spesifik.")


    langfusecontext.updatecurrentobservation(
        metadata={"intent": intent, "found": intent in KNOWLEDGEBASE}

    )

    return context

@observe(astype="generation")
def generateanswer(query: str, context: str, intent: str) -> str:

    """Generate jawaban menggunakan LLM."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": f"Anda adalah asisten customer support. "
                           f"Gunakan context berikut untuk menjawab: {context}"
            },
            {"role": "user", "content": query}
        ],
        temperature=0.7,
        maxtokens=500
    )

    langfusecontext.updatecurrentobservation(

        model="gpt-4o",
        usage={
            "input": response.usage.prompttokens,
            "output": response.usage.completiontokens

        },
        metadata={"intent": intent}
    )

    return response.choices[0].message.content

@observe()
def autoevaluate(query: str, response: str) -> dict:
    """Evaluasi otomatis kualitas respons."""
    evalresponse = client.chat.completions.create(

        model="gpt-4o-mini",
        messages=[
            {
                "role": "user",
                "content": f"Rate this support response (0-1) for: "
                           f"helpfulness, clarity, accuracy.\n"
                           f"Question: {query}\nResponse: {response}\n"
                           f"Return JSON only."
            }
        ],
        responseformat={"type": "jsonobject"},

        temperature=0
    )

    scores = json.loads(evalresponse.choices[0].message.content)

    for metric, value in scores.items():
        langfusecontext.scorecurrenttrace(

            name=metric,
            value=float(value)
        )

    return scores

@observe()
def handlesupportquery(userid: str, query: str) -> dict:
    """Pipeline utama customer support."""
    langfusecontext.updatecurrenttrace(

        userid=userid,

        tags=["customer-support", "production"],
        metadata={"source": "web-chat"}
    )

    # Step 1: Klasifikasi intent
    intent = classifyintent(query)

    # Step 2: Retrieve knowledge
    context = retrieveknowledge(intent)


    # Step 3: Generate jawaban
    answer = generateanswer(query, context, intent)

    # Step 4: Auto-evaluate
    scores = autoevaluate(query, answer)


    return {
        "intent": intent,
        "answer": answer,
        "scores": scores
    }

Jalankan
if name == "main":
    result = handlesupportquery(

        userid="user-456",
        query="Saya lupa password, bagaimana cara resetnya?"
    )

    print(f"Intent: {result['intent']}")
    print(f"Answer: {result['answer']}")
    print(f"Scores: {result['scores']}")

# Contoh sampling
import random

langfuse = Langfuse()

Hanya trace 10% request di production
if random.random() < 0.1:
    trace = langfuse.trace(name="sampled-trace")