Observability & Tracing for Langchain (Python & JS/TS)

Langfuse Tracing integrates with Langchain using Langchain Callbacks (Python, JS). The Langfuse SDK automatically captures detailed traces of your Langchain executions, creating properly nested observations for chains, LLMs, tools, and retrievers. This allows you to monitor, analyze and debug your LangChain applications with full observability.

This documentation has been updated to show examples for the new Python SDK v3. If you are looking for documentation for the Python SDK version 2, see here.

Add Langfuse to your Langchain Application

You can configure the integration via (1) constructor arguments or (2) environment variables. Get your Langfuse credentials from the Langfuse dashboard.

Set environment variables:

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

.env

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] });

✨

Done. Now you can explore detailed traces and metrics in the Langfuse dashboard.

Prefer end-to-end examples?

Example (Python)Example (JS/TS)Example LangGraph (Python)Example LangServe (Python)

Supported LangChain interfaces

Feature/interface	Python	JS/TS
LCEL	✅	✅
`invoke()`	✅	✅
`run()`	✅	✅
`call()`	✅	✅
`predict()`	✅	✅
async	✅	✅
`batch()`	✅	(✅)
streaming	✅	✅

We are interested in your feedback! Raise an issue on GitHub to request support for additional interfaces.

Supported LangChain features

🕸️ LangGraph: Works with Langfuse Integration. Requires Python 3.11+ (GH issue). See notebook for example integration.
🏓 LangServe: See notebook for example integration.

Additional Configuration

Configuration Options

The CallbackHandler does not accept any constructor arguments for trace attributes or global settings.

Global settings (like sample_rate, tracing_enabled) must be set when initializing the Langfuse client via Langfuse() constructor or environment variables
Trace attributes (like user_id, session_id, release, version) must be set via an enclosing span using span.update_trace() as shown in the examples above

Parameter	Type	Description
`userId`	string	The current user.
`sessionId`	string	The current session.
`release`	string	The release tag of your application.
`version`	string	The version of your application.
`enabled`	boolean	Enable or disable the Langfuse integration. Defaults to `true`.

Dynamic Trace Attributes

You can set trace attributes dynamically for each LangChain execution. The approach differs between SDK versions:

For Python SDK v3, use enclosing spans to set trace attributes dynamically:

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})

For Python SDK v2, you can set trace attributes through the runnable configuration:

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

Setting trace attributes dynamically requires the execution of a Langchain chain. Otherwise the dynamically provided chain configuration is not passed to the Langfuse CallbackHandler instance. Thus when invoking Langchain LLM class instances with dynamic trace metadata, an additional wrapping is required to force the execution as a chain:

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

Predefined Trace ID + Add Evaluation or User Feedback Score

Predefined Trace ID

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

Add Score to Trace

There are multiple ways to score a trace in Python SDK v3. See Scoring documentation for more details.

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"
)

Interoperability with Langfuse SDKs

The Langchain integration works seamlessly with the Langfuse SDK to create comprehensive traces that combine Langchain operations with other application logic.

Common use cases:

Add non-Langchain related observations to the trace
Group multiple Langchain runs into a single trace
Set trace-level attributes (user_id, session_id, etc.)
Add custom spans for business logic around LLM calls

Learn more about the structure of a trace here.

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] }
);

If you want to add the input/output of the Langchain run to the trace or span itself, use the updateRoot flag in the CallbackHandler constructor.

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

Decorator Approach (v2):

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()

Low-level Approach (v2):

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]})

Note: v2 SDK is legacy. Consider migrating to v3 for better performance and features.

If you pass these callback handlers to your Langchain code, the events will be nested under the respective trace or span in the Langfuse.

See the Langchain observability cookbook for an example of this in action (Python).

Queuing and flushing

The Langfuse SDKs queue and batch events in the background to reduce the number of network requests and improve overall performance. In a long-running application, this works without any additional configuration.

If you are running a short-lived application, you need to shutdown Langfuse to ensure that all events are flushed before the application exits.

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

If you want to flush events synchronously at a certain point, you can use the flush method. This will wait for all events that are still in the background queue to be sent to the Langfuse API. This is usually discouraged in production environments.

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

Serverless environments (JS/TS)

Since Langchain version > 0.3.0, the callbacks on which Langfuse relies have been backgrounded. This means that execution will not wait for the callback to either return before continuing. Prior to 0.3.0, this behavior was the opposite. If you are running code in serverless environments such as Google Cloud Functions, AWS Lambda or Cloudflare Workers you should set your callbacks to be blocking to allow them time to finish or timeout. This can be done either by

setting the LANGCHAIN_CALLBACKS_BACKGROUND environment variable to “false”
importing the global awaitAllCallbacks method to ensure all callbacks finish if necessary

Read more about awaiting callbacks here in the Langchain docs.

Azure OpenAI model names

Please add the model keyword argument to the AzureOpenAI or AzureChatOpenAI class to have the model name parsed correctly in Langfuse.

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