Workflow Control Primitives — Control-flow Design#

Problem#

Code generated by the current Planner must implement validation, retries, and routing manually:

# Manual version: many details and opportunities for mistakes
files = agent("Find files")
if len(files.split('\n')) < 3:
    files = agent("Expand the scope")
analysis = agent("Analyze: " + files)
if "missing" in analysis:
    extra = agent("Add missing information")
    analysis = analysis + extra

Problems:

• The Planner (LLM) struggles to write correct conditions (len(...) < 3).
• Validation and retry logic is repeated for every step.
• Intent is unclear: readers must inspect the code to identify validation and retries.

Approach: Provide Control-flow Primitives#

Give the Planner declarative control-flow helper functions to compose instead of implementing the logic itself.

New Primitives#

1. validate_and_retry — Validation and Retry#

files = validate_and_retry(
    action=lambda: agent("Find auth-related files"),
    check="File count >= 3",  # Condition description evaluated by the LLM
    retry=lambda: agent("Expand the scope to include oauth and openid files"),
    max_retries=2
)

Implementation:

• Execute action().
• Use llm() to determine whether the result satisfies check (return YES/NO).
• If not, execute retry(), repeating up to max_retries times.
• If all attempts fail, return the last result.

2. route — Route Selection#

strategy = route(
    question="Which migration strategy suits this task?",
    options=["Direct migration", "Refactor then migrate", "Phased migration"],
    context=analysis  # Optional context supplied to the LLM
)
# strategy = "Direct migration" (the option selected by the LLM)

Implementation:

• Use llm() to select one of options.
• Return the selected string.

3. conditional — Conditional Branching#

result = conditional(
    condition="Is the analysis complex?",
    context=analysis,
    if_true=lambda: agent("Analyze each module in detail"),
    if_false=lambda: agent("Give a brief conclusion")
)

Implementation:

• Use llm() to evaluate condition (return YES/NO).
• YES → execute if_true().
• NO → execute if_false().

4. repeat_until — Repeat Until Satisfied#

complete_files = repeat_until(
    action=lambda: agent("Find the remaining files"),
    condition="All relevant files have been found",
    max_rounds=5
)

Implementation:

• Execute action().
• Use llm() to evaluate condition.
• If satisfied, return the result.
• Otherwise, repeat up to max_rounds times.

5. parallel — Parallel Execution (Optional Later Extension)#

results = parallel([
    lambda: agent("Analyze frontend code"),
    lambda: agent("Analyze backend code"),
    lambda: agent("Analyze the database"),
])
# results = [frontend analysis, backend analysis, database analysis]

Example Code Generated by the Planner#

Task: “Migrate the auth module to the new client.”

Generated code:

def workflow():
    # Step 1: Find files, with validation and retries
    files = validate_and_retry(
        action=lambda: agent("Find all relevant files under openprogram/auth/"),
        check="File count >= 5 and includes oauth-related files",
        retry=lambda: agent("Expand the scope to include openid and token files"),
        max_retries=2
    )
    
    # Step 2: Analyze
    analysis = agent("Analyze dependencies and interfaces of these files: " + files)
    
    # Step 3: Select a strategy
    strategy = route(
        question="Choose a migration strategy based on the analysis",
        options=["Direct migration", "Refactor then migrate", "Phased migration"],
        context=analysis
    )
    
    # Step 4: Execute the conditional branch
    plan = conditional(
        condition="Is the selected strategy complex (refactoring or phased migration)?",
        context=strategy,
        if_true=lambda: agent("Write a detailed multi-phase migration plan: " + analysis),
        if_false=lambda: agent("Write brief direct-migration steps: " + analysis)
    )
    
    return plan

Comparison with a manual version:

# Manual version: 38 lines with extensive condition logic
def workflow():
    files = agent("Find auth-related files")
    if len(files.split('\n')) < 5 or "oauth" not in files.lower():
        files = agent("Expand the scope to include oauth")
        if len(files.split('\n')) < 5:
            files = agent("Expand again to include openid and token")
    
    analysis = agent("Analyze: " + files)
    
    strategies = ["Direct migration", "Refactor then migrate", "Phased migration"]
    strategy_prompt = f"Choose a strategy ({', '.join(strategies)}): {analysis}"
    strategy = llm(strategy_prompt)
    
    if "Refactor" in strategy or "Phased" in strategy:
        plan = agent("Write a detailed plan: " + analysis)
    else:
        plan = agent("Write brief steps: " + analysis)
    
    return plan

# Control-flow primitive version: 20 lines with clear intent
def workflow():
    files = validate_and_retry(
        action=lambda: agent("Find all relevant files under openprogram/auth/"),
        check="File count >= 5 and includes oauth-related files",
        retry=lambda: agent("Expand the scope to include openid and token files"),
        max_retries=2
    )
    
    analysis = agent("Analyze dependencies and interfaces of these files: " + files)
    
    strategy = route(
        question="Choose a migration strategy based on the analysis",
        options=["Direct migration", "Refactor then migrate", "Phased migration"],
        context=analysis
    )
    
    plan = conditional(
        condition="Is the selected strategy complex (refactoring or phased migration)?",
        context=strategy,
        if_true=lambda: agent("Write a detailed multi-phase migration plan: " + analysis),
        if_false=lambda: agent("Write brief direct-migration steps: " + analysis)
    )
    
    return plan

File Structure Changes#

New Files#

