Quickstart

Installation and Setup

pip install videosdk-conversational-graph

info

Check out the latest version of videosdk-conversational-graph on PyPI.

Conversational Graph is only supported in cascade mode.

Set your environment variables in a .env file:

VIDEOSDK_AUTH_TOKEN="your_token"
GOOGLE_API_KEY="your_key"           # LLM provider
DEEPGRAM_API_KEY="your_key"         # STT provider
CARTESIA_API_KEY="your_key"         # TTS provider

Minimal Example

example_graph.py

from videosdk.conversational_graph import (
    ConversationalGraph, GraphState, Context,
    Route, END, START, GraphConfig,
)
from pydantic import Field
from typing import Optional

# 1. Define your state (data model)
class GreetingState(GraphState):
    name: Optional[str] = Field(None, description="User's name")

# 2. Define node functions
async def welcome(state: GreetingState, ctx: Context):
    return await ctx.ask("Greet the user and ask for their name.")

async def farewell(state: GreetingState, ctx: Context):
    await ctx.say(f"Nice to meet you, {state.name}! Goodbye!")
    return END

# 3. Build the graph
config = GraphConfig(name="greeter", debug=True)
graph = ConversationalGraph(GreetingState, config)

graph.add_node("welcome", welcome)
graph.add_node("farewell", farewell)
graph.add_transition(START, "welcome", "farewell", END)

Running with VideoSDK Pipeline

main.py

from videosdk.agents import Agent, Pipeline, AgentSession, JobContext, WorkerJob, RoomOptions
from videosdk.plugins.google import GoogleLLM
from videosdk.plugins.deepgram import DeepgramSTT
from videosdk.plugins.cartesia import CartesiaTTS
# import graph
from example_graph import graph

class GreeterAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful voice assistant.")

    async def on_enter(self):
        await self.session.say("Hello! Welcome!")

async def entrypoint(ctx: JobContext):
    agent = GreeterAgent()

    pipeline = Pipeline(
        stt=DeepgramSTT(model="nova-2", language="en"),
        llm=GoogleLLM(model="gemini-2.5-flash"),
        tts=CartesiaTTS(model="sonic-3"),
        conversational_graph=graph,   # Add graph in Pipeline
    )

    session = AgentSession(agent=agent, pipeline=pipeline)
    # compile your graph
    await graph.compile()

    await session.start(wait_for_participant=True, run_until_shutdown=True)

if __name__ == "__main__":
    job = WorkerJob(entrypoint=entrypoint, jobctx=lambda: JobContext(
        room_options=RoomOptions(name="Greeter", playground=True)
    ))
    job.start()

Now Run your main file.

$ python main.py