Wskazówki praktyczne: Czyszczenie danych i przetwarzanie wstępne dla RAG produkcyjnego: Budowanie

Niech to służy jako wersja przeznaczona dla operatorów, zawierająca zasady przedstawione w „Data Cleaning and Preprocessing for Production RAG: Building Retrieval-Ready Documents”: wyraźne etapy, uporządkowane sekcje kodu oraz notatki dotyczące napraw, które przetrwają przeniesienie obowiązków. Etap Przeglądu działa najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Wolimy 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.

CONFIDENTIAL - INTERNAL USE ONLY
Group Financial Crime Compliance
Page 47 of 132

Gdzie mieści się czyszczenie danych w pipeline RAG

W etapie „Gdzie odbywa się czyszczenie danych” należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten 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 ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij powstałe pliki, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wskazuj fragmenty tekstu, które faktycznie stanowią podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.

Normalizacja kodowania: Popraw tekst przed jego analizą

Aby naprawić problem normalizacji kodowania, należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić tę czynność na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować 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ć fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie mogą odróżnić halucynacji od luki w indeksowaniu.

Customer’s identity must be verified before account opening.
The customer must provide proof of address.
Enhanced Due Diligence â€" High Risk Customers

Naprawa uszkodzonego Unicode za pomocą ftfy

Dla procesu naprawy uszkodzonego Unicode 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 od 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ć bez konieczności przeglądania całej struktury. Należy podawać konkretne fragmenty tekstu, na których opiera się odpowiedź. Bez tych odniesień operatorzy nie są w stanie odróżnić halucynacji od braku danych w indeksie. Dla procesu naprawy uszkodzonego Unicode 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 od 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, powinien wskazywać na konkretną odpowiedzialność, a nie na całą strukturę.

splątana ścieżka przetwarzania.

import re
import unicodedata
from dataclasses import dataclass, field
import ftfy

@dataclass
class TextCleaningResult:
    original_text: str
    cleaned_text: str
    transformations: list[str] = field(default_factory=list)
    quality_flags: list[str] = field(default_factory=list)
def normalize_encoding(text: str) -> TextCleaningResult:
    """
    Repair common encoding problems while preserving meaningful Unicode.
    Suitable for policy, AML/KYC, payment, and regulatory documents where
    accented names, currency symbols, and multilingual text must survive.
    """
    transformations: list[str] = []
    quality_flags: list[str] = []
    if not text:
        return TextCleaningResult(original_text=text, cleaned_text=text)
    cleaned = text
    # Repair mojibake and common Unicode encoding mistakes.
    repaired = ftfy.fix_text(cleaned)
    if repaired != cleaned:
        transformations.append("ftfy_encoding_repair")
        cleaned = repaired
    # NFC preserves characters while producing a consistent Unicode
    # representation. Safer than aggressively stripping accents.
    normalized = unicodedata.normalize("NFC", cleaned)
    if normalized != cleaned:
        transformations.append("unicode_nfc_normalization")
        cleaned = normalized
    # Replace non-breaking spaces with normal spaces.
    if "\u00a0" in cleaned:
        cleaned = cleaned.replace("\u00a0", " ")
        transformations.append("non_breaking_space_normalization")
    # Remove zero-width characters that frequently leak from PDFs,
    # web pages, and copied Office content.
    zero_width_chars = {"\u200b", "\u200c", "\u200d", "\ufeff"}
    if any(char in cleaned for char in zero_width_chars):
        cleaned = "".join(char for char in cleaned if char not in zero_width_chars)
        transformations.append("zero_width_character_removal")
    # Remove control characters; preserve newline and tab because
    # they may still carry document structure needed downstream.
    cleaned_without_controls = "".join(
        char for char in cleaned
        if char in "\n\t" or unicodedata.category(char) != "Cc"
    )
    if cleaned_without_controls != cleaned:
        transformations.append("control_character_removal")
        cleaned = cleaned_without_controls
    # Normalize horizontal whitespace without flattening paragraphs.
    whitespace_normalized = re.sub(r"[ \t]+", " ", cleaned)
    whitespace_normalized = re.sub(r"\n{3,}", "\n\n", whitespace_normalized)
    if whitespace_normalized != cleaned:
        transformations.append("whitespace_normalization")
        cleaned = whitespace_normalized
    cleaned = cleaned.strip()
    # Keep suspicious replacement characters observable rather than silently deleting them.
    if "\ufffd" in cleaned:
        quality_flags.append("unicode_replacement_character_detected")
    return TextCleaningResult(
        original_text=text,
        cleaned_text=cleaned,
        transformations=transformations,
        quality_flags=quality_flags,
    )

