Langfuse

# Local secrets — never commit the real .env
.env
# ...but keep the committed template (root .gitignore's `.env.*` would hide it).
!.env.sample

"""Langfuse: trace one LLM call into a self-hosted Langfuse, then read it back.

The bundled docker-compose starts a full Langfuse stack and bootstraps a project
with fixed API keys. This app wraps a Claude call in a Langfuse generation span,
flushes it, then queries the Langfuse API until the trace shows up — proving the
round trip end to end (not just "it didn't error").

Run (with the bundled Langfuse):
    docker compose up --build
"""

import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request

from anthropic import Anthropic
from dotenv import load_dotenv
from langfuse import Langfuse

load_dotenv()

HOST = os.environ.get("LANGFUSE_HOST", "http://localhost:3000")
PUBLIC = os.environ["LANGFUSE_PUBLIC_KEY"]
SECRET = os.environ["LANGFUSE_SECRET_KEY"]
MODEL = os.environ.get("MODEL", "claude-opus-4-8")


def _get(path: str):
    auth = base64.b64encode(f"{PUBLIC}:{SECRET}".encode()).decode()
    req = urllib.request.Request(f"{HOST}{path}", headers={"Authorization": f"Basic {auth}"})
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.status, json.load(r)


def wait_until_ready(seconds: int = 300) -> None:
    """Langfuse runs DB migrations on first boot — wait for it to come up."""
    for _ in range(seconds):
        try:
            req = urllib.request.Request(f"{HOST}/api/public/health")
            with urllib.request.urlopen(req, timeout=5) as r:
                if r.status == 200:
                    return
        except (urllib.error.URLError, OSError):
            pass
        time.sleep(1)
    raise SystemExit(f"Langfuse did not become ready at {HOST}")


def fetch_latest_trace():
    """Ingestion is async (worker -> ClickHouse), so poll for the trace."""
    for _ in range(30):
        _, data = _get("/api/public/traces?limit=1")
        if data.get("data"):
            return data["data"][0]
        time.sleep(2)
    return None


def main() -> None:
    question = " ".join(sys.argv[1:]) or "Summarize what Langfuse does in one sentence."

    print(f"waiting for Langfuse at {HOST} ...")
    wait_until_ready()

    langfuse = Langfuse()  # reads LANGFUSE_HOST / _PUBLIC_KEY / _SECRET_KEY
    client = Anthropic()  # reads ANTHROPIC_API_KEY

    # Wrap the model call in a generation span so it lands as a trace in Langfuse.
    with langfuse.start_as_current_observation(
        as_type="generation", name="plan", model=MODEL, input=question
    ) as gen:
        resp = client.messages.create(
            model=MODEL, max_tokens=256, messages=[{"role": "user", "content": question}]
        )
        answer = resp.content[0].text
        gen.update(
            output=answer,
            usage_details={"input": resp.usage.input_tokens, "output": resp.usage.output_tokens},
        )

    langfuse.flush()
    print("model says:", answer)

    print("\nreading the trace back from Langfuse ...")
    trace = fetch_latest_trace()
    if trace is None:
        print(f"trace not ingested yet (it is async) — check the UI at {HOST}")
        return
    print(f"  trace id: {trace['id']}")
    print(f"  name:     {trace.get('name')}")
    print(f"  input:    {trace.get('input')}")
    print(f"  output:   {trace.get('output')}")
    print(f"  UI:       {HOST}/project/{os.environ.get('LANGFUSE_PROJECT_ID', '')}/traces/{trace['id']}")


if __name__ == "__main__":
    main()

