ACAI - Chapter 2: Intelligent Planning Layer

 

2.1 Objective

Chapter 1 created the basic ACAI request pipeline:

User
 ↓
API
 ↓
Orchestrator
 ↓
Model Service
 ↓
Response

Chapter 2 adds the Planner.

The new architecture becomes:

User
 ↓
API
 ↓
Orchestrator
 ↓
Planner
 ↓
Model Service
 ↓
Response

The Planner's job is to examine a request and determine whether it is simple or requires multiple steps.

The goal is not to make every request complicated. A good planner should recognize when planning is unnecessary.

---

2.2 Planning Concept

A simple request:

"What is 2 + 2?"

may use:

User
 ↓
Planner
 ↓
Direct Answer

A complex request:

"Compare three research papers and explain their differences."

could produce:

User
 ↓
Planner
 ↓
1. Identify documents
2. Extract relevant information
3. Compare methodologies
4. Compare conclusions
5. Identify disagreements
6. Generate synthesis

This creates the foundation for later Retrieval, Memory, Tools, and Verification components.

---

2.3 Updated Architecture

                    USER
                      │
                      ▼
                 FastAPI API
                      │
                      ▼
               ACAI Orchestrator
                      │
                      ▼
                  ┌────────┐
                  │ Planner│
                  └───┬────┘
                      │
             ┌────────┴────────┐
             ▼                 ▼
        Simple Task       Complex Task
             │                 │
             └────────┬────────┘
                      ▼
                Model Service
                      │
                      ▼
                  Response

---

2.4 New Project Structure

Add a planner service:

ACAI/
└── backend/
    ├── app/
    │   ├── __init__.py
    │   ├── main.py
    │   ├── config.py
    │   ├── schemas.py
    │   ├── orchestrator.py
    │   │
    │   └── services/
    │       ├── __init__.py
    │       ├── model_service.py
    │       └── planner.py
    │
    └── tests/
        ├── test_api.py
        └── test_planner.py

---

2.5 Planner Data Model

Create app/services/planner.py.

from dataclasses import dataclass


@dataclass
class Plan:
    task_type: str
    complexity: str
    steps: list[str]

This represents the result of planning.

For example:

task_type = research
complexity = complex
steps =
    1. Identify sources
    2. Extract evidence
    3. Compare findings
    4. Generate synthesis

---

2.6 Planner Implementation

Add the following to app/services/planner.py:

from dataclasses import dataclass


@dataclass
class Plan:
    task_type: str
    complexity: str
    steps: list[str]


class Planner:

    def create_plan(self, message: str) -> Plan:
        text = message.strip().lower()

        if not text:
            raise ValueError("Cannot create a plan for an empty message.")

        task_type = self._detect_task_type(text)
        complexity = self._estimate_complexity(text)
        steps = self._generate_steps(
            task_type=task_type,
            complexity=complexity,
        )

        return Plan(
            task_type=task_type,
            complexity=complexity,
            steps=steps,
        )

    def _detect_task_type(self, text: str) -> str:

        research_keywords = [
            "research",
            "paper",
            "study",
            "compare",
            "analyze",
            "analysis",
            "literature",
        ]

        coding_keywords = [
            "code",
            "program",
            "python",
            "javascript",
            "debug",
            "function",
            "api",
        ]

        writing_keywords = [
            "write",
            "article",
            "essay",
            "report",
            "email",
            "document",
        ]

        if any(word in text for word in research_keywords):
            return "research"

        if any(word in text for word in coding_keywords):
            return "coding"

        if any(word in text for word in writing_keywords):
            return "writing"

        return "general"

    def _estimate_complexity(self, text: str) -> str:

        word_count = len(text.split())

        complex_indicators = [
            "step by step",
            "compare",
            "analyze",
            "research",
            "design",
            "build",
            "implement",
            "explain in detail",
        ]

        indicator_count = sum(
            indicator in text
            for indicator in complex_indicators
        )

        if word_count > 40 or indicator_count >= 2:
            return "complex"

        if word_count > 15 or indicator_count == 1:
            return "moderate"

        return "simple"

    def _generate_steps(
        self,
        task_type: str,
        complexity: str,
    ) -> list[str]:

        if complexity == "simple":
            return [
                "Understand the request",
                "Generate the response",
            ]

        if task_type == "research":
            return [
                "Understand the research question",
                "Identify relevant information",
                "Analyze the evidence",
                "Compare important findings",
                "Generate a structured response",
            ]

        if task_type == "coding":
            return [
                "Understand the technical requirement",
                "Identify the implementation approach",
                "Design the solution",
                "Generate the implementation",
                "Check the result",
            ]

        if task_type == "writing":
            return [
                "Understand the requested format",
                "Identify the main points",
                "Create an appropriate structure",
                "Generate the content",
                "Review the final output",
            ]

        return [
            "Understand the request",
            "Break the problem into logical steps",
            "Generate the response",
            "Review the result",
        ]


