ACAI - Chapter 7: Workflow Orchestration and Agent Execution

ACAI -Workflow Orchestration and Agent Execution 7.1 Objective 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. 7.2 From Single Request to Workflow 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. 7.3 Workflow Graph 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. 7.4 Task Data Model Create: app/services/workflow.py Start with: 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 7.5 Task States 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 7.6 Workflow Container Add: 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 7.7 Adding Tasks Add: 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 7.8 Example Workflow Create: 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 7.9 Dependency Validation A workflow should reject invalid dependencies. Add: 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. 7.10 Circular Dependency Detection A more dangerous problem is: Task A ↓ Task B ↓ Task A This creates a cycle. Add: 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: 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." ) 7.11 Finding Ready Tasks The executor needs to determine which tasks can run. A task is ready when: status = pending and every dependency is: completed Add: 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 7.12 Workflow Executor Create: 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 7.13 Task Execution Add: 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 7.14 Task Runner For the first prototype: 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 7.15 Complete Workflow Executor The prototype can therefore be: 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." ) 7.16 Sequential Execution For: Research ↓ Analysis ↓ Report execution becomes: Research ↓ COMPLETED ↓ Analysis ↓ COMPLETED ↓ Report ↓ COMPLETED 7.17 Parallel Execution 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 7.18 Parallel Task Execution Python's asyncio can execute independent asynchronous tasks concurrently. Add: 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. 7.19 Why Parallelism Matters 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. 7.20 Retry Policy 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 7.21 Retry Implementation async def execute_task( self, task: Task, workflow: Workflow, ) -> None: while task.attempts = task.max_attempts ): task.status = "failed" raise task.status = "retrying" This gives: Attempt 1 ↓ Fail ↓ Attempt 2 ↓ Fail ↓ Attempt 3 ↓ Success / Failure 7.22 Retry Is Not Always Correct Retries should not be automatic for every error. For example: Invalid input may not become valid by repeating the same request. But: Temporary n