AI System Architecture

Architecture is the skeleton of every application. Just as in nature - a strong structure allows an organism to survive and thrive!

Clean Architecture for AI Applications

1┌─────────────────────────────────────────────────────────────┐
2│                    Presentation Layer                        │
3│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
4│  │   REST API  │  │  WebSocket  │  │    CLI      │         │
5│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘         │
6├─────────┼────────────────┼────────────────┼─────────────────┤
7│         └────────────────┼────────────────┘                 │
8│                          ▼                                   │
9│                  Application Layer                           │
10│  ┌─────────────────────────────────────────────────────┐   │
11│  │              Use Cases / Services                    │   │
12│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐          │   │
13│  │  │  Query   │  │  Upload  │  │  Search  │          │   │
14│  │  │ Service  │  │ Service  │  │ Service  │          │   │
15│  │  └──────────┘  └──────────┘  └──────────┘          │   │
16│  └─────────────────────────────────────────────────────┘   │
17├─────────────────────────────────────────────────────────────┤
18│                     Domain Layer                             │
19│  ┌─────────────────────────────────────────────────────┐   │
20│  │              Entities & Business Logic               │   │
21│  │  Document │ Query │ Response │ User │ Embedding     │   │
22│  └─────────────────────────────────────────────────────┘   │
23├─────────────────────────────────────────────────────────────┤
24│                  Infrastructure Layer                        │
25│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
26│  │ Vector DB│  │   LLM    │  │ Database │  │   Cache  │  │
27│  │  Qdrant  │  │  OpenAI  │  │ Postgres │  │  Redis   │  │
28│  └──────────┘  └──────────┘  └──────────┘  └──────────┘  │
29└─────────────────────────────────────────────────────────────┘

Domain Layer Implementation

1from dataclasses import dataclass, field
2from datetime import datetime
3from uuid import UUID, uuid4
4from abc import ABC, abstractmethod
5
6# Entities
7@dataclass
8class Document:
9    id: UUID = field(default_factory=uuid4)
10    title: str = ""
11    content: str = ""
12    metadata: dict = field(default_factory=dict)
13    created_at: datetime = field(default_factory=datetime.now)
14    chunks: list["DocumentChunk"] = field(default_factory=list)
15
16@dataclass
17class DocumentChunk:
18    id: UUID = field(default_factory=uuid4)
19    document_id: UUID = None
20    content: str = ""
21    embedding: list[float] = field(default_factory=list)
22    metadata: dict = field(default_factory=dict)
23
24@dataclass
25class Query:
26    id: UUID = field(default_factory=uuid4)
27    text: str = ""
28    user_id: UUID = None
29    created_at: datetime = field(default_factory=datetime.now)
30
31@dataclass
32class Response:
33    query_id: UUID = None
34    answer: str = ""
35    sources: list[DocumentChunk] = field(default_factory=list)
36    confidence: float = 0.0
37    latency_ms: float = 0.0

Repository Pattern

1from abc import ABC, abstractmethod
2from typing import Optional
3
4class DocumentRepository(ABC):
5    @abstractmethod
6    async def save(self, document: Document) -> Document:
7        pass
8
9    @abstractmethod
10    async def get_by_id(self, doc_id: UUID) -> Optional[Document]:
11        pass
12
13    @abstractmethod
14    async def search(self, query_embedding: list[float], limit: int) -> list[DocumentChunk]:
15        pass
16
17class QdrantDocumentRepository(DocumentRepository):
18    def __init__(self, client, collection_name: str):
19        self.client = client
20        self.collection = collection_name
21
22    async def save(self, document: Document) -> Document:
23        # Implementation
24        pass
25
26    async def get_by_id(self, doc_id: UUID) -> Optional[Document]:
27        # Implementation
28        pass
29
30    async def search(self, query_embedding: list[float], limit: int) -> list[DocumentChunk]:
31        results = await self.client.search(
32            collection_name=self.collection,
33            query_vector=query_embedding,
34            limit=limit
35        )
36        return [self._to_chunk(r) for r in results]

Dependency Injection

1from functools import lru_cache
2
3class Container:
4    """Simple DI container."""
5
6    def __init__(self):
7        self._services = {}
8
9    def register(self, interface: type, implementation):
10        self._services[interface] = implementation
11
12    def resolve(self, interface: type):
13        return self._services.get(interface)
14
15# Setup
16container = Container()
17container.register(DocumentRepository, QdrantDocumentRepository(...))
18container.register(LLMService, OpenAILLMService(...))
19
20# Usage in FastAPI
21@lru_cache
22def get_container() -> Container:
23    return container
24
25def get_document_repo(container: Container = Depends(get_container)):
26    return container.resolve(DocumentRepository)

Good architecture is the foundation of a scalable application. In the next lesson, we will cover testing!