feat: migrate to Langfuse observability

.gitignore

@@ -71,4 +71,6 @@ Frontend_API_Updates.md
 *.npy
 section_titles.json
 test_rpcs.py
-test_api.py
+test_api.py
+misc
+test_langfuse.py

main.py

@@ -1,13 +1,28 @@
 import os
-import numpy as np
+import uvicorn
 from fastapi import FastAPI
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from langfuse import get_client
+from openinference.instrumentation.groq import GroqInstrumentor
+
+# Initialize Langfuse client and verify connectivity
+langfuse = get_client()
+if not langfuse.auth_check():
+    print("WARNING: Langfuse auth failed - check your keys ✋")
+
+# Initialize OpenTelemetry instrumentation for Groq
+GroqInstrumentor().instrument()
+
+from services.observability import setup_logger
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from sentence_transformers import SentenceTransformer
+import numpy as np
 from groq import Groq
-from dotenv import load_dotenv
 from kb_docs import KB_DOCS
-
 load_dotenv()
 app = FastAPI()
 

requirements.txt

@@ -7,4 +7,6 @@ python-dotenv==1.0.1
 supabase==2.7.4
 appwrite==22.2.0
 pydantic==2.13.4
-httpx==0.27.2
+httpx==0.27.2
+langfuse>=2.0.0
+openinference-instrumentation-groq

routes/real_estate_chat.py

@@ -5,7 +5,6 @@ from pydantic import BaseModel
 
 from services.groq_service import process_chat_message
 from services.rate_limiter import limiter
-from services.observability import log_conversation_turn, ConversationTurn
 
 router = APIRouter(prefix="/api/v1/real-estate", tags=["Real Estate Intelligence Layer"])
 
@@ -21,7 +20,7 @@ class ChatResponse(BaseModel):
     suggested_actions: Optional[List[str]] = []
 
 @router.post("/chat", response_model=ChatResponse)
-async def handle_real_estate_chat(payload: ChatRequest, request: Request, background_tasks: BackgroundTasks):
+async def handle_real_estate_chat(payload: ChatRequest, request: Request):
     # Enforce token bucket rate limiting by session ID or client IP
     client_key = payload.session_id or request.client.host
     limiter.check_rate_limit(client_key)
@@ -37,16 +36,7 @@ async def handle_real_estate_chat(payload: ChatRequest, request: Request, backgr
             reply="Unable to reach the intelligence layer. Please try again shortly.",
             path_used="ERROR",
             tools_called=[],
-            suggested_actions=[]
         )
