Notatki praktyczne: Budowa lokalnego pipeline RAG klasy produkcyjnej — 100% darmowy, bez
Szczegółowy przewodnik po Notatkach praktycznych: Budowa lokalnego pipeline RAG klasy produkcyjnej — 100% darmowy, bez umów, weryfikacji oraz gotowych miejsc na kod dla zespołów stosujących ten model.
To przewodnik pokazuje, jak odtworzyć proces od surowców do działającego systemu w celu stworzenia lokalnego pipeline RAG o standardach produkcyjnych — 100% bezpłatnie, bez konieczności korzystania z chmury. Skupiamy się na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu umieścić w repozytorium, bez konieczności domyślania się intencji. Na etapie przeglądu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności domyślania się ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
Dlaczego to ma znaczenie
Gdy przechodzisz przez etap „Dlaczego to ma znaczenie”, 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. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
Pełny zestaw technologiczny
Gdy przechodzisz przez etap The Complete Tech Stack, 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. Traktuj 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. Zmierz stopień pamięci na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Wymagania wstępne
Podczas przechodzenia przez etap Wstępne wymagania, 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 czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zmierz dokładność odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania. Podczas przechodzenia przez etap Wstępne wymagania, 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. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dopiero późniejszej optymalizacji.
Część 1 — Konfiguracja środowiska
Etap ustawiania środowiska w Części 1 działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
Krok 1: Stwórz projekt
Krok 1 – tworzenie scenariusza działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przykład realizacji, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Traktuj 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 zadań. Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
mkdir local-rag
cd local-rag
uv init
Krok 2: Stworzenie i aktywacja środowiska wirtualnego
Krok 2 – tworzenie i przygotowanie do użycia – działa najlepiej, gdy traktuje się go jako coś mierzalnego. Zanim rozszerzy się zakres, należy zarejestrować jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Obok wyników funkcjonalnych należy zapisywać czas trwania operacji oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy oddzielić zasadę dzielenia na fragmenty od zasady wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
uv venv
.venv\Scripts\activate
Krok 2 – tworzenie i przygotowanie do użycia – działa najlepiej, gdy traktuje się go jako coś mierzalnego. Zanim rozszerzy się zakres, należy zarejestrować jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Należy razem udokumentować ścieżkę prawidłowego działania oraz ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
Krok 3: Zainstaluj wszystkie zależności
W trzecim kroku, przed zmianą kodu, należy zainstalować wszystkie niezbędne elementy, określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę procesu. Należy podawać konkretne fragmenty tekstu, które stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić fałszywych informacji od braków w indeksowaniu.
uv add google-genai pypdf chromadb rich python-dotenv huggingface_hub fpdf2
Krok 4: Stworzenie struktury projektu
W kroku 4 „Stworzenie etapu” należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Podaj fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą mogli odróżnić halucynacji od luki w indeksowaniu.
mkdir pdfs
mkdir pdfs\versions
type nul > local_rag.ipynb
type nul > .env
type nul > .gitignore
local-rag/
├── .venv/ ← virtual environment (never commit)
├── pdfs/ ← drop your PDFs here
│ └── versions/ ← test PDFs for CDC testing
├── chroma_db/ ← auto-created on first ingest
├── memory_checkpoints/ ← auto-created on first memory session
├── staleness_registry.json ← auto-created
├── chunk_registry.json ← auto-created
├── local_rag.ipynb ← your notebook
├── .env ← API keys (never commit)
├── .gitignore
└── pyproject.toml
Krok 5: Konfiguracja kluczy API
W etapie Krok 5 – Konfiguracja API należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy odnotować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy podawać konkretne fragmenty tekstu, które stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie mogą odróżnić efektu halucynacji od luki w indeksowaniu. W etapie Krok 5 – Konfiguracja API należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżki naprawcze. Próby ponownego wykonania, kontrole ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
GEMINI_API_KEY=your_gemini_key_here
HF_API_KEY=your_huggingface_token_here
.env
chroma_db/
memory_checkpoints/
staleness_registry.json
chunk_registry.json
__pycache__/
.venv/
*.pyc
Krok 6: Konfiguracja VS Code
Podczas pracy nad krokiem 6, najpierw zapisz specyfikację: 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. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Ctrl+Shift+P → Python: Select Interpreter → .venv\Scripts\python.exe
Część 2 — Szczegółowe omówienie głównego procesu
Gdy przechodzisz przez etap Part 2 Core Pipeline, 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. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Zmierz stopień przywoływalności na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Komórka 1 — Zależności (w notatniku)
Gdy pracujesz nad zależnościami komórki 1 w ramach danej fazy, 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. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zmierz stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabą efektywność wyszukiwania.
# Run once inside the notebook if uv add was not used externally
# %pip install google-genai pypdf chromadb rich python-dotenv huggingface_hub fpdf2
Gdy pracujesz nad zależnościami komórki 1 w ramach danej fazy, 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. Dokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne etapy przeprowadzane przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
Komórka 2 — Konfiguracja
Etap konfiguracji Komórki 2 działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przekaz, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań przed rozszerzaniem zakresu. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Oddziel zasadę dzielenia na fragmenty od zasady pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej w momencie zmian wskaźników jakości.
import os
from dotenv import load_dotenv
load_dotenv()
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
EMBED_DIM = 768
HF_API_KEY = os.environ.get("HF_API_KEY", "")
GEMINI_EMBED_MODEL = "gemini-embedding-001"
HF_LLM_MODEL = "openai/gpt-oss-20b:groq"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "local_rag"
CHUNK_SIZE = 800
CHUNK_OVERLAP = 120
TOP_K = 5
EMBED_BATCH_SIZE = 50
BATCH_SLEEP_SEC = 0.3
LLM_MAX_NEW_TOKENS = 1024
LLM_TEMPERATURE = 0.1
assert GEMINI_API_KEY, "❌ GEMINI_API_KEY not set"
assert HF_API_KEY, "❌ HF_API_KEY not set"
Komórka 3 — Wydobywanie tekstu z PDF
Etap tekstowy PDF w Cell 3 funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis, jeden przypadek niepowodzenia oraz notatkę o cofnięciu zmian przed rozszerzaniem zakresu. Traktuj 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 pracy. Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
from pypdf import PdfReader
def extract_text_from_pdf(pdf_path: str) -> tuple[str, int]:
reader = PdfReader(pdf_path)
pages = []
for i, page in enumerate(reader.pages):
text = page.extract_text()
if text and text.strip():
pages.append(f"[Page {i + 1}]\n{text.strip()}")
return "\n\n".join(pages), len(reader.pages)
Cell 4 — Dzielenie na fragmenty metodą okna przesuwającego się
Etap Cell 4 Sliding Window funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Rozdziel politykę dzielenia na fragmenty od polityki wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.
def chunk_text(text: str) -> list[str]:
chunks, start = [], 0
while start < len(text):
end = start + CHUNK_SIZE
chunk = text[start:end].strip()
if len(chunk) >= 80:
chunks.append(chunk)
start += CHUNK_SIZE - CHUNK_OVERLAP
return chunks
Etap Cell 4 Sliding Window funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zdokumentuj zarówno pomyślną ścieżkę działania, jak i ścieżkę przywracania. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
Cell 5 — Gemini Embeddings
W etapie Cell 5 Gemini Embeddings należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki nad rozbudowanymi skryptami. Gdy jakiś krok zawiedzie, powinno to wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu. Należy oddzielić budowę klienta od pętli przekazywania wiadomości, aby można było wymieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy.
from google import genai
from google.genai import types
genai_client = genai.Client(api_key=GEMINI_API_KEY)
def embed_documents_batch(chunks: list[str]) -> list[list[float]]:
all_embeddings = []
for i, chunk in enumerate(chunks, 1):
result = genai_client.models.embed_content(
model=GEMINI_EMBED_MODEL,
contents=chunk,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBED_DIM,
),
)
all_embeddings.append(result.embeddings[0].values)
return all_embeddings
def embed_query(text: str) -> list[float]:
result = genai_client.models.embed_content(
model=GEMINI_EMBED_MODEL,
contents=text,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_QUERY",
output_dimensionality=EMBED_DIM,
),
)
return result.embeddings[0].values
Cell 6 — Przechowywanie i odzyskiwanie danych w ChromaDB
Dla etapu przechowywania w ChromaDB w komórce 6 należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij pliki artefaktów, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymień fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
import uuid, chromadb
def get_collection():
client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
return client.get_or_create_collection(
name=COLLECTION_NAME,
metadata={"hnsw:space": "cosine"},
)
def store_in_chroma(chunks, embeddings, doc_name):
collection = get_collection()
ids = [str(uuid.uuid4()) for _ in chunks]
metadatas = [{"source": doc_name, "chunk_index": i}
for i in range(len(chunks))]
collection.add(ids=ids, embeddings=embeddings,
documents=chunks, metadatas=metadatas)
return len(chunks)
def retrieve_context(query: str) -> list[dict]:
collection = get_collection()
query_embedding = embed_query(query)
results = collection.query(
query_embeddings=[query_embedding],
n_results=TOP_K,
include=["documents", "metadatas", "distances"],
)
chunks = []
for doc, meta, dist in zip(results["documents"][0],
results["metadatas"][0],
results["distances"][0]):
chunks.append({
"text": doc,
"source": meta.get("source", "unknown"),
"chunk_index": meta.get("chunk_index", -1),
"score": round(1 - dist, 4),
})
return sorted(chunks, key=lambda x: x["score"], reverse=True)
Komórka 7 — Uruchamianie procesu pobierania danych
Dla etapu Cell 7 Running Ingestion należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu systemu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy podawać konkretne fragmenty tekstu, na których opiera się odpowiedź. Bez tych odniesień operatorzy nie są w stanie odróżnić efektu halucynacji od luki w indeksowaniu.
PDF_PATH = "./pdfs/attention.pdf"
raw_text, page_count = extract_text_from_pdf(PDF_PATH)
chunks = chunk_text(raw_text)
embeddings = embed_documents_batch(chunks)
stored = store_in_chroma(chunks, embeddings,
os.path.basename(PDF_PATH))
W fazie pobierania danych dla komórki 7 należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
Komórka 8 — LLM: gpt-oss-20b przez HuggingFace
Gdy pracujesz nad etapem Cell 8 LLM gpt-oss-20b, 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. Wolij małe, testowalne jednostki od rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Zachowuj w pamięci tymczasowej stabilne instrukcje systemu oraz schematy narzędzi. Ponowne wysyłanie identycznego wstępu to częsty powód marnotrawstwa zasobów.
from huggingface_hub import InferenceClient
hf_client = InferenceClient(api_key=HF_API_KEY)
def build_messages(query: str, context_chunks: list[dict]) -> list[dict]:
context_str = "\n\n---\n\n".join([
f"[Source: {c['source']} | Chunk #{c['chunk_index']} | "
f"Relevance: {c['score']}]\n{c['text']}"
for c in context_chunks
])
return [
{"role": "system", "content": SYSTEM_MSG},
{"role": "user", "content":
f"CONTEXT:\n{context_str}\n\nQUESTION:\n{query}"},
]
def generate_answer(messages: list[dict]) -> str:
completion = hf_client.chat.completions.create(
model=HF_LLM_MODEL,
messages=messages,
max_tokens=LLM_MAX_NEW_TOKENS,
temperature=LLM_TEMPERATURE,
)
return completion.choices[0].message.content.strip()
Cell 9–11 — Pipeline ask() i REPL
Gdy pracujesz nad etapem Cell 9 11 The stage, 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. Traktuj 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. Zmierz stopień pamięci na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
query → embed_query() → ChromaDB cosine search → top-5 chunks
→ build_messages() → generate_answer() → printed answer
1. attention.pdf chunk #34 [██████████████████████░░░░░░░░] 0.7335
Część 3 — Pamięć konwersacji
Gdy przechodzisz przez etap Pamięci Rozmów w Części 3, 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 czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zmierz zdolność przywoływania informacji na podstawie ustalonego zestawu pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
Problem
Gdy przechodzisz przez etap definiowania problemu, 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. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zmierz stopę odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Rozwiązanie: ConversationMemory
Gdy pracujesz nad etapem ConversationMemory, 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. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dopinane później. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
class ConversationMemory:
def __init__(self, session_id: str = None):
self.session_id = session_id or datetime.now().strftime("%Y%m%d_%H%M%S")
self.filepath = os.path.join(MEMORY_DIR, f"{self.session_id}.json")
self.history = []
# Auto-loads if resuming an existing session
if os.path.exists(self.filepath):
self._load()
def add_turn(self, question: str, answer: str, chunks: list[dict]):
# Append user + assistant turns, checkpoint immediately
...
self._save()
def get_messages_with_history(self, query, context_chunks, system_msg):
# Injects last 6 Q&A pairs into the message list before the current turn
...
# New session
memory = ConversationMemory()
# Resume yesterday's session
memory = ConversationMemory("20260413_104959")
Turn 4 question: "How does that compare to what you said about the BLEU score?"
Turn 4 answer: "The context also reports a BLEU score of 28.4 for the
Transformer (big) on WMT 2014 English-to-German. This matches
exactly what I previously stated."
Część 4 — Śledzenie starzenia się danych, CDC i ważenie aktualności
Gdy przechodzisz przez etap śledzenia starych danych z Części 4, 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. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na jedną konkretne odpowiedzialność, a nie na skomplikowany łańcuch operacji. Zmierz stopień przywoływania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
Rzeczywisty problem
Gdy przechodzisz przez etap „Rzeczywisty problem”, 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. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Zmierz stopień przywoływalności na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.
Śledzenie przestarzałości (Komórka 14A)
Podczas pracy nad etapem Staleness Tracking Cell 14A najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Zmierz dokładność odzyskiwania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania. Podczas pracy nad etapem Staleness Tracking Cell 14A najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wysiłków, kontrola przez ludzi oraz obsługa wiadomości nieodpowiednich to elementy produktu, a nie dodatkowe ulepszenia późniejsze.
def compute_file_hash(pdf_path: str) -> str:
sha = hashlib.sha256()
with open(pdf_path, "rb") as f:
for block in iter(lambda: f.read(65536), b""):
sha.update(block)
return sha.hexdigest()
CDC Engine (Cell 14B)
Cel silnika CDC 14B funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden przykład udanego wyniku, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań przed rozszerzaniem zakresu. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej w przypadku zmian wskaźników jakości.
def compute_chunk_hash(text: str) -> str:
return hashlib.md5(text.encode("utf-8")).hexdigest()
def diff_chunks(old_registry: dict, new_chunks: list[str]) -> dict:
new_hash_map = {compute_chunk_hash(c): c for c in new_chunks}
old_hashes = set(old_registry.keys())
new_hashes = set(new_hash_map.keys())
return {
"added": {h: new_hash_map[h] for h in (new_hashes - old_hashes)},
"removed": {h: old_registry[h] for h in (old_hashes - new_hashes)},
"unchanged": {h: old_registry[h] for h in (old_hashes & new_hashes)},
}
v1 → v2 CDC result:
✅ Unchanged : 8 (kept — zero re-embedding cost)
➕ Added : 6 (embedded + inserted)
➖ Removed : 4 (deleted from ChromaDB)
💰 API calls saved: 8/14 (57% reuse)
v2 → v3 CDC result:
✅ Unchanged : 10 (kept - zero re-embedding cost)
➕ Added : 4 (embedded + inserted)
➖ Removed : 2 (deleted from ChromaDB)
💰 API calls saved: 10/14 (71% reuse)
Pobieranie danych z uwzględnieniem aktualności (Cell 14C)
Faza Recency-Weighted Retrieval Cell 14C działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden przykład udanego wyniku, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia działań przed rozszerzeniem zakresu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Oddziel zasady dzielenia na fragmenty od zasad wyszukiwania. Zmiana jednych nie powinna zmuszać do przepisania drugich, gdy zmieniają się metryki jakości.
def recency_decay(ingested_at_str: str, half_life_days: float = 30) -> float:
ingested = datetime.fromisoformat(ingested_at_str)
days_gone = (datetime.now() - ingested).total_seconds() / 86400
λ = math.log(2) / half_life_days
return round(math.exp(-λ * days_gone), 4)
blended = alpha * cosine_score + (1 - alpha) * recency_score
# Default: 0.85 * cosine + 0.15 * recency
Część 5 — Testowanie z PDF-ami wielowersjonowymi
Testowanie z użyciem etapów w Części 5 działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres testów. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Rozdziel politykę dzielenia na fragmenty od polityki wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości. Testowanie z użyciem etapów w Części 5 działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres testów. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wysiłków, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
TEST 1: Full ingest of v1
TEST 2: Staleness check — same file, correctly skipped
TEST 3: Baseline queries against v1
TEST 4: Copy v2 over active file → CDC kicks in
TEST 5: Same queries now return v2 content, newer chunks visible in recency scores
TEST 6: Copy v3 over active file → second CDC cycle
TEST 7: Recency verification — v3 chunks score highest across the board
TEST 8: Full stack test — weighted retrieval + conversation memory combined
Wyniki i zweryfikowane odpowiedzi
W fazie Wyniki i zweryfikowane odpowiedzi należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę procesu. Należy podawać fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
Podsumowanie efektywności
W fazie podsumowania efektywności należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Podaj fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.
Znane ograniczenia
W fazie Znanych Ograniczeń należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy odnotować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy podać fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie są w stanie odróżnić halucynacji od luki w indeksowaniu. W fazie Znanych Ograniczeń należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno ścieżkę pomyślnego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrole ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
To, co stworzyłeś
Podczas przechodzenia przez etap „To, co stworzyłeś”, 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. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Zmierz stopień przywoływania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania.
PDF on disk
└► SHA256 hash check (staleness)
├► Unchanged → skip
└► Changed → CDC diff
├► Unchanged chunks → kept in ChromaDB (zero API cost)
├► Removed chunks → deleted from ChromaDB
└► Added chunks → embed (Gemini) → store (ChromaDB)
↓
User question
└► embed_query() [RETRIEVAL_QUERY task type]
└► ChromaDB cosine search (TOP_K × 3 candidates)
└► recency_decay() per chunk
└► blended score re-ranking
└► top-5 chunks as context
└► ConversationMemory.get_messages_with_history()
└► gpt-oss-20b via Groq/HuggingFace
└► grounded answer + checkpoint to disk
Lista kontrolna operacyjna
Etap listy kontrolnej operacyjnej działa najlepiej, gdy jest traktowany jako mierzalna powierzchnia. Zapisz jeden idealny przykład transkrypcji, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy.
Zachowaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.
Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej w przypadku zmian wskaźników jakości.
Gdy budżet na to pozwala, dodaj test sprawdzający kluczową ścieżkę w procesie CI przy użyciu narzędzi testowych, a nie rzeczywistych płatnych API.
Zdokumentuj zarówno prawidłową ścieżkę działania, jak i ścieżkę odzyskiwania po awarii. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędowych stanowią część produktu, a nie element późniejszej optymalizacji.
Rozdziel politykę dzielenia na fragmenty od polityki pobierania danych. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej w przypadku zmian wskaźników jakości.
Zanim uruchomisz cały system, zamroź wersje, utwórz dokładny zapis dla kluczowych etapów i potwierdź kroki odwracania zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację haseł. Wolimy nudną niezawodność od pomysłowych, jednorazowych demonstracji.
Uwaga dotycząca wersji 8d172e929623: 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 pozostały porównywalne.