Skip to main content

LangChain integration guide

How to use LangGraph workflows and LangChain agents with LiveKit.

Available inPython

Overview

This plugin allows you to use LangGraph and other graph-based LangChain agents as an LLM provider for your voice agents.

Installation

Install the LiveKit LangChain plugin from PyPI:

uv add "livekit-agents[langchain]~=1.5"

Usage

To use a LangGraph workflow within an AgentSession, compile your graph and wrap it with the LLMAdapter. The following example builds a single-node graph that calls a chat model and uses it as the agent's LLM. For a full agent you can run, see the complete example.

from typing import Annotated, TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from livekit.agents import AgentSession
from livekit.plugins import langchain
class State(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
def create_graph():
llm = init_chat_model("openai:gpt-4.1-mini")
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
return builder.compile()
# Use the compiled graph as the LLM
session = AgentSession(
llm=langchain.LLMAdapter(graph=create_graph()),
# ... stt, tts, vad, turn_handling, etc.
)

The LLMAdapter automatically converts the LiveKit chat context to LangChain messages . The mapping is as follows:

  • system and developer messages to SystemMessage.
  • user messages to HumanMessage.
  • assistant messages to AIMessage.

Parameters

This section describes the available parameters for the LLMAdapter. See the plugin reference for a complete list of all available parameters.

graph
Required
PregelProtocol

The LangGraph workflow to use as an LLM. Must be a locally compiled graph. To learn more, see Graph Definitions .

configRunnableConfig | NoneDefault: None

Configuration options for the LangGraph workflow execution. This can include runtime configuration, callbacks, and other LangGraph-specific options. To learn more, see RunnableConfig .

contextContextT | NoneDefault: None

Static runtime context passed to the graph on each invocation. Use it to supply values your nodes read at runtime, such as a user ID or feature flags. To learn more, see Runtime context .

subgraphsboolDefault: False

Whether to stream outputs from nested subgraphs. Enable this when your graph composes other graphs and you want their tokens and custom data to reach the agent. To learn more, see subgraphs .

stream_modeStreamMode | list[StreamMode]Default: messages

Which LangGraph streaming mode the adapter consumes. Only messages and custom are supported. Other values raise a ValueError.

Streaming modes

The LLMAdapter consumes your graph's output using LangGraph streaming  and forwards it to the agent's TTS in realtime. Use the stream_mode parameter to control which output the adapter streams. Only two modes are supported:

  • messages (default): Streams LLM token chunks as the graph's chat model generates them. This minimizes time to first token and is the right choice for most voice agents.
  • custom: Streams data your nodes emit explicitly with LangGraph's get_stream_writer. Use this to speak text that doesn't come directly from an LLM, such as a fixed acknowledgment or the result of a tool call.

To consume both at once, pass a list:

llm = langchain.LLMAdapter(graph=create_graph(), stream_mode=["messages", "custom"])

Stream custom data

In custom mode, call get_stream_writer() inside a node to get a writer, then pass it the data you want spoken. The adapter speaks a string directly, or the content field of a dictionary that includes one. It skips payloads without text content.

from langchain_core.messages import AIMessage
from langgraph.config import get_stream_writer
def lookup_status(state: State):
writer = get_stream_writer()
writer("Let me check on that for you.") # spoken immediately
status = fetch_status() # your own long-running call
message = f"Your order status is {status}."
writer(message) # spoken when ready
# Record the turn in graph state. Only the writer output is spoken.
return {"messages": [AIMessage(content=message)]}

Enable custom mode, alone or in a list, when you construct the adapter:

llm = langchain.LLMAdapter(graph=create_graph(), stream_mode=["messages", "custom"])

Supported LangChain agent types

The LiveKit LangChain plugin supports LangGraph  and other graph-based LangChain agents:

All of these return a CompiledStateGraph (Pregel-compatible), which the LLMAdapter accepts directly.

To add voice to an existing LangChain agent, pass the compiled graph to LLMAdapter and use it as the Agent's LLM: llm=langchain.LLMAdapter(graph=your_compiled_graph).

The plugin does not support non-graph patterns such as plain LCEL chains (prompt | llm) or bare chat models.

For complete examples including LangGraph, LangChain agents, and deep agents, see the recipes page.

Complete example

The following agent uses a LangGraph workflow as its LLM. The graph handles the model call and any tool use, while the AgentSession handles voice orchestration such as turns and interruptions. To run it, install the plugin with the langchain extra and add langchain[openai] and langgraph to your dependencies.

import logging
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from livekit import agents
from livekit.agents import (
Agent,
AgentServer,
AgentSession,
JobContext,
JobProcess,
TurnHandlingOptions,
inference,
)
from livekit.plugins import langchain, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
logger = logging.getLogger("langgraph-agent")
load_dotenv()
class State(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
def create_graph():
llm = init_chat_model("openai:gpt-4.1-mini")
def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
return builder.compile()
server = AgentServer()
def prewarm(proc: JobProcess):
proc.userdata["vad"] = silero.VAD.load()
server.setup_fnc = prewarm
@server.rtc_session()
async def entrypoint(ctx: JobContext):
# Instructions and tool calls live in the LangGraph workflow, so the
# Agent's own instructions can be left empty.
agent = Agent(
instructions="",
llm=langchain.LLMAdapter(graph=create_graph()),
)
session = AgentSession(
vad=ctx.proc.userdata["vad"],
stt=inference.STT("deepgram/nova-3", language="multi"),
tts=inference.TTS("cartesia/sonic-3"),
turn_handling=TurnHandlingOptions(
turn_detection=MultilingualModel(),
),
)
await session.start(agent=agent, room=ctx.room)
await session.generate_reply(instructions="Ask the user how they're doing.")
if __name__ == "__main__":
agents.cli.run_app(server)

Latency

This plugin uses LangGraph's streaming mode to minimize time to first token as much as possible, but take care when porting LangChain workflows that were not originally designed for voice use cases. For more information on handling long-running operations and providing a better user experience, see the user feedback documentation.

Additional resources

The following resources provide more information about using LangChain with LiveKit Agents.