Observability makes a deployed system’s internal behavior inferable from outputs, telemetry, and application-specific information flows. In production ML, it connects monitoring, drift detection, model performance, lineage, and operational decisions. For LLM applications, infrastructure metrics alone are insufficient: useful visibility follows retrieved documents, assembled context, generation, failures, costs, and latency. section 2.1.7 section 12.4
A practical observability system should expose:
- Service health through health checks, liveness and readiness probes, request latency, and Prometheus metrics. section 10.9
- Model behavior through performance metrics, drift signals, experiment lineage, version information, and alerts. section 2.1.7 section 9.0
- LLM reasoning chains through prompts, responses, token usage, latency, retrieval operations, document metadata, intermediate steps, errors, and estimated cost. section 12.4.2
- Business context through query analysis, retrieval success, context utilization, and user-session metadata rather than technical metrics alone. section 12.4.3
Instrumenting a RAG service
A DakkaBotCore service initializes Langfuse with environment-based credentials and attaches a callback handler to its LangChain pipeline. One handler automatically traces retrieval and generation, while enriched metadata connects each query to retrieved documents, context use, performance, failures, and cost. The resulting traces support retrieval optimization and early detection of degradation without changing the application’s core behavior. section 12.4.3 section 12.4.2
Listing 12.19 Langfuse observability integration and tracing
class DakkaBotCore:
def __init__(self):
... #1
self.langfuse_handler = None #2
async def initialize(self):
try:
...
self.llm = ...
langfuse = Langfuse(
public_key=os.environ[ #4
"LANGFUSE_PUBLIC_KEY"], #4
secret_key=os.environ[ #4
"LANGFUSE_SECRET_KEY"], #4
host=os.environ.get( #4
"LANGFUSE_HOST", #4
"http://localhost:3000") #4
) #4
self.langfuse_handler = CallbackHandler() #5
self.initialized = True
except Exception as e:
raise RuntimeError(f"Failed to initialize DakkaBot: {str(e)}")
async def process_query(self, query: str) -> Dict[str, Any]:
try:
docs = ...
context = ...
prompt = ...
response = self.llm.invoke(
prompt,
config={
"callbacks": [self.langfuse_handler], #6
"metadata": {
"query": query, #7
"doc_count": len(docs), #7
"context_length": len(context), #7
"user_session": "demo" #7
}
}
)
return {
"response": response.content,
"sources": [...], # Source processing
"doc_count": len(docs),
"error": None
}
except Exception as e:
return {
"response": "Sorry! Please try again!",
"sources": [],
"doc_count": 0,
"error": str(e)
}
Inspect the Langfuse integration listing to see how initialization and callback-based tracing fit into a production RAG service. section 12.4.3
