AI System Architecture

A savanna camp where tents, water supplies and the campfire stand in one big mess works only until the first storm. After that, nobody knows what to move and what to leave. An AI application without architecture looks similar: the endpoint connects to the vector database itself, calls the language model itself and computes embeddings itself. Changing the database provider then means rewriting half of the project. Architecture is the skeleton that lets you swap parts without tearing down the whole camp.

Clean Architecture for AI Applications

Clean Architecture splits a system into layers. The key rule: dependencies point inward, so an outer layer may know an inner one, but never the other way round. Here are the four layers of our document assistant, from the interface at the top to the infrastructure at the bottom:

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└─────────────────────────────────────────────────────────────┘

The Presentation Layer accepts requests (REST, WebSocket, CLI), the Application Layer holds the use cases, the Domain Layer keeps the entities and business logic, and the Infrastructure Layer is the concrete tooling: Qdrant, OpenAI, Postgres, Redis. In Robert C. Martin's original description, the same circles are called Entities, Use Cases, Interface Adapters and Frameworks & Drivers. The names differ, the idea is the same: business logic does not know which database you use.

Domain Layer Implementation

The domain is the heart of the camp. We start with entities, objects that have an identity. We use field(default_factory=...) because a default value of type list or dict must be created separately for every object, and uuid4 generates a random identifier.

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 = None
20    content: str = ""
21    embedding: list[float] = field(default_factory=list)
22    metadata: dict = field(default_factory=dict)

DocumentChunk is a fragment of a document together with its embedding, a vector of numbers. The UUID | None annotation honestly says that a chunk may not belong to any document yet. Writing "DocumentChunk" in quotes refers to a class defined further down. Now the query and the response:

1@dataclass
2class Query:
3    id: UUID = field(default_factory=uuid4)
4    text: str = ""
5    user_id: UUID | None = None
6    created_at: datetime = field(default_factory=datetime.now)
7
8@dataclass
9class Response:
10    query_id: UUID | None = None
11    answer: str = ""
12    sources: list[DocumentChunk] = field(default_factory=list)
13    confidence: float = 0.0
14    latency_ms: float = 0.0

None of these classes imports Qdrant or FastAPI. That is deliberate: you can test the domain without starting anything.

Repository Pattern

The Repository Pattern is an abstraction over data access. The domain says "save a document" or "find similar chunks", but it does not know where the data lives. An abstract class based on ABC with the @abstractmethod decorator defines a contract: the methods every implementation must provide.

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

You cannot create a DocumentRepository object directly, Python raises TypeError until a subclass implements all abstract methods. The methods are async because talking to a database is an input/output operation. Here is the Qdrant implementation:

1class QdrantDocumentRepository(DocumentRepository):
2    def __init__(self, client, collection_name: str):
3        self.client = client
4        self.collection = collection_name
5
6    async def save(self, document: Document) -> Document:
7        # Implementation
8        pass
9
10    async def get_by_id(self, doc_id: UUID) -> Optional[Document]:
11        # Implementation
12        pass
13
14    async def search(self, query_embedding: list[float], limit: int) -> list[DocumentChunk]:
15        response = await self.client.query_points(
16            collection_name=self.collection,
17            query=query_embedding,
18            limit=limit
19        )
20        return [self._to_chunk(point) for point in response.points]

In the current qdrant-client, search is done with the query_points method, which returns an object with a list of points. The older search method was first deprecated and is gone from recent client versions, so you may still see it in older tutorials. For await to work, client must be an AsyncQdrantClient instance. The save and get_by_id methods and the _to_chunk helper are left as a sketch to complete in the project.

Dependency Injection

Dependency Injection means that a class receives its dependencies from outside instead of creating them itself. Thanks to that, you pass a fake in tests and the real Qdrant in production. Let's start with the simplest container, a dictionary of "interface, implementation" pairs:

1from functools import lru_cache
2
3from fastapi import Depends
4
5class Container:
6    """Simple DI container."""
7
8    def __init__(self):
9        self._services = {}
10
11    def register(self, interface: type, implementation):
12        self._services[interface] = implementation
13
14    def resolve(self, interface: type):
15        return self._services.get(interface)

register stores an object under a type key, resolve hands it back. Registration looks like this (LLMService and OpenAILLMService are a matching interface and implementation pair, and qdrant_client and llm_client are clients created earlier at application startup):

1# Setup
2container = Container()
3container.register(DocumentRepository, QdrantDocumentRepository(qdrant_client, "documents"))
4container.register(LLMService, OpenAILLMService(llm_client))

Notice that nothing changed in the domain classes, only the configuration changed. Finally, we plug the container into FastAPI through Depends, the framework's built-in dependency injection mechanism:

1# Usage in FastAPI
2@lru_cache
3def get_container() -> Container:
4    return container
5
6def get_document_repo(container: Container = Depends(get_container)):
7    return container.resolve(DocumentRepository)

@lru_cache makes get_container always return the same object. An endpoint that declares a parameter with Depends(get_document_repo) receives the repository without knowing where it comes from. In tests you swap it through app.dependency_overrides.

My advice: a hand-written container like this plus Depends is enough to start. Add DI libraries only when dependencies really start to multiply. In the next lesson you will see how this architecture makes testing easier.

Remember: architecture is the camp layout in which every tent can be moved without tearing down the others.