Instrument before importing Langfuse so the HTTP client is patched.
Overview
Use the Respan Langfuse instrumentor to redirect Langfuse traces to Respan without changing your Langfuse usage. You can keep using @observe and the Langfuse SDK as usual.
Quickstart
Step 1: Get a Respan API key
Create an API key in the Respan dashboard.
Step 2: Install packages
pip install langfuse respan-instrumentation-langfuse
Step 3: Set environment variables
Create a .env or export environment variables:
RESPAN_API_KEY=your-respan-api-key
RESPAN_BASE_URL=https://api.respan.ai/api
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key
LANGFUSE_SECRET_KEY=your-langfuse-secret-key
LANGFUSE_HOST=https://cloud.langfuse.com
Examples
Below are three common Langfuse patterns using the Respan instrumentor.
Basic decorator
Instrument the Langfuse client and use @observe.
import dotenv
dotenv.load_dotenv(".env", override=True)
import os
from respan_instrumentation_langfuse import LangfuseInstrumentor
# Instrument BEFORE importing Langfuse
instrumentor = LangfuseInstrumentor()
instrumentor.instrument(
api_key=os.environ["RESPAN_API_KEY"],
endpoint=os.environ["RESPAN_BASE_URL"] + "/v1/traces/ingest",
)
from langfuse import Langfuse, observe
langfuse = Langfuse(
public_key="pk-lf-...",
secret_key="sk-lf-...",
host="https://cloud.langfuse.com",
)
@observe()
def process_query(query: str):
return f"Response to: {query}"
result = process_query("Hello World")
print(result)
langfuse.flush()
Generation tracing
Mark a function as a generation with @observe(as_type="generation").
import os
from respan_instrumentation_langfuse import LangfuseInstrumentor
os.environ["RESPAN_API_KEY"] = "your-api-key"
instrumentor = LangfuseInstrumentor()
instrumentor.instrument(
api_key=os.environ["RESPAN_API_KEY"],
)
from langfuse import Langfuse, observe
langfuse = Langfuse(
public_key="pk-lf-...",
secret_key="sk-lf-...",
)
@observe(as_type="generation")
def generate_response(prompt: str):
return f"Generated: {prompt}"
result = generate_response("Write a poem")
print(result)
langfuse.flush()
Nested traces
Create parent-child relationships with nested @observe functions.
import os
from respan_instrumentation_langfuse import LangfuseInstrumentor
os.environ["RESPAN_API_KEY"] = "your-api-key"
instrumentor = LangfuseInstrumentor()
instrumentor.instrument(
api_key=os.environ["RESPAN_API_KEY"],
)
from langfuse import Langfuse, observe
langfuse = Langfuse(
public_key="pk-lf-...",
secret_key="sk-lf-...",
)
@observe()
def subtask(name: str):
return f"Completed: {name}"
@observe()
def main_workflow(task: str):
result1 = subtask("step 1")
result2 = subtask("step 2")
return f"Workflow done: {task}"
result = main_workflow("Process request")
print(result)
langfuse.flush()
