lang graph example

Thanks for reaching out, Ajay. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

one working example basic chat

Acknowledged — you want a minimal, working example of a basic chat using LangGraph with LiveKit Agents. Below is a concise Python example that shows: • a simple LangGraph/StateGraph with a single echo node • wrapping it in LiveKit's

LLMAdapter

• creating an

AgentSession

and sending one user message to get a response Replace placeholders (API keys, endpoints) with your values.

# Minimal LangGraph + LiveKit Agents basic chat example

from langgraph.graph import StateGraph, Node
from langgraph.actions import ReturnAction
from livekit.agents import AgentSession
from livekit.plugins import langchain as lk_langchain

# 1) Define a simple StateGraph that echoes user input
def create_workflow():
    g = StateGraph(name="echo_workflow")

    # a single node that returns a message based on the user's input
    def echo_behavior(context):
        # context.message contains the incoming user text
        user_text = context.message.content if context.message else ""
        return ReturnAction({"response": f"Echo: {user_text}"})

    echo_node = Node(name="echo", behavior=echo_behavior)
    g.add_node(echo_node)
    g.set_start("echo")

    return g.compile()

# 2) Create an LLM adapter around the compiled graph
compiled_graph = create_workflow()
llm_adapter = lk_langchain.LLMAdapter(graph=compiled_graph)

# 3) Create an AgentSession using the adapter
session = AgentSession(
    llm=llm_adapter,
    # Optionally configure other plugins: stt, tts, vad, turn_detection
)

# 4) Run a simple chat turn and get the assistant response
user_message = "Hello, LangGraph!"
result = session.send_user_message(user_message)

# `result` typically contains the assistant message(s); print response
print("Assistant response:", result)

Notes: • This example uses a single node returning an immediate response via

ReturnAction

. It keeps the example minimal and self-contained. • If you want a variant that maps messages to LangChain message types or integrates STT/TTS/turn detection, tell me which parts you want and I’ll extend the example.

from dotenv import load_dotenv from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage, HumanMessage from langchain_openai import ChatOpenAI from livekit import agents, rtc from livekit.agents import ( AgentServer, AgentSession, Agent, room_io, RoomInputOptions, RoomOutputOptions, ) from livekit.plugins import noise_cancellation, silero, deepgram, langchain from livekit.plugins.turn_detector.multilingual import MultilingualModel import asyncio import os # ----------------------------------------- # 1️⃣ Load environment # ----------------------------------------- load_dotenv(".env.local") # Use a lightweight, low-latency OpenAI model for real-time generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6) # ----------------------------------------- # 2️⃣ LangGraph State + Node # ----------------------------------------- class ChatState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] async def chat_node(state: ChatState): """Streaming chat node for faster live responses.""" messages = state["messages"] # Stream the model output asynchronously for minimal latency print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>") print(state) async for chunk in generator_llm.astream(messages): yield {"messages": [chunk]} def create_workflow(): """Builds the minimal reactive chat graph.""" graph = StateGraph(ChatState) graph.add_node("chat_node", chat_node) graph.add_edge(START, "chat_node") graph.add_edge("chat_node", END) return graph.compile() # ----------------------------------------- # 3️⃣ Assistant Persona # ----------------------------------------- class Assistant(Agent): def __init__(self): super().__init__( instructions=( "You are a helpful real-time voice assistant. " "You respond immediately with partial thoughts, " "then refine as you continue speaking. " "Keep responses short, natural, and conversational." ) ) # ----------------------------------------- # 4️⃣ LiveKit Agent Server # ----------------------------------------- server = AgentServer() initial_state = { 'messages': [HumanMessage(content='What is the capital of india')] } @breezy-insurance-66772.rtc_session() async def my_agent(ctx: agents.JobContext): print("🎙️ Real-time Gemini/OpenAI assistant started...") # Create workflow and adapter workflow = create_workflow() llm_adapter = langchain.LLMAdapter(graph=workflow) # Build agent session for streaming conversation session = AgentSession( stt="assemblyai/universal-streaming:en", llm=llm_adapter, tts=deepgram.TTS(model="aura-asteria-en"), # ✅ streaming TTS vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) await session.start( room=ctx.room, agent=Assistant(), room_options=room_io.RoomOptions( audio_input=room_io.AudioInputOptions( noise_cancellation=lambda params: ( noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC() ), ), ), room_input_options=RoomInputOptions( text_enabled=True, audio_enabled=True, ), room_output_options=RoomOutputOptions( transcription_enabled=True, audio_enabled=True, ), ) # Send greeting asynchronously so startup is instant # asyncio.create_task(session.generate_reply("Hi there! How can I help you today?")) if name == "__main__": agents.cli.run_app(server) correct example as live kit support