# A self-contained Langfuse v3 stack + an app that traces one LLM call into it.
#   docker compose up --build
#
# Adapted from the official Langfuse self-host compose. The infra secrets below
# are demo defaults — fine for a local throwaway stack, NOT for production. The
# project + API keys are bootstrapped on first boot via LANGFUSE_INIT_* so the
# app can authenticate without anyone clicking through the UI.
services:
  app:
    build: .
    depends_on:
      langfuse-web:
        condition: service_started
    environment:
      # Point the SDK at the bundled Langfuse and the bootstrapped project keys.
      LANGFUSE_HOST: http://langfuse-web:3000
      LANGFUSE_PUBLIC_KEY: pk-lf-00000000-0000-0000-0000-000000000000
      LANGFUSE_SECRET_KEY: sk-lf-00000000-0000-0000-0000-000000000000
      LANGFUSE_PROJECT_ID: demo-project
      MODEL: claude-opus-4-8
      ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:-}
    # The question to trace — override on the command line if you like.
    command: ["Summarize what Langfuse does in one sentence."]

  langfuse-worker:
    image: docker.io/langfuse/langfuse-worker:3
    restart: always
    depends_on: &langfuse-depends-on
      postgres:
        condition: service_healthy
      minio:
        condition: service_healthy
      redis:
        condition: service_healthy
      clickhouse:
        condition: service_healthy
    environment: &langfuse-worker-env
      DATABASE_URL: postgresql://postgres:postgres@postgres:5432/postgres
      SALT: mysalt
      ENCRYPTION_KEY: "0000000000000000000000000000000000000000000000000000000000000000"
      TELEMETRY_ENABLED: "false"
      CLICKHOUSE_MIGRATION_URL: clickhouse://clickhouse:9000
      CLICKHOUSE_URL: http://clickhouse:8123
      CLICKHOUSE_USER: clickhouse
      CLICKHOUSE_PASSWORD: clickhouse
      CLICKHOUSE_CLUSTER_ENABLED: "false"
      LANGFUSE_S3_EVENT_UPLOAD_BUCKET: langfuse
      LANGFUSE_S3_EVENT_UPLOAD_REGION: auto
      LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: minio
      LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: miniosecret
      LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://minio:9000
      LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true"
      LANGFUSE_S3_EVENT_UPLOAD_PREFIX: events/
      REDIS_HOST: redis
      REDIS_PORT: 6379
      REDIS_AUTH: myredissecret

  langfuse-web:
    image: docker.io/langfuse/langfuse:3
    restart: always
    depends_on: *langfuse-depends-on
    ports:
      - "3000:3000"
    environment:
      <<: *langfuse-worker-env
      NEXTAUTH_URL: http://localhost:3000
      NEXTAUTH_SECRET: mysecret
      # Bootstrap a project with fixed API keys on first boot (no UI needed).
      LANGFUSE_INIT_ORG_ID: demo-org
      LANGFUSE_INIT_ORG_NAME: Demo Org
      LANGFUSE_INIT_PROJECT_ID: demo-project
      LANGFUSE_INIT_PROJECT_NAME: Demo Project
      LANGFUSE_INIT_PROJECT_PUBLIC_KEY: pk-lf-00000000-0000-0000-0000-000000000000
      LANGFUSE_INIT_PROJECT_SECRET_KEY: sk-lf-00000000-0000-0000-0000-000000000000
      LANGFUSE_INIT_USER_EMAIL: demo@example.com
      LANGFUSE_INIT_USER_NAME: Demo
      LANGFUSE_INIT_USER_PASSWORD: demodemodemo

  clickhouse:
    image: docker.io/clickhouse/clickhouse-server
    restart: always
    user: "101:101"
    environment:
      CLICKHOUSE_DB: default
      CLICKHOUSE_USER: clickhouse
      CLICKHOUSE_PASSWORD: clickhouse
    volumes:
      - langfuse_clickhouse_data:/var/lib/clickhouse
      - langfuse_clickhouse_logs:/var/log/clickhouse-server
    healthcheck:
      test: wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1
      interval: 5s
      timeout: 5s
      retries: 10
      start_period: 1s

  minio:
    image: cgr.dev/chainguard/minio
    restart: always
    entrypoint: sh
    command: -c 'mkdir -p /data/langfuse && minio server --address ":9000" --console-address ":9001" /data'
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniosecret
    volumes:
      - langfuse_minio_data:/data
    healthcheck:
      test: ["CMD", "mc", "ready", "local"]
      interval: 1s
      timeout: 5s
      retries: 5
      start_period: 1s

  redis:
    image: docker.io/redis:7
    restart: always
    command: >
      --requirepass myredissecret
      --maxmemory-policy noeviction
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 3s
      timeout: 10s
      retries: 10

  postgres:
    image: docker.io/postgres:17
    restart: always
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 3s
      timeout: 3s
      retries: 10
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: postgres
      TZ: UTC
      PGTZ: UTC
    volumes:
      - langfuse_postgres_data:/var/lib/postgresql/data

volumes:
  langfuse_postgres_data:
  langfuse_clickhouse_data:
  langfuse_clickhouse_logs:
  langfuse_minio_data:

# Minimal image for the Langfuse sample app (the Langfuse stack itself runs as
# separate compose services).
FROM python:3.13-slim

ENV PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

ENTRYPOINT ["python", "app.py"]

langfuse>=3,<4
anthropic>=0.40,<1
python-dotenv>=1.1,<2