Tutorial Logfire: Platform Observability untuk Aplikasi Python dan AI

Logfire adalah platform observability buatan tim Pydantic yang dirancang khusus untuk aplikasi Python. Dengan integrasi native ke ekosistem Python termasuk Pydantic, FastAPI, OpenAI, LangChain, dan banyak library AI/ML lainnya, Logfire memberikan kemampuan monitoring, tracing, dan debugging yang mendalam untuk aplikasi modern. Tutorial ini akan membahas cara menggunakan Logfire dari instalasi hingga penggunaan lanjutan untuk monitoring aplikasi AI.

Mengapa Logfire?

Dalam pengembangan aplikasi AI dan ML, observability menjadi kebutuhan penting. Anda perlu memahami bagaimana model merespons, berapa lama setiap request diproses, dan di mana bottleneck terjadi. Logfire menawarkan beberapa keunggulan:

• Native Python Integration: Dibangun oleh tim Pydantic, terintegrasi langsung dengan ekosistem Python
• OpenTelemetry-based: Menggunakan standar OpenTelemetry sehingga kompatibel dengan tools monitoring lainnya
• LLM Observability: Tracking khusus untuk panggilan ke OpenAI, Anthropic, dan provider LLM lainnya
• SQL Query Interface: Query data observability menggunakan SQL langsung dari dashboard
• Auto-instrumentation: Otomatis mendeteksi dan instrument library populer Python

Instalasi dan Setup

Prasyarat

• Python 3.9 atau lebih baru
• pip atau uv sebagai package manager
• Akun Logfire (gratis untuk memulai)

Instalasi Package

pip install logfire

Atau menggunakan uv:

uv add logfire

Untuk instalasi dengan integrasi tambahan:

pip install "logfire[fastapi]"
pip install "logfire[openai]"
pip install "logfire[anthropic]"
pip install "logfire[langchain]"
pip install "logfire[httpx]"

Autentikasi

Setelah instalasi, lakukan autentikasi dengan akun Logfire:

logfire auth

Perintah ini akan membuka browser untuk login ke akun Logfire Anda. Setelah berhasil login, token akan disimpan secara lokal.

Membuat Project

logfire projects new my-ai-app

Atau gunakan environment variable untuk CI/CD:

export LOGFIRETOKEN=your-write-token-here

import logfire

logfire.configure()

logfire.info("Aplikasi dimulai")

with logfire.span("proses-data"):
    logfire.info("Memproses data pengguna", jumlahrecord=100)

    with logfire.span("validasi"):
        logfire.info("Validasi selesai", valid=95, invalid=5)

    with logfire.span("transformasi"):
        logfire.info("Transformasi selesai", outputformat="parquet")


logfire.info("Proses selesai")

import logfire

logfire.configure()

userdata = {
    "userid": "usr12345",
    "email": "user@example.com",
    "plan": "pro"
}

logfire.info("User login berhasil", *userdata)


logfire.warn(
    "Rate limit mendekati batas",
    currentusage=450,
    maxlimit=500,

    remaining=50
)

logfire.error(
    "Gagal memproses pembayaran",
    orderid="ord789",

    errorcode="INSUFFICIENTFUNDS"

)

logfire.trace("Detail trace untuk debugging mendalam")
logfire.debug("Informasi debugging")
logfire.info("Informasi umum")
logfire.warn("Peringatan")
logfire.error("Error")
logfire.fatal("Fatal error, aplikasi harus berhenti")

from fastapi import FastAPI
import logfire

app = FastAPI()
logfire.configure()
logfire.instrumentfastapi(app)

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/users/{userid}")

async def getuser(userid: str):

    with logfire.span("fetch-user", userid=userid):

        user = await fetchuserfromdb(userid)

        logfire.info("User ditemukan", username=user["name"])
        return user

@app.post("/predict")
async def predict(data: dict):
    with logfire.span("ml-prediction"):
        logfire.info("Input diterima", features=len(data))
        result = runprediction(data)

        logfire.info("Prediksi selesai", confidence=result["confidence"])
        return result

from pydantic import BaseModel
import logfire

logfire.configure()
logfire.instrumentpydantic()

class UserRequest(BaseModel):
    name: str
    email: str
    age: int

try:
    user = UserRequest(name="Ruby", email="ruby@example.com", age="invalid")
