68e889b14c39 для виробничих систем — контракти та перевірки

Наведені нижче примітки описують практичний підхід до роботи з „“. Основна увага приділяється контрактам, перевіркам та місцям для вставки коду, а не мотиваційним аспектам. Під час роботи на етапі огляду спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ.

1. Ембеддинги перетворюють значення на геометрію

Етап формування значень за допомогою 1 Embedding працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок невдачі та примітку про скасування змін перед розширенням обсягу роботи. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Розділіть політику часткового оброблення даних від політики їх пошуку. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.

# 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. Сховище векторів — це список, який можна ранжувати

Етап зберігання векторів 2A працює найкраще, якщо його розглядати як вимірювану поверхню. Зафіксуйте один ідеальний варіант виконання, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Документуйте як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Визначте версії залежностей та зафіксуйте хеш-значення зображення, яке використовувалося під час демонстрації. Відтворюваність краща за „колективні знання“.

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. Повний процес: отримання, доповнення, генерація

Повна стадія пайплайну працює найкраще, коли її розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якась крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутаний пайплайн. Розділяйте політику часткової обробки даних та політику їх отримання. Зміна однієї з них не повинна змушувати переписувати іншу, коли змінюються показники якості. Повна стадія пайплайну працює найкраще, коли її розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ.

# 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}")

Де ламається демонстраційна версія

На етапі «Де версія іграшки» необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Конфігурацію слід тримати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевіряти, не читаючи весь код. Коли це дозволяє бюджет, слід додати тест на базову функціональність, який перевіряє критичний шлях у процесі CI за допомогою фікстур, а не реальних платних API.

Чек-лист операцій

На етапі чек-листу операцій також необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан.

Розглядайте цю стадію як контракт між вхідними даними та перевіреними результатами. Дайте назви кожним елементам, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань.

Як тільки дозволяє бюджет, додайте тест на базову функціональність, який перевіряє критичний шлях у процесі CI за допомогою фікстур, а не реальних платних API.

Віддавайте перевагу невеликим, тестованим одиницям коду перед складними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру процесу.

Фіксуйте версії залежностей та записуйте хеш-значення образу, який використовувався для демонстрації. Відтворюваність важливіша за „колективні знання“.

Документуйте як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.

Перш ніж запускати стек у продакшн, заморозьте версії, створіть «золотий» запис для критичного шляху та підтвердьте кроки відкату. У спільних середовищах необхідні обмеження швидкості, перевірки прав власності та чіткий власник для зміни секретів. Віддавайте перевагу надійності перед креативними одноразовими демонстраціями.

Примітка до 68e889b14c39: не включайте ключі постачальника до репозиторію, встановіть ліміт токенів на сеанс та зберігайте записи поруч із фікстурами для оцінки, щоб подальша заміна моделей залишалася порівнянною.