Principle:Microsoft Agent framework Edge Condition Pattern

Overview

The Edge Condition Pattern defines how users implement branching logic in workflow graphs by attaching user-defined condition functions to edges between executors. Each edge in a workflow can carry an optional condition -- a callable that receives the output of the source executor and returns a boolean value indicating whether the edge should be traversed. This mechanism provides fine-grained, programmatic control over workflow routing without requiring modifications to the framework itself.

This is a user-defined pattern. The framework provides the workflow execution infrastructure and the ability to associate conditions with edges; the user supplies the condition logic that determines which paths are taken at runtime.

Problem Statement

In multi-step agent workflows, the output of one executor often determines which downstream executor should run next. Without a structured mechanism for conditional routing, users must either embed branching logic inside executors themselves or build ad-hoc post-processing layers. The Edge Condition Pattern establishes a consistent, declarative interface for attaching routing predicates to edges, keeping branching concerns separate from executor logic.

Pattern Description

The pattern consists of three elements:

Source executor completion -- An executor in the workflow finishes and produces output data (typically an AgentExecutorResponse or another result type).
Condition evaluation -- For each outgoing edge from the completed executor, the workflow engine invokes the edge's condition function, passing it the source executor's output.
Edge traversal decision -- If the condition function returns True, the edge is traversed and the target executor is scheduled for execution. If it returns False, the edge is skipped.

Condition Function Interface

Edge condition functions must conform to the following callable signature:

# Synchronous condition
def condition(data: Any) -> bool:
    ...

# Asynchronous condition
async def condition(data: Any) -> bool:
    ...

The formal type for the condition callable is:

Callable[[Any], bool | Awaitable[bool]]

The data parameter receives the message data produced by the source executor. The return value must be a boolean: True to traverse the edge, False to skip it.

Evaluation Semantics

Scenario	Behavior
Condition returns `True`	The edge is traversed; the target executor is scheduled.
Condition returns `False`	The edge is skipped; the target executor is not scheduled via this edge.
No condition attached to edge	The edge is traversed unconditionally (default behavior).
Multiple outgoing edges with conditions	Each edge's condition is evaluated independently; multiple edges may be traversed if multiple conditions return `True`.

Design Principles

Separation of Routing and Execution

By placing routing logic in edge conditions rather than inside executors, the pattern maintains a clean separation of concerns. Executors focus on their task; conditions focus on deciding what happens next. This makes both components independently testable and reusable.

User Ownership of Condition Logic

The framework deliberately does not prescribe what conditions should evaluate. Users retain full control over:

Value inspection -- Checking specific fields of the executor's structured output.
Threshold evaluation -- Comparing numeric results against configurable thresholds.
External lookups -- Querying databases, APIs, or caches to influence routing decisions (via async conditions).
Complex predicates -- Combining multiple checks using standard Python boolean logic.

Composability

Edge condition functions are ordinary Python callables. They can be composed using higher-order functions, wrapped with decorators, or parameterized with closures. A library of reusable condition predicates can be built and shared across workflows.

Usage Guidelines

Keep conditions pure when possible -- Prefer conditions that inspect their input without side effects.
Use async conditions sparingly -- Only use asynchronous conditions when external I/O is genuinely required for the routing decision.
Return explicit booleans -- Always return True or False; avoid relying on truthy/falsy coercion to prevent subtle routing bugs.
Handle missing fields defensively -- If the source executor's output structure may vary, guard against AttributeError or KeyError within the condition function.
Document expected input types -- Annotate the data parameter with the specific type produced by the source executor to improve readability and type safety.

Example

A basic edge condition that routes based on an approval flag in the executor's response:

from agent_framework.workflows import AgentExecutorResponse

def is_approved(data: AgentExecutorResponse) -> bool:
    return data.agent_response.value.approved == True

# Attach to an edge in a workflow definition
workflow.add_edge(source="reviewer", target="publisher", condition=is_approved)

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment