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.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
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 network failure
might succeed on retry.
Therefore future versions should classify errors:
Transient
Permanent
Unknown
Then retry only appropriate failures.
7.23 Timeout Protection
A task that never completes can block an entire workflow.
Use:
import asyncio
and:
result = await asyncio.wait_for(
self.run_task(
task,
workflow,
),
timeout=60,
)
This creates a maximum execution window.
7.24 Workflow Failure Handling
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
7.25 Workflow Result
The workflow can produce a final result from completed tasks.
Example:
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
7.26 Integrating Verification
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
7.27 Workflow-Level Verification
Architecture:
USER GOAL
│
▼
PLANNER
│
▼
TASK GRAPH
│
┌──────────┼──────────┐
▼ ▼ ▼
Task A Task B Task C
│ │ │
└──────────┼──────────┘
▼
Draft
│
▼
Verification
│
┌────┴────┐
▼ ▼
PASS REVISE
│ │
▼ ▼
Final Retry
7.28 Agent Execution
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
7.29 Agent State
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.
7.30 Agent Loop
The conceptual loop is:
Goal
↓
Observe
↓
Plan
↓
Act
↓
Observe Result
↓
Verify
↓
Continue / Stop
Implementation:
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.
7.31 Termination Conditions
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.
7.32 Maximum Steps
Add:
MAX_AGENT_STEPS = 10
Then:
for step in range(
MAX_AGENT_STEPS
):
...
This gives the system a hard upper bound.
7.33 Tool Execution
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.
7.34 Tool Registry
Create:
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.
7.35 Permission Layer
Before tool execution:
Agent
↓
Requested Tool
↓
Permission Policy
↓
Allowed?
┌──┴──┐
YES NO
│ │
▼ ▼
Run Reject
Example:
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.
7.36 Observability
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.
7.37 Workflow Tests
Create:
tests/test_workflow.py
Add:
import pytest
from app.services.workflow import (
Task,
Workflow,
WorkflowBuilder,
get_ready_tasks,
validate_workflow,
)
Test task dependencies:
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"
7.38 Dependency Validation Test
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
)
7.39 Cycle Detection Test
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
)
7.40 End-to-End Workflow Test
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"
)
7.41 Run Tests
Run:
pytest
You should now have coverage for:
API
Planner
Retrieval
Memory
Router
Verification
Workflow
7.42 Full ACAI Architecture After Chapter 7
USER
│
▼
FastAPI API
│
▼
ORCHESTRATOR
│
▼
PLANNER
│
▼
WORKFLOW GRAPH
│
┌────────────────┼────────────────┐
▼ ▼ ▼
TASK A TASK B TASK C
│ │ │
└────────────────┼────────────────┘
│
▼
MODEL ROUTER
│
▼
MODEL SERVICE
│
▼
GENERATION
│
▼
VERIFICATION
│
┌──────┴──────┐
▼ ▼
PASS REVISE
│ │
▼ ▼
RESULT RETRY
7.43 What ACAI Can Do Now
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.
7.44 What It Still Cannot Claim
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.
7.45 Next Chapter
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:
Chapter 8 — Persistent Storage, API Reliability, and Production Infrastructure
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
Top comments (0)