Tracing makes an LLM application observable by recording operations, inputs, outputs, intermediate results, latency, and failures. It helps teams investigate poor answers and identify opportunities to improve retrieval and prompts. section 12.4.3 section 12.5
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)
}
See how Langfuse instrumentation is added to DakkaBotCore to capture LangChain operations and enrich traces with business context. section 12.4.3
