LangChain / LangGraph

How it works

DunetraceCallbackHandler plugs into LangChain's callback system and translates LangChain events into Dunetrace events automatically.

Install

pip install 'dunetrace[langchain]'

# plus your LangChain stack
pip install langchain-openai langgraph

Create the handler

Instantiate once at startup, reuse across all invocations. The handler is thread-safe.

from dunetrace import Dunetrace
from dunetrace.integrations.langchain import DunetraceCallbackHandler

dt = Dunetrace(
    endpoint="https://your-dunetrace-ingest",
    api_key="dt_live_...",
)

callback = DunetraceCallbackHandler(
    dt,
    agent_id="my-langchain-agent",   # must match the api_keys row
    system_prompt=SYSTEM_PROMPT,      # used for version hash
    model="gpt-4o",
    tools=["web_search", "calculator"],
)

Constructor parameters

LangGraph (create_react_agent)

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o", temperature=0)

@tool
def web_search(query: str) -> str:
    """Search the web for information on a topic."""
    return f"Results for: {query}"

agent = create_react_agent(llm, [web_search], prompt=SYSTEM_PROMPT)

result = agent.invoke(
    {"messages": [("human", "What is 42 * 17?")]},
    config={"callbacks": [callback]},   # ← add this
)

Async

result = await agent.ainvoke(
    {"messages": [("human", "What is 42 * 17?")]},
    config={"callbacks": [callback]},
)

The same handler works for both invoke() and ainvoke().

LangChain AgentExecutor (older pattern)

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)

result = executor.invoke(
    {"input": "What is the capital of France?"},
    config={"callbacks": [callback]},
)

RAG agents

If your agent uses a LangChain retriever, retrieval events are captured automatically — no extra code.

from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever, "search_docs", "Search product documentation."
)
agent = create_react_agent(llm, [retriever_tool], prompt=SYSTEM_PROMPT)
result = agent.invoke(
    {"messages": [("human", "How do I configure feature X?")]},
    config={"callbacks": [callback]},
)

The handler extracts result_count and top_score from document metadata fields (score, relevance_score, or similarity). These feed the RAG_EMPTY_RETRIEVAL detector.

Concurrency

One handler can be shared across concurrent invoke() calls. Each invocation is tracked by LangChain's root run_id.

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
    futures = [
        pool.submit(agent.invoke, {"messages": [("human", q)]}, {"callbacks": [callback]})
        for q in queries
    ]
    results = [f.result() for f in futures]

Stale runs (invocations that never completed after 30 minutes) are pruned automatically on each new on_chain_start.

Shutdown

import atexit
atexit.register(dt.shutdown)
# or
dt.shutdown(timeout=5)

What is captured, and where it lives

Captured automatically:

• Every LLM call: model, token counts (prompt + completion), latency, finish reason
• Every tool call: tool name, success/failure, output length
• Every retriever call: index name, result count, top similarity score
• Run-level: total steps, tool call count, exit reason

Where your data lives: when you self-host, full event payloads, including prompts, completions, tool arguments, and outputs, are stored in your own PostgreSQL database. Nothing is sent to a third party.

Connect Langfuse to pull its evaluation results in

Root-cause analysis (POST /v1/signals/{signal_id}/explain) is fully native — it runs against Dunetrace's own stored events and needs no Langfuse credentials or trace lookup at all, with or without Langfuse running alongside it.

If you already run Langfuse for evaluation, connect it separately to pull those results into the same Dunetrace dashboard, correlated to the same runs via trace_id:

POST /v1/orgs/integrations/langfuse
Authorization: Bearer dt_live_...
Content-Type: application/json

{
  "endpoint_url": "https://cloud.langfuse.com",
  "public_key": "pk-lf-...",
  "secret_key": "sk-lf-...",
  "poll_interval_secs": 60
}

A background poller checks for new Langfuse evaluation results every 60 seconds (configurable) and writes them alongside your structural and semantic signals, tagged source: "langfuse". See the Langfuse integration guide for full detail.

Troubleshooting

No runs appear in the dashboard

• Call dt.shutdown() — events are flushed by the drain thread; without shutdown some may be lost on process exit.
• Confirm the api_key matches an active = TRUE row in api_keys.
• Pass debug=True to Dunetrace() for verbose logs.

Token counts missing

Some providers and LangChain versions don't populate token_usage in llm_output. The handler falls back to usage_metadata on the message. If both are absent, token fields are omitted — expected behavior, detectors still run.

Detectors fire too aggressively

Tune detectors.yml on the server — see threshold tuning.