Thanks — I reviewed your snippet and corrected it into a minimal, working LiveKit Agents + LangGraph chat example. I fixed decorator usage, async flow, imports, and a few naming issues. Replace environment values (API keys, endpoints) as needed.

# Minimal LiveKit Agents + LangGraph streaming chat example
# Replace env values in .env.local before running

from dotenv import load_dotenv
import asyncio
import os
from typing import TypedDict, Annotated

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI

from livekit import agents, rtc
from livekit.agents import AgentServer, AgentSession, Agent, room_io, RoomInputOptions, RoomOutputOptions
from livekit.plugins import noise_cancellation, silero, deepgram, langchain
from livekit.plugins.turn_detector.multilingual import MultilingualModel

# 1) Load environment
load_dotenv(".env.local")

# lightweight, low-latency OpenAI model for real-time streaming
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6)

# 2) LangGraph State + Node
class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

async def chat_node(state: ChatState):
    """Streaming chat node that yields partial model outputs."""
    messages = state["messages"]
    # stream tokens/chunks from the LLM
    async for chunk in generator_llm.astream(messages):
        # yield chunks back into the LangGraph state as incremental messages
        yield {"messages": [chunk]}

def create_workflow():
    """Builds the minimal reactive chat graph."""
    graph = StateGraph(ChatState)
    graph.add_node("chat_node", chat_node)
    graph.add_edge(START, "chat_node")
    graph.add_edge("chat_node", END)
    return graph.compile()

# 3) Assistant persona
class Assistant(Agent):
    def __init__(self):
        super().__init__(
            instructions=(
                "You are a helpful real-time voice assistant. "
                "Respond immediately with partial thoughts, then refine as you continue. "
                "Keep responses short and conversational."
            )
        )

# 4) LiveKit Agent Server
server = AgentServer()

@server.rtc_session()  # correct decorator usage
async def my_agent(ctx: agents.JobContext):
    print("🎙️ Real-time assistant started...")

    # initial state example (optional)
    initial_state = {
        "messages": [HumanMessage(content="What is the capital of India?")]
    }

    # create workflow and adapter
    workflow = create_workflow()
    llm_adapter = langchain.LLMAdapter(graph=workflow)

    # build agent session with streaming STT/TTS and turn detection
    session = AgentSession(
        stt="assemblyai/universal-streaming:en",
        llm=llm_adapter,
        tts=deepgram.TTS(model="aura-asteria-en"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    # start session attached to the room and agent
    await session.start(
        room=ctx.room,
        agent=Assistant(),
        initial_state=initial_state,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: (
                    noise_cancellation.BVCTelephony()
                    if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                    else noise_cancellation.BVC()
                ),
            ),
        ),
        room_input_options=RoomInputOptions(
            text_enabled=True,
            audio_enabled=True,
        ),
        room_output_options=RoomOutputOptions(
            transcription_enabled=True,
            audio_enabled=True,
        ),
    )

    # keep the session running while connected
    await session.wait()  # yields until session ends

if __name__ == "__main__":
    agents.cli.run_app(server)

If you want, I can: • Show a plain-text-only variant (no STT/TTS) for quick local testing. • Add a sample .env.local with the exact keys/env variables to set. • Walk through how to send a test text message into the session and read the assistant reply. Which would help most?

Traceback (most recent call last): File "/Users/ajaykumar/Work/Projects/PythonLearning/LangGraphDemo/.venv/lib/python3.11/site-packages/opentelemetry/util/_ decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/ajaykumar/Work/Projects/PythonLearning/LangGraphDemo/.venv/lib/python3.11/site-packages/livekit/agents/ipc/j ob_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "/Users/ajaykumar/Work/Projects/PythonLearning/LangGraphDemo/agent.py", line 82, in my_agent await session.start( ^^^^^^^^^^^^^^ TypeError: AgentSession.start() got an unexpected keyword argument 'initial_state'

