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-..."
# 🇪🇺 EU region
LANGFUSE_BASEURL="https://cloud.langfuse.com"
# 🇺🇸 US region
# LANGFUSE_BASEURL="https://us.cloud.langfuse.com"

import { CallbackHandler } from "langfuse-langchain";
// Deno: import CallbackHandler from "https://esm.sh/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})

from langfuse.callback import CallbackHandler
 
langfuse_handler = CallbackHandler()
 
trace_name = "langchain_trace_name"
session_id = "random-session"
user_id = "random-user"
tags = ["random-tag-1", "random-tag-2"]
 
# Your existing Langchain code to create the chain
 
# Pass config to the chain invocation to be parsed as Langfuse trace attributes
 
chain.invoke(
    {"animal": "dog"},
    config={
        "callbacks": [langfuse_handler],
        "run_name": trace_name,
        "tags": tags,
        "metadata": {
            "langfuse_session_id": session_id,
            "langfuse_user_id": user_id,
        },
    },
)

from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
 
from langfuse.callback import CallbackHandler
 
handler = CallbackHandler()
 
 
def main():
    llm = ChatOpenAI()
 
    runnable = RunnablePassthrough() | llm # force execution as chain
    result = runnable.invoke(
        "Whassup?",
        {
            "callbacks": [handler],
            "metadata": {
                "langfuse_session_id": "my-session",
                "langfuse_user_id": "my-user",
            },
        },
    )
 
    handler.flush()

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 { Langfuse } from "langfuse";
 
const langfuse = new Langfuse();
 
await langfuse.score({
  traceId: predefinedRunId,
  name: "user-feedback",
  value: 1,
  comment: "This was correct, thank you",
});

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.decorators import langfuse_context, observe
 
@observe() # automatically log function as a trace to Langfuse
def main():
    # update trace attributes
    langfuse_context.update_current_trace(
        name="custom-trace",
        session_id="user-1234",
        user_id="session-1234",
    )
    # get the langchain handler for the current trace
    langfuse_handler = langfuse_context.get_current_langchain_handler()
 
    # Your Langchain code
    chain.invoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})
 
main()

from langfuse import Langfuse
langfuse = Langfuse()
 
# Get Langchain handler for a trace
trace = langfuse.trace()
langfuse_handler_trace = trace.get_langchain_handler()
 
# Add Langfuse handler as callback
chain.invoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler_trace]})

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",
});