Logfire Tutorial: Observability Platform for Python and AI Applications

Logfire is an observability platform built by the Pydantic team, designed specifically for Python applications. With native integrations into the Python ecosystem including Pydantic, FastAPI, OpenAI, LangChain, and many other AI/ML libraries, Logfire provides deep monitoring, tracing, and debugging capabilities for modern applications. This tutorial covers how to use Logfire from installation to advanced usage for monitoring AI applications.

Why Logfire?

In AI and ML application development, observability is a critical need. You need to understand how models respond, how long each request takes to process, and where bottlenecks occur. Logfire offers several advantages:

• Native Python Integration: Built by the Pydantic team, it integrates directly with the Python ecosystem
• OpenTelemetry-based: Uses the OpenTelemetry standard, making it compatible with other monitoring tools
• LLM Observability: Dedicated tracking for calls to OpenAI, Anthropic, and other LLM providers
• SQL Query Interface: Query observability data using SQL directly from the dashboard
• Auto-instrumentation: Automatically detects and instruments popular Python libraries

Installation and Setup

Prerequisites

• Python 3.9 or later
• pip or uv as package manager
• Logfire account (free to start)

Package Installation

pip install logfire

Or using uv:

uv add logfire

For installation with additional integrations:

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

Authentication

After installation, authenticate with your Logfire account:

logfire auth

This command will open a browser for you to log in to your Logfire account. After successful login, the token will be stored locally.

Creating a Project

logfire projects new my-ai-app

Or use environment variables for CI/CD:

export LOGFIRETOKEN=your-write-token-here

import logfire

logfire.configure()

logfire.info("Application started")

with logfire.span("process-data"):
    logfire.info("Processing user data", recordcount=100)

    with logfire.span("validation"):
        logfire.info("Validation complete", valid=95, invalid=5)

    with logfire.span("transformation"):
        logfire.info("Transformation complete", outputformat="parquet")


logfire.info("Process completed")

import logfire

logfire.configure()

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

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


logfire.warn(
    "Rate limit approaching threshold",
    currentusage=450,
    maxlimit=500,

    remaining=50
)

logfire.error(
    "Payment processing failed",
    orderid="ord789",

    errorcode="INSUFFICIENTFUNDS"

)

logfire.trace("Detailed trace for deep debugging")
logfire.debug("Debugging information")
logfire.info("General information")
logfire.warn("Warning")
logfire.error("Error")
logfire.fatal("Fatal error, application must stop")

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 found", username=user["name"])
        return user

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

        logfire.info("Prediction complete", 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("Validation failed", 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": "You are a helpful assistant."},
        {"role": "user", "content": "Explain what machine learning is in 2 sentences."}
    ],
    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": "What are the benefits of using 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", "You are an AI expert helping developers."),
    ("user", "{question}")
])

chain = prompt | llm

response = chain.invoke({
    "question": "How do you optimize a 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("Starting document processing")

    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()

Database Integration (SQLAlchemy and 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 complete",
                trainsize=len(Xtrain),
                testsize=len(Xtest)
            )

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


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

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


            logfire.info(
                "Model evaluation",
                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 created", dimensions=len(embedding))

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

            logfire.info(
                "Documents found",
                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 assembled", charcount=len(context))


        with logfire.span("llm-generation"):
            response = await generateresponse(question, context)
            logfire.info(
                "Response generated",
                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. Use Meaningful Spans

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

Give spans descriptive names and add attributes useful for filtering.

2. Structured Logging

Always use structured data instead of 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 for LLMs

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. Use Sampling in Production

For high-volume applications, enable sampling to avoid overloading storage:

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

Deployment and 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

Conclusion

Logfire provides a powerful and easy-to-use observability solution for Python applications, especially for AI/LLM applications. With native integrations into Pydantic, FastAPI, OpenAI, and other popular Python libraries, Logfire reduces the boilerplate code typically required for monitoring. The SQL query feature on the dashboard enables flexible observability data analysis, and the auto-instrumentation capability makes setup extremely fast. For teams building AI applications in the Python ecosystem, Logfire is a worthy choice as a primary observability platform.