Thanks — I see the error:

AgentSession.start()

does not accept

initial_state

. You're passing

initial_state

in your call, which causes the TypeError. What to do next: • Remove

initial_state

from

await session.start(...)

. • If you need to seed LangGraph state, initialize it inside the workflow or feed messages at runtime (for example, by calling

session.generate_reply()

or sending a user message after start). The

langchain.LLMAdapter

maps LiveKit chat messages to LangChain message types so you can provide initial messages via the normal chat flow rather than via

start()

. Quick corrected snippet (showing removal of

initial_state

):

# ... build session as before ...
await session.start(
    room=ctx.room,
    agent=Assistant(),
    room_options=room_io.RoomOptions(...),
    room_input_options=RoomInputOptions(text_enabled=True, audio_enabled=True),
    room_output_options=RoomOutputOptions(transcription_enabled=True, audio_enabled=True),
)
# send an initial message after start if desired
await session.generate_reply("What is the capital of India?")

If you want, tell me whether you prefer: • Seeding the conversation with an initial user message after start (I can show exact call sequence), or • Embedding initialization logic in the LangGraph workflow (I can show how to set default messages in the graph). Which option should I provide code for? Sources: Agent session | LiveKit docs | LangChain integration guide | LiveKit docs

hain/langgraph.py124 LangChainDeprecationWarning: Calling .text() as a method is deprecated. Use .text as a property instead (e.g., message.text).

Thanks — that warning means a LangChain message object's

.text()

method is deprecated. Update your code to use the

.text

property (or

.content

for LangChain's message types) instead. In your snippet, change any calls like: •

message.text()

->

message.text

• or use

message.content

for `HumanMessage`/`AIMessage` Example fix inside

chat_node

(if you were accessing the user message text):

# before (deprecated)
user_text = state["messages"][-1].text()

# after
user_text = state["messages"][-1].text  # or .content

Search your code for

.text()

and replace them. If you want, share the specific line(s) and I’ll show the exact replacement.

create node where i do llm call

Acknowledged — you want a LangGraph node that performs the LLM call. Below is a minimal, practical example showing a node that invokes an async LLM (using LangChain-style

ChatOpenAI.astream

) and returns the model reply into the LangGraph state. Notes: • Use

.content

or

.text

(property) on LangChain messages, not

.text()

. • The node yields incremental outputs for streaming; return a final value for a non-streaming call. • Replace

generator_llm

with whichever LLM client you use.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI  # or your LLM client

# LLM client (async streaming)
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6)

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# Streaming node: yields partial chunks as they arrive
async def llm_node(state: ChatState):
    messages = state["messages"]
    # Pass LangChain message list directly to the LLM
    async for chunk in generator_llm.astream(messages):
        # chunk is typically a BaseMessage-like partial; yield incremental state updates
        yield {"messages": [chunk]}

# Non-streaming node: await full response and return once
async def llm_node_full(state: ChatState):
    messages = state["messages"]
    response = await generator_llm.agenerate(messages)  # or .generate() depending on client
    # response format varies; extract message content accordingly
    # Example: response.generations[0].message or response[0].message depending on client
    model_msg = response.generations[0].message
    return {"messages": [model_msg]}

def create_workflow_streaming():
    graph = StateGraph(ChatState)
    graph.add_node("llm_stream", llm_node)
    graph.add_edge(START, "llm_stream")
    graph.add_edge("llm_stream", END)
    return graph.compile()

def create_workflow_full():
    graph = StateGraph(ChatState)
    graph.add_node("llm_full", llm_node_full)
    graph.add_edge(START, "llm_full")
    graph.add_edge("llm_full", END)
    return graph.compile()

If you want, I can: • Show how to seed

state["messages"]

from LiveKit chat inside the

AgentSession

flow, or • Adapt the node to extract only the last user message and send a system prompt + user message to the LLM. Which would help most?

