Observability & Tracing for LangChain & LangGraph (Python & JS/TS)

export LANGFUSE_PUBLIC_KEY="your-public-key" export LANGFUSE_SECRET_KEY="your-secret-key" export LANGFUSE_HOST="https://cloud.langfuse.com" # Optional: defaults to https://cloud.langfuse.com

pip install langfuse

from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate   # Initialize the Langfuse handler langfuse_handler = CallbackHandler()   # Create your LangChain components llm = ChatOpenAI(model_name="gpt-4o") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") chain = prompt | llm   # Run your chain with Langfuse tracing response = chain.invoke({"topic": "cats"}, config={"callbacks": [langfuse_handler]}) print(response.content)

pip install langfuse

from langfuse import Langfuse, get_client from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI # Example LLM from langchain_core.prompts import ChatPromptTemplate   # Initialize Langfuse client with constructor arguments Langfuse(  public_key="your-public-key",  secret_key="your-secret-key",  host="https://cloud.langfuse.com" # Optional: defaults to https://cloud.langfuse.com )   # Get the configured client instance langfuse = get_client()   # Initialize the Langfuse handler langfuse_handler = CallbackHandler()   # Create your LangChain components llm = ChatOpenAI(model_name="gpt-4o") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") chain = prompt | llm   # Run your chain with Langfuse tracing response = chain.invoke({"topic": "cats"}, config={"callbacks": [langfuse_handler]}) print(response.content)   # Flush events to Langfuse in short-lived applications langfuse.flush()

npm i langfuse-langchain

import { CallbackHandler } from "langfuse-langchain"; // Deno: import CallbackHandler from "https://esm.sh/langfuse-langchain";   const langfuseHandler = new CallbackHandler({  secretKey: "sk-lf-...",  publicKey: "pk-lf-...",  baseUrl: "https://cloud.langfuse.com", // 🇪🇺 EU region  // baseUrl: "https://us.cloud.langfuse.com", // 🇺🇸 US region });   // Your Langchain code   // Add Langfuse handler as callback to `run` or `invoke` await chain.invoke({ input: "<user_input>" }, { callbacks: [langfuseHandler] });

npm i @langfuse/langchain

LANGFUSE_SECRET_KEY = "sk-lf-..." LANGFUSE_PUBLIC_KEY = "pk-lf-..." LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region # LANGFUSE_BASE_URL = "https://us.cloud.langfuse.com" # 🇺🇸 US region

import { CallbackHandler } from "@langfuse/langchain";   const langfuseHandler = new CallbackHandler();   // Your Langchain code   // Add Langfuse handler as callback to `run` or `invoke` await chain.invoke({ input: "<user_input>" }, { callbacks: [langfuseHandler] });

from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate   langfuse_handler = CallbackHandler()   llm = ChatOpenAI(model_name="gpt-4o") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") chain = prompt | llm   # Set trace attributes dynamically via metadata response = chain.invoke(  {"topic": "cats"},  config={  "callbacks": [langfuse_handler],  "metadata": {  "langfuse_user_id": "random-user",  "langfuse_session_id": "random-session",  "langfuse_tags": ["random-tag-1", "random-tag-2"]  }  } )

from langfuse import get_client from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate   langfuse = get_client() langfuse_handler = CallbackHandler()   llm = ChatOpenAI(model_name="gpt-4o") prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") chain = prompt | llm   # Set trace attributes dynamically via enclosing span with langfuse.start_as_current_span(name="dynamic-langchain-trace") as span:  span.update_trace(  user_id="random-user",  session_id="random-session",  tags=["random-tag-1", "random-tag-2"],  input={"animal": "dog"}  )    response = chain.invoke({"topic": "cats"}, config={"callbacks": [langfuse_handler]})    span.update_trace(output={"response": response.content})

import { CallbackHandler } from "langfuse-langchain";   const langfuseHandler = new CallbackHandler();   const traceName = "langchain_trace_name"; const sessionId = "random-session"; const userId = "random-user"; const tags = ["random-tag-1", "random-tag-2"];   await chain.invoke(  { animal: "dog" },  {  callbacks: [langfuseHandler],  runName: traceName,  tags,  metadata: { langfuseUserId: userId, langfuseSessionId: sessionId },  } );

from langfuse import get_client, Langfuse from langfuse.langchain import CallbackHandler   langfuse = get_client()   # Generate deterministic trace ID from external system external_request_id = "req_12345" predefined_trace_id = Langfuse.create_trace_id(seed=external_request_id)   langfuse_handler = CallbackHandler()   # Use the predefined trace ID with trace_context with langfuse.start_as_current_span(  name="langchain-request",  trace_context={"trace_id": predefined_trace_id} ) as span:  span.update_trace(  user_id="user_123",  input={"person": "Ada Lovelace"}  )    # LangChain execution will be part of this trace  response = chain.invoke(  {"person": "Ada Lovelace"},  config={"callbacks": [langfuse_handler]}  )    span.update_trace(output={"response": response})   print(f"Trace ID: {predefined_trace_id}") # Use this for scoring later print(f"Trace ID: {langfuse_handler.last_trace_id}") # Care needed in concurrent environments where handler is reused

