Budowa asystenta dokumentów

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

Orientacyjny czas: 15 min · Tryb pracy: najpierw własna próba, potem przykład

sprawdz.py jest gotowy w obu wariantach pakietu. Testy zastępują RAG kontrolowaną odpowiedzią tylko po to, aby sprawdzić logikę programu. Nie są pomiarem jakości modelu. Jeśli w wersji Start widzisz NotImplementedError, najpierw uzupełnij wskazaną funkcję.

Plik: sprawdz.py. Poniżej znajduje się kompletna wersja do porównania po własnej próbie.

"""Testy kontraktów bez Ollama: python -m unittest sprawdz -v."""
import unittest
from unittest.mock import patch
from copy import deepcopy
from pamiec import nowy_stan, sprawdz_wiadomosc, rozwin_pytanie
from decyzja import scisly_json, sprawdz_propozycje
from narzedzia import wykonaj, katalog_dokumentow
from asystent import krok

CATALOG = {"DOC-INSTR": {"document_id": "DOC-INSTR", "title": "Instrukcja", "source": "instrukcja.txt", "version": "1.4", "status": "active"}}
SOURCE = {"label": "S1", "chunk_id": "A", "text": "Dane", "score": 0.8, "metadata": {**CATALOG["DOC-INSTR"], "section": "Stan", "page_number": None}}
RAG = {"status": "answered", "answer": "Odpowiedź [S1]", "sources": [SOURCE], "warnings": []}


class Pamiec(unittest.TestCase):
    def test_empty_and_long(self):
        for value in ("  ", None, "x" * 1001):
            with self.assertRaises(ValueError):
                sprawdz_wiadomosc(value)

    def test_independent_states(self):
        first, second = nowy_stan(), nowy_stan()
        first["history"].append({})
        self.assertFalse(second["history"])
        self.assertNotEqual(first["conversation_id"], second["conversation_id"])

    def test_ambiguous_without_history(self):
        self.assertIsNotNone(rozwin_pytanie("Jak sprawdzić jej stan?", [])["clarification"])

    def test_resolve_and_ambiguity(self):
        history = [{"tool": "answer_from_documents", "status": "answered", "standalone": "Stan docassist-api?"}]
        result = rozwin_pytanie("Jak sprawdzić jej stan?", history)
        self.assertTrue(result["history_used"])
        self.assertIn("docassist-api", result["standalone"])
        history.append({"tool": "answer_from_documents", "status": "answered", "standalone": "Gdzie jest baza danych?"})
        self.assertIsNone(rozwin_pytanie("Jak sprawdzić jej stan?", history)["standalone"])

    def test_failed_turn_does_not_hide_topic_change(self):
        history = [{"tool": "answer_from_documents", "status": "answered", "standalone": "Stan docassist-api?"},
                   {"tool": None, "status": "rejected", "standalone": "Gdzie znajduje się baza danych?"}]
        self.assertIsNone(rozwin_pytanie("Jak sprawdzić jej stan?", history)["standalone"])


class Decyzja(unittest.TestCase):
    def test_invalid_json(self):
        for value in ('{"x":1,"x":2}', '{"x":NaN}', '```json\n{}\n```', '{} trailing'):
            with self.assertRaises(ValueError):
                scisly_json(value)

    def test_exact_fields(self):
        with self.assertRaises(ValueError):
            sprawdz_propozycje({"decision": "tool_call", "action": "x", "arguments": {}, "message": None, "extra": 1})

    def test_non_tool_schema(self):
        with self.assertRaises(ValueError):
            sprawdz_propozycje({"decision": "rejected", "action": "x", "arguments": {}, "message": "Nie"})