def create_workflow(): """Builds the minimal reactive chat graph.""" graph = StateGraph(...) graph.add_node("chat_node", chat_node) graph.add_edge(START, "chat_node") graph.add_edge("chat_node", END) return graph.compile() explain this code

You're asking what each part of this function does. Here's a concise explanation: •

StateGraph(...)

Creates a new workflow graph type. The type argument (e.g., a TypedDict like

ChatState

) defines the shape of the graph's state that nodes will read and update. •

graph.add_node("chat_node", chat_node)

Registers a node named "chat_node" and associates it with the callable

chat_node

. A node is a unit of behavior that receives the current state, performs work (sync or async), and returns or yields updated state fragments. •

graph.add_edge(START, "chat_node")

Connects the special

START

entry point to the "chat_node". This tells the graph to begin execution at that node when the workflow runs. •

graph.add_edge("chat_node", END)

Connects the "chat_node" to the special

END

exit point, indicating the node is terminal and the workflow should finish after it completes. •

return graph.compile()

Validates and compiles the graph into a runnable workflow object that adapters (like

langchain.LLMAdapter

) or the runtime can execute. Compilation checks node/edge consistency and prepares the graph for execution. In short: this function builds a minimal workflow with a single node that runs when the graph starts and then ends. The node receives and can modify the typed

state

you defined in

StateGraph(...)

.

class ChatState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] create object of this pass graph

You're asking how to instantiate your

ChatState

TypedDict and pass it into a compiled LangGraph

StateGraph

. Below is a concise, correct example and a few brief notes. Example — create state, compile graph, and invoke it:

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, add_messages, START, END

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# Example node that appends an AI reply
async def chat_node(state: ChatState):
    # read last user message
    last = state["messages"][-1].content
    # return a new AIMessage (add_messages reducer will merge it)
    return {"messages": [AIMessage(content=f"Echo: {last}")]}  # or yield for streaming

# Build and compile the graph
graph = StateGraph(ChatState)
graph.add_node("chat_node", chat_node)
graph.add_edge(START, "chat_node")
graph.add_edge("chat_node", END)
compiled = graph.compile()

# Create an initial ChatState instance
initial_state: ChatState = {
    "messages": [
        HumanMessage(content="Hello, how are you?")
    ]
}

# Invoke the compiled graph with the state (sync or async depending on graph)
result = compiled.invoke(initial_state)  # or await compiled.ainvoke(initial_state)
print(result)
# result["messages"] will include the original HumanMessage plus the appended AIMessage

Notes: • Use

add_messages

as the reducer so returned message lists are merged into the existing

messages

list. • For async graphs use

await compiled.ainvoke(initial_state)

. • Ensure you import and use `HumanMessage`/`AIMessage`/`BaseMessage` types your LLM client expects. • If you want streaming partial outputs, the node should

yield

incremental updates;

add_messages

will merge them. Sources: examples/drive-thru | livekit/livekit_composite | DeepWiki | LangChain integration guide | LiveKit docs

complete above example with langgraph

Below is a complete, minimal, end-to-end Python example using LangGraph with a node that calls an async LLM, compiles the graph, creates a ChatState instance, and invokes the graph. It includes both a non-streaming and streaming node variant. Replace API keys/env values and the LLM client as needed.

# complete_langgraph_example.py
from typing import Annotated, List
from typing_extensions import TypedDict
import asyncio
import os

# LangChain message types (used as state items)
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# LangGraph graph primitives
from langgraph.graph import StateGraph, START, END, add_messages

# Example LLM client (async). Replace with your async LLM client.
# This example uses a fake async LLM for illustration.
class FakeAsyncLLM:
    async def agenerate(self, messages: List[BaseMessage]):
        # Simulate async call and return an object shaped like LangChain response
        await asyncio.sleep(0.1)
        class Gen:
            def __init__(self, msg): self.message = msg
        class Resp:
            def __init__(self, gens): self.generations = [gens]
        return Resp(Gen(AIMessage(content="Fake reply to: " + messages[-1].content)))

    async def astream(self, messages: List[BaseMessage]):
        # Simulate streaming chunks (yield partial AIMessage objects)
        for i in range(3):
            await asyncio.sleep(0.05)
            yield AIMessage(content=f"Partial {i+1} for: {messages[-1].content}")

