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:
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:
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:
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:
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:
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:
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."
)
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
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
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
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
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."
)
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:
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:
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:
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:
async def run_agent(
goal: str,
) -> AgentState:
state = AgentState(
goal=goal
)
while True:
observation = (
"Current workflow state"
)
state.observations.append(
observation
)
task = choose_next_task(
state
)
if task is None:
break
state.current_task = task
result = await execute_agent_task(
task
)
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:
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:
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:
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"
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