Architektura Systemu AI

Obóz na sawannie, w którym namioty, zapasy wody i ognisko stoją w jednym bałaganie, działa tylko do pierwszej burzy. Potem nie wiadomo, co przenieść, a co zostawić. Aplikacja AI bez architektury wygląda podobnie: endpoint sam łączy się z bazą wektorową, sam woła model językowy i sam liczy embeddingi. Zmiana dostawcy bazy oznacza wtedy przepisywanie połowy projektu. Architektura to szkielet, który pozwala wymieniać części bez burzenia całego obozu.

Clean Architecture dla aplikacji AI

Clean Architecture dzieli system na warstwy. Najważniejsza zasada: zależności wskazują do środka, czyli warstwa zewnętrzna może znać wewnętrzną, ale nigdy odwrotnie. Oto cztery warstwy naszego asystenta dokumentów, od interfejsu na górze do infrastruktury na dole:

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

Presentation Layer przyjmuje żądania (REST, WebSocket, CLI), Application Layer zawiera przypadki użycia, Domain Layer trzyma encje i logikę biznesową, a Infrastructure Layer to konkretne narzędzia: Qdrant, OpenAI, Postgres, Redis. W oryginalnym opisie Roberta C. Martina te same kręgi nazywają się Entities, Use Cases, Interface Adapters oraz Frameworks & Drivers. Nazwy się różnią, idea jest ta sama: logika biznesowa nie wie, jakiej bazy używasz.

Implementacja Domain Layer

Domena to serce obozu. Zaczynamy od encji, czyli obiektów, które mają tożsamość. Użyjemy field(default_factory=...), bo domyślna wartość typu list albo dict musi być tworzona osobno dla każdego obiektu, a uuid4 generuje losowy identyfikator.

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 to fragment dokumentu razem z jego embeddingiem, czyli wektorem liczb. Adnotacja UUID | None mówi uczciwie, że fragment może jeszcze nie należeć do żadnego dokumentu. Zapis "DocumentChunk" w cudzysłowie to odwołanie do klasy zdefiniowanej niżej. Teraz pytanie i odpowiedź:

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

Żadna z tych klas nie importuje Qdranta ani FastAPI. To celowe: domenę możesz przetestować bez uruchamiania czegokolwiek.

Repository Pattern

Repository Pattern to abstrakcja dostępu do danych. Domena mówi "zapisz dokument" albo "znajdź podobne fragmenty", ale nie wie, gdzie te dane leżą. Klasa abstrakcyjna ABC z dekoratorem @abstractmethod definiuje kontrakt: metody, które każda implementacja musi dostarczyć.

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

Nie da się utworzyć obiektu DocumentRepository bezpośrednio, Python zgłosi TypeError, dopóki podklasa nie zaimplementuje wszystkich metod abstrakcyjnych. Metody są async, bo rozmowa z bazą to operacja wejścia-wyjścia. Oto implementacja dla Qdranta:

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]

W bieżącym qdrant-client wyszukiwanie robi się metodą query_points, która zwraca obiekt z listą points. Starsza metoda search była najpierw oznaczona jako przestarzała, a w nowych wersjach klienta już jej nie ma, więc w starszych tutorialach możesz ją jeszcze spotkać. Żeby await działał, client musi być instancją AsyncQdrantClient. Metody save, get_by_id i pomocnicze _to_chunk zostawiamy jako szkic do uzupełnienia w projekcie.

Dependency Injection

Dependency Injection (wstrzykiwanie zależności) oznacza, że klasa dostaje swoje zależności z zewnątrz, zamiast tworzyć je sama w środku. Dzięki temu w testach podajesz atrapę, a w produkcji prawdziwy Qdrant. Zacznijmy od najprostszego kontenera, czyli słownika "interfejs, implementacja":

1from functools import lru_cache
2
3from fastapi import Depends
4
5class Container:
6    """Prosty kontener DI."""
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 zapamiętuje obiekt pod kluczem typu, resolve go oddaje. Rejestracja wygląda tak (LLMService i OpenAILLMService to analogiczna para interfejsu i implementacji, a qdrant_client i llm_client to klienci utworzeni wcześniej przy starcie aplikacji):

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

Zauważ, że nic tu się nie zmieniło w klasach domeny, zmieniła się tylko konfiguracja. Na koniec podłączamy kontener do FastAPI przez Depends, czyli wbudowany mechanizm wstrzykiwania zależności w tym frameworku:

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 sprawia, że get_container zwraca zawsze ten sam obiekt. Endpoint, który zadeklaruje parametr z Depends(get_document_repo), dostanie repozytorium bez wiedzy, skąd ono pochodzi. W testach podmienisz je przez app.dependency_overrides.

Moja rada: na początek wystarczy taki ręczny kontener i Depends. Biblioteki DI dodawaj dopiero, gdy zależności naprawdę zaczną się mnożyć. W następnej lekcji zobaczysz, jak ta architektura ułatwia testowanie.

Pamiętaj: architektura to plan obozu, w którym każdy namiot da się przenieść bez burzenia pozostałych.