Conversational Graph

The Conversational Graph is a powerful tool that allows you to define complex, structured conversation flows for your AI agents. Instead of relying solely on an LLM's inherent reasoning, which can sometimes be unpredictable, you can use a graph-based approach to guide the conversation through specific states and transitions.

Installation

To use the Conversational Graph, you need to install the videosdk-conversational-graph package.

pip install videosdk-conversational-graph

note

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

Core Concepts

The Conversational Graph is built around a few key concepts:

1. ConversationalGraph: The main object that manages the states and transitions.
2. State: A specific point in the conversation (e.g., "Greeting", "Asking for Name"). Each state has instructions for the agent.
3. Transition: Logic that dictates how the agent moves from one state to another based on user input or collected data.

Example: Loan Application

Let's walk through a complete example of building a Loan Application agent. This agent will guide the user through selecting a loan type (Personal, Home, or Car) and collecting the necessary details.

Step 1: Define the Data Model

1

Define the Data Model

First, define the data you want to collect using ConversationalDataModel. This ensures the agent knows exactly what information to extract.

main.py

from pydantic import Field
from conversational_graph import ConversationalGraph,ConversationalDataModel

class LoanFlow(ConversationalDataModel):
    loan_type: str = Field(None, description="Type of loan: personal, home, car")
    annual_income: int = Field(None, description="Annual income of the applicant in INR")
    credit_score: int = Field(None, description="Credit score of the applicant. Must be between 300 and 850")
    property_value: int = Field(None, description="Value of the property for home loan in INR")
    vehicle_price: int = Field(None, description="Price of the vehicle for car loan in INR")
    loan_amount: int = Field(None, description="Desired loan amount in INR. MUST be greater than ₹11 lakh for approval")

Step 2: Initialize the Graph

2

Initialize the Graph

Create an instance of ConversationalGraph and pass your data model.

main.py

loan_application = ConversationalGraph(
    name="Loan Application",
    DataModel=LoanFlow,
    off_topic_threshold=5
)

Step 3: Define States

3

Define States

Define the various states of your conversation. Each state has a name and instruction that tells the agent what to do in that state. You can also define specific tools available to the agent within that state.

main.py

# Start Greeting
q0 = loan_application.state(
    name="Greeting",
    instruction="Welcome user and start the conversation about loan application. Ask if they are ready to apply for a loan.",
)

# Loan Type Selection
q1 = loan_application.state(
    name="Loan Type Selection",
    instruction="Ask user to select loan type. We only offer personal loan, home loan, and car loan at the moment.",
)


# Tool for state q2
submit_loan_application = PreDefinedTool().http_tool(HttpToolRequest(
        name="submit_loan_application",
        description="Called when loan request is approved and sumitted.",
        url="https://videosdk.free.beeceptor.com/apply",
        method="POST"
    )
)

q2 = loan_application.state(
    name="Review and Confirm",
    instruction="Review all loan details with the user and get confirmation.",
    tool=submit_loan_application
)

# Master / Off-topic handler
q_master = loan_application.state(
    name="Off-topic Handler",
    instruction="Handle off-topic or inappropriate inputs respectfully and end the call politely",
    master=True
)

Step 4: Define Transitions

4

Define Transitions

Now, link the states together using transitions. You specify the from_state, to_state, and a condition that must be met to trigger the transition.

main.py

# Greeting → Loan Type Selection
loan_application.transition(
    from_state=q0,
    to_state=q1,
    condition="User ready to apply for loan"
)

# Branch from Loan Type Selection
loan_application.transition(
    from_state=q1,
    to_state=q1a,
    condition="User wants personal loan"
)

# Merge all branches → Loan Amount Collection
loan_application.transition(
    from_state=q1a,
    to_state=q2,
    condition="Personal loan details collected and verified"
)
# ... More Transitions ...

Step 5: Integrate with the Agent Pipeline

5

Integrate with the Agent Pipeline

Finally, pass the conversational_graph to your CascadingPipeline.

main.py

class VoiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful voice assistant that can assist you with your loan applications")
        
    async def on_enter(self) -> None:
        await self.session.say("Hello, I am here to help with your loan application. How can I help you today?", interruptible=False)
    
    async def on_exit(self) -> None:
        await self.session.say("Goodbye!")
  
async def entrypoint(ctx: JobContext):
    
    agent = VoiceAgent()
    conversation_flow = ConversationFlow(agent)

    pipeline = CascadingPipeline(
        stt= DeepgramSTT(),
        llm=OpenAILLM(),
        tts=GoogleTTS(),
        vad=SileroVAD(),
        turn_detector=TurnDetector(),
        conversational_graph = loan_application
    )
    
    session = AgentSession(
        agent=agent, 
        pipeline=pipeline,
        conversation_flow=conversation_flow
    )

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

def make_context() -> JobContext:
    room_options = RoomOptions(room_id="<room_id>", name="Workflow Agent", playground=True)
    
    return JobContext(room_options=room_options)

if __name__ == "__main__":
    job = WorkerJob(entrypoint=entrypoint, jobctx=make_context)
    job.start()

Working Example

Conversational Graph Example

Check out the complete working example of a Loan Application agent using Conversational Graph.