Dlaczego agresywne czyszczenie tekstu szkodzi RAG

Podczas prace nad etapem „Dlaczego agresywne czyszczenie tekstu szkodzi RAG”, 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ń odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.

Customer must NOT be classified as low risk.

Usuwanie szablonów: z uwzględnieniem struktury, a nie słów kluczowych

Gdy przechodzisz przez etap Structure-Aware Not w ramach usuwania Boilerplate, 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. Zapisz czas wykonywania oraz koszt tokena lub zapytania obok wyników funkcjonalnych. 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ą skuteczność wyszukiwania.

KYC Policy | Version 7.2 | Page 41 of 132
KYC Policy | Version 7.2 | Page 42 of 132

import re
from collections import Counter
from dataclasses import dataclass

@dataclass
class BoilerplatePattern:
    normalized_text: str
    occurrences: int
    page_ratio: float
    position: str
def normalize_boilerplate_candidate(line: str) -> str:
    """
    Normalize variable fields so structurally identical headers and
    footers can be compared across pages.
    """
    normalized = line.strip()
    normalized = re.sub(
        r"\bpage\s+\d+\s+of\s+\d+\b", "page <n> of <n>",
        normalized, flags=re.IGNORECASE,
    )
    normalized = re.sub(r"\bpage\s+\d+\b", "page <n>", normalized, flags=re.IGNORECASE)
    normalized = re.sub(r"\s+", " ", normalized)
    return normalized.casefold()
def detect_repeated_marginal_text(
    pages: list[dict],
    margin_lines: int = 3,
    min_page_ratio: float = 0.6,
) -> list[BoilerplatePattern]:
    """
    Detect repeated text in page-header or page-footer regions.
    A candidate is considered boilerplate only when it appears in the same
    marginal position across a substantial fraction of the document.
    """
    if not pages:
        return []
    header_counts: Counter = Counter()
    footer_counts: Counter = Counter()
    for page in pages:
        lines = [line.strip() for line in page["text"].splitlines() if line.strip()]
        if not lines:
            continue
        header_counts.update(normalize_boilerplate_candidate(l) for l in lines[:margin_lines])
        footer_counts.update(normalize_boilerplate_candidate(l) for l in lines[-margin_lines:])
    total_pages = len(pages)
    patterns: list[BoilerplatePattern] = []
    for position, counts in (("header", header_counts), ("footer", footer_counts)):
        for normalized_text, occurrences in counts.items():
            page_ratio = occurrences / total_pages
            if page_ratio >= min_page_ratio:
                patterns.append(BoilerplatePattern(
                    normalized_text=normalized_text,
                    occurrences=occurrences,
                    page_ratio=page_ratio,
                    position=position,
                ))
    return patterns

Deduplikacja: Dokładne dopasowania to najłatwiejsza część

Gdy przechodzisz przez etap Deduplication Exact Matches Are, 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. Trzymaj 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 stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania. Gdy przechodzisz przez etap Deduplication Exact Matches Are, 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 skomplikowaną sekwencję operacji.

AML_Policy_v8_Final.docx
AML_Policy_v8_Final_Approved.docx
AML_Policy_v8_Final_Copy.docx
AML_Policy_v8_Approved_2026.docx

Dokładna eliminacja duplikatów za pomocą haszowania treści