openprogram/agentic_programming/control_flow/
├── __init__.py          # Export all control-flow primitives
├── validate.py          # validate_and_retry implementation
├── route.py             # route implementation
├── conditional.py       # conditional implementation
├── repeat.py            # repeat_until implementation
└── parallel.py          # parallel implementation (optional)

Modified Files#

1. openprogram/agentic_programming/__init__.py

   • Export control-flow primitives: from .control_flow import validate_and_retry, route, conditional, repeat_until.

2. openprogram/programs/workflow/__init__.py

   • Inject control-flow primitives through _build_namespace() alongside llm/agent/goal.
   • Apply checkpoint wrappers.
   • Explain primitives and usage constraints in existing Planner instructions; do not assume a separate planner.py.

3. docs/capabilities/agentic-workflow.md and .zh.md

   • Document control-flow primitives.
   • Update generated-code examples.

4. docs/reference/design/runtime/self-programmed-agentic-workflow.html

   • Add a “Control-flow primitives” section.
   • Update the table of injected execution-environment symbols.

New Tests#

tests/unit/
├── test_control_flow_validate.py    # validate_and_retry unit tests
├── test_control_flow_route.py       # route unit tests
├── test_control_flow_conditional.py # conditional unit tests
├── test_control_flow_repeat.py      # repeat_until unit tests
└── test_workflow_with_control_flow.py  # Integration: workflow uses control-flow primitives

Implementation Sequence#

Phase 1: Core Primitives (validate, route, conditional)#

1. Implement three core control-flow functions

   • validate_and_retry
   • route
   • conditional
   • Write unit tests for each.

2. Inject them into the workflow execution environment

   • Modify _build_namespace().
   • Apply checkpoint wrappers.
   • Write integration tests proving they can be called.

3. Update the Planner prompt

   • Add primitive documentation.
   • Include example code.
   • Explain when to use each primitive.

4. End-to-end tests

   • Real task: “Migrate the auth module.”
   • Verify that the Planner generates code using control-flow primitives.
   • Verify correct execution.

Phase 2: Loop Primitive (repeat_until)#

5. Implement repeat_until
   • Unit tests.
   • Update the Planner prompt.
   • End-to-end tests.

Phase 3 (Optional): Parallel Primitive#

6. Implement parallel
   • Requires a thread or process pool.
   • Handle concurrent checkpoint writes.
   • More complex; may remain deferred.

Design Decisions#

Why Use lambda?#

validate_and_retry(
    action=lambda: agent("..."),  # Why not call agent("...") directly?
    retry=lambda: agent("...")
)

Reason: deferred execution.

• With action=agent("..."), the call executes immediately, before validate_and_retry.
• lambda: agent("...") passes a function describing what to execute; validate_and_retry decides when to execute it.

Why Is the Condition a String Instead of a Function?#

check="File count >= 5"  # String
# Instead of:
check=lambda r: len(r.split('\n')) >= 5  # Function

Reason: the Planner (LLM) struggles to write correct Python expressions.

• len(r.split('\n')) >= 5 is prone to mistakes in variable names, method names, and boundary conditions.
• "File count >= 5" is a natural-language description that an LLM can generate easily.
• The framework uses llm() to turn the description into a judgment (YES/NO).

How Are Checkpoints Recorded?#

Every llm()/agent() call inside a control-flow primitive is recorded normally:

[
  {"call_index": 0, "fn": "agent", "prompt": "Find files", "result": "auth.py"},
  {"call_index": 1, "fn": "llm", "prompt": "Evaluate: file count >= 3?", "result": "NO"},
  {"call_index": 2, "fn": "agent", "prompt": "Expand the scope", "result": "auth.py\noauth.py\nlogin.py"},
  {"call_index": 3, "fn": "llm", "prompt": "Evaluate: file count >= 3?", "result": "YES"}
]

validate_and_retry itself is not recorded; it only composes agent and llm calls.

What If the LLM Judgment Is Wrong?#

For example, llm("File count >= 5?") returns YES when only three files exist.

Current design: trust the LLM judgment.

Rationale:

• Exact judgments (len(...) >= 5) require the Planner to write correct Python code, which it cannot reliably do.
• LLM judgments are imprecise but usually correct (80–90% accuracy).
• An incorrect judgment may be detected by later steps, causing the entire workflow to be revised and rerun.

Future improvement: allow a hybrid mode.

check=lambda r: len(r.split('\n')) >= 5,  # Exact judgment, written by the user
# Or:
check="File count >= 5"  # LLM judgment, generated by the Planner

The framework inspects the parameter type: invoke callable directly, or evaluate str through llm().

Relationship to Existing Mechanisms#

Boundaries#

Out of scope:

1. Visual orchestration: this is not a drag-and-drop workflow editor; the Planner still generates code.
2. DAG execution engine: execution still uses exec() on Python code, not a data-driven scheduler.
3. Runtime step insertion: changes still require revision and rerun; code cannot change during execution.

Those capabilities require architectural changes. Control-flow primitives only help the Planner write better code; they do not change the execution model.

Summary#

Problem: manually generated validation, retry, and routing logic is verbose and error-prone.

Approach: provide control-flow primitives (validate_and_retry, route, conditional, repeat_until) for the Planner to compose.

Benefits:

• Shorter code with clearer intent.
• The Planner need not implement Python details such as conditions and string manipulation.
• Consistent patterns simplify debugging and maintenance.

Implementation:

• Add openprogram/agentic_programming/control_flow/.
• Inject primitives into the workflow execution environment.
• Update the Planner prompt.
• Keep the execution model unchanged: Python code still runs through exec.