-    
-    if "telemetry" in result:
-        try:
-            turn_obj = ConversationTurn(**result["telemetry"])
-            background_tasks.add_task(log_conversation_turn, turn_obj)
-        except Exception as e:
-            # Optionally log validation errors
-            pass
 
     return ChatResponse(
         reply=result["reply"],

services/groq_service.py

@@ -7,8 +7,10 @@ from groq import Groq
 from services.tools import REAL_ESTATE_TOOLS
 from services.supabase_service import execute_tool_rpc, search_methodology_rag
 from services.embedding_service import get_embedding_model
-from services.observability import setup_logger, ConversationTurn, RoutingDecision, EmbeddingsInfo, ToolExecution
+from services.observability import setup_logger
 from config import GROQ_API_KEY
+from langfuse import observe, propagate_attributes, get_client
+import langfuse
 
 logger = setup_logger(__name__)
 
@@ -65,14 +67,16 @@ OPERATIONAL RULES:
 ```
 """
 
+@observe(name="process-chat")
 async def process_chat_message(user_query: str, session_id: str, session_context: dict) -> dict:
     global section_title_embeddings, section_titles
     start_time = time.time()
     
-    embeddings_info = EmbeddingsInfo()
-
-    if session_id not in session_history:
-        session_history[session_id] = []
+    with propagate_attributes(session_id=session_id, tags=["real-estate-chat"]):
+        get_client().update_current_span(input=user_query)
+        
+        if session_id not in session_history:
+            session_history[session_id] = []
 
     # STEP 1: Pre-Router Local Vector Search
     pre_check_hint = ""
@@ -91,9 +95,8 @@ async def process_chat_message(user_query: str, session_id: str, session_context
         matched_titles = [section_titles[i] for i in top_idx if sims[i] >= 0.45]
         
         if matched_titles:
-            embeddings_info.matched_section_titles = matched_titles
-            embeddings_info.similarity_scores = [float(sims[i]) for i in top_idx if sims[i] >= 0.45]
             pre_check_hint = f"\n\nLocal Methodology Pre-Check: High similarity match with section titles: {matched_titles}. Consider classifying as PATH_B or BOTH."
+            get_client().update_current_span(metadata={"matched_section_titles": matched_titles, "similarity_scores": [float(sims[i]) for i in top_idx if sims[i] >= 0.45]})
 
     router_sys_prompt = ROUTER_PROMPT + pre_check_hint
     router_messages = [{"role": "system", "content": router_sys_prompt}]
@@ -122,42 +125,28 @@ async def process_chat_message(user_query: str, session_id: str, session_context
     if classification not in ["OUT_OF_SCOPE", "PATH_A", "PATH_B", "BOTH", "GREETING"]:
         classification = "PATH_A"
 
-    routing_decision = RoutingDecision(classification=classification, reason=reason)
+    get_client().update_current_span(metadata={"classification": classification, "reason": reason})
     snapshot_history = session_history[session_id].copy() if session_id in session_history else []
 
-    def build_telemetry(reply_text: str, tools_exec: list, docs: list, suggested: list) -> dict:
-        return ConversationTurn(
-            session_id=session_id,
-            user_message=user_query,
-            model_response=reply_text,
-            time_taken_ms=int((time.time() - start_time) * 1000),
-            routing_decision=routing_decision,
-            embeddings_info=embeddings_info,
-            docs_retrieved=docs,
-            tools_executed=tools_exec,
-            history_context=snapshot_history,
-            suggested_actions=suggested
-        ).model_dump()
-
     # Guardrail: Immediate short-circuit if Out of Scope
     if classification == "OUT_OF_SCOPE":
         reply_out = "I apologize, but I am currently scoped exclusively to the Real Estate Rate Monitor page. I cannot assist with other topics like lead generation, pricing automation, or general knowledge outside of real estate data."
+        get_client().update_current_span(output=reply_out)
         return {
             "reply": reply_out,
             "path_used": "OUT_OF_SCOPE",
             "tools_called": [],
-            "suggested_actions": [],
-            "telemetry": build_telemetry(reply_out, [], [], [])
+            "suggested_actions": []
         }
 
     if classification == "GREETING":
         reply_greeting = "Hello! I'm the Joule Dynamics Real Estate Intelligence Assistant. I can help you with rate spikes, market trends, and availability data. How can I assist you today?"
+        get_client().update_current_span(output=reply_greeting)
         return {
             "reply": reply_greeting,
             "path_used": "GREETING",
             "tools_called": [],
-            "suggested_actions": [],
-            "telemetry": build_telemetry(reply_greeting, [], [], [])
+            "suggested_actions": []
         }
 
     tool_results = []
@@ -209,7 +198,7 @@ async def process_chat_message(user_query: str, session_id: str, session_context
             else:
                 db_result = await execute_tool_rpc(func_name, func_args)
                 
-            tool_results.append(ToolExecution(tool_name=func_name, args=func_args, db_response=db_result))
+            tool_results.append({"tool": func_name, "args": func_args})
 
             messages.append({
                 "tool_call_id": tool_call.id,
@@ -248,10 +237,11 @@ async def process_chat_message(user_query: str, session_id: str, session_context
         except json.JSONDecodeError:
             pass
 
+    get_client().update_current_span(output=final_reply)
+
     return {
         "reply": final_reply,
         "path_used": classification,
-        "tools_called": [{"tool": t.tool_name, "args": t.args} for t in tool_results],
-        "suggested_actions": suggested_actions,
-        "telemetry": build_telemetry(final_reply, tool_results, rag_chunks, suggested_actions)
+        "tools_called": tool_results,
+        "suggested_actions": suggested_actions
     }

services/observability.py

@@ -4,75 +4,30 @@ import logging
 from logging.handlers import RotatingFileHandler
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
-from pydantic import BaseModel, Field
 
-# Ensure logs directory exists
-LOGS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logs")
-os.makedirs(LOGS_DIR, exist_ok=True)
-
-# -----------------------------------------------------------------------------
-# General Project Logger Setup
-# -----------------------------------------------------------------------------
 def setup_logger(name: str) -> logging.Logger:
-    """Creates a rotating file logger for general project traceability."""
+    """
+    Creates and returns a standard Python logger that writes to a rotating file.
+    Max file size: 5MB. Keeps up to 3 backups.
+    """
     logger = logging.getLogger(name)
+    
     if not logger.handlers:
         logger.setLevel(logging.INFO)
         
-        # Rotating File Handler (Max 5MB, keep 3 backups)
-        log_file = os.path.join(LOGS_DIR, "app.log")
-        file_handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=3)
+        log_dir = "logs"
+        if not os.path.exists(log_dir):
+            os.makedirs(log_dir)
+            
+        file_handler = RotatingFileHandler(
+            os.path.join(log_dir, "app.log"),
+            maxBytes=5 * 1024 * 1024,
+            backupCount=3
+        )
         
-        # Console Handler
-        console_handler = logging.StreamHandler()
-
         formatter = logging.Formatter('%(asctime)s | %(name)s | %(levelname)s | %(message)s')
         file_handler.setFormatter(formatter)
-        console_handler.setFormatter(formatter)
-
+        
         logger.addHandler(file_handler)
-        logger.addHandler(console_handler)
         
     return logger
-
-# -----------------------------------------------------------------------------
-# JSONL Conversation Telemetry Setup
-# -----------------------------------------------------------------------------
-class RoutingDecision(BaseModel):
-    classification: str
-    reason: Optional[str] = None
-
-class EmbeddingsInfo(BaseModel):
-    matched_section_titles: List[str] = Field(default_factory=list)
-    similarity_scores: List[float] = Field(default_factory=list)
-
-class ToolExecution(BaseModel):
-    tool_name: str
-    args: Dict[str, Any]
-    db_response: Any
-
-class ConversationTurn(BaseModel):
-    timestamp: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
-    session_id: str
-    user_message: str
-    model_response: str
-    time_taken_ms: int
-    routing_decision: RoutingDecision
-    embeddings_info: EmbeddingsInfo
-    docs_retrieved: List[str] = Field(default_factory=list)
-    tools_executed: List[ToolExecution] = Field(default_factory=list)
-    history_context: List[Dict[str, str]] = Field(default_factory=list)
-    suggested_actions: List[str] = Field(default_factory=list)
-
-def log_conversation_turn(turn: ConversationTurn):
-    """
-    Appends a structured JSON string to the conversations.jsonl file.
-    Designed to be run via FastAPI BackgroundTasks.
-    """
-    jsonl_file = os.path.join(LOGS_DIR, "conversations.jsonl")
-    try:
-        with open(jsonl_file, "a", encoding="utf-8") as f:
-            f.write(turn.model_dump_json() + "\n")
-    except Exception as e:
-        logger = setup_logger(__name__)
-        logger.error(f"Failed to write conversation telemetry to JSONL: {e}")