import { CallbackHandler } from "langfuse-langchain"; import { v4 as uuidv4 } from "uuid";   const langfuseHandler = new CallbackHandler();   const predefinedRunId = uuid4();   await chain.invoke(  { animal: "dog" },  {  callbacks: [langfuseHandler],  runId: predefinedRunId,  } );

from langfuse import get_client   langfuse = get_client()   # Option 1: Use the yielded span object from the context manager with langfuse.start_as_current_span(  name="langchain-request",  trace_context={"trace_id": predefined_trace_id} ) as span:  # ... LangChain execution ...    # Score using the span object  span.score_trace(  name="user-feedback",  value=1,  data_type="NUMERIC",  comment="This was correct, thank you"  )   # Option 2: Use langfuse.score_current_trace() if still in context with langfuse.start_as_current_span(name="langchain-request") as span:  # ... LangChain execution ...    # Score using current context  langfuse.score_current_trace(  name="user-feedback",  value=1,  data_type="NUMERIC"  )   # Option 3: Use create_score() with trace ID (when outside context) langfuse.create_score(  trace_id=predefined_trace_id,  name="user-feedback",  value=1,  data_type="NUMERIC",  comment="This was correct, thank you" )

import { LangfuseClient } from "@langfuse/client";   const langfuse = new LangfuseClient();   langfuse.score.create({  id: "unique_id", // optional, can be used as an idempotency key to update the score subsequently  traceId: message.traceId,  observationId: message.generationId, // optional  name: "correctness",  value: 0.9,  dataType: "NUMERIC", // optional, inferred if not provided  comment: "Factually correct", // optional });   // Flush the scores in short-lived environments await langfuse.flush();

from langfuse import observe, get_client from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate   @observe() # Automatically log function as a trace to Langfuse def process_user_query(user_input: str):  langfuse = get_client()    # Update trace attributes  langfuse.update_current_trace(  name="user-query-processing",  session_id="session-1234",  user_id="user-5678",  input={"query": user_input}  )    # Initialize the Langfuse handler - automatically inherits the current trace context  langfuse_handler = CallbackHandler()    # Your Langchain code - will be nested under the @observe trace  llm = ChatOpenAI(model_name="gpt-4o")  prompt = ChatPromptTemplate.from_template("Respond to: {input}")  chain = prompt | llm    result = chain.invoke({"input": user_input}, config={"callbacks": [langfuse_handler]})    # Update trace with final output  langfuse.update_current_trace(output={"response": result.content})    return result.content   # Usage answer = process_user_query("What is the capital of France?")

from langfuse import get_client from langfuse.langchain import CallbackHandler from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate   langfuse = get_client()   # Create a trace via Langfuse spans and use Langchain within it with langfuse.start_as_current_span(name="multi-step-process") as root_span:  # Update trace attributes  root_span.update_trace(  session_id="session-1234",  user_id="user-5678",  input={"user_query": "Explain quantum computing"}  )    # Initialize the Langfuse handler  langfuse_handler = CallbackHandler()    # Step 1: Initial processing (custom logic)  with langfuse.start_as_current_span(name="input-preprocessing") as prep_span:  processed_input = "Simplified: Explain quantum computing"  prep_span.update(output={"processed_query": processed_input})    # Step 2: LangChain processing  llm = ChatOpenAI(model_name="gpt-4o")  prompt = ChatPromptTemplate.from_template("Answer this question: {input}")  chain = prompt | llm    result = chain.invoke(  {"input": processed_input},  config={"callbacks": [langfuse_handler]}  )    # Step 3: Post-processing (custom logic)  with langfuse.start_as_current_span(name="output-postprocessing") as post_span:  final_result = f"Response: {result.content}"  post_span.update(output={"final_response": final_result})    # Update trace output  root_span.update_trace(output={"final_answer": final_result})

import { CallbackHandler, Langfuse } from "langfuse-langchain"; const langfuse = new Langfuse();   // Get Langchain handler for a trace const trace = langfuse.trace(); const langfuseHandlerTrace = new CallbackHandler({ root: trace });   // Add Langfuse handler as callback (classic and LCEL) await chain.invoke(  { input: "<user_input>" },  { callbacks: [langfuseHandlerTrace] } );   // Get Langchain handler for a span const span = trace.span(); const langfuseHandlerSpan = new CallbackHandler({ root: span });   // Add Langfuse handler as callback (classic and LCEL) await chain.invoke(  { input: "<user_input>" },  { callbacks: [langfuseHandlerSpan] } );

const langfuseHandlerTrace = new CallbackHandler({  root: trace,  updateRoot: true, });

from langfuse import get_client   # Shutdown the underlying singleton instance get_client().shutdown()

await langfuseHandler.shutdownAsync();

from langfuse import get_client   # Flush the underlying singleton instance get_client().flush()

await langfuseHandler.flushAsync();

from langchain_openai import AzureChatOpenAI   llm = AzureChatOpenAI( azure_deployment="my-gpt-4o-deployment", model="gpt-4o", )

import { AzureChatOpenAI } from "@langchain/openai";   const llm = new AzureChatOpenAI({  azureOpenAIApiDeploymentName: "my-gpt-4o-deployment",  model: "gpt-4o", });