Architecture d'un système d'IA

Un campement dans la savane où les tentes, les réserves d'eau et le feu de camp sont entassés dans le désordre ne tient que jusqu'au premier orage. Ensuite, plus personne ne sait quoi déplacer et quoi laisser. Une application d'IA sans architecture ressemble à ça : l'endpoint se connecte lui-même à la base vectorielle, appelle lui-même le modèle de langage et calcule lui-même les embeddings. Changer de fournisseur de base signifie alors réécrire la moitié du projet. L'architecture, c'est le squelette qui permet de remplacer des pièces sans démonter tout le campement.

Clean Architecture pour les applications d'IA

La Clean Architecture découpe un système en couches. La règle clé : les dépendances pointent vers l'intérieur, une couche externe peut connaître une couche interne, mais jamais l'inverse. Voici les quatre couches de notre assistant de documents, de l'interface en haut à l'infrastructure en bas :

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

La Presentation Layer reçoit les requêtes (REST, WebSocket, CLI), l'Application Layer contient les cas d'usage, la Domain Layer garde les entités et la logique métier, et l'Infrastructure Layer regroupe les outils concrets : Qdrant, OpenAI, Postgres, Redis. Dans la description originale de Robert C. Martin, les mêmes cercles s'appellent Entities, Use Cases, Interface Adapters et Frameworks & Drivers. Les noms diffèrent, l'idée est la même : la logique métier ne sait pas quelle base de données tu utilises.

Implémentation de la couche domaine

Le domaine, c'est le cœur du campement. On commence par les entités, des objets qui ont une identité. On utilise field(default_factory=...) parce qu'une valeur par défaut de type list ou dict doit être créée séparément pour chaque objet, et uuid4 génère un identifiant aléatoire.

1from dataclasses import dataclass, field
2from datetime import datetime
3from uuid import UUID, uuid4
4from abc import ABC, abstractmethod
5
6# Entités
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 est un fragment de document avec son embedding, c'est-à-dire un vecteur de nombres. L'annotation UUID | None dit honnêtement qu'un fragment peut n'appartenir encore à aucun document. Écrire "DocumentChunk" entre guillemets permet de faire référence à une classe définie plus bas. Passons à la requête et à la réponse :

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

Aucune de ces classes n'importe Qdrant ni FastAPI. C'est voulu : tu peux tester le domaine sans rien lancer.

Repository Pattern

Le Repository Pattern est une abstraction de l'accès aux données. Le domaine dit « enregistre un document » ou « trouve des fragments similaires », mais il ne sait pas où se trouvent les données. Une classe abstraite basée sur ABC avec le décorateur @abstractmethod définit un contrat : les méthodes que chaque implémentation doit fournir.

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

Impossible de créer directement un objet DocumentRepository, Python lève une TypeError tant qu'une sous-classe n'implémente pas toutes les méthodes abstraites. Les méthodes sont async car parler à une base de données est une opération d'entrée-sortie. Voici l'implémentation pour Qdrant :

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        # Implémentation
8        pass
9
10    async def get_by_id(self, doc_id: UUID) -> Optional[Document]:
11        # Implémentation
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]

Dans la version actuelle de qdrant-client, la recherche se fait avec la méthode query_points, qui renvoie un objet contenant une liste points. L'ancienne méthode search a d'abord été dépréciée, puis a disparu des versions récentes du client, donc tu la croiseras peut-être encore dans d'anciens tutoriels. Pour que await fonctionne, client doit être une instance d'AsyncQdrantClient. Les méthodes save et get_by_id ainsi que l'utilitaire _to_chunk restent une ébauche à compléter dans le projet.

Dependency Injection

La Dependency Injection (injection de dépendances) signifie qu'une classe reçoit ses dépendances de l'extérieur au lieu de les créer elle-même. Grâce à ça, tu passes une doublure dans les tests et le vrai Qdrant en production. Commençons par le conteneur le plus simple, un dictionnaire « interface, implémentation » :

1from functools import lru_cache
2
3from fastapi import Depends
4
5class Container:
6    """Conteneur DI simple."""
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 mémorise un objet sous une clé de type, resolve le rend. L'enregistrement ressemble à ceci (LLMService et OpenAILLMService forment une paire interface et implémentation analogue, et qdrant_client et llm_client sont des clients créés plus tôt au démarrage de l'application) :

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

Remarque que rien n'a changé dans les classes du domaine, seule la configuration a changé. Enfin, on branche le conteneur sur FastAPI avec Depends, le mécanisme d'injection de dépendances intégré au framework :

1# Utilisation dans 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 fait en sorte que get_container renvoie toujours le même objet. Un endpoint qui déclare un paramètre avec Depends(get_document_repo) reçoit le repository sans savoir d'où il vient. Dans les tests, tu le remplaces via app.dependency_overrides.

Mon conseil : pour démarrer, un conteneur écrit à la main comme celui-ci et Depends suffisent. N'ajoute une bibliothèque de DI que lorsque les dépendances commencent vraiment à se multiplier. Dans la prochaine leçon, tu verras comment cette architecture facilite les tests.

Souviens-toi : l'architecture, c'est le plan du campement où chaque tente peut être déplacée sans démonter les autres.