LangChain Tracing & Callbacks — Open Source Observability for LangChain & LangGraph - Langfuse

LangChain Tracing & LangGraph Integration

Langfuse integrates with LangChain using LangChain Callbacks — the standard mechanism for hooking into the execution of LangChain components. The Langfuse CallbackHandler automatically captures detailed traces of your LangChain executions, LLMs, tools, and retrievers to evaluate and debug your application.

What is LangChain? LangChain is an open-source framework that helps developers build applications powered by large language models (LLMs) by providing tools to connect models with external data, APIs, and logic.

What is LangGraph? LangGraph is a framework built on top of LangChain that makes it easier to design and run stateful, multi-step AI agents using a graph-based architecture.

What is Langfuse? Langfuse is a platform for observability and tracing of LLM applications. It captures everything happening during an LLM interaction: inputs, outputs, tool usage, retries, latencies and costs and allows you to evaluate and debug your application.

Getting Started

Install Dependencies

pip install langfuse langchain langchain_openai langgraph

Initialize Langfuse Callback Handler

Next, set up your Langfuse API keys. You can get these keys by signing up for a free Langfuse Cloud account or by self-hosting Langfuse. These environment variables are essential for the Langfuse client to authenticate and send data to your Langfuse project.

.env

LANGFUSE_SECRET_KEY = "sk-lf-..."
LANGFUSE_PUBLIC_KEY = "pk-lf-..."
LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region
# Other Langfuse data regions include 🇺🇸 US: https://us.cloud.langfuse.com, 🇯🇵 Japan: https://jp.cloud.langfuse.com and ⚕️ HIPAA: https://hipaa.cloud.langfuse.com

OPENAI_API_KEY = "sk-proj-..."

With the environment variables set, we can now initialize the Langfuse Client and the CallbackHandler. You can also use constructor arguments to initialize the Langfuse client.

from langfuse import get_client
from langfuse.langchain import CallbackHandler

# Initialize Langfuse client
langfuse = get_client()

# Initialize Langfuse CallbackHandler for Langchain (tracing)
langfuse_handler = CallbackHandler()

LangChain Example

💡

Instrumenting LangGraph follows the same pattern. Simply pass the langfuse_handler to the agent invocation. ( Example Notebook).

from langchain.agents import create_agent

def add_numbers(a: int, b: int) -> int:
    """Add two numbers together and return the result."""
    return a + b

agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[add_numbers],
    system_prompt="You are a helpful math tutor who can do calculations using the provided tools.",
)

# Run the agent
agent.invoke(
    {"messages": [{"role": "user", "content": "what is 42 + 58?"}]},
    config={"callbacks": [langfuse_handler]}
)

See Traces in Langfuse

After executing the application, navigate to your Langfuse Trace Table. You will find detailed traces of the application’s execution, providing insights into the LLM calls, retrieval operations, inputs, outputs, and performance metrics.

Install Dependencies

npm install @langfuse/core @langfuse/langchain

Set Up Environment

.env

OPENAI_API_KEY = "sk-proj-..."


With the environment variables set, we can now initialize the `langfuseSpanProcessor` which is passed to the main OpenTelemetry SDK that orchestrates tracing.

import { NodeSDK } from "@opentelemetry/sdk-node"; import { LangfuseSpanProcessor } from "@langfuse/otel";

const sdk = new NodeSDK({ spanProcessors: [new LangfuseSpanProcessor()], });

sdk.start();


### Initialize Langfuse Callback Handler

Instantiate the CallbackHandler. Optionally, you can pass additional attributes like `sessionId`, `userId`, and `tags`.

import { CallbackHandler } from "@langfuse/langchain";

// Initialize the Langfuse CallbackHandler const langfuseHandler = new CallbackHandler({ sessionId: "user-session-123", userId: "user-abc", tags: ["langchain-test"], });


### LangChain Example

💡

Instrumenting LangGraph follows the same pattern. Simply pass the `langfuseHandler` to the agent invocation.

import { createAgent, tool } from "@langchain/core/agents"; import * as z from "zod";

