langfuse

LangFuse Tracing Setup Guide for FastAPI + LangChain/LangGraph

This guide provides step-by-step instructions for implementing comprehensive LangFuse tracing in a FastAPI application using LangChain/LangGraph agents.

🎯 Overview

LangFuse provides observability for LLM applications through automatic tracing of LangChain operations. This guide covers the CallbackHandler method which integrates naturally with LangChain's callback system.

🚨 Critical Fix: CallbackHandler Auth Check Error

IMPORTANT: If you encounter 'LangchainCallbackHandler' object has no attribute 'auth_check' error:

# ❌ WRONG - Don't do this with langfuse.langchain import
from langfuse.langchain import CallbackHandler
handler = CallbackHandler()
handler.auth_check()  # This will fail!

# ✅ CORRECT - Remove auth_check() call entirely
from langfuse.langchain import CallbackHandler
handler = CallbackHandler()  # No auth_check needed

Root Cause: The CallbackHandler from langfuse.langchain doesn't have auth_check() method. Only the generic CallbackHandler from langfuse.callback has this method.

📦 Step 1: Install Dependencies

Add to your pyproject.toml:

[project]
dependencies = [
    "langfuse",
    # ... other dependencies
]

⚙️ Step 2: Environment Configuration

Add to your settings class (typically in src/utils/config.py):

# LangFuse Configuration
LANGFUSE_PUBLIC_KEY: str = os.getenv("LANGFUSE_PUBLIC_KEY", "")
LANGFUSE_SECRET_KEY: str = os.getenv("LANGFUSE_SECRET_KEY", "")
LANGFUSE_HOST: str = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")

Add to your .env file:

LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.com

🔧 Step 3: Create LangFuse Configuration Module

Create src/core/langfuse_config.py:

"""
LangFuse tracing configuration and utilities for FastAPI + LangChain applications.
"""

# Python Standard Library
import time
import logging
from typing import Optional, Dict, Any
from contextlib import asynccontextmanager

# Third-Party Packages
from langfuse.langchain import CallbackHandler
from langfuse import Langfuse

# Local Imports
from utils.config import get_settings

logger = logging.getLogger(__name__)

def get_langfuse_client() -> Optional[Langfuse]:
    """Get initialized LangFuse client"""
    try:
        settings = get_settings()
        
        if not settings.LANGFUSE_PUBLIC_KEY or not settings.LANGFUSE_SECRET_KEY:
            logger.warning("LangFuse keys not configured, tracing disabled")
            return None
        
        client = Langfuse(
            public_key=settings.LANGFUSE_PUBLIC_KEY,
            secret_key=settings.LANGFUSE_SECRET_KEY,
            host=settings.LANGFUSE_HOST
        )
        
        # Test connection (client has auth_check, handler doesn't)
        try:
            client.auth_check()
            logger.info("LangFuse connection verified successfully")
            return client
        except Exception as e:
            logger.error(f"LangFuse connection failed: {e}")
            return None
            
    except Exception as e:
        logger.error(f"Failed to initialize LangFuse client: {e}")
        return None

def get_langfuse_handler() -> Optional[CallbackHandler]:
    """Get initialized LangFuse CallbackHandler"""
    try:
        settings = get_settings()
        
        if not settings.LANGFUSE_PUBLIC_KEY or not settings.LANGFUSE_SECRET_KEY:
            logger.warning("LangFuse keys not configured, callback handler disabled")
            return None
        
        # CRITICAL: Don't call auth_check() on CallbackHandler from langfuse.langchain
        handler = CallbackHandler()
        logger.info("LangFuse CallbackHandler initialized successfully")
        
        return handler
        
    except Exception as e:
        logger.error(f"Failed to create LangFuse CallbackHandler: {e}")
        return None

def get_langfuse_config(
    conversation_id: Optional[str] = None,
    endpoint_name: Optional[str] = None,
    trace_name: Optional[str] = None,
    additional_metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Get config dict with CallbackHandler for LangChain/LangGraph
    
    Args:
        conversation_id: Unique conversation identifier
        endpoint_name: API endpoint name for tagging
        trace_name: Custom trace name
        additional_metadata: Extra metadata to include
    
    Returns:
        Config dictionary for LangChain/LangGraph agents
    """
    try:
        settings = get_settings()
        handler = get_langfuse_handler()
        
        if not handler:
            return {"callbacks": []}
        
        # Build metadata
        metadata = {
            "environment": settings.ENVIRONMENT.lower(),
            "timestamp": time.time()
        }
        
        if conversation_id:
            metadata["conversation_id"] = conversation_id
        
        if endpoint_name:
            metadata["endpoint"] = endpoint_name
        
        if additional_metadata:
            metadata.update(additional_metadata)
        
        # Build tags
        tags = [
            f"environment:{settings.ENVIRONMENT.lower()}",
        ]
        
        if endpoint_name:
            tags.append(f"endpoint:{endpoint_name}")
        
        # Configure the handler with trace context
        config = {
            "callbacks": [handler],
            "metadata": metadata,
            "tags": tags
        }
        
        # Add trace name if provided
        if trace_name:
            config["run_name"] = trace_name
        
        return config
        
    except Exception as e:
        logger.error(f"Failed to create LangFuse config: {e}")
        return {"callbacks": []}

@asynccontextmanager
async def langfuse_trace_context(
    trace_name: str,
    conversation_id: Optional[str] = None,
    endpoint_name: Optional[str] = None,
    additional_metadata: Optional[Dict[str, Any]] = None
):
    """
    Context manager for LangFuse trace with custom metrics
    
    Usage:
        async with langfuse_trace_context("rag_chat", conversation_id="123") as metrics:
            # Your agent code here
            result = await agent.ainvoke(input, config=metrics.config)
    """
    class Metrics:
        def __init__(self):
            self.start_time = time.time()
            self.config = get_langfuse_config(
                conversation_id=conversation_id,
                endpoint_name=endpoint_name,
                trace_name=trace_name,
                additional_metadata=additional_metadata
            )
        
        def add_tagger_timing(self, tagger_name: str, duration: float):
            """Track individual AI tagger performance"""
            logger.info(f"Tagger {tagger_name} completed in {duration:.2f}s")
        
        def add_error(self, error_type: str, error_message: str):
            """Track errors for failure rate calculation"""
            logger.error(f"Error in trace: {error_type} - {error_message}")
    
    metrics = Metrics()
    
    try:
        yield metrics
    except Exception as e:
        logger.error(f"Error in LangFuse trace context: {e}")
        raise
    finally:
        total_time = time.time() - metrics.start_time
        logger.info(f"Trace '{trace_name}' completed in {total_time:.2f}s")