LangChain Tracing 与 LangGraph 集成

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 Litefuse
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]})
 
      # Set trace I/O (deprecated — only for backward compat with legacy trace-level LLM-as-a-judge evaluators)
      langfuse.set_current_trace_io(
        input={"query": user_input},
        output={"response": result.content},
        )
 
    return result.content
 
# Usage
answer = process_user_query("What is the capital of France?")

from langfuse import get_client, propagate_attributes
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_observation(as_type="span", name="multi-step-process") as root_span:
    # Update trace attributes
    with propagate_attributes(
        session_id="session-1234",
        user_id="user-5678",
    ):
 
      # Initialize the Langfuse handler
      langfuse_handler = CallbackHandler()
 
      # Step 1: Initial processing (custom logic)
      with langfuse.start_as_current_observation(as_type="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_observation(as_type="span", name="output-postprocessing") as post_span:
          final_result = f"Response: {result.content}"
          post_span.update(output={"final_response": final_result})
 
      # Set trace I/O (deprecated — only for backward compat with legacy trace-level LLM-as-a-judge evaluators)
      root_span.set_trace_io(
        input={"user_query": "Explain quantum computing"},
        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_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 = 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_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();

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