Budowa asystenta dokumentów
3. Pamięć rozmowy i pełne pytanie
3.1. Porównaj rozwiązanie pamięci
Lokalne chatboty, agenci i systemy RAG z własną bazą wiedzy
Najpierw porównaj pola i warunki z własną implementacją. Jeśli testy nie przechodzą, znajdź konkretną różnicę w zachowaniu zamiast wymieniać cały plik bez analizy. Przykład jest celowo wąski: nie interpretuje dowolnego „tego”, „tam” ani swobodnych nawiązań do wielu obiektów.
Plik: pamiec.py. Poniżej znajduje się kompletna wersja do porównania po własnej próbie.
"""Mała pamięć sesji i jawny, regułowy resolver odniesień DocAssist."""
import re
import uuid
from copy import deepcopy
from lokalny import CONFIG
def nowy_stan():
return {"conversation_id": uuid.uuid4().hex, "next_turn": 1,
"history": [], "last_sources": [], "source_origin": None,
"tool_attempted": False}
def sprawdz_wiadomosc(message):
if not isinstance(message, str) or not message.strip():
raise ValueError("empty_or_invalid_message")
if len(message.strip()) > CONFIG["question_chars"]:
raise ValueError("message_too_long")
return message.strip()
def prosba_o_zrodla(message):
return message.casefold().strip(" .?!") in {
"pokaż źródła", "pokaż źródła tej odpowiedzi",
"pokaż źródła poprzedniej odpowiedzi", "skąd ta odpowiedź"}
def jawne_id(message, catalog):
return [key for key in sorted(catalog)
if re.search(r"(?<![\w.-])" + re.escape(key) + r"(?![\w.-])", message)]
def rozwin_pytanie(question, history):
result = {"original": question, "standalone": question,
"history_used": False, "clarification": None}
low = question.casefold()
if prosba_o_zrodla(question) or "metadan" in low:
return result
dependent = re.search(r"\b(jej|jego|tej usługi|tej zmianie|co teraz|a potem)\b", low)
if not dependent or "docassist-api" in low:
return result
subjects = set()
for turn in history:
# Czytamy pytania, także po błędzie wyboru. Zmiana tematu nie znika po błędzie.
# Odpowiedź modelu nie jest dowodem ani źródłem nowego tematu.
text = turn["standalone"].casefold()
if prosba_o_zrodla(text) or "metadan" in text:
continue
if "docassist-api" in text or "konfiguracji docassist" in text:
subjects.add("usluga")
if re.search(r"baz(?:a|y|ę|ie|ą) danych", text):
subjects.add("baza")
if subjects == {"usluga"} and "stan" in low and "failed" not in low:
result.update(standalone="Jak sprawdzić stan usługi docassist-api?", history_used=True)
elif subjects == {"usluga"} and "failed" in low:
result.update(standalone="Co zrobić, jeśli po zmianie konfiguracji usługa docassist-api nie uruchamia się?", history_used=True)
else:
result.update(standalone=None, history_used=bool(history),
clarification="Wskaż obiekt i zadaj pełne pytanie, np. o stan usługi docassist-api.")
return result
def zapamietaj(state, result):
limit = CONFIG["history_turns"]
if type(limit) is not int or not 1 <= limit <= 5:
raise ValueError("history_turns musi być liczbą od 1 do 5.")
if result["valid_input"]:
state["history"].append({"question": result["original_question"],
"standalone": result["standalone_question"] or result["original_question"],
"answer": result["answer"][:1200], "status": result["status"],
"tool": result["tool"].get("name"), "result_id": result["result_id"]})
state["history"] = state["history"][-limit:]
# Nowe wykonane RAG zastępuje źródła. Błąd lub pusty kontekst je czyści.
if result["tool"].get("name") == "answer_from_documents" and result["tool"].get("executed"):
state["last_sources"] = deepcopy(result["sources"] if result["status"] == "answered" else [])
state["source_origin"] = ({"turn_id": result["turn_id"], "result_id": result["result_id"]}
if state["last_sources"] else None)
state["next_turn"] += 1
Sprawdź przed przejściem dalej
- Każda sesja ma niezależne listy i identyfikator.
- Niejasna wiadomość kończy się doprecyzowaniem.
- Rozwinięte pytanie jest jawne i możliwe do sprawdzenia w wyniku.
Materiał uzupełniający do modułów 1–6 · wersja 1.0 · 30.09.2026