ACAI — Chapter 20: Data Engineering, Knowledge Systems, RAG, Vector Search, Memory, Data Pipelines, and Knowledge Quality

Image
  20.1 Objective An advanced AI system is only as useful as the information it can reliably access. ACAI therefore needs a complete knowledge architecture: DATA ↓ INGESTION ↓ PROCESSING ↓ STORAGE ↓ INDEXING ↓ RETRIEVAL ↓ RERANKING ↓ CONTEXT ↓ MODEL ↓ VERIFICATION ↓ ANSWER The purpose of this chapter is to explain how ACAI can turn raw information into searchable, trustworthy context. 20.2 Data Sources ACAI may receive information from many sources: Documents Web pages Databases APIs User uploads Internal knowledge Application records Structured datasets Images Audio Video Different sources require different processing pipelines. 20.3 Data Ingestion Ingestion means bringing information into the system. SOURCE ↓ INGESTION SERVICE ↓ RAW DATA ↓ PROCESSING PIPELINE The ingestion layer should record where the information came from. Example metadata: { "source_id": "source_001", "source_type": "document", "created_at": ...

ACAI — Chapter 8: Persistent Storage, API Reliability, and Production Infrastructure

 

Post cover


8.1 Objective

Up to Chapter 7, ACAI has evolved into a modular AI orchestration system:

Chapter 1 → Core API
Chapter 2 → Planner
Chapter 3 → Retrieval
Chapter 4 → Memory
Chapter 5 → Model Router
Chapter 6 → Verification
Chapter 7 → Workflow Engine

However, most of the current prototype state exists only while the application is running.

If the server stops:

Memory
   ↓
Lost

Workflow State
   ↓
Lost

Runtime Data
   ↓
Lost

A production system therefore needs persistent infrastructure.

Chapter 8 introduces:

Database
Persistent Memory
Workflow Persistence
Request IDs
Error Handling
Rate Limiting
Caching
Health Checks
Background Jobs
Production Configuration

8.2 Production Architecture

The architecture now becomes:

                           USER
                             │
                             ▼
                        API Gateway
                             │
                             ▼
                       FastAPI Server
                             │
                    ┌────────┴────────┐
                    ▼                 ▼
               Rate Limiter       Request ID
                    │                 │
                    └────────┬────────┘
                             ▼
                       ORCHESTRATOR
                             │
        ┌────────────────────┼────────────────────┐
        ▼                    ▼                    ▼
     Planner              Memory              Retrieval
        │                    │                    │
        └────────────────────┼────────────────────┘
                             ▼
                       Model Router
                             │
                             ▼
                       Model Service
                             │
                             ▼
                         Workflow
                             │
                             ▼
                       Verification
                             │
                             ▼
                          Response
                             │
             ┌───────────────┼───────────────┐
             ▼               ▼               ▼
          Database         Cache          Monitoring

8.3 Why Persistence Matters

Consider a user interaction:

User:
"I prefer Python."

ACAI stores it in memory.

Then the server restarts.

If the memory was only:

self.items = {}

the information disappears.

A persistent implementation changes this:

Memory
 ↓
Database
 ↓
Restart
 ↓
Database
 ↓
Memory restored

8.4 Database Layer

For the first production-oriented prototype, SQLite is a convenient starting point.

It requires no separate database server.

Create:

app/database.py

Add:

import sqlite3
from pathlib import Path


DATABASE_PATH = Path(
    "data/acai.db"
)


def get_connection():

    DATABASE_PATH.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    connection = sqlite3.connect(
        DATABASE_PATH
    )

    connection.row_factory = (
        sqlite3.Row
    )

    return connection

This creates:

data/
    acai.db

8.5 Database Initialization

Add:

def initialize_database():

    connection = get_connection()

    cursor = connection.cursor()

    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS memories (
            memory_id TEXT PRIMARY KEY,
            content TEXT NOT NULL,
            memory_type TEXT NOT NULL,
            importance REAL NOT NULL,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )

    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS workflows (
            workflow_id TEXT PRIMARY KEY,
            status TEXT NOT NULL,
            result TEXT,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )

    connection.commit()

    connection.close()

8.6 Application Startup

Update the FastAPI startup process.

For example:

from fastapi import FastAPI

from app.database import (
    initialize_database,
)


app = FastAPI(
    title="ACAI",
    version="0.8.0",
)


@app.on_event("startup")
def startup():

    initialize_database()

The database is initialized when the application starts.


8.7 Persistent Memory

The existing memory service can now write to SQLite.

Create:

app/services/persistent_memory.py

Add:

from datetime import datetime
from uuid import uuid4

from app.database import (
    get_connection,
)


class PersistentMemoryService:

    def remember(
        self,
        content: str,
        memory_type: str = "general",
        importance: float = 0.5,
    ) -> str:

        memory_id = str(uuid4())

        now = datetime.utcnow().isoformat()

        connection = get_connection()

        connection.execute(
            """
            INSERT INTO memories (
                memory_id,
                content,
                memory_type,
                importance,
                created_at,
                updated_at
            )
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                memory_id,
                content,
                memory_type,
                importance,
                now,
                now,
            ),
        )

        connection.commit()

        connection.close()

        return memory_id

8.8 Reading Persistent Memories

Add:

    def search(
        self,
        query: str,
        limit: int = 5,
    ):

        connection = get_connection()

        rows = connection.execute(
            """
            SELECT *
            FROM memories
            ORDER BY importance DESC
            LIMIT ?
            """,
            (limit,),
        ).fetchall()

        connection.close()

        return [
            dict(row)
            for row in rows
        ]

This is a simple baseline.

A production semantic-memory system would later add embeddings and vector search.


8.9 Memory Architecture Upgrade

The memory architecture is now:

                    MEMORY
                       │
              ┌────────┴────────┐
              ▼                 ▼
        Short-Term          Long-Term
          Memory              Memory
              │                 │
              │                 ▼
              │              Database
              │                 │
              └────────┬────────┘
                       ▼
                  Memory Context

8.10 Workflow Persistence

Long-running workflows also need persistence.

Add:

def save_workflow(
    workflow_id: str,
    status: str,
    result: str | None = None,
):

    from datetime import datetime

    now = datetime.utcnow().isoformat()

    connection = get_connection()

    connection.execute(
        """
        INSERT INTO workflows (
            workflow_id,
            status,
            result,
            created_at,
            updated_at
        )
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(workflow_id)
        DO UPDATE SET
            status = excluded.status,
            result = excluded.result,
            updated_at = excluded.updated_at
        """,
        (
            workflow_id,
            status,
            result,
            now,
            now,
        ),
    )

    connection.commit()

    connection.close()

Now a workflow can survive process restarts.


8.11 Workflow Lifecycle

The persistent lifecycle becomes:

Created
   ↓
Saved
   ↓
Running
   ↓
Task A
   ↓
Task B
   ↓
Task C
   ↓
Completed
   ↓
Saved

If the process crashes:

Database
   ↓
Last known state
   ↓
Recovery
   ↓
Resume / Retry

A full recovery system requires additional task-level persistence, which can be added later.


8.12 Request IDs

Every API request should have a unique identifier.

Example:

request_id:
9c4d...

This makes debugging much easier.

Create:

import uuid


def generate_request_id():

    return str(
        uuid.uuid4()
    )

8.13 Request Middleware

In FastAPI:

from fastapi import Request


@app.middleware("http")
async def request_tracking(
    request: Request,
    call_next,
):

    request_id = (
        request.headers.get(
            "X-Request-ID"
        )
        or generate_request_id()
    )

    response = await call_next(
        request
    )

    response.headers[
        "X-Request-ID"
    ] = request_id

    return response

Now every response carries a request ID.


8.14 Why Request IDs Matter

Suppose a user reports:

"My request failed."

Without an ID, debugging may require searching through thousands of logs.

With:

Request ID:
abc123

developers can search:

abc123

and find the complete request path.


8.15 Structured Error Handling

Errors should be predictable.

Create:

app/errors.py

Add:

class ACAIError(Exception):

    def __init__(
        self,
        message: str,
        code: str = "ACAI_ERROR",
    ) -> None:

        super().__init__(message)

        self.message = message

        self.code = code

Specific errors can inherit from it:

class ValidationError(ACAIError):

    def __init__(
        self,
        message: str,
    ) -> None:

        super().__init__(
            message,
            code="VALIDATION_ERROR",
        )


class ModelError(ACAIError):

    def __init__(
        self,
        message: str,
    ) -> None:

        super().__init__(
            message,
            code="MODEL_ERROR",
        )


class WorkflowError(ACAIError):

    def __init__(
        self,
        message: str,
    ) -> None:

        super().__init__(
            message,
            code="WORKFLOW_ERROR",
        )

8.16 Error Response

The API should return structured errors.

Example:

{
  "success": false,
  "error": {
    "code": "MODEL_ERROR",
    "message": "Model request failed.",
    "request_id": "abc123"
  }
}

Avoid exposing internal stack traces to normal users.


8.17 Global Exception Handler

FastAPI can handle ACAI-specific errors:

from fastapi.responses import JSONResponse


@app.exception_handler(
    ACAIError
)
async def acai_error_handler(
    request,
    exc: ACAIError,
):

    request_id = (
        request.headers.get(
            "X-Request-ID",
            "unknown",
        )
    )

    return JSONResponse(
        status_code=400,
        content={
            "success": False,
            "error": {
                "code": exc.code,
                "message": exc.message,
                "request_id": request_id,
            },
        },
    )

8.18 Rate Limiting

Public APIs need protection against excessive requests.

Conceptually:

User
 ↓
Rate Limiter
 ↓
Allowed?
 ┌──┴──┐
YES    NO
 │      │
 ▼      ▼
API    429

A simple conceptual policy:

100 requests / minute / client

The actual production limit should be selected according to expected traffic and infrastructure capacity.


8.19 Simple Rate-Limit Prototype

For demonstration:

import time


class RateLimiter:

    def __init__(
        self,
        limit: int = 100,
        window: int = 60,
    ):

        self.limit = limit

        self.window = window

        self.requests = {}

    def allow(
        self,
        client_id: str,
    ) -> bool:

        now = time.time()

        timestamps = (
            self.requests
            .setdefault(
                client_id,
                [],
            )
        )

        timestamps[:] = [
            timestamp
            for timestamp in timestamps
            if now - timestamp
            < self.window
        ]

        if len(timestamps) >= self.limit:

            return False

        timestamps.append(now)

        return True

This is a prototype only.

For multiple application servers, a shared store such as Redis is more appropriate.


8.20 Caching

Some operations are repeated frequently.

For example:

Same query
 ↓
Same retrieval
 ↓
Repeated computation

A cache can reduce unnecessary work.

Architecture:

Request
 ↓
Cache
 ↓
┌───────┴───────┐
▼               ▼
HIT             MISS
│               │
▼               ▼
Result        Compute
                │
                ▼
              Cache

8.21 Simple Cache

Create:

class SimpleCache:

    def __init__(self):

        self.data = {}

    def get(
        self,
        key: str,
    ):

        return self.data.get(
            key
        )

    def set(
        self,
        key: str,
        value,
    ):

        self.data[key] = value

    def delete(
        self,
        key: str,
    ):

        self.data.pop(
            key,
            None,
        )

    def clear(self):

        self.data.clear()

8.22 Cache Safety

Not every operation should be cached.

Good candidates can include:

Stable retrieval results
Configuration
Public metadata
Repeated deterministic calculations

Potentially unsafe candidates include:

Private user data
Sensitive responses
Time-sensitive information
Requests whose result depends on changing state

A production cache therefore needs:

TTL
Invalidation
User isolation
Permission checks
Size limits

8.23 Health Checks

A production API should expose:

/health

Example:

@app.get("/health")
async def health():

    return {
        "status": "ok",
        "service": "acai",
    }

This confirms that the application process is responding.


8.24 Readiness vs Liveness

These should eventually be separate.

Liveness

"Is the application process alive?"

Readiness

"Can the application actually serve requests?"

For example:

/health/live
/health/ready

Readiness can verify:

Database
Model provider
Required services
Configuration

8.25 Production Configuration

Configuration should not be hardcoded.

Create:

.env

Example:

ACAI_ENV=development
ACAI_LOG_LEVEL=INFO
ACAI_DATABASE_URL=sqlite:///data/acai.db
ACAI_MODEL_PROVIDER=mock
ACAI_MAX_WORKFLOW_STEPS=10

Do not commit real secrets.

For example:

API_KEY=actual-secret

should never be placed into source control.


8.26 Configuration Class

Update:

app/config.py

Example:

import os


class Settings:

    environment = os.getenv(
        "ACAI_ENV",
        "development",
    )

    log_level = os.getenv(
        "ACAI_LOG_LEVEL",
        "INFO",
    )

    database_url = os.getenv(
        "ACAI_DATABASE_URL",
        "sqlite:///data/acai.db",
    )

    model_provider = os.getenv(
        "ACAI_MODEL_PROVIDER",
        "mock",
    )

    max_workflow_steps = int(
        os.getenv(
            "ACAI_MAX_WORKFLOW_STEPS",
            "10",
        )
    )


settings = Settings()

8.27 Logging

Create:

import logging


logging.basicConfig(
    level=logging.INFO,
)

logger = logging.getLogger(
    "acai"
)

Then:

logger.info(
    "Workflow started: %s",
    workflow_id,
)

Errors:

logger.exception(
    "Workflow failed: %s",
    workflow_id,
)

8.28 Logging Architecture

                     ACAI
                       │
                       ▼
                    Events
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
        INFO         ERROR       METRICS
          │            │            │
          └────────────┼────────────┘
                       ▼
                  Log System

8.29 Background Jobs

Some tasks should not block the API request.

For example:

Large document processing
Long workflow
Batch evaluation
Embedding generation
Dataset indexing

Instead of:

HTTP Request
 ↓
Wait 10 minutes
 ↓
Response

use:

HTTP Request
 ↓
Create Job
 ↓
Return Job ID

Then:

Worker
 ↓
Process Job
 ↓
Save Result

The user can query:

/jobs/{job_id}

8.30 Job Architecture

                  API
                   │
                   ▼
                Job Queue
                   │
          ┌────────┼────────┐
          ▼        ▼        ▼
       Worker 1 Worker 2 Worker 3
          │        │        │
          └────────┼────────┘
                   ▼
                Database
                   │
                   ▼
                 Result

For a simple prototype, FastAPI background tasks can be used.

For larger systems, a dedicated queue/worker architecture is more appropriate.


8.31 Production Deployment

A typical deployment could be:

Internet
   │
   ▼
Reverse Proxy
   │
   ▼
FastAPI
   │
   ├── Database
   ├── Cache
   ├── Model APIs
   └── Workers

Multiple application instances can be added:

             Load Balancer
              /    |    \
             ▼     ▼     ▼
          API-1  API-2  API-3
             \     |     /
              \    |    /
               Database

This is where shared state becomes important.


8.32 Stateless API Design

The API layer should ideally be stateless.

Instead of:

API Server
 ↓
Own local memory

use:

API Server
 ↓
Shared Database / Cache

Then any API instance can process the next request.


8.33 Containerization

A production deployment can use Docker.

Example:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install \
    --no-cache-dir \
    -r requirements.txt

COPY app ./app

COPY data ./data

CMD [
    "uvicorn",
    "app.main:app",
    "--host",
    "0.0.0.0",
    "--port",
    "8000"
]

Build:

docker build -t acai .

Run:

docker run -p 8000:8000 acai

The exact deployment configuration depends on the target environment.


8.34 Database Migration Strategy

As the schema evolves, manually editing production tables becomes dangerous.

A production system should use migrations.

Conceptually:

Schema v1
 ↓
Migration 001
 ↓
Schema v2
 ↓
Migration 002
 ↓
Schema v3

This allows controlled database evolution.


8.35 Security Fundamentals

ACAI should implement basic security before public deployment.

Important areas:

Authentication
Authorization
Input validation
Rate limiting
Secret management
TLS
Database permissions
Audit logging
Dependency updates

Never assume:

"AI application"

means security is automatically handled by the model.


8.36 Input Validation

Existing Pydantic validation should remain enabled.

Example:

class ChatRequest(BaseModel):

    message: str = Field(
        ...,
        min_length=1,
        max_length=10000,
    )

This prevents obviously invalid requests from entering the pipeline.


8.37 Resource Limits

Every expensive operation should have a limit.

Examples:

Maximum input size
Maximum output tokens
Maximum workflow steps
Maximum retries
Maximum execution time
Maximum uploaded file size
Maximum concurrent jobs

This protects the system from accidental or malicious resource exhaustion.


8.38 Persistent Architecture

After Chapter 8:

                         USER
                           │
                           ▼
                      API Gateway
                           │
                     Rate Limiter
                           │
                     Request ID
                           │
                           ▼
                     ORCHESTRATOR
                           │
          ┌────────────────┼────────────────┐
          ▼                ▼                ▼
       Planner           Memory          Retrieval
                           │
                           ▼
                        Database
                           │
          ┌────────────────┼────────────────┐
          ▼                ▼                ▼
       Workflows         Cache          Logs
          │
          ▼
      Model Router
          │
          ▼
      Model Service
          │
          ▼
      Verification
          │
          ▼
        Result

8.39 Testing Production Infrastructure

Create:

tests/test_database.py

Example:

def test_database_initialization():

    from app.database import (
        initialize_database,
        get_connection,
    )

    initialize_database()

    connection = get_connection()

    rows = connection.execute(
        """
        SELECT name
        FROM sqlite_master
        WHERE type='table'
        """
    ).fetchall()

    tables = {
        row["name"]
        for row in rows
    }

    assert "memories" in tables

    assert "workflows" in tables

    connection.close()

8.40 Test Persistent Memory

def test_persistent_memory():

    from app.services.persistent_memory import (
        PersistentMemoryService,
    )

    service = (
        PersistentMemoryService()
    )

    memory_id = service.remember(
        content=(
            "The project uses "
            "FastAPI."
        ),
        memory_type="project",
        importance=0.8,
    )

    assert memory_id

8.41 Test Rate Limiting

def test_rate_limiter():

    from app.services.rate_limiter import (
        RateLimiter,
    )

    limiter = RateLimiter(
        limit=2,
        window=60,
    )

    assert limiter.allow(
        "client"
    )

    assert limiter.allow(
        "client"
    )

    assert not limiter.allow(
        "client"
    )

8.42 Test Health Endpoint

def test_health(client):

    response = client.get(
        "/health"
    )

    assert response.status_code == 200

    assert response.json()[
        "status"
    ] == "ok"

8.43 Full Test Command

Run:

pytest -v

For coverage:

pytest --cov=app

The exact coverage percentage should come from the actual test run.

Do not claim a coverage percentage without running the command.


8.44 Production Readiness Checklist

Before deployment:

[ ] Environment variables configured
[ ] Secrets removed from source
[ ] Database persistence tested
[ ] Database backups planned
[ ] Rate limiting enabled
[ ] Input limits enabled
[ ] Request IDs implemented
[ ] Error handling implemented
[ ] Logging configured
[ ] Health checks available
[ ] Workflow limits configured
[ ] Retry limits configured
[ ] Timeout limits configured
[ ] Authentication implemented
[ ] Authorization implemented
[ ] Dependencies reviewed
[ ] HTTPS configured
[ ] Monitoring configured
[ ] Recovery strategy tested

8.45 Important Reality Check

A production architecture is not automatically production-ready simply because it contains:

Database
Docker
FastAPI
Redis
Workers
AI Models

Production readiness requires testing.

For example:

Architecture
       ↓
Implementation
       ↓
Unit Tests
       ↓
Integration Tests
       ↓
Load Tests
       ↓
Failure Tests
       ↓
Security Tests
       ↓
Monitoring
       ↓
Production

8.46 Failure Testing

ACAI should deliberately test failures.

Examples:

Model unavailable
Database unavailable
Retrieval timeout
Worker crashes
Invalid model response
Rate limit reached
Workflow cycle
Memory database error
Network failure

The objective is to determine:

Does the system fail safely and predictably?


8.47 Recovery Testing

A useful experiment:

Start Workflow
      ↓
Execute Task A
      ↓
Crash Application
      ↓
Restart
      ↓
Read Persistent State
      ↓
Resume / Retry

If the workflow cannot recover, identify exactly where state was lost.


8.48 Chapter 8 Success Criteria

Chapter 8 is complete when:

[✓] Persistent database exists
[✓] Memory can be stored persistently
[✓] Workflow state can be persisted
[✓] Request IDs exist
[✓] Structured errors exist
[✓] Rate limiting is designed
[✓] Caching layer is defined
[✓] Health checks exist
[✓] Background jobs are defined
[✓] Configuration is externalized
[✓] Logging is implemented
[✓] Deployment architecture is defined
[✓] Infrastructure tests exist

8.49 Current ACAI Architecture

At this point:

                              USER
                                │
                                ▼
                           API GATEWAY
                                │
                    ┌───────────┴───────────┐
                    ▼                       ▼
               RATE LIMITER             REQUEST ID
                    │                       │
                    └───────────┬───────────┘
                                ▼
                         ORCHESTRATOR
                                │
       ┌────────────────────────┼────────────────────────┐
       ▼                        ▼                        ▼
    PLANNER                   MEMORY                 RETRIEVAL
       │                        │                        │
       │                        ▼                        │
       │                    DATABASE                    │
       │                        │                        │
       └────────────────────────┼────────────────────────┘
                                ▼
                         MODEL ROUTER
                                │
                                ▼
                         MODEL SERVICE
                                │
                                ▼
                           WORKFLOW
                                │
                                ▼
                         VERIFICATION
                                │
                         ┌──────┴──────┐
                         ▼             ▼
                      ACCEPT        REVISE
                         │             │
                         └──────┬──────┘
                                ▼
                             RESULT
                                │
               ┌────────────────┼────────────────┐
               ▼                ▼                ▼
             CACHE            LOGS           METRICS

8.50 What Comes Next?

ACAI now has architecture, execution, persistence, and basic production infrastructure.

The next major step is to connect the system to real external model providers and tools while keeping the architecture provider-independent.

That leads to:

Chapter 9 — Multi-Provider AI Integration and Fallback Architecture

The next layer will cover:

Provider Interface
      ↓
OpenAI-compatible providers
      ↓
Cloud Models
      ↓
Local Models
      ↓
Fallback Models
      ↓
Timeouts
      ↓
Retries
      ↓
Provider Health
      ↓
Automatic Routing

The goal is not to claim that every provider will always work.

The goal is to design ACAI so that:

Provider A fails
      ↓
Health Check
      ↓
Fallback Policy
      ↓
Provider B
      ↓
Response

can happen in a controlled and measurable way.

End of Chapter 8


Comments

Popular posts from this blog

Adaptive Cognitive AI (ACAI): Chapter 1 — Introduction & System Vision

Chapter 2 (Part 2) Knowledge Retrieval Engine

Adaptive Cognitive AI (ACAI) Chapter 2 (Part 1).