Budowa chatbota CLI

Lokalne chatboty, agenci i systemy RAG z własną bazą wiedzy

Odniesienie: M2 R2–R5   |   Czas: 25 min, opcjonalnie po przykładzie

Utwórz prosty moduł przygotowania danych. W pierwszym przejściu zapisz ten plik, ale zachowaj dostarczoną bazę. Własne dokumenty podłączysz po sprawdzeniu całej aplikacji.

Utwórz baza.py

Kod zawiera dwie ścieżki: kontrolę istniejącego JSONL oraz podział plików TXT/MD według nagłówków ## i okien znakowych. Kopiuj cały plik:

"""Odczyt zatwierdzonych fragmentów i prosty podział własnych plików TXT/MD."""
import argparse
import hashlib
import json
import re
from pathlib import Path
from lokalny import ROOT

CHUNKS = ROOT / "dane" / "chunks.jsonl"


def read_chunks(path=CHUNKS):
    records = [json.loads(line) for line in Path(path).read_text(
        encoding="utf-8-sig").splitlines() if line.strip()]
    if not records:
        raise ValueError("Baza jest pusta.")
    seen = set()
    for record in records:
        if not isinstance(record.get("chunk_id"), str) or not record["chunk_id"].strip():
            raise ValueError("Brak identyfikatora fragmentu.")
        if record["chunk_id"] in seen:
            raise ValueError("Powtórzony chunk_id: " + record["chunk_id"])
        seen.add(record["chunk_id"])
        if not isinstance(record.get("text"), str) or not record["text"].strip():
            raise ValueError("Pusty tekst fragmentu.")
        meta = record.get("metadata", {})
        for key in ("document_id", "source", "title", "version", "status"):
            if not isinstance(meta.get(key), str) or not meta[key].strip():
                raise ValueError("Brak metadanych: " + key)
        if meta["status"] != "active":
            raise ValueError("Baza zawiera dokument nieaktywny: " + meta["source"])
    return records


def split_sections(text, size=900, overlap=150):
    section = "Początek dokumentu"
    for part in re.split(r"(?m)^(## .+)$", text.replace("\r\n", "\n")):
        if part.startswith("## "):
            section = part[3:].strip()
            continue
        part = part.strip()
        start = 0
        while start < len(part):
            end = min(start + size, len(part))
            # Staraj się kończyć na granicy wyrazu, bez utraty znaków.
            if end < len(part):
                boundary = part.rfind(" ", start + size // 2, end)
                if boundary > start:
                    end = boundary
            yield section, part[start:end].strip()
            if end == len(part):
                break
            start = end - overlap


def prepare(folder, version):
    files = sorted(p for p in Path(folder).iterdir()
                   if p.is_file() and p.suffix.lower() in (".txt", ".md"))
    if not files:
        raise ValueError("Brak plików TXT/MD w wybranym folderze.")
    records = []
    for file in files:
        text = file.read_text(encoding="utf-8-sig")
        digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:10]
        doc = file.name
        for i, (section, fragment) in enumerate(split_sections(text), 1):
            if not fragment:
                continue
            records.append({"chunk_id": f"{doc}-{digest}-{i:04d}", "text": fragment,
                "metadata": {"document_id": doc, "source": file.name,
                    "title": file.stem, "version": version, "status": "active",
                    "section": section, "page_number": None}})
    if not records:
        raise ValueError("W plikach nie znaleziono treści.")
    return records


def main():
    parser = argparse.ArgumentParser()
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--folder", type=Path)
    group.add_argument("--import-m2", type=Path)
    parser.add_argument("--version", default="1.0")
    args = parser.parse_args()
    if CHUNKS.exists():
        raise FileExistsError("chunks.jsonl już istnieje. Pracuj w nowej kopii projektu "
                              "albo zmień nazwę dotychczasowego pliku na chunks_poprzednie.jsonl.")
    records = read_chunks(args.import_m2) if args.import_m2 else prepare(args.folder, args.version)
    CHUNKS.parent.mkdir(parents=True, exist_ok=True)
    with CHUNKS.open("x", encoding="utf-8") as output:
        for record in records:
            output.write(json.dumps(record, ensure_ascii=False) + "\n")
    print("Zapisano fragmentów:", len(records))
    print("Przejrzyj teksty i metadane przed budowaniem indeksu:", CHUNKS)


if __name__ == "__main__":
    main()

Jak użyć go do własnych dokumentów

1. Pracuj na kolejnej kopii projektu. Dotychczasowe dane/chunks.jsonl zachowaj pod nazwą chunks_docassist.jsonl. Jeśli istnieje index.json, zachowaj go jako index_docassist.json.
2. Utwórz folder moje_dokumenty i umieść w nim 2–5 krótkich plików TXT lub MD zapisanych w UTF-8. Użyj jednej aktualnej wersji każdego dokumentu. Program nie przeszukuje podfolderów.
3. Dodaj nagłówki ## i zachowaj kompletne procedury wraz z wyjątkami. Ustal wspólną wersję tego małego zestawu.
4. Uruchom:

.\.venv\Scripts\python.exe baza.py --folder moje_dokumenty --version 1.0

Program tworzy rekordy z limitem około 900 znaków i nakładaniem 150 znaków wewnątrz sekcji. Nie jest to podział tokenowy. Może przeciąć zdanie lub procedurę, więc przejrzyj wynik. Status active jest tu Twoją deklaracją wynikającą z doboru plików, a nie automatyczną oceną ich aktualności.

Punkt kontrolny

• Po użyciu własnych plików widzisz niepuste fragmenty i poprawne sekcje.
• Umiesz wskazać miejsce, w którym długi warunek mógł zostać przecięty.
• Po zmianie bazy planujesz budowę nowego indeksu.

Instrukcja uczestnika · Python i Ollama · wydanie 29.09.2026