Langfuse

Langfuse's drop-in OpenAI wrapper sends each call to plugboard and records it as a generation in Langfuse.

Connect

Import OpenAI from langfuse.openai instead of openai. The rest is the OpenAI SDK.

client = OpenAI(base_url=os.environ["PLUGBOARD_BASE_URL"], api_key=os.environ["PLUGBOARD_KEY"], max_retries=0)

The example sets PLUGBOARD_BASE_URL to http://127.0.0.1:4010/v1.

Tracing

The SDK sends traces to the Langfuse server named by LANGFUSE_BASE_URL, with LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY. Set all three before the first call. Without them the wrapper has nothing to send to, and LANGFUSE_TRACING_ENABLED=false turns tracing off.

os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-example"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-example"
os.environ["LANGFUSE_BASE_URL"] = fake.url  # your Langfuse URL, for example https://cloud.langfuse.com

The SDK exports each generation as OTLP spans to <LANGFUSE_BASE_URL>/api/public/otel/v1/traces. A generation carries the model name you sent (fast), the prompt, the answer and the token usage plugboard returned. The SDK exports in the background, so flush before a short script exits:

response = client.chat.completions.create(model=model, messages=[{"role": "user", "content": PROMPT}], max_completion_tokens=1024, name=TRACE_NAME)
# The SDK exports in the background: send what is queued before the script ends.
get_client().flush()

A tool call. TOOL is a function schema and TOOL_RESULT the tool's output:

messages = [{"role": "user", "content": TOOL_PROMPT}]
first = client.chat.completions.create(model=model, messages=messages, tools=[TOOL], max_completion_tokens=1024)
call = first.choices[0].message.tool_calls[0]
# The assistant turn goes back as the SDK typed it, then the result under the same call id.
messages.append(first.choices[0].message.model_dump(exclude_none=True))
messages.append({"role": "tool", "tool_call_id": call.id, "content": TOOL_RESULT})
second = client.chat.completions.create(model=model, messages=messages, tools=[TOOL], max_completion_tokens=1024)

Streaming, with usage:

stream = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": PROMPT}],
    max_completion_tokens=1024,
    stream=True,
    stream_options={"include_usage": True},
)

A script with a text reply, a stream and a tool round trip is in examples/langfuse/:

cd examples/langfuse
PLUGBOARD_KEY=<app key> uv run example.py

The script traces only when the three Langfuse variables are set.

What works

Everything the OpenAI SDK does through plugboard, with the wrapper's name argument labelling each generation. Streamed calls are traced with their usage when the request asks for it (stream_options={"include_usage": True}).

Limits

• Beta OpenAI APIs are not traced by the wrapper.