Etap dokładnej eliminacji duplikatów działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań przed rozszerzeniem zakresu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy artefaktom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Oddziel politykę dzielenia na fragmenty od polityki wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.

import hashlib
from dataclasses import dataclass

@dataclass
class DeduplicationRecord:
    document_id: str
    source: str
    content_hash: str
    duplicate_of: str | None = None
def canonicalize_for_hashing(text: str) -> str:
    """
    Deterministic normalization before hashing.
    Does not lowercase or remove punctuation because those transformations
    could collapse documents that are not actually identical.
    """
    lines = [" ".join(line.split()) for line in text.splitlines()]
    return "\n".join(lines).strip()
def calculate_content_hash(text: str) -> str:
    canonical_text = canonicalize_for_hashing(text)
    return hashlib.sha256(canonical_text.encode("utf-8")).hexdigest()
def find_exact_duplicates(
    documents: list[dict],
) -> tuple[list[dict], list[DeduplicationRecord]]:
    """
    Keep one canonical copy of each identical document while preserving
    duplicate lineage for auditability.
    """
    seen_hashes: dict[str, str] = {}
    unique_documents: list[dict] = []
    records: list[DeduplicationRecord] = []
    for document in documents:
        content_hash = calculate_content_hash(document["clean_text"])
        if content_hash in seen_hashes:
            records.append(DeduplicationRecord(
                document_id=document["document_id"],
                source=document["source"],
                content_hash=content_hash,
                duplicate_of=seen_hashes[content_hash],
            ))
            continue
        seen_hashes[content_hash] = document["document_id"]
        unique_documents.append(document)
        records.append(DeduplicationRecord(
            document_id=document["document_id"],
            source=document["source"],
            content_hash=content_hash,
        ))
    return unique_documents, records

Detekcja niemal identycznych treści za pomocą MinHash LSH

Rozpoznawanie prawie identycznych treści za pomocą MinHash działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Oddziel politykę dzielenia na fragmenty od polityki wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej, gdy zmieniają się metryki jakości.

import re
from datasketch import MinHash, MinHashLSH

def create_word_shingles(text: str, shingle_size: int = 5) -> set[str]:
    """
    Five-word shingles work well for long policy and regulatory documents
    because they capture local textual structure without being overly
    sensitive to isolated formatting changes.
    """
    tokens = re.findall(r"\b\w+\b", text.casefold())
    if len(tokens) < shingle_size:
        return {" ".join(tokens)} if tokens else set()
    return {
        " ".join(tokens[i:i + shingle_size])
        for i in range(len(tokens) - shingle_size + 1)
    }
def create_minhash(text: str, num_perm: int = 128) -> MinHash:
    shingles = create_word_shingles(text)
    minhash = MinHash(num_perm=num_perm)
    for shingle in shingles:
        minhash.update(shingle.encode("utf-8"))
    return minhash
def build_duplicate_index(
    documents: list[dict],
    threshold: float = 0.85,
    num_perm: int = 128,
) -> tuple[MinHashLSH, dict[str, MinHash]]:
    """
    Build an LSH index for candidate near-duplicate discovery.
    The threshold identifies candidates. It does not automatically
    determine whether a document is deleted.
    """
    lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
    signatures: dict[str, MinHash] = {}
    for document in documents:
        document_id = document["document_id"]
        signature = create_minhash(document["clean_text"], num_perm=num_perm)
        signatures[document_id] = signature
        lsh.insert(document_id, signature)
    return lsh, signatures

Korekta błędów OCR: Gdy tekst wygląda poprawnie, ale jest błędny

Faza korekty błędów OCR funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny transkrypt, jeden przypadek awarii oraz notatkę o cofnięciu zmian przed rozszerzaniem zakresu. 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. Oddziel zasadę dzielenia na fragmenty od zasady wyszukiwania. Zmiana jednej z nich nie powinna zmuszać do przepisywania drugiej wraz ze zmianami wskaźników jakości. Faza korekty błędów OCR funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny transkrypt, jeden przypadek awarii oraz notatkę o cofnięciu zmian przed rozszerzaniem zakresu. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną ścieżkę przetwarzania.

