Skip to main content
Control execution flow with conditional routing. Route to different paths based on data, loop for agentic workflows, or terminate early.
  • @ifelse - Simple boolean routing: True goes one way, False goes another
  • @route - Route execution to one of several target nodes based on a function’s return value
  • END - Sentinel indicating execution should terminate along this path
  • multi_target - Route to multiple nodes in parallel when needed

​
When to Use Routing

Routing solves problems that pure DAGs cannot:
PatternExampleWhy DAGs Fail
Conditional pathsRoute based on document typeDAGs execute all branches
Early terminationStop if cache hitDAGs run to completion
Agentic loopsRetry until quality thresholdDAGs have no cycles
Multi-turn conversationContinue until user satisfiedDAGs are single-pass

​
Binary Branching with @ifelse

For simple true/false decisions, use @ifelse. It’s cleaner than @route when you only have two paths.

​
Basic If/Else

from hypergraph import Graph, node, ifelse, END, SyncRunner

@node(output_name="is_cached")
def check_cache(query: str) -> bool:
    return query in cache

@ifelse(when_true="use_cache", when_false="full_retrieval")
def cache_gate(is_cached: bool) -> bool:
    return is_cached

@node(output_name="response")
def use_cache(query: str) -> str:
    return cache[query]

@node(output_name="response")
def full_retrieval(query: str) -> str:
    return expensive_rag_pipeline(query)

graph = Graph([check_cache, cache_gate, use_cache, full_retrieval])
runner = SyncRunner()

result = runner.run(graph, {"query": "What is RAG?"})
print(result["response"])
The function returns True or False. Based on the result:
  • True → routes to when_true target
  • False → routes to when_false target

​
Early Termination with END

from hypergraph import ifelse, END

@ifelse(when_true="process", when_false=END)
def should_process(is_valid: bool) -> bool:
    return is_valid  # False terminates, True continues
When when_false=END, returning False terminates execution along this path.

​
When to Use @ifelse vs @route

Use @ifelse when…Use @route when…
Two mutually exclusive pathsThree or more paths
Decision is booleanDecision returns a name
No fallback neededNeed fallback for None
Simple branchingMulti-target routing

​
Flexible Routing with @route

​
Route to One of Several Targets

from hypergraph import Graph, node, route, END, SyncRunner

@node(output_name="complexity")
def analyze(document: str) -> str:
    """Classify document complexity."""
    if len(document) < 100:
        return "simple"
    elif "diagram" in document.lower():
        return "visual"
    return "complex"

@route(targets=["simple_path", "visual_path", "complex_path"])
def choose_path(complexity: str) -> str:
    """Route based on complexity analysis."""
    if complexity == "simple":
        return "simple_path"
    elif complexity == "visual":
        return "visual_path"
    return "complex_path"

@node(output_name="response")
def simple_path(document: str) -> str:
    return f"Quick answer: {document[:50]}"

@node(output_name="response")
def visual_path(document: str) -> str:
    return f"Processing visual content: {document}"

@node(output_name="response")
def complex_path(document: str) -> str:
    return f"Deep analysis: {document}"

graph = Graph([analyze, choose_path, simple_path, visual_path, complex_path])
runner = SyncRunner()

result = runner.run(graph, {"document": "This contains a diagram of the architecture"})
print(result["response"])  # "Processing visual content: ..."
The routing function examines data and returns the target node name. Only that node executes.

​
Terminate Early with END

from hypergraph import route, END

@route(targets=["process", END])
def check_cache(query: str) -> str:
    """Skip processing if answer is cached."""
    if query in cache:
        return END  # Stop here, don't run "process"
    return "process"
When a route returns END, execution terminates along that path. Other independent paths continue.

​
Real-World Example: RAG with Diagram Detection

Documents containing diagrams need special handling - convert pages with diagrams to images and send them alongside text to a multimodal LLM. 
from hypergraph import Graph, node, route, END, SyncRunner

@node(output_name="documents")
def retrieve(query: str, embedding: list[float]) -> list[dict]:
    """Retrieve relevant documents from vector store."""
    return vector_db.search(embedding, top_k=5)

@node(output_name="doc_analysis")
def analyze_documents(documents: list[dict]) -> dict:
    """Analyze documents to detect which pages contain diagrams."""
    analysis = {
        "has_diagrams": False,
        "diagram_pages": [],
        "text_content": [],
    }
    for doc in documents:
        for page_num, page in enumerate(doc["pages"]):
            if page.get("has_diagram"):
                analysis["has_diagrams"] = True
                analysis["diagram_pages"].append({
                    "doc_id": doc["id"],
                    "page_num": page_num,
                    "content": page["content"],
                })
            analysis["text_content"].append(page["text"])
    return analysis

@route(targets=["text_only_response", "multimodal_response"])
def route_by_content(doc_analysis: dict) -> str:
    """Route based on whether documents contain diagrams."""
    if doc_analysis["has_diagrams"]:
        return "multimodal_response"
    return "text_only_response"

