> ## Documentation Index
> Fetch the complete documentation index at: https://docs.usetuner.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# LangChain / LangGraph
> Capture what your LangGraph/LangChain brain layer did on every call — node transitions, timing, and tool calls — whether you're on a custom stack, LiveKit, or Pipecat.
## 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**, you attach a callback handler and include the resulting segments in your own [Create Call](/docs/api-reference/create-call) payload.
***
## How it works
Three pieces do the work. A simple way to picture them:
```text theme={null}
Handler = ears (listens while the graph runs)
Accumulator = notebook (stores what happened)
Segment builder = formatter (turns the notebook into Tuner transcript rows)
```
1. **Handler** (`TunerLangGraphHandler` or `TunerLangChainHandler`): a LangChain callback. It's passed to `invoke()`; you never call its methods yourself.
2. **Accumulator** (`TunerAccumulator`): one per call. It stores every event the brain layer produced.
3. **Segment builder** (`segments_from_invocation()`): converts one turn's events into Tuner transcript rows, with timestamps as **milliseconds since call start**.
On LiveKit and Pipecat, `wrap_graph()` wires all three for you — you never touch them directly.
**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.
You attach the handler yourself and include the segments in your own Create Call payload. See [Custom stack](/docs/integrations/custom-stack/overview) for the surrounding integration.
### Install
```bash theme={null}
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](https://pypi.org/project/tuner-langchain/).
### 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.
Create one accumulator (the notebook) and one handler (the ears).
```python theme={null}
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
```
Pass the handler in `config` each time you run the graph or chain. This is the only wiring you add.
```python LangGraph theme={null}
result = graph.invoke(state, config={"callbacks": [handler]})
response = result["response"]
```
```python LangChain theme={null}
from tuner_langchain.handlers import TunerLangChainHandler
handler = TunerLangChainHandler(accumulator)
chain.invoke(inputs, config={"callbacks": [handler]})
```
Turn the captured events into transcript rows, merge them with your own voice turns, and POST.
```python theme={null}
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](/docs/api-reference/create-call) request.
The [Tuner LiveKit plugin](/docs/integrations/livekit) handles the wiring and flushes the events automatically when the session ends.
### Install
No extra install — `tuner-langchain` ships 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 — no callbacks to wire up.
```python theme={null}
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:
```python theme={null}
agent = Agent(
instructions="",
llm=langchain.LLMAdapter(
plugin.wrap_chain(my_chain),
stream_mode="messages",
config={"configurable": {"thread_id": thread_id}},
),
)
```
The [Tuner Pipecat observer](/docs/integrations/pipecat) merges brain-layer events into the same `transcript_with_tool_calls` timeline it builds for plain pipecat pipelines — independent of which frames the driving processor emits.
### Install
```shell theme={null}
pip install tuner-pipecat-sdk[langchain]
```
This extra is optional. If `tuner-langchain` isn't installed, calling `wrap_graph` / `wrap_chain` raises a clear `ImportError` with install instructions — everything else about the SDK behaves identically.
### 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:
```python theme={null}
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/`](https://github.com/usetuner/tuner-pipecat-sdk-python/tree/main/examples/pizza_order_langchain) (`wrap_chain()`, a raw LangChain tool-calling chat model) and [`examples/nova_clinic_langgraph/`](https://github.com/usetuner/tuner-pipecat-sdk-python/tree/main/examples/nova_clinic_langgraph) (`wrap_graph()`, a LangGraph `create_react_agent` — demonstrates 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:
```python theme={null}
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 | `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**. On LiveKit and Pipecat, the host SDK calls them for you — you only need `CaptureConfig`.
```python theme={null}
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 (custom stack)
```python theme={null}
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
```
***
## Next steps