Budowa asystenta dokumentów

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

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

Przy czytaniu kodu prześledź osobno: zwykłe pytanie, niejasne dopowiedzenie, poprawne metadane, źródła oraz błąd RAG. Nie wszystkie pola muszą mieć wartość w każdej ścieżce: np. brak wyboru narzędzia przed doprecyzowaniem jest oczekiwanym zachowaniem.

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

"""Jedna wiadomość: walidacja, interpretacja, wybór, narzędzie, wynik, pamięć."""
import json
import re
import uuid
from datetime import datetime, timezone
from copy import deepcopy
from lokalny import CONFIG
from pamiec import sprawdz_wiadomosc, prosba_o_zrodla, jawne_id, rozwin_pytanie, zapamietaj
from decyzja import wybierz_reguly, wybierz_model, sprawdz_propozycje
from narzedzia import wykonaj
from rag import format_sources


def krok(message, state, records, index, catalog, selector="reguly"):
    state["tool_attempted"] = False
    result = {"schema": "poradnik-asystent-v1", "result_id": uuid.uuid4().hex,
        "conversation_id": state["conversation_id"], "turn_id": state["next_turn"],
        "timestamp": datetime.now(timezone.utc).isoformat(), "selector": selector,
        "original_question": message if isinstance(message, str) else None,
        "standalone_question": None, "history_used": False, "valid_input": False,
        "status": "error", "reason": None, "answer": "", "sources": [],
        "source_origin": None, "decision": None, "selection_trace": None,
        "tool": {}, "rag": None, "warnings": []}

    def finish(status, reason, answer):
        result.update(status=status, reason=reason, answer=answer)
        zapamietaj(state, result)
        return result

    try:
        question = sprawdz_wiadomosc(message)
    except (ValueError, TypeError) as exc:
        return finish("rejected", str(exc), "Podaj niepuste pytanie do 1000 znaków.")
    result["valid_input"] = True
    if selector not in ("reguly", "model"):
        return finish("error", "invalid_selector", "Wybierz tryb reguly albo model.")
    if re.match(r"^(?:proszę\s+)?(?:zrestartuj|uruchom|wykonaj|usuń|zmień|wyślij)\b", question, re.I):
        return finish("rejected", "forbidden_action", "Asystent objaśnia dokumenty, ale nie wykonuje operacji systemowych.")
    ids = jawne_id(question, catalog)
    if "metadan" in question.casefold() and len(ids) != 1:
        reason = "unknown_document" if re.search(r"\bDOC-[A-Z0-9_-]+\b", question) and not ids else "missing_or_ambiguous_document_id"
        status = "rejected" if reason == "unknown_document" else "clarification_needed"
        return finish(status, reason, "Podaj jeden znany identyfikator: " + ", ".join(sorted(catalog)))
    if prosba_o_zrodla(question) and not state["last_sources"]:
        return finish("clarification_needed", "missing_previous_sources", "Najpierw zadaj pytanie do dokumentów; nie mam źródeł poprzedniej odpowiedzi.")
    resolved = rozwin_pytanie(question, state["history"])
    result.update(standalone_question=resolved["standalone"], history_used=resolved["history_used"])
    if resolved["clarification"]:
        return finish("clarification_needed", "ambiguous_reference", resolved["clarification"])
    try:
        if selector == "reguly":
            proposal = wybierz_reguly(question, catalog)
        else:
            proposal, result["selection_trace"] = wybierz_model(question, resolved["standalone"], state, catalog)
        if proposal is None:
            return finish("error", "selector_parse_error", "Niepoprawna propozycja modelu; nie wykonano narzędzia.")
        proposal = sprawdz_propozycje(proposal)
    except (RuntimeError, ValueError, KeyError, TypeError) as exc:
        return finish("error", "selector_error", "Błąd wyboru działania: " + str(exc))
    result["decision"] = proposal
    if proposal["decision"] != "tool_call":
        status = "clarification_needed" if proposal["decision"] == "clarification_needed" else "rejected"
        return finish(status, proposal["decision"], proposal["message"])
    args = deepcopy(proposal["arguments"])
    if proposal["action"] == "answer_from_documents":
        if args:
            return finish("rejected", "unexpected_selector_arguments", "Pytanie RAG ustala aplikacja; selektor nie może go podmieniać.")
        args = {"question": resolved["standalone"]}
    context = {"state": state, "records": records, "index": index, "catalog": catalog,
               "original": question, "standalone": resolved["standalone"]}
    tool = wykonaj({"action": proposal["action"], "arguments": args}, context)
    result["tool"] = tool
    if tool["status"] != "completed":
        return finish("error" if tool["status"] == "error" else "rejected",
                      tool["reason"], "Nie udało się wykonać narzędzia.")
    data = tool["data"]
    if tool["name"] == "answer_from_documents":
        result["rag"] = data
        result["warnings"] = data["warnings"]
        if data["status"] == "answered":
            result["sources"] = deepcopy(data["sources"])
            result["source_origin"] = {"turn_id": result["turn_id"], "result_id": result["result_id"]}
        status = data["status"] if data["status"] in {"answered", "insufficient_context", "error"} else "rejected"
        return finish(status, data.get("reason") or "rag_completed", data["answer"])
    if tool["name"] == "show_last_sources":
        result["sources"] = data["sources"]
        result["source_origin"] = data["source_origin"]
        return finish("answered", "sources_returned", format_sources(data["sources"]))
    return finish("answered", "metadata_returned", json.dumps(data, ensure_ascii=False, indent=2))

Sprawdź przed przejściem dalej

• Nowe pytanie merytoryczne zawsze uruchamia świeży RAG.
• Metadane nie zastępują ostatnich źródeł.
• Po błędzie nowego RAG źródła nie pochodzą z wcześniejszej odpowiedzi.
• Nie ma ukrytego przejścia z selektora modelowego na reguły.

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