planner = Planner()

---

2.7 Integrating the Planner

Now update app/orchestrator.py.

Replace the previous implementation with:

from app.services.model_service import model_service
from app.services.planner import planner


class ACAIOrchestrator:

    async def process(self, message: str) -> dict:
        cleaned_message = message.strip()

        if not cleaned_message:
            raise ValueError("Message cannot be empty.")

        plan = planner.create_plan(cleaned_message)

        response = await model_service.generate(
            cleaned_message
        )

        return {
            "response": response,
            "plan": {
                "task_type": plan.task_type,
                "complexity": plan.complexity,
                "steps": plan.steps,
            },
        }


orchestrator = ACAIOrchestrator()

---

2.8 Update the API Schema

Update app/schemas.py:

from pydantic import BaseModel, Field


class ChatRequest(BaseModel):
    message: str = Field(
        ...,
        min_length=1,
        max_length=10000,
        description="User message",
    )


class PlanResponse(BaseModel):
    task_type: str
    complexity: str
    steps: list[str]


class ChatResponse(BaseModel):
    success: bool
    response: str
    model: str
    mode: str
    plan: PlanResponse

---

2.9 Update main.py

Because the orchestrator now returns the plan, update the endpoint.

from fastapi import FastAPI, HTTPException

from app.config import settings
from app.orchestrator import orchestrator
from app.schemas import ChatRequest, ChatResponse


app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    description="Adaptive Cognitive AI Architecture",
)


@app.get("/")
async def root():
    return {
        "name": settings.app_name,
        "version": settings.app_version,
        "status": "online",
    }


@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "environment": settings.environment,
    }


@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):

    try:
        result = await orchestrator.process(
            request.message
        )

        return ChatResponse(
            success=True,
            response=result["response"],
            model=settings.model_name,
            mode=settings.model_provider,
            plan=result["plan"],
        )

    except ValueError as exc:
        raise HTTPException(
            status_code=400,
            detail=str(exc),
        )

    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=f"ACAI processing error: {exc}",
        )

---

2.10 Test the Planner Directly

Create tests/test_planner.py:

from app.services.planner import planner


def test_simple_plan():

    plan = planner.create_plan(
        "What is Python?"
    )

    assert plan.complexity == "simple"
    assert plan.task_type == "general"
    assert len(plan.steps) >= 2


def test_research_plan():

    plan = planner.create_plan(
        "Compare three research papers and analyze their conclusions."
    )

    assert plan.task_type == "research"
    assert plan.complexity == "complex"
    assert len(plan.steps) >= 4


def test_coding_plan():

    plan = planner.create_plan(
        "Build a Python API and debug the function."
    )

    assert plan.task_type == "coding"
    assert len(plan.steps) >= 4


def test_writing_plan():

    plan = planner.create_plan(
        "Write an article about artificial intelligence."
    )

    assert plan.task_type == "writing"