llm = FakeAsyncLLM()

# 1) Define the typed state for the graph
class ChatState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# 2a) Non-streaming node: await full LLM response and return updated state
async def llm_node_full(state: ChatState):
    messages = state["messages"]
    resp = await llm.agenerate(messages)  # adapt to your LLM client
    model_msg = resp.generations[0].message
    # Return a list of messages; add_messages reducer will merge it into state
    return {"messages": [model_msg]}

# 2b) Streaming node: yield incremental partial messages
async def llm_node_stream(state: ChatState):
    messages = state["messages"]
    async for chunk in llm.astream(messages):
        # yield partial updates; add_messages merges them into the messages list
        yield {"messages": [chunk]}

# 3) Build and compile a graph for full (non-streaming) run
def build_full_graph():
    g = StateGraph(ChatState)
    g.add_node("llm_full", llm_node_full)
    g.add_edge(START, "llm_full")
    g.add_edge("llm_full", END)
    return g.compile()

# 4) Build and compile a graph for streaming run
def build_stream_graph():
    g = StateGraph(ChatState)
    g.add_node("llm_stream", llm_node_stream)
    g.add_edge(START, "llm_stream")
    g.add_edge("llm_stream", END)
    return g.compile()

# 5) Example usage: invoke graphs with an initial ChatState
async def main():
    initial_state: ChatState = {
        "messages": [HumanMessage(content="Hello LangGraph!")]
    }

    # Non-streaming invocation
    full_graph = build_full_graph()
    # If compiled graph exposes async invocation:
    result_full = await full_graph.ainvoke(initial_state)  # or compiled.invoke(initial_state) if sync
    print("Non-streaming result messages:")
    for m in result_full["messages"]:
        print(f"- {type(m).__name__}: {m.content}")

    # Streaming invocation
    stream_graph = build_stream_graph()
    print("\nStreaming results (incremental yields):")
    # For streaming, collect incremental results from ainvoke_stream if available,
    # otherwise simulate by driving node directly. Here we assume compiled.ainvoke yields updates.
    async for partial in stream_graph.ainvoke_stream(initial_state):
        # partial is a partial state fragment, e.g., {"messages": [AIMessage(...)]}
        for m in partial["messages"]:
            print(f"- STREAM PARTIAL: {m.content}")

if __name__ == "__main__":
    asyncio.run(main())

Notes: • Use

add_messages

as the reducer so returned/ yielded message lists are appended to the existing

messages

. • For real LLMs, replace

FakeAsyncLLM

with your client (e.g., LangChain's

ChatOpenAI

), and adapt to its response shapes (

agenerate

,

astream

, or `generate`/`astream` names may vary). • Use

compiled.ainvoke(initial_state)

for async full runs and

compiled.ainvoke_stream(initial_state)

(or similar) to receive incremental yields — method names depend on your LangGraph version. If your compiled graph only exposes `invoke`/`ainvoke`, use those accordingly. • If you need the exact integration into LiveKit's

LLMAdapter

, I can show how to wrap the compiled graph with

langchain.LLMAdapter(graph=compiled)

next.

with live kit agent

Thanks — below is a concise, runnable example showing a complete LangGraph workflow that calls an async LLM, compiles the graph, wraps it with LiveKit's

langchain.LLMAdapter

, and runs it inside a LiveKit

AgentSession

. It shows how to provide the initial ChatState (messages) by sending a message after starting the session (since

AgentSession.start()

does not accept

initial_state

). Replace env values and LLM client as needed.

# livekit_langgraph_agent.py
import asyncio
from typing import Annotated, List
from typing_extensions import TypedDict

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END, add_messages

from langchain_openai import ChatOpenAI  # or your async LLM client

from livekit import agents, rtc
from livekit.agents import AgentServer, AgentSession, Agent, room_io, RoomInputOptions, RoomOutputOptions
from livekit.plugins import langchain, deepgram, silero, noise_cancellation
from livekit.plugins.turn_detector.multilingual import MultilingualModel

# ---- LLM client (async streaming/generate) ----
# Use your async LLM client; ChatOpenAI has .agenerate and .astream in some builds.
# Adjust method names to your client (agenerate/astream vs generate/astream).
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6)