except Exception as e:
    logfire.error("Validasi gagal", error=str(e))

import openai
import logfire

logfire.configure()
logfire.instrumentopenai(openai)

client = openai.OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Kamu adalah asisten yang membantu."},
        {"role": "user", "content": "Jelaskan apa itu machine learning dalam 2 kalimat."}
    ],
    temperature=0.7
)

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

import anthropic
import logfire

logfire.configure()
logfire.instrumentanthropic(anthropic)

client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-sonnet-4-6",
    maxtokens=1024,

    messages=[
        {"role": "user", "content": "Apa keuntungan menggunakan observability?"}
    ]
)

print(message.content[0].text)

from langchainopenai import ChatOpenAI
from langchaincore.prompts import ChatPromptTemplate

import logfire

logfire.configure()
logfire.instrumentlangchain()

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

prompt = ChatPromptTemplate.frommessages([

    ("system", "Kamu adalah AI expert yang membantu developer."),
    ("user", "{question}")
])

chain = prompt | llm

response = chain.invoke({
    "question": "Bagaimana cara mengoptimalkan RAG pipeline?"
})

print(response.content)

import logfire

logfire.configure()

@logfire.instrument("process-document", extractargs=True)
def processdocument(docid: str, doctype: str, pagecount: int):
    logfire.info("Memulai pemrosesan dokumen")

    with logfire.span("extract-text") as span:
        text = extracttext(docid)
        span.setattribute("charcount", len(text))

    with logfire.span("analyze-content") as span:
        analysis = analyze(text)
        span.setattribute("sentiment", analysis["sentiment"])

        span.setattribute("topics", analysis["topics"])

    return analysis

import logfire

logfire.configure()

requestcounter = logfire.metriccounter("apirequests")

responsetime = logfire.metrichistogram("responsetimems")


async def handlerequest(request):
    requestcounter.add(1, {"endpoint": request.path, "method": request.method})


    import time
    start = time.time()

    response = await processrequest(request)

    durationms = (time.time() - start)  1000
    responsetime.record(durationms, {"endpoint": request.path})


    return response

import httpx
import logfire

logfire.configure()
logfire.instrumenthttpx(httpx)

async def fetchexternaldata():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

Integrasi Database (SQLAlchemy dan Psycopg)

from sqlalchemy import createengine
import logfire

logfire.configure()

engine = createengine("postgresql://user:pass@localhost/mydb")
logfire.instrumentsqlalchemy(engine)

import psycopg
import logfire

logfire.configure()
logfire.instrumentpsycopg(psycopg)

conn = psycopg.connect("postgresql://user:pass@localhost/mydb")
with conn.cursor() as cur:
    cur.execute("SELECT  FROM users WHERE active = %s", (True,))

    users = cur.fetchall()

import logfire

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(
        extrapatterns=["password", "secret", "token", "apikey", "creditcard"]

    )
)

logfire.info(
    "User registered",
    email="user@example.com",
    password="supersecret123"

)

import logfire
from logfire import SamplingOptions

logfire.configure(
    sampling=SamplingOptions(
        head=0.1,
        tail=logfire.TailSamplingOptions(
            level="error",
            durationthreshold=5.0
        )
    )
)

import logfire
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.modelselection import traintestsplit

from sklearn.metrics import accuracyscore, f1score


logfire.configure()

def trainmodel(X, y, modelname: str = "randomforest"):
    with logfire.span("ml-training-pipeline", modelname=modelname):

        with logfire.span("data-split"):
            Xtrain, Xtest, ytrain, ytest = traintestsplit(
                X, y, testsize=0.2, randomstate=42
            )
            logfire.info(
                "Data split selesai",
                trainsize=len(Xtrain),
                testsize=len(Xtest)
            )

        with logfire.span("model-training"):
            model = RandomForestClassifier(nestimators=100, randomstate=42)
            model.fit(Xtrain, ytrain)
            logfire.info("Training selesai", nestimators=100)


        with logfire.span("model-evaluation"):
            ypred = model.predict(Xtest)

            accuracy = accuracyscore(ytest, ypred)
            f1 = f1score(ytest, ypred, average="weighted")


            logfire.info(
                "Evaluasi model",
                accuracy=round(accuracy, 4),
                f1score=round(f1, 4)
            )

        return model, {"accuracy": accuracy, "f1score": f1}

import logfire

logfire.configure()

