Trace LangChain and LangGraph agents with OpenTelemetry

pip install "langchain>=1.4" "langgraph>=1.2" "langchain-openai>=1.6" \
  "openinference-instrumentation-langchain>=0.1.76" "openinference-instrumentation>=0.1.66" \
  "opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45"

uv add "langchain>=1.4" "langgraph>=1.2" "langchain-openai>=1.6" \
  "openinference-instrumentation-langchain>=0.1.76" "openinference-instrumentation>=0.1.66" \
  "opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45"

Keep the explicit openinference-instrumentation pin. Older versions don’t work with this setup.

npm install @arizeai/openinference-instrumentation-langchain @opentelemetry/sdk-node @opentelemetry/sdk-trace-base @opentelemetry/exporter-trace-otlp-proto

pnpm add @arizeai/openinference-instrumentation-langchain @opentelemetry/sdk-node @opentelemetry/sdk-trace-base @opentelemetry/exporter-trace-otlp-proto

bun add @arizeai/openinference-instrumentation-langchain @opentelemetry/sdk-node @opentelemetry/sdk-trace-base @opentelemetry/exporter-trace-otlp-proto

@opentelemetry/sdk-node has to be 0.209 or newer.

Add a tracing.py and import it at the top of your entry point, before the first invoke():

# tracing.py
from openinference.instrumentation import TraceConfig
from openinference.instrumentation.langchain import LangChainInstrumentor
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# The name= you gave create_agent(), and graph nodes that act as agents
AGENT_NAMES = {"assistant"}
# Your tool node
STEP_NAMES = {"tools"}


class AgentSpans(SpanProcessor):
    def on_start(self, span, parent_context=None):
        if span.instrumentation_scope.name != "openinference.instrumentation.langchain":
            return
        if span.name in AGENT_NAMES:
            span.set_attribute("gen_ai.operation.name", "invoke_agent")
            span.set_attribute("gen_ai.agent.name", span.name)
        elif span.name in STEP_NAMES:
            span.set_attribute("gen_ai.operation.name", "invoke_workflow")


provider = TracerProvider()
provider.add_span_processor(AgentSpans())
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

LangChainInstrumentor().instrument(
    tracer_provider=provider,
    config=TraceConfig(enable_genai_semconv=True),
)

Put every agent’s name= in AGENT_NAMES, sub-agents included, so each gets its own lane. If your tool node isn’t called tools, add its name to STEP_NAMES.

If the app already has a TracerProvider (from opentelemetry-instrument, Logfire or Sentry), add AgentSpans() and the exporter to it and pass it to instrument().

The TypeScript instrumentation only writes OpenInference attributes, so add a span processor that copies them into the GenAI attributes Maple reads. Save this file as genai-spans.ts:

// genai-spans.ts: adds the OpenTelemetry GenAI attributes Maple reads to OpenInference's LangChain spans
import type { Span, SpanProcessor } from "@opentelemetry/sdk-trace-base"

// Hand-built StateGraph agents, by compiled name. createAgent({ name }) is detected on its own.
const AGENT_NAMES = new Set<string>()

const ROLES: Record<string, string> = { human: "user", ai: "assistant" }

function parse(value: unknown): any {
	if (typeof value !== "string") return undefined
	try {
		return JSON.parse(value)
	} catch {
		return undefined
	}
}

function text(content: unknown): string {
	if (typeof content === "string") return content
	if (!Array.isArray(content)) return ""
	return content.map((block) => (block?.type === "text" ? block.text : "")).join("")
}

// LangChain messages are serialized as { lc, id: [..., "AIMessage"], kwargs }; plain inputs are { role, content }
function toGenAiMessage(message: any) {
	const fields = message?.lc ? message.kwargs : (message ?? {})
	const type = message?.lc
		? String(message.id.at(-1)).replace(/Message(Chunk)?$/, "").toLowerCase()
		: String(fields.role ?? fields.type)
	const role = ROLES[type] ?? type
	const parts: object[] = []
	if (role === "tool") {
		parts.push({ type: "tool_call_response", id: fields.tool_call_id, response: text(fields.content) })
	} else if (text(fields.content)) {
		parts.push({ type: "text", content: text(fields.content) })
	}
	for (const call of fields.tool_calls ?? []) {
		parts.push({ type: "tool_call", id: call.id, name: call.name, arguments: call.args })
	}
	return { role, parts }
}