# ---- State typed dict ----
class ChatState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# ---- Node that calls the LLM (streaming) ----
async def chat_node(state: ChatState):
    messages = state["messages"]
    # streaming: yield partial chunks as they arrive
    async for chunk in generator_llm.astream(messages):
        # chunk should be a message-like object; yield to have it merged by add_messages
        yield {"messages": [chunk]}

# Non-streaming alternative (await full response)
async def chat_node_full(state: ChatState):
    messages = state["messages"]
    resp = await generator_llm.agenerate(messages)
    # extract message (adapt to your client's response shape)
    model_msg = resp.generations[0].message
    return {"messages": [model_msg]}

# ---- Build and compile the LangGraph workflow ----
def create_workflow_streaming():
    g = StateGraph(ChatState)
    g.add_node("chat_node", chat_node)
    g.add_edge(START, "chat_node")
    g.add_edge("chat_node", END)
    return g.compile()

# ---- Agent persona ----
class Assistant(Agent):
    def __init__(self):
        super().__init__(
            instructions=(
                "You are a helpful real-time voice assistant. "
                "Respond concisely and conversationally."
            )
        )

# ---- LiveKit Agent Server entrypoint ----
server = AgentServer()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    print("Agent starting in room:", ctx.room.sid)

    # compile graph and wrap with LLMAdapter
    compiled = create_workflow_streaming()
    llm_adapter = langchain.LLMAdapter(graph=compiled)

    # build AgentSession with STT/TTS/VAD/turn detection (optional)
    session = AgentSession(
        stt="assemblyai/universal-streaming:en",
        llm=llm_adapter,
        tts=deepgram.TTS(model="aura-asteria-en"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    # start session (do NOT pass initial_state here)
    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: (
                    noise_cancellation.BVCTelephony()
                    if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                    else noise_cancellation.BVC()
                )
            )
        ),
        room_input_options=RoomInputOptions(text_enabled=True, audio_enabled=True),
        room_output_options=RoomOutputOptions(transcription_enabled=True, audio_enabled=True),
    )

    # Seed the conversation by sending a user message after start
    # This uses the session API to generate a reply or inject a message into the flow
    # Use generate_reply for a single turn; or session.send_user_message / session.send_text depending on API.
    await session.generate_reply("What is the capital of India?")

    # Keep the session alive until it ends
    await session.wait()

# ---- CLI runner ----
if __name__ == "__main__":
    agents.cli.run_app(server)

Key points: • Do not pass

initial_state

to

AgentSession.start()

— it is unsupported. • To seed messages, call

session.generate_reply(...)

or the session method that injects a user message after

start()

. • Use

add_messages

in

ChatState

so yielded/returned message lists are merged into

messages

. • For streaming behavior, the node yields partial chunks from

generator_llm.astream(messages)

. For a single final response use `agenerate`/`await`. • Replace

ChatOpenAI

calls with the exact async method names your LLM client provides. Would you like a plain-text-only variant (no STT/TTS) to simplify local testing?

user_away_timeout i want callback of user away time

You're asking how to get a callback when a user is marked "away" by

user_away_timeout

. Use the session's

user_state_changed

event — it fires when the user's state transitions (including to

"away"

). Set

user_away_timeout

on the session if you want a custom timeout. Minimal Python example:

# create session (user_away_timeout in seconds, or None to disable)
session = AgentSession(
    llm=llm_adapter,
    stt="assemblyai/universal-streaming:en",
    user_away_timeout=20.0,
    # ...
)

# register a listener
@session.on("user_state_changed")
def on_user_state_changed(ev):
    # ev.new_state is typically "active", "idle", or "away"
    if ev.new_state == "away":
        print("User is away — run callback logic here")
        # e.g. end session, notify, log, etc.

Notes: • Default

user_away_timeout

is 15.0s if not set. • Set

user_away_timeout=None

to disable away detection. • The event object provides

new_state

(check name in your SDK version); inspect

ev

in a print to confirm available fields if needed. Would you like a short example that ends the session or sends a message when the user goes away? Sources: Agent session | LiveKit docs | Turns overview | LiveKit docs