const getWeather = tool( (input) => It's always sunny in ${input.city}!, { name: "get_weather", description: "Get the weather for a given city", schema: z.object({ city: z.string().describe("The city to get the weather for"), }), } );

const agent = createAgent({ model: "openai:gpt-5-mini", tools: [getWeather], });

console.log( await agent.invoke( { messages: [{ role: "user", content: "What's the weather in San Francisco?" }] }, { callbacks: [langfuseHandler] } ) );


### See Traces in Langfuse

### Example Notebooks

[**LangChain (Python)**](/content/guides/cookbook/integration_langchain/index.html) [**LangChain (JS/TS)**](/content/guides/cookbook/js_integration_langchain/index.html) [**LangGraph (Python)**](/content/guides/cookbook/integration_langgraph/index.html) [**Evaluate LangGraph Agents (Python)**](/content/guides/cookbook/example_langgraph_agents/index.html) [**LangChain DeepAgents (Python)**](/content/integrations/frameworks/langchain-deepagents/index.html)

### Additional Configuration

### 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`](/content/docs/observability/features/users/index.html), [`session_id`](/content/docs/observability/features/sessions/index.html), [`tags`](/content/docs/observability/features/tags/index.html), etc.)

```python
from langfuse import observe, get_client, propagate_attributes
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()

# Propagate trace attributes to all child observations
    with propagate_attributes(
        trace_name="user-query-processing",
        session_id="session-1234",
        user_id="user-5678",
    ):

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

return result.content

# Usage
answer = process_user_query("What is the capital of France?")

See the Langchain + decorator observability cookbook for an example of this in action.

Trace Attributes

You can set trace attributes such as user_id, session_id, and tags dynamically for each LangChain execution.

Python SDKJS/TS SDK

With Python SDK, you have two options to set trace attributes dynamically:

Option 1: Via metadata fields in chain invocation (simplest approach):

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

Option 2: Using the Langfuse SDK

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

Trace IDs & Distributed Tracing

To pass a custom trace_id to a Langchain execution, you can wrap the execution in a span that sets a predefined trace ID. You can also retrieve the last trace ID a callback handler has created via langfuse_handler.last_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_observation(
    as_type="span",
    name="langchain-request",
    trace_context={"trace_id": predefined_trace_id}
) as span:
    # Set trace I/O (deprecated — only for backward compat with legacy trace-level LLM-as-a-judge evaluators)
    span.set_trace_io(
        input={"person": "Ada Lovelace"}
    )

with propagate_attributes(
        user_id="user_123",
    ):
        # LangChain execution will be part of this trace
        response = chain.invoke(
            {"person": "Ada Lovelace"},
            config={"callbacks": [langfuse_handler]}
        )

span.set_trace_io(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 = uuidv4();

await chain.invoke(
  { animal: "dog" },
  {
    callbacks: [langfuseHandler],
    runId: predefinedRunId,
  }
);

Score a Trace

There are multiple ways to score a LangChain trace in Langfuse. 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_observation(
    as_type="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_observation(as_type="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();

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

await langfuseHandler.shutdownAsync();

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

await langfuseHandler.flushAsync();

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.

AWS Bedrock AgentCore

When deploying LangChain applications to AWS Bedrock AgentCore, the runtime's ADOT (AWS Distro for OpenTelemetry) auto-instrumentation requires OTEL configuration instead of relying solely on the Langfuse callback handler. See Using Langfuse with an Existing OpenTelemetry Setup for configuration details, or the full Amazon Bedrock AgentCore integration guide.

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

import { AzureChatOpenAI } from "@langchain/openai";

const llm = new AzureChatOpenAI({
  azureOpenAIApiDeploymentName: "my-gpt-4o-deployment",
  model: "gpt-4o",
});

Upgrade Paths for Langchain Integration

This doc is a collection of upgrade paths for different versions of the integration. If you want to add the integration to your project, you should start with the latest version and follow the integration guide above.

Langfuse and Langchain are under active development. Thus, we are constantly improving the integration. This means that we sometimes need to make breaking changes to our APIs or need to react to breaking changes in Langchain. We try to keep these to a minimum and to provide clear upgrade paths when we do make them.

Python

JS/TS

FAQ

GitHub Discussions

Need per-API / per-trace control to disable media uploads