export class GenAiSpans implements SpanProcessor {
	// Tool call arguments by call id, from the model reply that requested them
	private toolArgs = new Map<string, unknown>()

	onEnding(span: Span) {
		if (span.instrumentationScope.name !== "@arizeai/openinference-instrumentation-langchain") return
		const attrs = span.attributes
		const set = (key: string, value: unknown) => {
			if (value === undefined || value === null || value === "") return
			span.setAttribute(key, typeof value === "object" ? JSON.stringify(value) : (value as string | number))
		}
		const metadata = parse(attrs["metadata"]) ?? {}
		const input = parse(attrs["input.value"])
		const output = parse(attrs["output.value"])
		const kind = attrs["openinference.span.kind"]

		if (kind === "LLM") {
			const generations = output?.generations?.[0] ?? []
			const reply = generations[0]?.message?.kwargs ?? {}
			const finish = generations[0]?.generationInfo?.finish_reason ?? reply.response_metadata?.finish_reason
			if (this.toolArgs.size > 1000) this.toolArgs.clear() // calls whose tool never ran
			for (const call of reply.tool_calls ?? []) this.toolArgs.set(call.id, call.args)
			set("gen_ai.operation.name", "chat")
			set("gen_ai.provider.name", metadata.ls_provider)
			set("gen_ai.request.model", attrs["llm.model_name"])
			set("gen_ai.response.model", reply.response_metadata?.model_name)
			set("gen_ai.response.id", reply.id)
			set("gen_ai.usage.input_tokens", attrs["llm.token_count.prompt"])
			set("gen_ai.usage.output_tokens", attrs["llm.token_count.completion"])
			if (finish) span.setAttribute("gen_ai.response.finish_reasons", [finish])
			set("gen_ai.input.messages", (input?.messages?.[0] ?? []).map(toGenAiMessage))
			set(
				"gen_ai.output.messages",
				generations.map((g: any) => ({ ...toGenAiMessage(g.message ?? { role: "assistant", content: g.text }), finish_reason: finish ?? "stop" })),
			)
		} else if (kind === "TOOL") {
			// A tool called by the model returns a serialized ToolMessage
			const message = output?.output?.lc ? output.output.kwargs : undefined
			const content = message ? text(message.content) : attrs["output.value"]
			const result = parse(content)
			set("gen_ai.operation.name", "execute_tool")
			set("gen_ai.tool.name", attrs["tool.name"])
			set("gen_ai.tool.call.id", message?.tool_call_id)
			set("gen_ai.tool.call.arguments", this.toolArgs.get(message?.tool_call_id))
			this.toolArgs.delete(message?.tool_call_id)
			if (content !== undefined) set("gen_ai.tool.call.result", typeof result === "object" && result !== null ? result : content)
		} else if (span.name === metadata.lc_agent_name || AGENT_NAMES.has(span.name)) {
			const messages = output?.messages ?? []
			set("gen_ai.operation.name", "invoke_agent")
			set("gen_ai.agent.name", span.name)
			set("gen_ai.input.messages", (input?.messages ?? []).map(toGenAiMessage))
			set("gen_ai.output.messages", messages.slice(-1).map(toGenAiMessage))
		} else if (kind !== "RETRIEVER" && kind !== "EMBEDDING") {
			// Graph nodes, prompts and other runnables: steps, not model or tool calls
			set("gen_ai.operation.name", "invoke_workflow")
		}
	}

	onStart() {}
	onEnd() {}
	forceFlush() {
		return Promise.resolve()
	}
	shutdown() {
		return Promise.resolve()
	}
}