@node(output_name="response")
def text_only_response(query: str, doc_analysis: dict) -> str:
    """Generate response using text-only LLM."""
    context = "\n".join(doc_analysis["text_content"])
    return llm.generate(
        model="gpt-5.2",
        messages=[
            {"role": "system", "content": f"Context:\n{context}"},
            {"role": "user", "content": query},
        ],
    )

@node(output_name="response")
def multimodal_response(query: str, doc_analysis: dict) -> str:
    """Generate response using multimodal LLM with diagram images."""
    # Convert diagram pages to images
    images = []
    for page_info in doc_analysis["diagram_pages"]:
        image = pdf_to_image(page_info["doc_id"], page_info["page_num"])
        images.append(image)

    # Combine text and images for multimodal LLM
    context = "\n".join(doc_analysis["text_content"])
    return multimodal_llm.generate(
        model="gpt-5.2",
        messages=[
            {"role": "system", "content": f"Context:\n{context}"},
            {"role": "user", "content": [
                {"type": "text", "text": query},
                *[{"type": "image", "image": img} for img in images],
            ]},
        ],
    )

# Build the graph
rag_with_diagrams = Graph([
    retrieve,
    analyze_documents,
    route_by_content,
    text_only_response,
    multimodal_response,
])

# Run
runner = SyncRunner()
result = runner.run(rag_with_diagrams, {
    "query": "Explain the system architecture",
    "embedding": [0.1, 0.2, ...],
})
print(result["response"])
The routing logic is clean and explicit:
1

Retrieve documents

Fetch relevant documents from vector store
2

Analyze for diagrams

Detect which pages contain visual content
3

Route to appropriate LLM

Text-only or multimodal based on content type
4

Generate response

Use the right model with the right inputs

​
Multi-Target Routing

Sometimes you need to run multiple paths in parallel. Use multi_target=True. 
@route(targets=["notify_slack", "notify_email", "log_event"], multi_target=True)
def choose_notifications(event_type: str, severity: str) -> list[str]:
    """Route to multiple notification channels based on event."""
    targets = ["log_event"]  # Always log

    if severity == "critical":
        targets.extend(["notify_slack", "notify_email"])
    elif severity == "warning":
        targets.append("notify_slack")

    return targets

@node(output_name="slack_sent")
def notify_slack(message: str) -> bool:
    return slack.send(message)

@node(output_name="email_sent")
def notify_email(message: str) -> bool:
    return email.send(message)

@node(output_name="logged")
def log_event(message: str) -> bool:
    return logger.info(message)
With multi_target=True, the function returns a list of targets to execute. All returned targets run (potentially in parallel with AsyncRunner).
When using multi_target=True, target nodes must have unique output names. If multiple nodes produce the same output name, you’ll get a GraphConfigError at build time.

​
Fallback Targets

When a routing function might return None, use fallback to specify a default: 
@route(targets=["premium_path", "standard_path"], fallback="standard_path")
def route_by_tier(user_tier: str | None) -> str | None:
    """Route premium users to premium path."""
    if user_tier == "premium":
        return "premium_path"
    return None  # Falls back to standard_path

​
Validation and Error Handling

​
Build-Time Validation

Hypergraph validates routing at graph construction: 
@route(targets=["step_a", "step_b", END])
def decide(x: int) -> str:
    return "step_c"  # Typo - not in targets

graph = Graph([decide, step_a, step_b])
# GraphConfigError: Route target 'step_c' not found.
# Valid targets: ['step_a', 'step_b', 'END']
# Did you mean 'step_a'?

​
Type Safety

Routing functions must be synchronous and non-generator: 
# This raises TypeError at decoration time:
@route(targets=["a", "b"])
async def async_decide(x: int) -> str:  # Can't be async
    return "a"

# Error: Routing function 'async_decide' cannot be async.
# Routing decisions should be fast and based on already-computed values.
Move complex async logic to a regular @node before the gate. Keep routing functions fast and deterministic.

​
Patterns and Best Practices

​
Keep Routing Functions Simple

Routing functions should be fast and deterministic. Move complex logic to regular nodes: 
# Good: Routing based on pre-computed value
@node(output_name="doc_type")
def classify(document: str) -> str:
    """Heavy classification logic here."""
    return classifier.predict(document)

@route(targets=["pdf_processor", "image_processor", "text_processor"])
def route_by_type(doc_type: str) -> str:
    """Simple routing based on classification result."""
    return f"{doc_type}_processor"

​
Use Descriptive Targets

You can provide descriptions for visualization and documentation: 
@route(targets={
    "text_only_response": "Use text-only LLM for documents without visual content",
    "multimodal_response": "Use vision LLM for documents containing diagrams",
    END: "Terminate if no relevant documents found",
})
def route_by_content(doc_analysis: dict) -> str:
    ...

​
What’s Next?

Build docs developers (and LLMs) love

Get started for freeTalk to us