user_away_timeout set user away time out

You're looking to set the user away timeout and receive a callback when a user is marked away. Here's a concise Python example that shows how to set

user_away_timeout

on AgentSession and handle the

user_state_changed

event to run custom logic (e.g., end the session or send a message).

# configure session with a 20-second away timeout
session = AgentSession(
    llm=llm_adapter,
    stt="assemblyai/universal-streaming:en",
    user_away_timeout=20.0,  # seconds; set to None to disable
    # ... other options
)

# listen for user state changes
@session.on("user_state_changed")
def on_user_state_changed(ev):
    # ev.new_state is typically "active", "idle", or "away"
    if ev.new_state == "away":
        print("User is away — running callback")
        # example: end the session gracefully
        asyncio.create_task(session.stop())
        # or send a message / log / notify:
        # asyncio.create_task(session.generate_reply("We detected you're away. Goodbye."))

Notes: • Default is 15.0s if you don't set

user_away_timeout

. • Use

user_away_timeout=None

to disable away detection. • Inspect

ev

(print it) if your SDK version returns different field names. Would you like a variant that sends a chat message before ending the session or one that marks multiple participants away?

disccnect the user if user_away_timeout is trigger

You can disconnect the user when

user_away_timeout

triggers by listening for the session's

user_state_changed

event and calling the room disconnect/stop method. Here's a concise Python example you can drop into your agent entrypoint:

# inside your rtc_session entrypoint (async)
session = AgentSession(
    llm=llm_adapter,
    stt="assemblyai/universal-streaming:en",
    user_away_timeout=20.0,  # seconds
    # ...
)

@session.on("user_state_changed")
def on_user_state_changed(ev):
    # ev.new_state is typically "active", "idle", or "away"
    if ev.new_state == "away":
        print("User is away — disconnecting")
        # stop the agent session
        asyncio.create_task(session.stop())
        # optionally disconnect the room/local participant
        # asyncio.create_task(ctx.room.disconnect())

Notes: • Default timeout is 15s if unspecified; set

user_away_timeout=None

to disable. • Use

session.stop()

to cleanly shut down the session; use

ctx.room.disconnect()

to force-disconnect the participant. • Ensure VAD/turn-detection is enabled so away detection works. • If you want, I can provide a variant that sends a farewell message before disconnecting. Which would you prefer? Sources: Agent session | LiveKit docs | Turns overview | LiveKit docs | Job lifecycle | LiveKit docs

which is best to give on session shutdown Use

session.stop()

to cleanly shut down the session; use

ctx.room.disconnect()

Good question — use

session.stop()

for a clean, graceful shutdown and

ctx.room.disconnect()

only when you must forcefully drop the participant. Why: • session.stop() ◦ Gracefully stops STT/TT S/VAD, flushes pending work, emits session lifecycle events, and lets the agent perform cleanup (logs, final messages). ◦ Use when you want an orderly end (e.g., after sending a farewell message, saving state, or releasing resources). • ctx.room.disconnect() ◦ Immediately disconnects the participant from the room (hard drop). ◦ Use only for emergency or administrative cases where you must force the client off (no guarantees of cleanup). Recommended pattern for user-away timeout: 1. On "away" event, send a short farewell/notification via session.generate_reply(...) (optional). 2. Wait a short grace period (1–3s) if you sent a message. 3. Call asyncio.create_task(session.stop()) to shut down cleanly. 4. If the client doesn’t disconnect after a timeout, call ctx.room.disconnect() as a fallback. Minimal example:

@session.on("user_state_changed")
def on_user_state_changed(ev):
    if ev.new_state == "away":
        # optional: notify user
        asyncio.create_task(session.generate_reply("We detected you're away. Goodbye."))
        # clean shutdown
        asyncio.create_task(session.stop())
        # fallback after N seconds if needed:
        # asyncio.create_task(delayed_force_disconnect(ctx, delay=5.0))

async def delayed_force_disconnect(ctx, delay=5.0):
    await asyncio.sleep(delay)
    # if still running, force disconnect
    await ctx.room.disconnect()

Would you like a ready-to-run snippet that includes the optional notification + fallback disconnect?