import re
from dataclasses import dataclass

@dataclass
class OCRIssue:
    token: str
    issue_type: str
    risk_level: str
    suggested_value: str | None
    requires_review: bool
CURRENCY_PATTERN = re.compile(r"\b(EUR|USD|GBP|INR)\s+([A-Za-z0-9.,]+)\b")
PERCENTAGE_PATTERN = re.compile(r"\b([A-Za-z0-9.,]+)\s*%")
def detect_numeric_ocr_issues(text: str) -> list[OCRIssue]:
    """
    Detect suspicious OCR substitutions inside financial values.
    This function identifies candidates. It does not silently rewrite
    high-risk values in the source text.
    """
    issues: list[OCRIssue] = []
    substitutions = {"O": "0", "o": "0", "I": "1", "l": "1"}
    def inspect_numeric_token(token: str, issue_type: str) -> None:
        if token.replace(",", "").replace(".", "").isdigit():
            return
        corrected = "".join(substitutions.get(char, char) for char in token)
        numeric_candidate = corrected.replace(",", "").replace(".", "")
        if numeric_candidate.isdigit() and corrected != token:
            issues.append(OCRIssue(
                token=token,
                issue_type=issue_type,
                risk_level="high",
                suggested_value=corrected,
                requires_review=True,
            ))
    for match in CURRENCY_PATTERN.finditer(text):
        inspect_numeric_token(match.group(2), issue_type="currency_value")
    for match in PERCENTAGE_PATTERN.finditer(text):
        inspect_numeric_token(match.group(1), issue_type="percentage")
    return issues

Multijęzyczne przetwarzanie wstępne: jeden dokument, wiele języków

W fazie multijęzycznego przetwarzania wstępnego dla jednego dokumentu należy zdefiniować dane wejściowe, osobę odpowiedzialną za tę czynność oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić tę czynność 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 powstałe pliki, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymień fragmenty tekstu, które faktycznie stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od luki w indeksowaniu.

from dataclasses import dataclass
from langdetect import DetectorFactory, LangDetectException, detect_langs

DetectorFactory.seed = 0  # Makes pipeline output reproducible.
@dataclass
class LanguagePrediction:
    language: str
    confidence: float
    alternatives: list[tuple[str, float]]
    needs_review: bool
def detect_document_language(
    text: str,
    confidence_threshold: float = 0.85,
) -> LanguagePrediction:
    sample = text.strip()
    if not sample:
        return LanguagePrediction(language="unknown", confidence=0.0, alternatives=[], needs_review=True)
    try:
        predictions = detect_langs(sample)
    except LangDetectException:
        return LanguagePrediction(language="unknown", confidence=0.0, alternatives=[], needs_review=True)
    best = predictions[0]
    alternatives = [(p.lang, round(p.prob, 4)) for p in predictions]
    return LanguagePrediction(
        language=best.lang,
        confidence=best.prob,
        alternatives=alternatives,
        needs_review=best.prob < confidence_threshold,
    )

metadata = {
    "primary_language": "en",
    "languages": ["en", "de", "fr"],
    "language_distribution": {"en": 0.72, "de": 0.21, "fr": 0.07},
    "is_multilingual": True,
}

Dane osobowe i dane wrażliwe: wykrywanie przed embedowaniem

W etapie przetwarzania danych osobowych i wrażliwych 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 systemu. Należy rejestrować 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. 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.

from dataclasses import dataclass, field
from presidio_analyzer import AnalyzerEngine

@dataclass
class SensitiveEntity:
    entity_type: str
    start: int
    end: int
    score: float
    original_value: str
@dataclass
class PIIScanResult:
    entities: list[SensitiveEntity] = field(default_factory=list)
    needs_review: bool = False
