Langfuse Integration

Integrate CoreValue AI Gateway with Langfuse to access LLM providers with observability and LLM tracing.

Introduction

Langfuse is an open-source LLM observability and analytics platform that provides tracing, monitoring, and analytics for LLM applications.

This integration requires only two changes to your existing Langfuse code - updating the base URL and API key.

Integration Steps

Log into CoreValue or create an account. Once you have an account, you can generate an API key here.

Create a .env file in your project:

COVA_API_KEY=sk-cova-XXXXXXXXXXXXXXXX

pip install langfuse python-dotenv

Use Langfuse's OpenAI client wrapper with CoreValue's base URL:

import os
from dotenv import load_dotenv
from langfuse.openai import openai

# Load environment variables
load_dotenv()

# Create an OpenAI client with CoreValue's base URL
client = openai.OpenAI(
    api_key=os.getenv("COVA_API_KEY"),
    base_url="https://gateway.corevalue.dev/v1"
)

Your existing Langfuse code continues to work without any changes:

# Make a chat completion request
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Tell me a fun fact about space."}
    ],
    name="fun-fact-request"  # Optional: Name of the generation in Langfuse
)

# Print the assistant's reply
print(response.choices[0].message.content)

You can find all supported models at corevalue.dev/models.

All your Langfuse requests are now visible in your CoreValue dashboard.

Request/response bodies
Latency metrics
Token usage and costs
Model performance analytics
Error tracking
LLM traces and spans in Langfuse
Session tracking

While you're here, why not give us a star on GitHub? It helps us a lot!

Complete Working Example

#!/usr/bin/env python3

import os
from dotenv import load_dotenv
from langfuse.openai import openai

# Load environment variables
load_dotenv()

# Create an OpenAI client with CoreValue's base URL
client = openai.OpenAI(
    api_key=os.getenv("COVA_API_KEY"),
    base_url="https://gateway.corevalue.dev/v1"
)

# Make a chat completion request
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Tell me a fun fact about space."}
    ],
    name="fun-fact-request"  # Optional: Name of the generation in Langfuse
)

# Print the assistant's reply
print(response.choices[0].message.content)

Streaming Responses

Langfuse supports streaming responses with full observability:

# Streaming example
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Write a short story about a robot learning to code."}
    ],
    stream=True,
    name="streaming-story"
)

print("🤖 Assistant (streaming):")
for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)
print("\n")

Nested Example

import os
from dotenv import load_dotenv
from langfuse import observe
from langfuse.openai import openai

load_dotenv()

client = openai.OpenAI(
    base_url="https://gateway.corevalue.dev/v1",
    api_key=os.getenv("COVA_API_KEY"),
)

@observe()  # This decorator enables tracing of the function
def analyze_text(text: str):
    # First LLM call: Summarize the text
    summary_response = summarize_text(text)
    summary = summary_response.choices[0].message.content

    # Second LLM call: Analyze the sentiment of the summary
    sentiment_response = analyze_sentiment(summary)
    sentiment = sentiment_response.choices[0].message.content

    return {
        "summary": summary,
        "sentiment": sentiment
    }

@observe()  # Nested function to be traced
def summarize_text(text: str):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You summarize texts in a concise manner."},
            {"role": "user", "content": f"Summarize the following text:\n{text}"}
        ],
        name="summarize-text"
    )

@observe()  # Nested function to be traced
def analyze_sentiment(summary: str):
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You analyze the sentiment of texts."},
            {"role": "user", "content": f"Analyze the sentiment of the following summary:\n{summary}"}
        ],
        name="analyze-sentiment"
    )

# Example usage
text_to_analyze = "OpenAI's GPT-4 model has significantly advanced the field of AI, setting new standards for language generation."
analyze_text(text_to_analyze)