LangChain / LangGraph

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. It records node transitions, timing, and tool calls automatically, with no hand-wrapping of individual nodes or tools. How you attach it depends on your stack. On LiveKit and Pipecat, the host SDK wraps your graph and flushes the events for you. On a custom stack with tuner-core, the TunerSession does the same. On a custom stack without tuner-core, you attach a callback handler and include the resulting segments in your own Create Call payload.

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. It’s passed to invoke(); you never call its methods yourself.
Accumulator (TunerAccumulator): one per call. It stores every event the brain layer produced.
Segment builder (segments_from_invocation()): converts one turn’s events into Tuner transcript rows, with timestamps as milliseconds since call start.

On LiveKit, Pipecat and tuner-core, wrap_graph() wires all three for you.

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`

Attach to your stack

Pick your host. Each tab covers the install and the full wiring for that stack.

Custom stack
LiveKit
Pipecat

See Custom stack for the surrounding integration. How you attach depends on whether you use tuner-core.

Install

pip install tuner-langchain

Requires Python ≥ 3.10 and langchain-core 1.x. Pin a version in production. See the package on PyPI.

With tuner-core

If you use a TunerSession from tuner-core (see the SDK quickstart), wrap the graph and attach it to the session. The session adds the graph’s segments to the call when you run flush().

from tuner_langchain import wrap_graph

graph = session.attach(wrap_graph(your_graph))

# then invoke as usual:
result = await graph.ainvoke({"messages": history})

Use wrap_chain() for a plain LangChain runnable. The wrapper adds its callback to ainvoke() and astream() calls only, so call the graph asynchronously.

Without tuner-core

You attach the handler yourself and include the segments in your own Create Call payload. 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.

The Tuner LiveKit plugin handles the wiring and flushes the events automatically when the session ends.

Install

No extra install. tuner-langchain is installed as a dependency of tuner-livekit-sdk.

Attach to a LangGraph graph

Save the plugin reference, then call wrap_graph() and pass the result into langchain.LLMAdapter. It returns a drop-in replacement for the graph you pass in, with no callbacks to wire up.

from tuner import TunerPlugin
from livekit.plugins import langchain

async def entrypoint(ctx: JobContext):
    session = AgentSession(...)
    plugin = TunerPlugin(session, ctx)      # save the reference

    agent = Agent(
        instructions="",
        llm=langchain.LLMAdapter(
            plugin.wrap_graph(my_graph),
            stream_mode="custom",
            config={"configurable": {"thread_id": thread_id}},
        ),
    )
    await session.start(agent=agent)  # plus your room/transport options

Attach to a plain LangChain chain

For a plain (non-graph) LangChain runnable, use wrap_chain() instead:

agent = Agent(
    instructions="",
    llm=langchain.LLMAdapter(
        plugin.wrap_chain(my_chain),
        stream_mode="messages",
        config={"configurable": {"thread_id": thread_id}},
    ),
)

The Tuner Pipecat observer merges brain-layer events into the same transcript_with_tool_calls timeline it builds for plain Pipecat pipelines, whichever frames the driving processor emits.

Install

pip install 'tuner-pipecat-sdk[langchain]'

This extra is optional. If tuner-langchain isn’t installed, calling wrap_graph / wrap_chain raises an ImportError with install instructions. The rest of the SDK works the same.

Attach a graph or chain

Wrap your compiled graph (or chain) with the observer, then pass the wrapped runnable into pipecat’s LangchainProcessor in place of a native LLM service:

from pipecat.processors.frameworks.langchain import LangchainProcessor
from tuner_pipecat_sdk import Observer

observer = Observer(api_key=..., workspace_id=..., agent_id=..., call_id=...)

wrapped_graph = observer.wrap_graph(my_compiled_graph)   # or observer.wrap_chain(my_chain)
lc = LangchainProcessor(chain=wrapped_graph)

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    lc,                 # in place of a native LLM service
    tts,
    transport.output(),
    context_aggregator.assistant(),
])

Runnable examples: examples/pizza_order_langchain/ (wrap_chain(), a raw LangChain tool-calling chat model) and examples/nova_clinic_langgraph/ (wrap_graph(), a LangGraph create_react_agent that shows node-transition tracking).

Data privacy

By default the library forwards node inputs, outputs, system prompts, and tool payloads. Pass a CaptureConfig to suppress what you don’t need:

from tuner_langchain import CaptureConfig

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
)

Where you pass it depends on your stack:

Stack	Pass it to
Custom stack with `tuner-core`	`wrap_graph(my_graph, capture=capture)`
Custom stack without `tuner-core`	`TunerAccumulator(capture=capture)`
LiveKit	`plugin.wrap_graph(my_graph, capture=capture)`
Pipecat	`observer.wrap_graph(my_graph, capture=capture)`

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

Public API

The symbols below are what you use on a custom stack without tuner-core. With tuner-core, LiveKit or Pipecat, you only need wrap_graph() or wrap_chain() and CaptureConfig.

from tuner_langchain import TunerAccumulator, CaptureConfig, wrap_graph, wrap_chain
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
`wrap_graph()`, `wrap_chain()`	Drop-in wrappers that add the handler for you; used with `tuner-core`
`GraphInvocation`, `NodeTransition`, `ToolCallEvent`	Data models, mostly for reading/debugging

Minimal starter (custom stack without tuner-core)

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

Integrations

Evaluation

Observability

Simulation

Datasets

MCP

Guides

Resources

What it does

How it works

Attach to your stack

Install

With tuner-core

Without tuner-core

Install

Attach to a LangGraph graph

Attach to a plain LangChain chain

Install

Attach a graph or chain

Data privacy

Public API

Minimal starter (custom stack without tuner-core)

Next steps

Custom stack

Connecting to LiveKit

Connecting to Pipecat

Create Call API Reference

​What it does

​How it works

​Attach to your stack

​Install

​With tuner-core

​Without tuner-core

​Install

​Attach to a LangGraph graph

​Attach to a plain LangChain chain

​Install

​Attach a graph or chain

​Data privacy

​Public API

​Minimal starter (custom stack without tuner-core)

​Next steps

Custom stack

Connecting to LiveKit

Connecting to Pipecat

Create Call API Reference

What it does

How it works

Attach to your stack

Install

With tuner-core

Without tuner-core

Install

Attach to a LangGraph graph

Attach to a plain LangChain chain

Install

Attach a graph or chain

Data privacy

Public API

Minimal starter (custom stack without tuner-core)

Next steps