Python SDK (Low-level)

This is a Python SDK used to send LLM data to Langfuse in a convenient way. It uses a worker Thread and an internal queue to manage requests to the Langfuse backend asynchronously. Hence, the SDK adds only minimal latency to your application.

For most use cases, you should check out the decorator-based SDK, which is more convenient and easier to use. This SDK is more low-level and is only recommended if you need more control over the request process.

Installation

%pip install langfuse --upgrade

Initialize Client

To start, initialize the client by providing your credentials. You can set the credentials either as environment variables or constructor arguments.

If you are self-hosting Langfuse or using the US data region, don’t forget to configure LANGFUSE_HOST.

To verify your credentials and host, use the langfuse.auth_check() function.

 import os
 
# get keys for your project from https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
 
# your openai key
os.environ["OPENAI_API_KEY"] = ""
 
# Your host, defaults to https://cloud.langfuse.com
# For US data region, set to "https://us.cloud.langfuse.com"
# os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

from langfuse import Langfuse
 
langfuse = Langfuse()

Tracing

The Langfuse SDK and UI are designed to support complex LLM features which contain for example vector database searches and multiple LLM calls. For that, it is very convenient to nest or chain the SDK. Understanding a small number of terms makes it easy to integrate with Langfuse.

Traces

A Trace represents a single execution of a LLM feature. It is a container for all succeeding objects.

Observations

Each Trace can contain multiple Observations to record individual steps of an execution. There are different types of Observations:

• Events are the basic building block. They are used to track discrete events in a Trace.
• Spans track time periods and include an end_time.
• Generations are a specific type of Spans which are used to record generations of an AI model. They contain additional metadata about the model, LLM token and cost tracking, and the prompt/completions are specifically rendered in the langfuse UI.

Example

TRACE
|
|-- SPAN: Retrieval
|   |
|   |-- GENERATION: Vector DB Query Creation
|   |
|   |-- SPAN: Data Fetching
|   |
|   |-- EVENT: Data Summary Creation
|
|-- GENERATION: Output Generation

trace = langfuse.trace(name = "llm-feature")
retrieval = trace.span(name = "retrieval")
retrieval.generation(name = "query-creation")
retrieval.span(name = "vector-db-search")
retrieval.event(name = "db-summary")
trace.generation(name = "user-output");

Traces

Traces are the top-level entity in the Langfuse API. They represent an execution flow in a LLM application usually triggered by an external event.

trace = langfuse.trace(
    name = "docs-retrieval",
    user_id = "user__935d7d1d-8625-4ef4-8651-544613e7bd22",
    metadata = {
        "email": "user@langfuse.com",
    },
    tags = ["production"]
)

Traces can be updated:

# option 1: using trace object
trace.update(
    input="Hi there"
)
 
# option 2: via trace_id, trace is upserted on id
langfuse.trace(id=trace.id, output="Hi 👋")

You can get the url of a trace in the Langfuse interface. Helpful in interactive use or when adding this url to your logs.

trace.get_trace_url()

Span

Spans represent durations of units of work in a trace.

Parameters of langfuse.span():

Use trace or observation objects to create child spans:

# create span, sets start_time
span = trace.span(
    name="embedding-search",
    metadata={"database": "pinecone"},
    input = {'query': 'This document entails the OKR goals for ACME'},
)
 
# function, mocked
# retrieved_documents = retrieveDoc()
retrieved_documents = {"response": "[{'name': 'OKR Engineering', 'content': 'The engineering department defined the following OKR goals...'},{'name': 'OKR Marketing', 'content': 'The marketing department defined the following OKR goals...'}]"}
 
# update span and sets end_time
span.end(
    output=retrieved_documents
);

Other span methods:

• span.update(), does not change end_time if not explicitly set

Alternatively, if using the Langfuse objects is not convenient, you can use the langfuse client, trace_id and (optionally) parent_observation_id to create spans, and id to upsert a span.

trace_id = trace.id
 
# create span
span = langfuse.span(
    trace_id=trace_id,
    name="initial name"
)
 
# update span, upserts on id
langfuse.span(
    id=span.id,
    name="updated name"
)
 
# create new nested span
langfuse.span(
    trace_id=trace_id,
    parent_observation_id=span.id,
    name="nested span"
)

Generation

Generations are used to log generations of AI models. They contain additional metadata about the model, the prompt/completion, the cost of executing the model and are specifically rendered in the langfuse UI.

