from kapso.builder import Agentfrom kapso.builder.nodes import DefaultNode, SubagentNode, WarmEndNodefrom kapso.builder.agent.constants import START_NODE, END_NODEagent = Agent(name="customer_service")# Add nodesagent.add_node(START_NODE)agent.add_node(END_NODE)triage = DefaultNode(name="triage", prompt="Understand what the user needs")orders = SubagentNode(name="orders", prompt="Handle order inquiries")support = SubagentNode(name="support", prompt="Provide technical support")goodbye = WarmEndNode(name="goodbye", timeout_minutes=5)agent.add_node(triage)agent.add_node(orders)agent.add_node(support)agent.add_node(goodbye)# Create edge flow with conditionsagent.add_edge(START_NODE, "triage")# From triage to specialized handlersagent.add_edge("triage", "orders", condition="user asks about orders, shipping, or tracking")agent.add_edge("triage", "support", condition="user has technical issues or needs help")agent.add_edge("triage", "goodbye", condition="user just saying hi or bye")# From handlers to endagent.add_edge("orders", "goodbye", condition="order inquiry resolved")agent.add_edge("support", "goodbye", condition="support issue resolved")# Final edgeagent.add_edge("goodbye", END_NODE)# Validate the complete flowagent.validate()
Add conditions to control flow based on conversation context:
Copy
# Multiple paths from one nodeagent.add_edge("greeting", "order_status", condition="user asks about their order or shipment")agent.add_edge("greeting", "product_info", condition="user asks about products or pricing")agent.add_edge("greeting", "support", condition="user has a problem or complaint")# Default path (no condition)agent.add_edge("greeting", "general_help")
The Edge class represents connections between nodes:
Copy
from kapso.builder import Edge, create_edge# Using the Edge class directlyedge = Edge( source="node1", target="node2", condition="user confirms")# Using the factory functionedge = create_edge( source="node1", target="node2", condition="user confirms")
# ✅ Good - Clear, non-overlapping conditionsagent.add_edge("triage", "billing", condition="user has billing or payment questions")agent.add_edge("triage", "technical", condition="user has technical issues or error messages")agent.add_edge("triage", "general", condition="user has general questions or other inquiries")
# Specific conditions firstagent.add_edge("router", "urgent", condition="user mentions emergency or urgent")agent.add_edge("router", "appointment", condition="user wants to schedule or reschedule")# Default catch-all (no condition)agent.add_edge("router", "general_inquiry")
# Simple A → B → C flowagent.add_edge("collect_info", "process_request")agent.add_edge("process_request", "show_results")agent.add_edge("show_results", "end")
Unreachable nodes: All nodes must have a path from START_NODE
Copy
# This will fail validation - node2 is unreachableagent.add_edge(START_NODE, "node1")agent.add_edge("node1", END_NODE)agent.add_node(DefaultNode("node2")) # No edges to/from node2!
Missing terminal paths: All paths must eventually reach END_NODE
Copy
# This will fail - no path to END_NODEagent.add_edge(START_NODE, "node1")agent.add_edge("node1", "node2")agent.add_edge("node2", "node1") # Infinite loop!
Multiple START edges: START_NODE can only have one outgoing edge
Copy
# This will fail validationagent.add_edge(START_NODE, "path1")agent.add_edge(START_NODE, "path2") # Error!
# During development, print the graph structurefor edge in agent.edges: print(f"{edge.source} → {edge.target}") if edge.condition: print(f" Condition: {edge.condition}")
# Create test scenarios for each edge conditiontest_cases = [ ("I need help with my order", "order_support"), ("What products do you have?", "product_info"), ("I want to speak to a human", "handoff")]