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

Zauważ, że rejestr ma trzy jawne pozycje. W komunikacie można poprosić o restart, ale w aplikacji nie istnieje narzędzie restartu. Tekstowa blokada wybranych poleceń ułatwia rozmowę; właściwe ograniczenie wynika z rejestru i sprawdzania argumentów.

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

"""Rejestr funkcji tylko do odczytu i jedna próba wykonania na turę."""
import json
from copy import deepcopy
from lokalny import CONFIG
from pamiec import sprawdz_wiadomosc, jawne_id
from rag import ask_rag


def katalog_dokumentow(records):
    catalog = {}
    fields = ("document_id", "source", "title", "version", "status")
    for record in records:
        meta = {key: record["metadata"][key] for key in fields}
        key = meta["document_id"]
        if key in catalog and catalog[key] != meta:
            raise ValueError("Niespójne metadane dokumentu: " + key)
        catalog[key] = meta
    return catalog


def answer_from_documents(args, context):
    return ask_rag(args["question"], context["records"], context["index"])


def show_last_sources(args, context):
    return {"sources": deepcopy(context["state"]["last_sources"]),
            "source_origin": deepcopy(context["state"]["source_origin"])}


def get_document_metadata(args, context):
    return deepcopy(context["catalog"][args["document_id"]])


REGISTRY = {"answer_from_documents": answer_from_documents,
            "show_last_sources": show_last_sources,
            "get_document_metadata": get_document_metadata}


def sprawdz_argumenty(name, args, context):
    if not isinstance(args, dict):
        raise ValueError("arguments_not_object")
    expected = {"question"} if name == "answer_from_documents" else ({"document_id"} if name == "get_document_metadata" else set())
    if set(args) != expected:
        raise ValueError("invalid_argument_fields")
    if name == "answer_from_documents":
        if sprawdz_wiadomosc(args["question"]) != context["standalone"]:
            raise ValueError("question_mismatch")
    elif name == "get_document_metadata":
        doc = args["document_id"]
        if not isinstance(doc, str) or doc not in context["catalog"]:
            raise ValueError("unknown_document")
        if doc not in jawne_id(context["original"], context["catalog"]):
            raise ValueError("document_not_explicit")
    elif not context["state"]["last_sources"]:
        raise ValueError("missing_previous_sources")
    return deepcopy(args)


def wykonaj(call, context):
    state = context["state"]
    result = {"name": None, "status": "rejected", "executed": False,
              "arguments": {}, "data": None, "reason": None}
    if state["tool_attempted"]:
        return dict(result, reason="attempt_limit")
    state["tool_attempted"] = True  # Nieudana próba też zużywa limit.
    try:
        if not isinstance(call, dict) or set(call) != {"action", "arguments"}:
            raise ValueError("invalid_call")
        name = call["action"]
        if not isinstance(name, str) or name not in REGISTRY:
            raise ValueError("unknown_tool")
        result["name"] = name
        args = sprawdz_argumenty(name, call["arguments"], context)
        result["arguments"] = args
    except (ValueError, TypeError) as exc:
        return dict(result, reason=str(exc))
    try:
        result["executed"] = True
        data = REGISTRY[name](args, context)
        if not isinstance(data, dict) or len(json.dumps(data, ensure_ascii=False, allow_nan=False)) > 120000:
            raise ValueError("invalid_or_large_tool_output")
        result.update(status="completed", data=data)
    except Exception as exc:
        result.update(status="error", reason=str(exc))
    return result

Jeśli narzędzie RAG zgłosi wyjątek, wykonanie zostaje zapisane jako error. Kolejny etap zwróci zrozumiały komunikat i wyczyści poprzednie źródła RAG. Nie ponawiamy operacji ukrycie i nie zamieniamy błędu w gotową odpowiedź z przykładu.

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