Feature Proposal: Langfuse/LangSmith Observability Integration for RAG Pipeline

[codingdoneraitt]

Problem/Use Case

We're running LibreChat with rag_api in production and have Langfuse tracing enabled for LibreChat's LLM calls. However, we're missing visibility into the RAG pipeline itself:

• File uploads: What documents were processed, how many chunks were created, embedding generation time
• Retrieval queries: Which chunks were retrieved, relevance scores, retrieval latency
• Pipeline performance: Embedding bottlenecks, vector search performance

Currently, Langfuse shows us the LLM conversation but not the RAG retrieval steps that happen before. This makes debugging poor retrieval quality or performance issues difficult.

Proposed Solution

Add optional LangChain Callbacks support (Langfuse/LangSmith) to trace:

1. Document Processing Pipeline (/upload endpoint):
   - File metadata (name, size, type)
   - Chunking stats (chunk count, avg size)
   - Embedding generation time & batches
2. Query/Retrieval Pipeline (/query endpoint):
   - User query
   - Query embedding generation
   - Vector search with top K results + scores
   - Retrieved chunk content (truncated)

Example Implementation

Optional Langfuse integration

from langfuse.callback import CallbackHandler

  if os.getenv("LANGFUSE_PUBLIC_KEY"):                                                                                                                                             
      langfuse_handler = CallbackHandler(                                                                                                                                          
          public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),                                                                                                                             
          secret_key=os.getenv("LANGFUSE_SECRET_KEY"),                                                                                                                             
          host=os.getenv("LANGFUSE_BASE_URL", "https://cloud.langfuse.com"),                                                                                                       
      )                                                                                                                                                                            
                                                                                                                                                                                   
  @router.post("/query")                                                                                                                                                           
  async def query_embeddings_by_file_id(body: QueryRequestBody, ...):                                                                                                              
      if langfuse_handler:                                                                                                                                                         
          trace = langfuse_handler.trace(                                                                                                                                          
              name="rag_query",                                                                                                                                                    
              input={"query": body.query, "k": body.k},                                                                                                                            
          )                                                                                                                                                                        
                                                                                                                                                                                   
      # ... existing query logic ...                                                                                                                                               
                                                                                                                                                                                   
      if langfuse_handler:                                                                                                                                                         
          trace.span(                                                                                                                                                              
              name="vector_search",                                                                                                                                                
              output={"results": [...]}  # Top results with scores                                                                                                                 
          )                                                                                                                                                                        
              

Questions

1. Does this already exist? I couldn't find any observability/tracing implementation in the current codebase.
2. Is this on the roadmap? Would the maintainers be interested in this feature?
3. Would you accept a PR? If not already planned, I'm happy to implement this with:
   - Optional configuration (no breaking changes)
   - Support for multiple backends (Langfuse, LangSmith, or any LangChain callback)
   - Minimal performance overhead (async logging)

Benefits

• Production debugging: See exactly which chunks were retrieved for failed queries
• Quality metrics: Track retrieval relevance scores over time
• Performance monitoring: Identify embedding/search bottlenecks
• LibreChat integration: Complete end-to-end trace from file upload → retrieval → LLM response