Langfuse

Langfuse Introduction¶

Langfuse is an open-source observability platform specifically designed for LLM (Large Language Model) applications, providing the following core functionalities:

End-to-End Trace Tracking
- Records LLM call chains (Prompt→LLM→Output)
- Supports multi-step complex workflow tracking
Metrics Monitoring
- Token usage statistics
- Request latency monitoring
- Cost calculation (based on model pricing)
Data Annotation & Analysis
- Manual annotation functionality
- Output quality scoring
- A/B testing support

Environment Preparation¶

Dependency Installation¶

# Core SDK
pip install langfuse

# Optional: Async support (recommended for production)
pip install langfuse[async]

# Development tools (for testing)
pip install pytest langfuse-test

Langfuse Integration Access Instructions

Langfuse supports a large number of integrations. Currently, we have tested the following data access methods, and support for more models is pending further testing.

Dify
LangChain
Ollama
Gemini
OpenAI

Python SDK Integration Examples¶

Client Initialization

Environment Variable Integration (Recommended)Parameter Construction

LANGFUSE_PUBLIC_KEY="YOUR_LLM_APP_NAME"
LANGFUSE_SECRET_KEY="YOUR_LLM_APP_TOKEN"
LANGFUSE_HOST="https://llm-openway.truewatch.com"

from langfuse import Langfuse
langfuse = Langfuse(
    public_key="YOUR_LLM_APP_NAME",
    secret_key="YOUR_LLM_APP_TOKEN",
    host="https://llm-openway.truewatch.com"
)

Integration Verification

from langfuse import Langfuse

# Initialize with constructor arguments
langfuse = Langfuse(
    public_key="YOUR_LLM_APP_NAME",
    secret_key="YOUR_LLM_APP_TOKEN",
    host="https://llm-openway.truewatch.com"
)

# Verify connection, do not use in production as this is a synchronous call
if langfuse.auth_check():
    print("Langfuse client is authenticated and ready!")

If failed, we'll get following error:

langfuse.api.resources.commons.errors.unauthorized_error.UnauthorizedError: status_code: 401, body: {}

Application Examples¶

Here are some simple examples.

Ollama Integration¶

If we have Ollama deployed locally, we can use Langfuse to track Ollama API calls:

import os

from langfuse.openai import OpenAI, AsyncOpenAI, AzureOpenAI, AsyncAzureOpenAI

os.environ["LANGFUSE_PUBLIC_KEY"] = "YOUR_LLM_APP_NAME"
os.environ["LANGFUSE_SECRET_KEY"] = "YOUR_LLM_APP_TOKEN"
os.environ["LANGFUSE_HOST"] = "https://llm-openway.truewatch.com"

# Configure the OpenAI client to use http://localhost:11434/v1 as base url

client = OpenAI(
    base_url = 'http://localhost:11434/v1', # local deployed ollama service
    api_key='ollama', # required, but unused
)

stream=False # use stream mode

response = client.chat.completions.create(
        model="gemma3:4b", # specify gemma3:4b or other model such as llama3.1:latest
        stream=stream,
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Explain how nuclear fusion/fission works."},
            ]
        )

if stream:
    for chk in response:
        content = chk.choices[0].delta.content
        if content is not None:
            print(content, end="", flush=True)
else:
    print(response)

DeepSeek Integration¶

import os
from langfuse.openai import OpenAI
from langfuse import observe
from langfuse import get_client

os.environ["LANGFUSE_PUBLIC_KEY"] = "YOUR_LLM_APP_NAME" 
os.environ["LANGFUSE_SECRET_KEY"] = "YOUR_LLM_APP_TOKEN" 
os.environ["LANGFUSE_HOST"] = "https://llm-openway.truewatch.com"

Your DeepSeek API key (get it from https://platform.deepseek.com/api_keys)

os.environ["DEEPSEEK_API_KEY"] = "YOUR_DEEPSEEK_API_KEY"  # Replace with your DeepSeek API key

client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.getenv('DEEPSEEK_API_KEY'),
)

langfuse = get_client()

@observe()
def my_llm_call(input):
    completion = client.chat.completions.create(
        name="story-generator",
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "You are a creative storyteller."},
            {"role": "user", "content": input}
        ],
        metadata={"genre": "adventure"},
    )
    return completion.choices[0].message.content

with langfuse.start_as_current_span(name="my-ds-trace") as span:
    # Run your application here
    output = my_llm_call("Tell me a short story about a token that got lost on its way to the language model. Answer in 100 words or less.")

    # Pass additional attributes to the span
    span.update_trace(
        input=input,
        output=output,
        user_id="user_123",
        session_id="session_abc",
        tags=["agent", "my-trace"],
        metadata={"email": "[email protected]"},
        version="1.0.0"
        )

# Flush events in short-lived applications
langfuse.flush()

For more Langfuse integration examples, see here.

Additional References¶

Langfuse Python SDK