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
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"]
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))
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
Next steps
Create Call API Reference
