ACAI — Chapter 7: Workflow Orchestration and Agent Execution

A developer introduced Chapter 7 of the ACAI project, adding a workflow orchestration engine that decomposes complex AI requests into a task graph. The new system replaces the linear planner-router-model-verifier pipeline with a workflow builder that manages task dependencies, states, and retries, enabling multi-step operations like research, analysis, and report generation.

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