---

2.11 Update API Tests

Update the chat test in tests/test_api.py:

def test_chat():

    response = client.post(
        "/api/chat",
        json={
            "message": "Compare two research papers."
        },
    )

    assert response.status_code == 200

    data = response.json()

    assert data["success"] is True
    assert "response" in data
    assert "plan" in data

    assert data["plan"]["task_type"] == "research"
    assert len(data["plan"]["steps"]) >= 4

---

2.12 Run the Tests

Run:

pytest

You should now have the original API tests plus the planner tests.

The exact number of tests may change depending on which tests from Chapter 1 you kept, but all implemented tests should pass.

---

2.13 Test Through Swagger

Start the server:

uvicorn app.main:app --reload

Open:

http://127.0.0.1:8000/docs

Send:

{
  "message": "Compare three research papers and analyze their conclusions."
}

The response should contain a plan similar to:

{
  "success": true,
  "response": "ACAI Demo Model Response ...",
  "model": "acai-demo-model",
  "mode": "mock",
  "plan": {
    "task_type": "research",
    "complexity": "complex",
    "steps": [
      "Understand the research question",
      "Identify relevant information",
      "Analyze the evidence",
      "Compare important findings",
      "Generate a structured response"
    ]
  }
}

---

2.14 What Has Actually Changed?

Before Chapter 2:

User
 ↓
Model
 ↓
Response

After Chapter 2:

User
 ↓
Planner
 ↓
Task Classification
 ↓
Complexity Estimation
 ↓
Plan
 ↓
Model
 ↓
Response

This is the first significant architectural improvement.

---

2.15 Important Limitation

The current planner uses deterministic keyword-based logic.

For example:

"research"

can trigger the research classification.

This is intentionally simple.

It is not being presented as an advanced AI planner.

Later, the planner can be upgraded to use:

Rule-Based Planner
       ↓
LLM-Assisted Planner
       ↓
Structured Planning
       ↓
Plan Validation
       ↓
Adaptive Planning

Each version can be benchmarked against the previous one.

---

2.16 Planner Evaluation

The planner should eventually be evaluated using a dataset such as:

Task
Expected Type
Expected Complexity
Expected Steps

Example:

Input	Expected Type	Expected Complexity
What is Python?	General	Simple
Write a Python API	Coding	Moderate/Complex
Compare research papers	Research	Complex
Write a business article	Writing	Moderate
Explain a complex architecture	General	Complex

The benchmark can then calculate classification accuracy.

---

2.17 Chapter 2 Architecture

The complete Chapter 2 flow is:

                       USER
                         │
                         ▼
                    FastAPI API
                         │
                         ▼
                 ACAI Orchestrator
                         │
                         ▼
                      Planner
                         │
             ┌───────────┴───────────┐
             ▼                       ▼
        Task Type              Complexity
             │                       │
             └───────────┬───────────┘
                         ▼
                    Plan Object
                         │
                         ▼
                   Model Service
                         │
                         ▼
                      Output

---

2.18 Chapter 2 Success Criteria

Chapter 2 is complete when:

[✓] Planner service exists
[✓] Task type detection works
[✓] Complexity estimation works
[✓] Plans contain multiple steps
[✓] Orchestrator uses Planner
[✓] API returns the generated plan
[✓] Planner tests pass
[✓] API tests pass

---

2.19 Next Architectural Upgrade

The next major problem is:

Where does the system obtain reliable external information?

That leads to Chapter 3:

Chapter 1
Core
   ↓
Chapter 2
Planner
   ↓
Chapter 3
Retrieval / RAG
   ↓
Documents
   ↓
Relevant Evidence
   ↓
Model

The next chapter will implement a real Retrieval-Augmented Generation foundation, including document ingestion, chunking, embeddings/vector search architecture, retrieval, and integration with the ACAI Orchestrator.

End of Chapter 2