async def ragquery(question: str):
    with logfire.span("rag-pipeline", question=question):

        with logfire.span("embedding-generation"):
            embedding = await generateembedding(question)

            logfire.info("Embedding dibuat", dimensions=len(embedding))

        with logfire.span("vector-search"):
            documents = await searchvectors(embedding, topk=5)

            logfire.info(
                "Dokumen ditemukan",
                count=len(documents),
                avgscore=np.mean([d["score"] for d in documents])
            )

        with logfire.span("context-assembly"):
            context = "\n".join([d["text"] for d in documents])
            logfire.info("Context dirakit", charcount=len(context))


        with logfire.span("llm-generation"):
            response = await generateresponse(question, context)
            logfire.info(
                "Response dihasilkan",
                tokencount=response["usage"]["totaltokens"]
            )

        return response["content"]

SELECT
    spanname,

    AVG(duration) as avgdurationms,

    COUNT() as totalcalls,
    SUM(CASE WHEN level = 'error' THEN 1 ELSE 0 END) as errorcount

FROM spans
WHERE spanname LIKE 'llm-%'
    AND createdat > NOW() - INTERVAL '24 hours'

GROUP BY spanname
ORDER BY avgdurationms DESC;

SELECT
    attributes->>'model' as model,
    AVG(CAST(attributes->>'prompttokens' AS INT)) as avgprompttokens,

    AVG(CAST(attributes->>'completiontokens' AS INT)) as avgcompletiontokens,
    AVG(duration) as avglatencyms
FROM spans
WHERE spanname = 'openai.chat.completions'

    AND createdat > NOW() - INTERVAL '7 days'
GROUP BY model
ORDER BY avglatencyms DESC;

Best Practices

1. Gunakan Span yang Bermakna

with logfire.span("process-order", orderid=orderid):
    ...

Beri nama span yang deskriptif dan tambahkan atribut yang berguna untuk filtering.

2. Structured Logging

Selalu gunakan structured data daripada string formatting:

logfire.info("Order processed", orderid=orderid, total=totalamount)

logfire.info(f"Order {orderid} processed with total {totalamount}")

import logfire

try:
    result = processdata(data)
except Exception as e:
    logfire.error(
        "Processing failed",
        errortype=type(e).name,

        errormessage=str(e),
        inputsize=len(data)

    )
    raise

import os
import logfire

env = os.getenv("ENVIRONMENT", "development")
logfire.configure(
    servicename=f"my-app-{env}",
    environment=env
)

5. Monitor Token Usage untuk LLM

import logfire

async def monitoredllmcall(prompt: str):
    with logfire.span("llm-call") as span:
        response = await callllm(prompt)


        span.setattribute("prompttokens", response.usage.prompttokens)
        span.setattribute("completiontokens", response.usage.completiontokens)

        span.setattribute("totaltokens", response.usage.totaltokens)

        estimatedcost = calculatecost(response.usage)
        span.setattribute("estimatedcostusd", estimatedcost)

        return response

6. Gunakan Sampling di Production

Untuk aplikasi dengan volume tinggi, aktifkan sampling agar tidak overload storage:

logfire.configure(
    sampling=SamplingOptions(
        head=0.1,
        tail=logfire.TailSamplingOptions(level="warn")
    )
)

Deployment dan Production

Docker Setup

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

ENV LOGFIRETOKEN=your-write-token

ENV LOGFIRESERVICENAME=my-app

ENV LOGFIREENVIRONMENT=production

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Environment Variables

LOGFIRETOKEN=your-write-token
LOGFIRESERVICENAME=my-app

LOGFIREENVIRONMENT=production
LOGFIRESENDTO_LOGFIRE=true

Kesimpulan

Logfire memberikan solusi observability yang powerful dan mudah digunakan untuk aplikasi Python, terutama untuk aplikasi AI/LLM. Dengan integrasi native ke Pydantic, FastAPI, OpenAI, dan library Python populer lainnya, Logfire mengurangi boilerplate code yang biasanya dibutuhkan untuk monitoring. Fitur SQL query pada dashboard memungkinkan analisis data observability yang fleksibel, dan kemampuan auto-instrumentation membuat setup menjadi sangat cepat. Untuk tim yang membangun aplikasi AI di ekosistem Python, Logfire adalah pilihan yang layak dipertimbangkan sebagai platform observability utama.