Transitions: Wire Nodes Together

Transitions define how the conversation moves from one node to another. They are the edges of your graph connecting nodes into a complete conversation flow. The graph engine follows these transitions deterministically, so the path is always predictable and auditable.

Linear Transitions

Use add_transition() to chain nodes in a fixed sequence. The conversation will move through each node in the order you specify, from START to END.

graph.add_transition(START, "welcome", "collect_name", "collect_email", "confirm", END)

START -> welcome -> collect_name -> collect_email -> confirm -> END

Conditional Transitions

Use add_conditional_transition() when a node needs to route to different targets based on runtime state. This lets you create branching logic. For example, approving or rejecting a loan based on the user's credit score.

1. The source node sets a state field:

async def credit_decision_node(state, ctx):
    score = state.credit_score or 0
    if score >= 700:
        return {"decision": "pending_docs"}
    elif score >= 600:
        return {"decision": "review"}
    else:
        return {"decision": "rejected"}

2. The transition reads that field and routes:

graph.add_conditional_transition(
    "credit_decision",
    mapping={
        "pending_docs": "document_check",
        "review":       "manual_review",
        "rejected":     "reject",
    },
    decide=lambda state: state.decision,
)

                        decide(state) returns:
                        +-- "pending_docs" --> document_check
credit_decision --------+-- "review"       --> manual_review
                        +-- "rejected"     --> reject

Parameter	Type	Description
from_node	str	The source node. decide runs after this node completes.
mapping	dict[str, str]	Maps decision keys to target node names.
decide	Callable(state) -> str	Returns one of the keys from mapping.

If decide returns a key not in mapping, an InvalidTransitionError is raised.

Registering Nodes

Before you can wire transitions, every node function must be registered with the graph. You can use add_node() for standard nodes, or the convenience methods add_start_node() and add_end_node() which automatically create the START or END transitions for you.

graph.add_node("name of node", func name)  

graph.add_node("collect_name", collect_name_node)       # standard node
graph.add_start_node("welcome", welcome_node)           # add_node + add_transition(START, name)
graph.add_end_node("farewell", farewell_node)            # add_node + add_transition(name, END)

GraphConfig

GraphConfig controls the runtime behavior of your graph including logging, retry limits, hangup timing, and persistence. Pass it when creating a ConversationalGraph instance to customize how the engine operates.

config = GraphConfig(
    name="loan_assistant",
    graph_id="loan_v1",
    debug=True,
    max_retries=5,
    hangup_delay=4.0,
    checkpointer=my_saver,
)
graph = ConversationalGraph(LoanState, config)

Option	Type	Default	Description
name	str	"conversational_graph"	Human-readable name for logs
graph_id	str	UUID	Unique version identifier
debug	bool	False	Enable verbose logging
max_retries	int	10	Max Interrupt retries per node
hangup_delay	float	4.0	Seconds before hangup after graph ends
checkpointer	BaseCheckpointer	InMemorySaver()	Persistence backend
stream	bool	False	Whether to give runtime updates of each node

Graph Methods

Once your graph is built and compiled, you can use these methods to inspect its topology, manage state, and control execution at runtime.

Method	Description
graph.compile(user_config)	Initialize runtime. Must be called before processing input. Agent is set automatically by the pipeline.
graph.visualize()	Print graph topology.
graph.get_graph_status()	Print current state, active node, pause status.
graph.update_states(updates)	Apply state updates and save a checkpoint.
graph.get_states()	Return dict of all state values.
graph.get_nodes()	Return list of node names.
graph.get_transitions()	Return dict of all edges.