Budowa chatbota CLI

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

Odniesienie: M5 R2   |   Czas: 25 min

Połącz gotowe funkcje RAG z prostym interfejsem. Każde zwykłe pytanie uruchamia nowe wyszukiwanie i generowanie.

Utwórz chatbot.py

Zapisz poniższy kod. Ten plik obsługuje wejście, komendy, drukowanie i zapis przebiegu. Logika wyszukiwania pozostaje w indeks.py, a logika odpowiedzi w rag.py.

"""Interfejs konsolowy. Komendy rozpoznaje kod; model nie wykonuje narzędzi."""
import json
import uuid
from datetime import datetime, timezone
from lokalny import ROOT, CONFIG, require_models, api
from indeks import load_index, file_hash, INDEX
from rag import ask_rag, format_sources


def main():
    records, index = load_index()
    installed = require_models([CONFIG["chat_model"]])
    session = ROOT / "wyniki" / (datetime.now().strftime("%Y%m%d_%H%M%S_") + uuid.uuid4().hex[:8])
    session.mkdir(parents=True)
    baseline = {"config": CONFIG, "index_sha256": file_hash(INDEX),
                "model_digests": {name: installed[name] for name in
                                  (CONFIG["chat_model"], CONFIG["embedding_model"])},
                "ollama_version": api("/api/version")["version"]}
    (session / "config_snapshot.json").write_text(
        json.dumps(baseline, ensure_ascii=False, indent=2), encoding="utf-8")
    history, last_sources = [], []
    print("Chatbot RAG. Pisz pełne pytania o dokumenty.")
    print("Komendy: /sources, /history, /clear, /exit. Wyniki:", session)
    while True:
        try:
            question = input("Ty: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nKoniec.")
            break
        command = question.casefold()
        if command == "/exit":
            break
        if command == "/clear":
            history.clear()
            last_sources = []
            print("Pamięć sesji wyczyszczona. Zapisane pliki pozostają na dysku.")
            continue
        if command == "/sources":
            print(format_sources(last_sources))
            continue
        if command == "/history":
            for row in history:
                print("Ty:", row["question"], "\nBot:", row["answer"])
            continue
        if question.startswith("/"):
            print("Nieznana komenda. Dostępne: /sources, /history, /clear, /exit.")
            continue
        last_sources = []  # Błąd nowej tury nie może pokazać starych źródeł.
        try:
            result = ask_rag(question, records, index)
        except KeyboardInterrupt:
            print("\nPrzerwano oczekiwanie. Koniec sesji.")
            break
        print("Bot:", result["answer"])
        print("Status:", result["status"])
        if result.get("reason"):
            print("Szczegóły:", result["reason"])
        for warning in result["warnings"]:
            print("Uwaga:", warning)
        if result["status"] == "answered":
            last_sources = result["sources"]
        print("ŹRÓDŁA PRZEKAZANE MODELOWI:\n" + format_sources(last_sources))
        # Źródła i surowa odpowiedź dotyczą tego samego wywołania RAG.
        target = session / (result["result_id"] + ".json")
        with target.open("x", encoding="utf-8") as output:
            json.dump(result, output, ensure_ascii=False, indent=2)
        if result["status"] == "answered":
            history = (history + [result])[-3:]


if __name__ == "__main__":
    try:
        main()
    except (OSError, RuntimeError, ValueError, KeyError) as exc:
        print("Nie można uruchomić chatbota:", exc)
        raise SystemExit(1)

Uruchom

.\.venv\Scripts\python.exe chatbot.py

1. Zadaj pełne pytanie: W jakich godzinach operator DocAssist przyjmuje zgłoszenia?
2. Odczytaj odpowiedź, status i źródła. Otwórz dokument i sprawdź zgodność.
3. Wpisz /sources. Powinna pojawić się ta sama lista, bez nowego wyszukiwania.
4. Zapytaj o parametr timeout_seconds, a potem ponownie wpisz /sources. Lista ma dotyczyć nowej odpowiedzi.
5. Sprawdź /history, następnie /clear i ponownie /sources. Na końcu wpisz /exit.

Pliki po sesji

W wyniki pojawia się oddzielny folder sesji. Zawiera konfigurację i po jednym pliku JSON dla każdego pytania. result_id łączy pytanie, ranking, kontekst, odpowiedź i źródła. Nie edytuj surowej odpowiedzi. Własny komentarz zapisz osobno.

Punkt kontrolny

• Program odpowiada na co najmniej dwa pytania i poprawnie kończy sesję.
• /sources pokazuje źródła właściwej tury.
• Na dysku jest konfiguracja sesji i ślad odpowiedzi.

Instrukcja uczestnika · Python i Ollama · wydanie 29.09.2026