class BankingPIIDetector:
    def __init__(self) -> None:
        self.analyzer = AnalyzerEngine()
    def detect(
        self,
        text: str,
        language: str = "en",
        score_threshold: float = 0.70,
    ) -> PIIScanResult:
        """
        Detect sensitive entities without modifying the source text.
        Detection is deliberately separated from anonymization because
        different entity types require different handling policies.
        """
        results = self.analyzer.analyze(
            text=text, language=language, score_threshold=score_threshold
        )
        entities = [
            SensitiveEntity(
                entity_type=r.entity_type,
                start=r.start,
                end=r.end,
                score=r.score,
                original_value=text[r.start:r.end],
            )
            for r in results
        ]
        high_risk_types = {"CREDIT_CARD", "IBAN_CODE", "US_SSN", "IP_ADDRESS"}
        return PIIScanResult(
            entities=entities,
            needs_review=any(e.entity_type in high_risk_types for e in entities),
        )

Customer IBAN: <IBAN>
Customer Email: <EMAIL_ADDRESS>

Całość w jedności: pipeline czyszczenia na poziomie produkcyjnym

W fazie Putting It Together A 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 od 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ć bez konieczności czytania całej struktury. Należy podawać konkretne fragmenty tekstu, na których opiera się odpowiedź. Bez tych odniesień operatorzy nie są w stanie odróżnić halucynacji od braku danych w indeksie. W fazie Putting It Together A 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 od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu danych.

e.

from dataclasses import dataclass, field
from enum import Enum
from typing import Any

class QualityStatus(str, Enum):
    PASS = "pass"
    REVIEW = "review"
    REJECT = "reject"
@dataclass
class Transformation:
    stage: str
    operation: str
    details: dict[str, Any] = field(default_factory=dict)
@dataclass
class QualityFlag:
    code: str
    severity: str
    details: dict[str, Any] = field(default_factory=dict)
@dataclass
class RetrievalReadyDocument:
    document_id: str
    source: str
    raw_text: str
    clean_text: str
    language: str | None = None
    language_confidence: float | None = None
    content_hash: str | None = None
    duplicate_of: str | None = None
    transformations: list[Transformation] = field(default_factory=list)
    quality_flags: list[QualityFlag] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    status: QualityStatus = QualityStatus.PASS

import logging
logger = logging.getLogger(__name__)

class ProductionDocumentCleaner:
    def __init__(
        self,
        pii_detector: BankingPIIDetector,
        pii_anonymizer,
        redact_pii_for_embeddings: bool = True,
    ) -> None:
        self.pii_detector = pii_detector
        self.pii_anonymizer = pii_anonymizer
        self.redact_pii_for_embeddings = redact_pii_for_embeddings
    def clean(self, document: dict) -> RetrievalReadyDocument:
        result = RetrievalReadyDocument(
            document_id=document["document_id"],
            source=document["source"],
            raw_text=document["text"],
            clean_text=document["text"],
            metadata=document.get("metadata", {}).copy(),
        )
        try:
            self._normalize_text(result)
            self._detect_language(result)
            self._validate_ocr(result)
            self._handle_sensitive_data(result)
            self._calculate_hash(result)
            self._apply_quality_gate(result)
        except Exception as exc:
            logger.exception("Preprocessing failed for document %s", result.document_id)
            result.quality_flags.append(QualityFlag(
                code="PREPROCESSING_EXCEPTION",
                severity="critical",
                details={"error": str(exc)},
            ))
            result.status = QualityStatus.REJECT
        return result
    def _apply_quality_gate(self, document: RetrievalReadyDocument) -> None:
        severities = {flag.severity for flag in document.quality_flags}
        if "critical" in severities:
            document.status = QualityStatus.REJECT
        elif "high" in severities:
            document.status = QualityStatus.REVIEW
        else:
            document.status = QualityStatus.PASS

Kompromisy w produkcji

Podczas przechodzenia przez etap Kompromisów w produkcji 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 poprawiają słabą skuteczność wyszukiwania.

Co to oznacza w praktyce

Gdy przechodzisz przez etap „Co to oznacza”, 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 nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. 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.

