# Workflow Control Primitives — Control-flow Design
## Problem
Code generated by the current Planner must implement validation, retries, and routing manually:
```python
# 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
```python
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
```python
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
```python
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
```python
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)
```python
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**:
```python
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**:
```python
# 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`.
4. **`docs/capabilities/agentic-workflow.md` and `.zh.md`**
- Document control-flow primitives.
- Update generated-code examples.
5. **`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`?
```python
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?
```python
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:
```json
[
{"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.
```python
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
| Mechanism | Relationship |
|---|---|
| Three primitive layers (llm/agent/goal) | Control-flow primitives build on them, calling llm/agent internally |
| Checkpoints | Transparent to control-flow primitives: internal llm/agent calls are recorded normally |
| Planner revision | Unchanged: on failure, the Planner examines the error, revises the code, and reruns |
| `goal()` | `repeat_until` is a simplified goal with a loop but no complex judgment |
## 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.