Use trace or observation objects to create child generations:

# creates generation
generation = trace.generation(
    name="summary-generation",
    model="gpt-3.5-turbo",
    model_parameters={"maxTokens": "1000", "temperature": "0.9"},
    input=[{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Please generate a summary of the following documents \nThe engineering department defined the following OKR goals...\nThe marketing department defined the following OKR goals..."}],
    metadata={"interface": "whatsapp"}
)
 
# execute model, mocked here
# chat_completion = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello world"}])
chat_completion = {
    "completion":"The Q3 OKRs contain goals for multiple teams...",
    "usage":{"input": 50, "output": 49, "unit":"TOKENS"}
}
 
# update span and sets end_time
generation.end(
    output=chat_completion["completion"],
    usage=chat_completion["usage"],
);

Other generation methods:

• generation.update(), does not change end_time if not explicitly set

See documentation of spans above on how to use the langfuse client and ids if you cannot use the Langfuse objects to trace your application. This also fully applies to generations.

Events

Events are used to track discrete events in a trace.

Use trace or observation objects to create child generations:

event = span.event(
    name="chat-docs-retrieval",
    metadata={"key": "value"},
    input = {"key": "value"},
    output = {"key": "value"}
)

See documentation of spans above on how to use the langfuse client and ids if you cannot use the Langfuse objects to trace your application. This also fully applies to events.

Scores

Scores are used to evaluate single executions/traces. They can be created via Annotation in the Langfuse UI or via the SDKs.

If the score relates to a specific step of the trace, specify the observation_id.

# via {trace, span, event, generation}.score
trace.score(
    name="user-explicit-feedback",
    value=1,
    comment="I like how personalized the response is",
)
 
# using the trace.id
langfuse.score(
    trace_id=trace.id,
    name="user-explicit-feedback",
    value=1,
    comment="I like how personalized the response is"
)
 
# scoring a specific observation
langfuse.score(
    trace_id=trace.id,
    observation_id=span.id,
    name="user-explicit-feedback",
    value=1,
    comment="I like how personalized the response is"
)

Additional configurations

Shutdown behavior

The Langfuse SDK executes network requests in the background on a separate thread for better performance of your application. This can lead to lost events in short lived environments like NextJs cloud functions or AWS Lambda functions when the Python process is terminated before the SDK sent all events to our backend.

To avoid this, ensure that the langfuse.flush() function is called before termination. This method is waiting for all tasks to have completed, hence it is blocking.

langfuse.flush()

Releases and versions

Track releases in Langfuse to relate traces in Langfuse with the versioning of your application. This can be done by either providing the environment variable LANGFUSE_RELEASE, instantiating the client with the release, or setting it as a trace parameter.

If no release is set, this defaults to common system environment names.

# The SDK will automatically include the env variable.
os.environ["LANGFUSE_RELEASE"] = "ba7816b..." # <- example, github sha
 
# Alternatively, use the constructor of the SDK
langfuse = Langfuse(release="ba7816b")
 
# Alternatively, set it when creating a trace
langfuse.trace(release="ba7816b")

To track versions of individual pieces of you application apart from releases, use the version parameter on all observations. This is for example useful to track the effect of changed prompts.

# works the same for spans, generations, events
langfuse.span(name="retrieval", version="<version>")

Troubleshooting

Debug mode

Enable debug mode to get verbose logs.

langfuse = Langfuse(debug=True)

Alternatively, set the debug mode via the environment variable LANGFUSE_DEBUG=True.

Configuration/authentication problems

Use auth_check() to verify that your host and api credentials are correct.

langfuse.auth_check()

Google Cloud Functions

When using Langfuse in a Google Cloud Function or a Firebase Function, the underlying managed Python runtime has issues with threading whenever threads are spawned off the main scope and not inside the actual function scope. See here. Since Langfuse uses background threads to deliver the observability events, this will lead to incomplete traces.

Make sure to initialize Langfuse always inside the function body. If you want to reuse the created Langfuse clients in different modules, use lazy initialization of the Langfuse client to ensure the actual initialization occurs inside the function execution context.

import functions_framework
from langfuse import Langfuse
 
 
# Lazy initialization of the Langfuse client to allow imports in other modules
def get_langfuse():
    if not hasattr(get_langfuse, "langfuse"):
        get_langfuse.langfuse = Langfuse(debug=True)
 
    return get_langfuse.langfuse
 
 
# Google Cloud Function
@functions_framework.http
def hello_world(request):
    langfuse = get_langfuse()
 
    response = "Hello world!"
    langfuse.trace(name="my-cloud-function", output=response)
 
    langfuse.flush()  # Ensure all events are sent before the function terminates
 
    return response

Upgrading from v1.x.x to v2.x.x

v2 is a major release with breaking changes to simplify the SDK and make it more consistent. We recommend to upgrade to v2 as soon as possible.

You can automatically migrate your codebase using grit, either online or with the following CLI command:

npx -y @getgrit/launcher apply langfuse_v2

The grit binary executes entirely locally with AST-based transforms. Be sure to audit its changes: we suggest ensuring you have a clean working tree beforehand, and running git add --patch afterwards.

If your Jupyter Notebooks are not in source control, it might be harder to track changes. You may want to copy each cell individually into grit’s web interface, and paste the output back in.

Remove Pydantic interfaces

We like Pydantic, but it made the Langfuse SDK interfaces messy. Therefore, we removed the objects from the function signatures and replaced them with named parameters.

All parameters are still validated using Pydantic internally. If the validation fails, errors are logged instead of throwing exceptions.

Pydantic objects

v1.x.x

from langfuse.model import CreateTrace
 
langfuse.trace(CreateTrace(name="My Trace"))

v2.x.x

langfuse.trace(name="My Trace")

Pydantic Enums

v1.x.x

from langfuse.model import InitialGeneration
from langfuse.api.resources.commons.types.observation_level import ObservationLevel
 
langfuse.generation(InitialGeneration(level=ObservationLevel.ERROR))

v2.x.x

langfuse.generation(level="ERROR")

Rename prompt and completion to input and output

To ensure consistency throughout Langfuse, we have renamed the prompt and completion parameters in the generation function to input and output, respectively. This change brings them in line with the rest of the Langfuse API.

Snake case parameters

To increase consistency, all parameters are snake case in v2.

• trace_id instead of traceId
• start_time instead of startTime
• end_time instead of endTime
• completion_start_time instead of completionStartTime
• status_message instead of statusMessage
• user_id instead of userId
• session_id instead of sessionId
• parent_observation_id instead of parentObservationId
• model_parameters instead of modelParameters

More generalized usage object

We improved the flexibility of the SDK by allowing you to ingest any type of usage while still supporting the OpenAI-style usage object.

v1.x.x

 
from langfuse.model import InitialGeneration, Usage
 
 langfuse.generation(
    InitialGeneration(
        name="my-generation",
        usage=Usage(promptTokens=50, completionTokens=49),
    )
)

v2.x.x

The usage object supports the OpenAi structure with {promptTokens, completionTokens, totalTokens} and a more generic version {input, output, total, unit} where unit can be of value "TOKENS" or "CHARACTERS". For some models the token counts and costs are automatically calculated by Langfuse. Create an issue to request support for other units and models.

# Generic style
langfuse.generation(
    name="my-claude-generation",
    usage={
        "input": 50,
        "output": 49,
        "total": 99,
        "unit": "TOKENS"
    },
)
 
# OpenAI style
langfuse.generation(
    name="my-openai-generation",
    usage={
        "promptTokens": 50,
        "completionTokens": 49,
        "totalTokens": 99
    }, # defaults to "TOKENS" unit
)
 
# set ((input and/or output) or total), total is calculated automatically if not set

FastAPI

For engineers working with FastAPI, we have a short example, of how to use it there.

%pip install fastapi --upgrade

Here is an example of how to initialise FastAPI and register the langfuse.flush() method to run at shutdown. With this, your Python environment will only terminate once Langfuse received all the events.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Query, BackgroundTasks
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Operation on startup
 
    yield  # wait until shutdown
 
    # Flush all events to be sent to Langfuse on shutdown and terminate all Threads gracefully. This operation is blocking.
    langfuse.flush()
 
 
app = FastAPI(lifespan=lifespan)

langfuse = Langfuse()
 
@app.get("/generate/",tags=["APIs"])
async def campaign(prompt: str = Query(..., max_length=20)):
  # call to a LLM
  generation = langfuse.generation(
      name="llm-feature",
      metadata="test",
      input=prompt
  )
  return True