Gdy przechodzisz przez etap A Production Checklist Before, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Ta lista sprawia, że późniejsze zmiany w kodzie są przejrzyste. Trzymaj 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 serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe mechanizmy wyszukiwania. Gdy przechodzisz przez etap A Production Checklist Before, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Ta lista sprawia, że późniejsze zmiany w kodzie są przejrzyste. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na jedną konkretne odpowiedzialność, a nie na skomplikowaną ścieżkę przetwarzania.

Lista kontrolna operacyjna

Gdy przechodzisz przez etap listy kontrolnej operacyjnej, 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ę przywracania stanu. Próby ponowne, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później w celu udoskonalenia.

Zmierz stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania.

Zamroź złoty zestaw parametrów przed modyfikacją promptów lub modeli. Zmiana zarówno systemu, jak i kryteriów oceny ukrywa regresje.

Gdy budżet na to pozwala, dodaj test dymny, który symuluje kluczową ścieżkę działania w procesie CI przy użyciu fixitów, a nie rzeczywistych płatnych API.

Zapisuj czas trwania oraz koszt tokena lub zapytania obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk udostępnionych.

Zanim przesuniesz całą architekturę do produkcji, zamroź wersje, utwórz dokładny zapis dla kluczowych ścieżek oraz potwierdź kroki odwracania zmian. Środowiska udostępnione wymagają ograniczeń szybkości, weryfikacji uprawnień użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację haseł. Wolisz nudną niezawodność od pomysłowych, jednorazowych demonstracji.

Uwaga dotycząca af00655fb2bc: 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.

Notatka dotycząca wzmocnienia bezpieczeństwa na etapie 0 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres prac. 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.

Szczegóły wzmocnienia bezpieczeństwa 0/902: zmierz czas wykonywania operacji, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie informacji anegdotycznych.

Dla etapu 1 notatki dotyczącej wzmocnienia bezpieczeństwa zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie wykonać ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu systemu. Zdokumentuj zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, mechanizmy kontroli ludzkiej oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.

Szczegół wzmocnienia 1/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Podczas przechodzenia przez drugi etap notatki dotyczącej wzmocnienia, 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.

Szczegół wzmocnienia 2/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Metoda zaostrzania bezpieczeństwa na etapie 3 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian przed rozszerzeniem zakresu prac. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny haseł oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności analizy całej struktury.

Szczegół zaostrzania bezpieczeństwa 3/902: zmierz czas wykonywania operacji, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę, opierając się na ustalonej serii pytań, a nie na indywidualnych obserwacjach.

Dla etapu 4 notatki dotyczącej wzmocnienia bezpieczeństwa 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, testowalne jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces.

Szczegół 4/902 dotyczący wzmocnienia bezpieczeństwa: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę, opierając się na ustalonej serii pytań, a nie na indywidualnych obserwacjach.

Gdy przechodzisz przez etap nr 5 notatki dotyczącej wzmocnienia bezpieczeństwa, najpierw zapisz warunki kontraktu: 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. Jasna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk.

Szczegóły wzmocnienia bezpieczeństwa 5/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie osobistych obserwacji.

Etap nr 6 notatki dotyczącej wzmocnienia bezpieczeństwa działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię do analizy. Zanim rozszerzysz zakres, zapisz jeden idealny przepływ działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Zdokumentuj razem ścieżkę prawidłowego działania oraz ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.

Szczegół wzmocnienia 6/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

W fazie 7 notatki dotyczącej wzmocnienia określ dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie wykonać 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. Nadaj nazwy artefaktom, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania.

Szczegół wzmocnienia 7/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie anegdoty.

Gdy przechodzisz przez etap nr 8 notatki dotyczącej wzmocnienia bezpieczeństwa, najpierw zapisz warunki umowy: 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. 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.

Szczegóły wzmocnienia bezpieczeństwa 8/902: zmierz czas wykonywania, klasę błędu oraz zużycie tokenów dla tej notatki, a następnie zdecyduj, czy zachować zmianę na podstawie ustalonego zestawu pytań, a nie jedynie informacji anegdotycznych.