# ACAI — Chapter 7: Workflow Orchestration and Agent Execution > Source: > Published: 2026-08-30 16:35:21+00:00 ACAI can now: ``` Chapter 1 → Core API Chapter 2 → Planner Chapter 3 → Retrieval Chapter 4 → Memory Chapter 5 → Model Router Chapter 6 → Verification ``` The next limitation is that a complex request may require **multiple dependent operations**. For example: ``` "Research a topic, summarize the evidence, compare the findings, and produce a report." ``` This is not one simple task. It can be represented as: ``` Research ↓ Collect Evidence ↓ Analyze ↓ Compare ↓ Write Report ↓ Verify ``` Chapter 7 introduces a workflow engine that represents these operations as a **task graph**. The previous system was approximately: ``` User ↓ Planner ↓ Router ↓ Model ↓ Verifier ↓ Response ``` The new system becomes: ``` User Goal ↓ Planner ↓ Workflow ↓ Task Graph ↓ Executor ↓ Verification ↓ Final Result ``` The key idea is: A complex AI task should be decomposed into smaller executable steps. A workflow can be represented as a directed graph. Example: ``` ┌───────────────┐ │ Research │ └───────┬───────┘ │ ┌───────▼───────┐ │ Extract Data │ └───────┬───────┘ │ ┌───────▼───────┐ │ Analyze │ └───────┬───────┘ │ ┌──────┴──────┐ ▼ ▼ ┌──────────┐ ┌──────────┐ │ Compare │ │ Validate │ └────┬─────┘ └────┬─────┘ │ │ └──────┬───────┘ ▼ ┌───────────┐ │ Report │ └─────┬─────┘ ▼ Verification ``` Some tasks depend on previous tasks. Others can execute independently. Create: ``` app/services/workflow.py ``` Start with: ``` python from dataclasses import dataclass, field @dataclass class Task: task_id: str name: str task_type: str dependencies: list[str] = field( default_factory=list ) status: str = "pending" result: str | None = None error: str | None = None ``` Each task contains: ``` task_id name task_type dependencies status result error ``` A task should have explicit states. ``` pending ↓ running ↓ completed ``` If something fails: ``` running ↓ failed ``` A retry can produce: ``` failed ↓ retrying ↓ running ``` The state machine is: ``` ┌──────────┐ │ pending │ └────┬─────┘ ▼ ┌──────────┐ │ running │ └────┬─────┘ ┌───┴───┐ ▼ ▼ ┌─────────┐ ┌────────┐ │complete │ │ failed │ └─────────┘ └───┬────┘ │ ▼ retry ``` Add: ``` python from dataclasses import dataclass, field @dataclass class Workflow: workflow_id: str tasks: dict[str, Task] = field( default_factory=dict ) status: str = "pending" result: str | None = None ``` Now ACAI can represent: ``` Workflow ├── Task A ├── Task B ├── Task C └── Task D ``` Add: ``` python class WorkflowBuilder: def __init__( self, workflow_id: str, ) -> None: self.workflow = Workflow( workflow_id=workflow_id ) def add_task( self, task_id: str, name: str, task_type: str, dependencies: list[str] | None = None, ) -> None: if task_id in self.workflow.tasks: raise ValueError( f"Task already exists: " f"{task_id}" ) self.workflow.tasks[task_id] = Task( task_id=task_id, name=name, task_type=task_type, dependencies=( dependencies or [] ), ) def build(self) -> Workflow: return self.workflow ``` Create: ``` python from uuid import uuid4 builder = WorkflowBuilder( workflow_id=str(uuid4()) ) builder.add_task( task_id="research", name="Research topic", task_type="research", ) builder.add_task( task_id="analysis", name="Analyze evidence", task_type="analysis", dependencies=[ "research" ], ) builder.add_task( task_id="report", name="Write report", task_type="writing", dependencies=[ "analysis" ], ) workflow = builder.build() ``` The dependency graph is: ``` research ↓ analysis ↓ report ``` A workflow should reject invalid dependencies. Add: ``` php def validate_workflow( workflow: Workflow, ) -> None: task_ids = set( workflow.tasks.keys() ) for task in workflow.tasks.values(): for dependency in task.dependencies: if dependency not in task_ids: raise ValueError( f"Unknown dependency " f"{dependency} for task " f"{task.task_id}" ) ``` This prevents: ``` Task A ↓ Missing Task X ``` from reaching execution. A more dangerous problem is: ``` Task A ↓ Task B ↓ Task A ``` This creates a cycle. Add: ``` php def detect_cycle( workflow: Workflow, ) -> bool: visiting = set() visited = set() def visit( task_id: str, ) -> bool: if task_id in visiting: return True if task_id in visited: return False visiting.add(task_id) task = workflow.tasks[task_id] for dependency in task.dependencies: if visit(dependency): return True visiting.remove(task_id) visited.add(task_id) return False for task_id in workflow.tasks: if visit(task_id): return True return False ``` Then: ``` php def validate_workflow( workflow: Workflow, ) -> None: task_ids = set( workflow.tasks.keys() ) for task in workflow.tasks.values(): for dependency in task.dependencies: if dependency not in task_ids: raise ValueError( f"Unknown dependency " f"{dependency}" ) if detect_cycle(workflow): raise ValueError( "Workflow contains a cycle." ) ``` The executor needs to determine which tasks can run. A task is ready when: ``` status = pending ``` and every dependency is: ``` completed ``` Add: ``` php def get_ready_tasks( workflow: Workflow, ) -> list[Task]: ready = [] for task in workflow.tasks.values(): if task.status != "pending": continue dependencies_completed = all( workflow.tasks[ dependency ].status == "completed" for dependency in task.dependencies ) if dependencies_completed: ready.append(task) return ready ``` Create: ``` python class WorkflowExecutor: async def execute( self, workflow: Workflow, ) -> Workflow: validate_workflow(workflow) workflow.status = "running" while True: ready_tasks = get_ready_tasks( workflow ) if not ready_tasks: unfinished = [ task for task in workflow.tasks.values() if task.status not in { "completed", "failed", } ] if unfinished: raise RuntimeError( "Workflow cannot make " "further progress." ) break for task in ready_tasks: await self.execute_task( task, workflow, ) workflow.status = "completed" return workflow ``` Add: ``` python async def execute_task( self, task: Task, workflow: Workflow, ) -> None: task.status = "running" try: result = await self.run_task( task, workflow, ) task.result = result task.status = "completed" except Exception as exc: task.error = str(exc) task.status = "failed" workflow.status = "failed" raise ``` For the first prototype: ``` python async def run_task( self, task: Task, workflow: Workflow, ) -> str: if task.task_type == "research": return ( "Research task completed." ) if task.task_type == "analysis": return ( "Analysis task completed." ) if task.task_type == "writing": return ( "Writing task completed." ) return ( f"Task {task.name} completed." ) ``` This is intentionally a mock implementation. Later it will call: ``` Research → Retrieval Service Analysis → Model Router + Model Writing → Model Router + Model Verification → Verification Service ``` The prototype can therefore be: ``` python class WorkflowExecutor: async def execute( self, workflow: Workflow, ) -> Workflow: validate_workflow(workflow) workflow.status = "running" while True: ready_tasks = get_ready_tasks( workflow ) if not ready_tasks: unfinished = [ task for task in workflow.tasks.values() if task.status not in { "completed", "failed", } ] if unfinished: raise RuntimeError( "Workflow cannot make " "further progress." ) break for task in ready_tasks: await self.execute_task( task, workflow, ) workflow.status = "completed" return workflow async def execute_task( self, task: Task, workflow: Workflow, ) -> None: task.status = "running" try: result = await self.run_task( task, workflow, ) task.result = result task.status = "completed" except Exception as exc: task.error = str(exc) task.status = "failed" workflow.status = "failed" raise async def run_task( self, task: Task, workflow: Workflow, ) -> str: if task.task_type == "research": return ( "Research task completed." ) if task.task_type == "analysis": return ( "Analysis task completed." ) if task.task_type == "writing": return ( "Writing task completed." ) return ( f"Task {task.name} completed." ) ``` For: ``` Research ↓ Analysis ↓ Report ``` execution becomes: ``` Research ↓ COMPLETED ↓ Analysis ↓ COMPLETED ↓ Report ↓ COMPLETED ``` Consider: ``` Research / \ ▼ ▼ Source A Source B │ │ └────┬─────┘ ▼ Analysis ``` Source A and Source B do not depend on each other. They can therefore run in parallel. The architecture becomes: ``` Research │ ┌──────┴──────┐ ▼ ▼ Source A Source B │ │ └──────┬──────┘ ▼ Analysis ``` Python's `asyncio` can execute independent asynchronous tasks concurrently. Add: ``` python import asyncio ``` Then replace the sequential loop: ``` for task in ready_tasks: await self.execute_task( task, workflow, ) ``` with: ``` await asyncio.gather( *[ self.execute_task( task, workflow, ) for task in ready_tasks ] ) ``` Now independent tasks can execute concurrently. Suppose: ``` Task A = 5 seconds Task B = 5 seconds ``` Sequential execution can take approximately: ``` 5 + 5 = 10 seconds ``` If they are independent and safely executed concurrently, idealized execution can approach: ``` max(5, 5) = 5 seconds ``` Real systems have overhead, rate limits, network latency, and resource constraints, so actual performance must be measured. Real workflows fail. Possible causes: ``` Network error Provider timeout Temporary API failure Rate limit Invalid response Dependency failure ``` A task should therefore support bounded retries. Add: ``` @dataclass class Task: task_id: str name: str task_type: str dependencies: list[str] = field( default_factory=list ) status: str = "pending" result: str | None = None error: str | None = None attempts: int = 0 max_attempts: int = 3 php async def execute_task( self, task: Task, workflow: Workflow, ) -> None: while task.attempts < task.max_attempts: task.attempts += 1 task.status = "running" try: result = await self.run_task( task, workflow, ) task.result = result task.status = "completed" return except Exception as exc: task.error = str(exc) if ( task.attempts >= task.max_attempts ): task.status = "failed" raise task.status = "retrying" ``` This gives: ``` Attempt 1 ↓ Fail ↓ Attempt 2 ↓ Fail ↓ Attempt 3 ↓ Success / Failure ``` Retries should not be automatic for every error. For example: ``` Invalid input ``` may not become valid by repeating the same request. But: ``` Temporary network failure ``` might succeed on retry. Therefore future versions should classify errors: ``` Transient Permanent Unknown ``` Then retry only appropriate failures. A task that never completes can block an entire workflow. Use: ``` python import asyncio ``` and: ``` result = await asyncio.wait_for( self.run_task( task, workflow, ), timeout=60, ) ``` This creates a maximum execution window. Suppose: ``` Task A → completed Task B → failed Task C → depends on B ``` Task C cannot safely execute. Therefore: ``` A → COMPLETE B → FAILED C → BLOCKED ``` The system should distinguish: ``` failed ``` from: ``` blocked ``` Add: ``` pending running retrying completed failed blocked ``` The workflow can produce a final result from completed tasks. Example: ``` php def collect_results( workflow: Workflow, ) -> dict[str, str]: return { task.task_id: task.result for task in workflow.tasks.values() if task.result is not None } ``` Then: ``` Workflow ↓ Task Results ↓ Result Aggregation ↓ Final Answer ``` Workflow execution should not end immediately after generation. A final verification task should be added. Example: ``` Research ↓ Analysis ↓ Draft ↓ Verification ↓ Final ``` The verification task can inspect: ``` Draft + Evidence + Original User Goal ``` Then return: ``` PASS ``` or: ``` REVISION_REQUIRED ``` Architecture: ``` USER GOAL │ ▼ PLANNER │ ▼ TASK GRAPH │ ┌──────────┼──────────┐ ▼ ▼ ▼ Task A Task B Task C │ │ │ └──────────┼──────────┘ ▼ Draft │ ▼ Verification │ ┌────┴────┐ ▼ ▼ PASS REVISE │ │ ▼ ▼ Final Retry ``` At this point ACAI begins to resemble an agentic workflow system. But an important distinction should be maintained: ``` Agent ≠ Uncontrolled autonomous process ``` A practical agent should have: ``` Goal + Tools + State + Constraints + Termination Conditions ``` Create: ``` @dataclass class AgentState: goal: str current_task: str | None = None completed_tasks: list[str] = field( default_factory=list ) failed_tasks: list[str] = field( default_factory=list ) observations: list[str] = field( default_factory=list ) ``` The agent can now maintain execution state. The conceptual loop is: ``` Goal ↓ Observe ↓ Plan ↓ Act ↓ Observe Result ↓ Verify ↓ Continue / Stop ``` Implementation: ``` php async def run_agent( goal: str, ) -> AgentState: state = AgentState( goal=goal ) while True: # Observe observation = ( "Current workflow state" ) state.observations.append( observation ) # Plan task = choose_next_task( state ) if task is None: break # Act state.current_task = task result = await execute_agent_task( task ) # Record state.completed_tasks.append( task ) return state ``` This is a simplified demonstration. An agent must have clear stopping conditions. For example: ``` Goal achieved OR Maximum steps reached OR Maximum time reached OR No valid action available OR Critical failure ``` Without termination conditions: ``` Agent ↓ Action ↓ Action ↓ Action ↓ ... ``` could continue indefinitely. Add: ``` MAX_AGENT_STEPS = 10 ``` Then: ``` for step in range( MAX_AGENT_STEPS ): ... ``` This gives the system a hard upper bound. A future ACAI agent can use controlled tools: ``` Retrieval File Search Calculator Code Executor Database External API Model ``` The architecture should be: ``` Agent │ ▼ Tool Selection │ ▼ Permission Check │ ▼ Tool Execution │ ▼ Result Validation ``` The permission layer is important. An agent should not automatically receive unrestricted access to arbitrary systems. Create: ``` php class ToolRegistry: def __init__(self) -> None: self.tools = {} def register( self, name: str, function, ) -> None: self.tools[name] = function def get( self, name: str, ): return self.tools.get(name) def list_tools(self) -> list[str]: return list( self.tools.keys() ) ``` Example: ``` registry = ToolRegistry() registry.register( "retrieval", retrieval_service, ) ``` Now the agent can discover available tools through a controlled registry. Before tool execution: ``` Agent ↓ Requested Tool ↓ Permission Policy ↓ Allowed? ┌──┴──┐ YES NO │ │ ▼ ▼ Run Reject ``` Example: ``` python class ToolPolicy: def __init__( self, allowed_tools: set[str], ) -> None: self.allowed_tools = ( allowed_tools ) def allowed( self, tool_name: str, ) -> bool: return ( tool_name in self.allowed_tools ) ``` This makes tool access explicit. Workflow execution needs detailed logs. For each task: ``` workflow_id task_id task_type start_time end_time duration status attempt error ``` Example: ``` event = { "workflow_id": workflow.workflow_id, "task_id": task.task_id, "status": task.status, "attempt": task.attempts, } ``` These events can later be sent to a logging system. Create: ``` tests/test_workflow.py ``` Add: ``` python import pytest from app.services.workflow import ( Task, Workflow, WorkflowBuilder, get_ready_tasks, validate_workflow, ) ``` Test task dependencies: ``` python def test_ready_tasks(): workflow = Workflow( workflow_id="test" ) workflow.tasks["a"] = Task( task_id="a", name="A", task_type="general", ) workflow.tasks["b"] = Task( task_id="b", name="B", task_type="general", dependencies=["a"], ) ready = get_ready_tasks( workflow ) assert len(ready) == 1 assert ready[0].task_id == "a" python def test_invalid_dependency(): workflow = Workflow( workflow_id="test" ) workflow.tasks["a"] = Task( task_id="a", name="A", task_type="general", dependencies=["missing"], ) with pytest.raises(ValueError): validate_workflow( workflow ) python def test_cycle_detection(): workflow = Workflow( workflow_id="test" ) workflow.tasks["a"] = Task( task_id="a", name="A", task_type="general", dependencies=["b"], ) workflow.tasks["b"] = Task( task_id="b", name="B", task_type="general", dependencies=["a"], ) with pytest.raises(ValueError): validate_workflow( workflow ) python import pytest from app.services.workflow import ( WorkflowBuilder, WorkflowExecutor, ) @pytest.mark.asyncio async def test_workflow_execution(): builder = WorkflowBuilder( workflow_id="demo" ) builder.add_task( task_id="research", name="Research", task_type="research", ) builder.add_task( task_id="analysis", name="Analysis", task_type="analysis", dependencies=[ "research" ], ) workflow = builder.build() executor = WorkflowExecutor() result = await executor.execute( workflow ) assert result.status == "completed" assert ( result.tasks["research"] .status == "completed" ) assert ( result.tasks["analysis"] .status == "completed" ) ``` Run: ``` pytest ``` You should now have coverage for: ``` API Planner Retrieval Memory Router Verification Workflow USER │ ▼ FastAPI API │ ▼ ORCHESTRATOR │ ▼ PLANNER │ ▼ WORKFLOW GRAPH │ ┌────────────────┼────────────────┐ ▼ ▼ ▼ TASK A TASK B TASK C │ │ │ └────────────────┼────────────────┘ │ ▼ MODEL ROUTER │ ▼ MODEL SERVICE │ ▼ GENERATION │ ▼ VERIFICATION │ ┌──────┴──────┐ ▼ ▼ PASS REVISE │ │ ▼ ▼ RESULT RETRY ``` After Chapter 7, the architecture can conceptually: ``` [✓] Receive a request [✓] Analyze the task [✓] Retrieve information [✓] Use memory [✓] Select a model [✓] Create multiple tasks [✓] Handle dependencies [✓] Execute independent tasks concurrently [✓] Retry bounded failures [✓] Apply timeouts [✓] Verify outputs [✓] Produce a final result ``` This is substantially more capable than a simple: ``` Prompt → Model → Answer ``` pipeline. The architecture should **not** yet be described as: ``` AGI Human-level intelligence Fully autonomous intelligence Guaranteed factual AI Self-improving superintelligence ``` Those claims would require evidence far beyond the architecture described here. A technically defensible description is: ACAI is a modular AI orchestration architecture that combines planning, retrieval, memory, model routing, workflow execution, and output verification. The architecture now has execution capabilities. The next major requirement is **persistent data and production infrastructure**. Currently: ``` Memory → In-memory Python objects Workflow → Runtime objects Logs → Basic application logging ``` These disappear when the process stops unless persistent storage is added. Therefore the next chapter will introduce: The architecture will move toward: ``` ACAI │ ┌──────────┼──────────┐ ▼ ▼ ▼ Compute Storage Observability │ │ │ ▼ ▼ ▼ Workers Database Metrics │ ┌─────┴─────┐ ▼ ▼ Memory Workflows ``` The next stage will cover: ``` Database schema Persistent memory Workflow persistence Request IDs Error handling Rate limiting Caching Background jobs Health checks Production configuration ``` **End of Chapter 7**