class Narzedzia(unittest.TestCase):
    def setUp(self):
        self.ctx = {"state": nowy_stan(), "records": [], "index": {}, "catalog": CATALOG,
                    "original": "Pokaż metadane DOC-INSTR", "standalone": "Stan?"}

    def test_allowlist_and_attempt_limit(self):
        self.assertEqual(wykonaj({"action": "run_shell", "arguments": {}}, self.ctx)["reason"], "unknown_tool")
        self.assertEqual(wykonaj({"action": "get_document_metadata", "arguments": {"document_id": "DOC-INSTR"}}, self.ctx)["reason"], "attempt_limit")

    def test_extra_argument(self):
        result = wykonaj({"action": "get_document_metadata", "arguments": {"document_id": "DOC-INSTR", "path": "x"}}, self.ctx)
        self.assertFalse(result["executed"])

    def test_metadata_id_must_be_explicit(self):
        self.ctx["original"] = "Pokaż metadane"
        result = wykonaj({"action": "get_document_metadata", "arguments": {"document_id": "DOC-INSTR"}}, self.ctx)
        self.assertEqual(result["reason"], "document_not_explicit")

    def test_question_cannot_change(self):
        result = wykonaj({"action": "answer_from_documents", "arguments": {"question": "Inne?"}}, self.ctx)
        self.assertEqual(result["reason"], "question_mismatch")

    def test_catalog_conflict(self):
        records = [{"metadata": CATALOG["DOC-INSTR"]}, {"metadata": {**CATALOG["DOC-INSTR"], "version": "2"}}]
        with self.assertRaises(ValueError):
            katalog_dokumentow(records)


class Przeplyw(unittest.TestCase):
    def setUp(self):
        self.state = nowy_stan()

    def run_turn(self, question, selector="reguly"):
        return krok(question, self.state, [], {}, CATALOG, selector)

    @patch("narzedzia.ask_rag", return_value=RAG)
    def test_followup_runs_fresh_rag(self, mock):
        self.run_turn("Jak sprawdzić stan docassist-api?")
        result = self.run_turn("Jak sprawdzić jej stan?")
        self.assertEqual(mock.call_count, 2)
        self.assertTrue(result["history_used"])
        self.assertEqual(mock.call_args.args[0], "Jak sprawdzić stan usługi docassist-api?")

    @patch("narzedzia.ask_rag", return_value=RAG)
    def test_metadata_preserves_source_origin(self, mock):
        self.run_turn("Stan docassist-api?")
        self.run_turn("Pokaż metadane DOC-INSTR")
        result = self.run_turn("Pokaż źródła")
        self.assertEqual(result["source_origin"]["turn_id"], 1)
        self.assertEqual(mock.call_count, 1)

    @patch("narzedzia.ask_rag", side_effect=[RAG, RuntimeError("Awaria")])
    def test_error_clears_previous_sources(self, mock):
        self.run_turn("Stan docassist-api?")
        self.assertEqual(self.run_turn("Nowe pytanie?")["status"], "error")
        self.assertFalse(self.state["last_sources"])
        self.assertEqual(self.run_turn("Pokaż źródła")["status"], "clarification_needed")

    @patch("narzedzia.ask_rag", return_value=RAG)
    def test_history_bound(self, mock):
        for number in range(6):
            self.run_turn(f"Pytanie {number}?")
        self.assertEqual(len(self.state["history"]), 3)
        self.assertEqual(self.state["next_turn"], 7)

    @patch("asystent.wybierz_model", return_value=(None, {"parse_error": "bad"}))
    @patch("narzedzia.ask_rag")
    def test_selector_error_has_no_fallback(self, rag, selector):
        self.assertEqual(self.run_turn("Stan?", "model")["status"], "error")
        rag.assert_not_called()

    @patch("asystent.wybierz_model")
    def test_guards_before_model(self, selector):
        for message in ("Pokaż metadane", "Zrestartuj usługę", "Pokaż źródła"):
            self.assertIn(self.run_turn(message, "model")["status"], ("rejected", "clarification_needed"))
        selector.assert_not_called()

    @patch("asystent.wybierz_model", return_value=({"decision": "tool_call", "action": "answer_from_documents", "arguments": {"question": "Podmienione"}, "message": None}, {}))
    @patch("narzedzia.ask_rag")
    def test_model_cannot_replace_question(self, rag, selector):
        self.assertEqual(self.run_turn("Stan?", "model")["status"], "rejected")
        rag.assert_not_called()


if __name__ == "__main__":
    unittest.main(verbosity=2)

Dodaj własny test, jeżeli rozszerzasz kontrakt. Przykład: po zmianie obsługi źródeł sprawdź nie tylko obecność listy, ale również turn_id i result_id. Zachowaj test drugiej próby narzędzia, ponieważ łatwo przypadkowo usunąć ograniczenie podczas rozbudowy pętli.

Materiał uzupełniający do modułów 1–6 · wersja 1.0 · 30.09.2026