Then create an instrumentation.ts and import it as the first line of your entry point (import "./instrumentation"):

// instrumentation.ts
import { LangChainInstrumentation } from "@arizeai/openinference-instrumentation-langchain"
import * as CallbackManagerModule from "@langchain/core/callbacks/manager"
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto"
import { NodeSDK } from "@opentelemetry/sdk-node"
import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base"
import { GenAiSpans } from "./genai-spans"

// Reads OTEL_SERVICE_NAME, OTEL_RESOURCE_ATTRIBUTES and OTEL_EXPORTER_OTLP_*
export const spanProcessor = new BatchSpanProcessor(new OTLPTraceExporter())
export const sdk = new NodeSDK({
	spanProcessors: [new GenAiSpans(), spanProcessor],
	// Room for long chats: OpenInference writes several attributes per message
	spanLimits: { attributeCountLimit: 1000 },
})
sdk.start()

new LangChainInstrumentation().manuallyInstrument(CallbackManagerModule)

Keep the manuallyInstrument() call. Without it, nothing from LangChain is traced.

Give every createAgent() a name, sub-agents included, so each gets its own lane. For a graph you build with StateGraph, compile it with a name (.compile({ name: "planner" })) and add that name to AGENT_NAMES.

If the app already starts a NodeSDK or another tracer provider (auto-instrumentations, Sentry), add new GenAiSpans() and the exporter’s processor to its span processors instead of starting a second SDK.

Pass your app’s conversation id as thread_id on every invoke(), stream() and Command(resume=...):

import tracing  # first, before the first invoke()

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    ChatOpenAI(model="gpt-4o-mini", stream_usage=True),
    tools=[get_weather, calculate],
    name="assistant",
    checkpointer=InMemorySaver(),
)


def handle_message(conversation_id: str, text: str) -> str:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": text}]},
        {"configurable": {"thread_id": conversation_id}},
    )
    return result["messages"][-1].content

Use the id your app stores the chat under. A new UUID per request gives you one session per message. A plain chain (prompt | model, no graph) ignores configurable, so pass {"metadata": {"thread_id": conversation_id}} instead.

Keep stream_usage=True if ChatOpenAI uses a custom base_url, vLLM or a gateway. Without it, streamed replies have no tokens.

Pass your app’s conversation id as thread_id on every invoke(), stream() and new Command({ resume }):

import "./instrumentation" // first, before the first invoke()

import { ChatOpenAI } from "@langchain/openai"
import { MemorySaver } from "@langchain/langgraph"
import { createAgent } from "langchain"

const agent = createAgent({
	model: new ChatOpenAI({ model: "gpt-4o-mini" }),
	tools: [getWeather, calculate],
	name: "assistant",
	checkpointer: new MemorySaver(),
})

export async function handleMessage(conversationId: string, text: string) {
	const result = await agent.invoke(
		{ messages: [{ role: "user", content: text }] },
		{ configurable: { thread_id: conversationId } },
	)
	return result.messages.at(-1)?.content
}

Use the id your app stores the chat under. A new UUID per request gives you one session per message. A plain chain (prompt.pipe(model), no graph) ignores configurable, so pass { metadata: { thread_id: conversationId } } instead.

Long-running servers need nothing. In serverless handlers, notebooks and task workers, flush after each run:

from tracing import provider

try:
    handle_message("conv-42", "What's the weather in Berlin?")
finally:
    provider.force_flush()

On LangGraph Server (langgraph dev or self-hosted), import tracing at the top of the graph module that langgraph.json points to, and set the OTEL_* variables in the server’s environment. Server threads group into sessions without extra code.

Long-running servers need nothing. At the end of a script, call await sdk.shutdown(). In serverless handlers and queue workers, flush after each run:

import { spanProcessor } from "./instrumentation"

try {
	await handleMessage("conv-42", "What's the weather in Berlin?")
} finally {
	await spanProcessor.forceFlush()
}