68e889b14c39 dla systemów produkcyjnych — umowy i weryfikacje

Poniższe notatki przedstawiają praktyczną ścieżkę postępowania wokół „”. Nacisk kładziony jest na umowy, sprawdzenia oraz miejsca zastępcze dla kodu, a nie na motywacyjne aspekty. Podczas przechodzenia przez etap przeglądu najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas trwania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych.

1. Embeddings przekształcają znaczenie w geometrię

Faza przekształcania znaczeń za pomocą 1 Embeddingów działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przekaz, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, skrytki z danymi poufnymi oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całego grafu. Oddziel zasadę dzielenia na fragmenty od zasady wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.

# pip install openai
# export OPENAI_API_KEY=sk-...
import math

from openai import OpenAI

client = OpenAI()


def embed(texts):
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in resp.data]


def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb)


q, hit, miss = embed([
    "How do I get my money back?",
    "Refunds are possible within 14 days "
    "of purchase.",
    "Data is stored in Frankfurt and "
    "encrypted at rest.",
])

print(f"refund doc:  {cosine(q, hit):.3f}")
print(f"storage doc: {cosine(q, miss):.3f}")

2. Magazyn wektorów to lista, którą można posortować

Faza przechowywania wektorów 2 A działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zdokumentuj zarówno prawidłowy przebieg operacji, jak i ścieżkę przywracania stanu. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Zaznacz wersje zależności i zapisz hash obrazu, który został użyty do uruchomienia demonstracji. Reprodukowalność jest ważniejsza od lokalnej wiedzy zespołu.

class VectorStore:
    def __init__(self):
        self.texts = []
        self.vectors = []

    def add(self, texts):
        self.texts.extend(texts)
        self.vectors.extend(embed(texts))

    def top_k(self, query, k=3):
        q = embed([query])[0]
        scored = [
            (cosine(q, v), t)
            for v, t in zip(
                self.vectors, self.texts
            )
        ]
        scored.sort(
            key=lambda s: s[0], reverse=True
        )
        return scored[:k]


store = VectorStore()
store.add([
    "Refunds are possible within 14 days.",
    "The Pro plan costs 12 EUR per month.",
    "Data is stored in Frankfurt.",
])

for score, text in store.top_k(
    "can I cancel and get a refund?", k=2
):
    print(f"{score:.3f}  {text}")

3. Pełny proces: pobieranie, uzupełnianie, generowanie

Pełny etap pipeline funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zanim rozszerzysz zakres, zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany pipeline. Oddziel zasadę dzielenia na fragmenty od zasady pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości. Pełny etap pipeline funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zanim rozszerzysz zakres, zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z wersji demonstracyjnej do środowisk współdzielonych.

# examples/part07_rag.py
# Mini RAG in ~90 lines: embed -> store -> top-k
# -> stitch into messages -> generate.
#
# Setup:
#   pip install openai
#   export OPENAI_API_KEY=sk-...
#
# Run:
#   python part07_rag.py

import math

from openai import OpenAI

client = OpenAI()

EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"

# A tiny corpus. In real life these are chunks
# of your docs, not whole documents.
DOCS = [
    "Support hours are 9am to 6pm CET, "
    "Monday to Friday.",
    "Refunds are possible within 14 days "
    "of purchase, no questions asked.",
    "The Pro plan costs 12 EUR per month "
    "and includes API access.",
    "The Free plan allows 3 projects and "
    "community support only.",
    "Data is stored in Frankfurt and "
    "encrypted at rest with AES-256.",
    "You can export all your data as "
    "JSON from the settings page.",
]


def embed(texts):
    resp = client.embeddings.create(
        model=EMBED_MODEL,
        input=texts,
    )
    return [d.embedding for d in resp.data]


def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb)


class VectorStore:
    """A list of (vector, text) pairs. That is
    all a vector store is, before you need
    scale."""

    def __init__(self):
        self.texts = []
        self.vectors = []

    def add(self, texts):
        self.texts.extend(texts)
        self.vectors.extend(embed(texts))

    def top_k(self, query, k=3):
        q = embed([query])[0]
        scored = [
            (cosine(q, v), t)
            for v, t in zip(
                self.vectors, self.texts
            )
        ]
        scored.sort(key=lambda s: s[0], reverse=True)
        return scored[:k]



def build_messages(question, hits):
    context = "\n\n".join(
        f"[{i + 1}] {text}"
        for i, (_, text) in enumerate(hits)
    )
    system = (
        "Answer using only the context below. "
        "If the answer is not in the context, "
        "say you do not know.\n\n"
        "Context:\n" + context
    )
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": question},
    ]


def ask(store, question):
    hits = store.top_k(question, k=3)
    print(f"Q: {question}")
    for score, text in hits:
        print(f"  {score:.3f}  {text[:44]}...")
    messages = build_messages(question, hits)
    resp = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=messages,
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    store = VectorStore()
    store.add(DOCS)

    answer = ask(
        store,
        "Where is my data stored, and can "
        "I get it out?",
    )
    print(f"A: {answer}")

Gdzie psuje się wersja testowa

W fazie „Gdzie wersja zabawki”, zanim zmieni się kod, należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić, nie musząc czytać całej struktury. Gdy budżet na to pozwala, należy dodać test dymny, który symuluje kluczową ścieżkę w procesie CI przy użyciu narzędzi pomocniczych, a nie rzeczywistych, płatnych API.

Lista kontrolna operacyjna

W fazie „Lista kontrolna operacyjna”, zanim zmieni się kod, należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.

Rozpatruj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania.

Dodaj test sprawdzający podstawową funkcjonalność w środowisku CI, wykorzystujący przygotowane przykłady danych, a nie rzeczywiste, płatne API, o ile pozwala na to budżet.

Wolij małe, łatwe do przetestowania jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces.

Zdefiniuj konkretne wersje zależności i zapisz identyfikator obrazu, który został użyty do uruchomienia demonstracji. Reprodukowalność jest ważniejsza od lokalnej wiedzy zespołu.

Zdokumentuj zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownego wykonania, kontrola przez ludzi oraz obsługa błędów stanowią część produktu, a nie elementy dodawane później.

Zanim wdrożysz tę architekturę, zamroź wersje, utwórz dokładny zapis dla kluczowych etapów realizacji oraz potwierdź kroki odwracania zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji przynależności użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację haseł. Wolimy nudną niezawodność od pomysłowych, jednorazowych demonstracji.

Uwaga dotycząca 68e889b14c39: unikaj przechowywania kluczy dostawcy w repozytorium, ustaw ograniczenie liczby tokenów na sesję oraz przechowuj zapisy obok plików testowych, aby późniejsze zmiany modeli pozostawały porównywalne.