LangChain / LangGraph

This is a Custom Integration library. If you’re using LiveKit, use the built-in LangChain / LangGraph observability support in the Tuner LiveKit plugin instead.

What it does

Every call has two layers:

Voice layer: STT hears the user and TTS speaks the reply (your transport). You already track this.
Brain layer: LangGraph decides which step runs next, and LangChain runs the LLM and tools inside each step. This is invisible unless you instrument it.

tuner-langchain makes the brain layer visible. Instead of hand-wrapping every node and tool, you attach one callback handler to graph.invoke(). It records node transitions, timing, and tool calls automatically. At hang-up you format those events and include them in your Create Call payload.

The library owns brain-layer tracing. You still own the voice transport and the API POST.

How it works

Three pieces do the work. A simple way to picture them:

Handler          = ears       (listens while the graph runs)
Accumulator      = notebook   (stores what happened)
Segment builder  = formatter  (turns the notebook into Tuner transcript rows)

Handler (TunerLangGraphHandler or TunerLangChainHandler): a LangChain callback. You pass it to invoke() and never call its methods yourself.
Accumulator (TunerAccumulator): one per call. It stores every event the brain layer produced. Read it back with get_invocations().
Segment builder (segments_from_invocation()): converts one turn’s events into Tuner transcript rows, with timestamps as milliseconds since call start.

One handler sees both frameworks. LangGraph is built on LangChain, so a single handler on graph.invoke() captures the graph’s nodes and the LLM/tool calls nested inside them, with no extra wiring per LLM.

Framework	Role	You’ll recognize
LangGraph	Orchestration (which step runs next)	`StateGraph`, conditional edges
LangChain	Execution (LLM calls, tools, prompts)	`ChatOpenAI.invoke()`, `@tool`

Install

pip install tuner-langchain

Requires Python ≥ 3.10 and langchain-core ≥ 1.0 (langgraph ≥ 1.0 for graphs). Pin a version in production. See the package on PyPI.

The flow

You touch the library at three moments in a call: set up once, attach the handler on every turn, then format and send when the call ends.

Set up once per call

Create one accumulator (the notebook) and one handler (the ears).

from tuner_langchain import TunerAccumulator
from tuner_langchain.handlers import TunerLangGraphHandler

accumulator = TunerAccumulator()              # collects events for this call
handler = TunerLangGraphHandler(accumulator)  # listens while the graph runs

Attach the handler on every turn

Pass the handler in config each time you run the graph or chain. This is the only wiring you add.

result = graph.invoke(state, config={"callbacks": [handler]})
response = result["response"]

from tuner_langchain.handlers import TunerLangChainHandler

handler = TunerLangChainHandler(accumulator)
chain.invoke(inputs, config={"callbacks": [handler]})

Format and send when the call ends

Turn the captured events into transcript rows, merge them with your own voice turns, and POST.

from tuner_langchain.segment_builder import segments_from_invocation

# segments_from_invocation expects nanoseconds, so convert your call start
call_start_ns = call_start_ms * 1_000_000

graph_segments = []
for invocation in accumulator.get_invocations():   # one entry per user turn
    graph_segments.extend(segments_from_invocation(invocation, call_start_ns))

# merge with your STT/TTS turns and sort the timeline
transcript = sorted(voice_turns + graph_segments, key=lambda e: e.get("start_ms", 0))

Then include transcript in your Create Call request.

Data privacy

By default the library forwards prompts, inputs, and tool payloads. To keep sensitive data out, pass a CaptureConfig to the accumulator:

from tuner_langchain import TunerAccumulator, CaptureConfig

accumulator = TunerAccumulator(
    capture=CaptureConfig(
        node_instructions=False,   # system prompts
        node_inputs=False,         # graph state in
        node_outputs=False,        # graph state out
        tool_inputs=False,         # tool parameters
        tool_outputs=False,        # tool results
    )
)

Tool error output is always captured, regardless of tool_outputs. Errors aren’t treated as sensitive and are needed for debugging.

Public API

from tuner_langchain import TunerAccumulator, CaptureConfig
from tuner_langchain.handlers import TunerLangGraphHandler, TunerLangChainHandler
from tuner_langchain.segment_builder import segments_from_invocation
from tuner_langchain.models import GraphInvocation, NodeTransition, ToolCallEvent

Symbol	Role
`TunerLangGraphHandler`	Callback for a LangGraph `StateGraph`
`TunerLangChainHandler`	Callback for a plain LangChain chain
`TunerAccumulator`	Per-call event storage; read with `get_invocations()`
`segments_from_invocation()`	Turns raw events into Tuner transcript rows
`CaptureConfig`	Privacy / redaction flags
`GraphInvocation`, `NodeTransition`, `ToolCallEvent`	Data models, mostly for reading/debugging

Minimal starter

import time
from tuner_langchain import TunerAccumulator, CaptureConfig
from tuner_langchain.handlers import TunerLangGraphHandler
from tuner_langchain.segment_builder import segments_from_invocation

# Per call
call_start_ms = int(time.time() * 1000)
accumulator = TunerAccumulator(capture=CaptureConfig(tool_inputs=False))
handler = TunerLangGraphHandler(accumulator)

voice_turns = []   # your own STT/TTS turns

# Per turn
def handle_turn(user_text: str, state: dict) -> str:
    voice_turns.append({"role": "user", "text": user_text, "start_ms": ...})
    result = compiled_graph.invoke(state, config={"callbacks": [handler]})
    response = result["response"]
    voice_turns.append({"role": "agent", "text": response, "start_ms": ...})
    return response

# At hang-up
def build_transcript() -> list[dict]:
    call_start_ns = call_start_ms * 1_000_000
    segments = []
    for inv in accumulator.get_invocations():
        segments.extend(segments_from_invocation(inv, call_start_ns))
    return sorted(voice_turns + segments, key=lambda e: e.get("start_ms", 0))

# POST build_transcript() inside your Tuner Create Call payload

Getting Started

Simulation

Integrations

MCP

API Reference

Agent Configurations

What it does

How it works

Install

The flow

Data privacy

Public API

Minimal starter

Next steps

Custom Integration

Create Call API Reference

​What it does

​How it works

​Install

​The flow

​Data privacy

​Public API

​Minimal starter

​Next steps

Custom Integration

Create Call API Reference

What it does

How it works

Install

The flow

Data privacy

Public API

Minimal starter

Next steps