tuner-langchain 0.1.1

LangChain and LangGraph observability for Tuner SDKs

tuner-langchain

LangChain and LangGraph observability for Tuner SDKs.

Captures node transitions, tool calls, and tool results from LangChain and LangGraph agents and feeds them into the Tuner transcript pipeline — giving you full visibility into what your orchestration layer did on every call.

---

Overview

When users run LangChain or LangGraph as the orchestration layer inside a LiveKit or Pipecat voice agent, tool calls and node transitions happen inside the graph — invisible to the voice framework. This package bridges that gap.

It works by attaching a LangChain callback handler to the graph invocation. The handler captures every node transition and tool call with wall-clock timestamps, and the Tuner SDK mappers inject them into the transcript at flush time — in the correct position between the user message and the bot response.

---

Installation

Not yet on PyPI. Install locally from the repo:

pip install -e /path/to/tuner-langchain

Once published:

pip install tuner-langchain

Requirements: Python ≥ 3.10, langchain-core >= 1.0, < 2.0

---

Usage

With LiveKit

from livekit.agents import AgentSession, JobContext, Agent
from livekit.plugins import langchain
from tuner import TunerPlugin

async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    plugin = TunerPlugin(session, ctx)
    wrapped_graph = plugin.wrap_graph(my_compiled_graph)

    agent = Agent(
        instructions="",
        llm=langchain.LLMAdapter(
            wrapped_graph,
            stream_mode="custom",
            config={"configurable": {"thread_id": thread_id}},
        ),
    )

    await session.start(agent=agent, room=ctx.room)

For plain LangChain (non-graph):

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

---

What gets captured

Node transitions (node_transition)

Every named LangGraph node or LangChain chain step:

Field	Description
node_name	The node/step name as defined in the graph
start_ms	Start time relative to call start
end_ms	End time relative to call start
duration_ms	Execution time in milliseconds
inputs	Node inputs (omitted when empty)
outputs	Node outputs (omitted when empty)
error	Error message if the node failed
node_instructions	System prompt active during this node's LLM call

LangGraph internal nodes (__start__, __end__, compiled graph root) are filtered out automatically.

Tool calls (agent_function + agent_result)

Every tool invocation inside the graph:

Field	Description
tool_name	Tool name
inputs	Raw input string passed to the tool
output	Tool result or error message
is_error	Whether the tool raised an error
duration_ms	Tool execution time in milliseconds
start_ms	Invocation time relative to call start

---

Data Privacy

By default, tuner-langchain forwards the following data to the Tuner ingestion API:

Field	Captured by default	How to disable
Node instructions	✅	node_instructions=False
Tool inputs	✅	tool_inputs=False
Tool outputs	✅	tool_outputs=False
Node inputs	✅	node_inputs=False
Node outputs	✅	node_outputs=False

Node instructions are capped at 300 characters. Tool error output is always captured regardless of tool_outputs — errors are not considered sensitive and are required for debugging.

To disable specific fields, pass a CaptureConfig to wrap_graph() or wrap_chain():

from tuner_langchain import CaptureConfig

wrapped_graph = plugin.wrap_graph(
    my_graph,
    capture=CaptureConfig(
        node_instructions=False,
        tool_inputs=False,
        tool_outputs=False,
        node_inputs=False,
        node_outputs=False,
    )
)

---

How it fits in the transcript

All segments are sorted chronologically by start_ms — the correct execution order:

user message
  node_transition  "intent_classifier"   (start_ms, end_ms, duration_ms)
  node_transition  "booking_node"        (start_ms, end_ms, inputs, outputs)
  agent_function   "get_patient_info"    (start_ms, inputs)
  agent_result     "get_patient_info"    (start_ms, output, duration_ms)
agent text response
user message
  ...

The segment shape is identical to what Tuner produces for non-LangGraph tool calls — same role, same tool object structure, same timing fields. The Tuner API and frontend handle both paths transparently.

---

Architecture

src/tuner_langchain/
├── __init__.py          public API
├── models.py            NodeTransition, ToolCallEvent, GraphInvocation
├── accumulator.py       TunerAccumulator — stores events per session
├── segment_builder.py   segments_from_invocation() — used by SDK mappers at flush
└── handlers/
    ├── __init__.py
    ├── base.py          TunerBaseHandler — shared tool + chain-end logic
    ├── langgraph.py     TunerLangGraphHandler — filters __start__/__end__
    └── langchain.py     TunerLangChainHandler — filters anonymous wrappers

TunerBaseHandler holds all shared logic. TunerLangGraphHandler and TunerLangChainHandler only override on_chain_start — the one place where the two frameworks differ in how chain names are interpreted.

---

Public API

from tuner_langchain import (
    # Handlers — pass to graph/chain config
    TunerLangGraphHandler,
    TunerLangChainHandler,

    # Accumulator — one per session, created by wrap_graph() / wrap_chain()
    TunerAccumulator,

    # Segment builder — used internally by SDK mappers at flush time
    segments_from_invocation,

    # Models
    GraphInvocation,
    NodeTransition,
    ToolCallEvent,
)

---

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy