Langfuse

pixeltable.insert
└─ pixeltable.store.insert_rows
   ├─ pixeltable.row                      (one per row, debug)
   ├─ pixeltable.udf.chat_completions     (one per LLM cell, debug)
   │  └─ ChatCompletion                    (generation: model, tokens, cost)
   └─ pixeltable.sa.insert_rows

pip install 'pixeltable[otel]' openai langfuse openinference-instrumentation-openai

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

import base64
import os

import opentelemetry.instrumentation.pixeltable as pxt_otel
from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry import trace

base_url = os.environ.get('LANGFUSE_BASE_URL', 'https://cloud.langfuse.com').rstrip('/')
token = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

pxt_otel.init(
    endpoint=f'{base_url}/api/public/otel',  # http/protobuf exporter appends /v1/traces
    headers=f'Authorization=Basic {token},x-langfuse-ingestion-version=4',  # real-time ingestion
    service_name='pixeltable-app',  # default service.name is pixeltable
    span_level='debug',  # info: operation, catalog, store, SQL; debug adds row, UDF, and file spans
    metrics=False,  # an endpoint turns metrics on; Langfuse ingests traces
)
OpenAIInstrumentor().instrument()  # generation spans with model, tokens, and cost

export OTEL_EXPORTER_OTLP_ENDPOINT="$LANGFUSE_BASE_URL/api/public/otel"
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic $(printf '%s' "$LANGFUSE_PUBLIC_KEY:$LANGFUSE_SECRET_KEY" | base64 | tr -d '\n'),x-langfuse-ingestion-version=4"
export OTEL_METRICS=false
export OTEL_SPAN_LEVEL=debug

from langfuse import Langfuse, propagate_attributes
from langfuse.span_filter import is_default_export_span


def export_span(span) -> bool:
    scope = span.instrumentation_scope
    return is_default_export_span(span) or (scope is not None and scope.name == 'pixeltable')


langfuse = Langfuse(should_export_span=export_span)  # reads LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_BASE_URL

from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.instrumentation.pixeltable import PixeltableInstrumentor

PixeltableInstrumentor().instrument(span_level='debug')
OpenAIInstrumentor().instrument()

import pixeltable as pxt
import pixeltable.functions as pxtf

# my_udfs.py
import openai
from langfuse import observe, propagate_attributes

import pixeltable as pxt


@pxt.udf
@observe(name='word_count')
def word_count(text: str) -> int:
    return len(text.split())


@pxt.udf
@observe(name='headline')
async def headline(text: str, row_id: int) -> str:
    # metadata values are strings of 200 characters or fewer
    with propagate_attributes(metadata={'row_id': str(row_id)}):
        # a client per call: a notebook or served app runs each query on a new event loop
        async with openai.AsyncOpenAI() as client:
            response = await client.chat.completions.create(
                model='gpt-4o-mini',
                max_tokens=20,
                messages=[{'role': 'user', 'content': f'Write a 5-word headline for: {text}'}],
            )
    return response.choices[0].message.content

import pixeltable as pxt
import pixeltable.functions as pxtf

from my_udfs import headline, word_count

TableModel = pxt.model_base()


class Docs(TableModel, name='docs'):
    id: pxt.Int
    body: pxt.String
    summary = pxtf.openai.chat_completions(
        messages=[{'role': 'user', 'content': 'Summarize in one sentence: ' + body}],
        model='gpt-4o-mini',
    ).choices[0].message.content
    words = word_count(body)
    title = headline(body, id)  # headline() adds row_id to its span metadata

rows = [
    {'id': 1, 'body': 'Pixeltable is declarative data infrastructure for multimodal AI apps.'},
    {'id': 2, 'body': 'Langfuse records prompts, completions, token usage, and cost.'},
]
t = pxt.get_table('my_app.docs')
with langfuse.start_as_current_observation(name='ingest-docs', as_type='span'):
    with propagate_attributes(
        trace_name='ingest-docs',
        tags=['pixeltable'],
        metadata={'table': 'my_app.docs'},
    ):
        t.insert(rows)
    print(langfuse.get_current_trace_id())
langfuse.flush()