LangChain & LangGraph Integration

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

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