4. Historia i pytania zależne

„Jak sprawdzić jej stan?” nie identyfikuje usługi. Historia może pomóc ustalić odniesienie, ale wcześniejsza odpowiedź nie staje się dokumentem źródłowym. Po ustaleniu pełnego pytania trzeba ponownie wyszukać fragmenty.

1. Przechowuj ograniczoną historię

Do pliku kod/moj_m5.py: get_recent_history()

def get_recent_history(messages, max_history_turns=MAX_HISTORY_TURNS):
    if max_history_turns <= 0:
        return []
    return deepcopy([m for m in messages if m.get('role') in ('user', 'assistant')][-2 * max_history_turns:])

Przy naprzemiennych parach user/assistant sześć ostatnich wiadomości oznacza trzy tury. Funkcja odrzuca inne role i zwraca kopię, aby jej modyfikacja nie zmieniała stanu. To limit wiadomości, nie dokładny budżet tokenów. Dłuższa rozmowa wymagałaby dodatkowej kontroli rozmiaru.

2. Rozwiąż odniesienie albo poproś o doprecyzowanie

Resolver jest regułowy: obsługuje określone wzorce DocAssist i konflikt „usługa” kontra „baza danych”. Nie jest ogólnym rozumieniem języka. Wynik oddziela oryginalną wiadomość od samodzielnego pytania; przy niejednoznaczności standalone_question ma wartość None.

Do pliku kod/moj_m5.py: resolve_followup_question()

def resolve_followup_question(question, recent_history):
    q = validate_chat_message(question)
    result = {'original_question': q, 'standalone_question': q, 'history_used': False, 'resolution_status': 'unchanged', 'reason': None, 'clarification_question': None}
    low = q.casefold()
    if is_last_sources_request(q) or 'metadan' in low:
        return result
    dependent = bool(re.search('\\b(jej|jego|tej usługi|tej zmianie|co teraz|stan to failed)\\b', low))
    if not dependent or 'docassist-api' in low:
        return result
    history = get_recent_history(recent_history)
    subjects = set()
    for message in history:
        text = message.get('content', '').casefold()
        if 'docassist-api' in text or 'konfiguracji docassist' in text:
            subjects.add('usługa docassist-api')
        if re.search('baz(?:a|y|ę|ie|ą) danych', text):
            subjects.add('baza danych DocAssist')
    result['history_used'] = bool(history)
    if len(subjects) != 1 or 'usługa docassist-api' not in subjects:
        result.update(standalone_question=None, resolution_status='clarification_needed', reason='ambiguous_reference', clarification_question='Czy chodzi o usługę docassist-api, czy o inny obiekt? Wskaż go w pełnym pytaniu.')
        return result
    if 'failed' in low or 'co teraz' in low:
        standalone = 'Co zrobić, jeśli po zmianie konfiguracji usługa docassist-api nie uruchamia się?'
    else:
        standalone = re.sub('\\b(jej|jego|tej usługi)\\b', 'usługi docassist-api', q, flags=re.I)
        standalone = re.sub('\\btej zmianie\\b', 'zmianie konfiguracji DocAssist', standalone, flags=re.I)
    result.update(standalone_question=standalone, resolution_status='resolved', history_used=True)
    return result

Test R3: odniesienie obecne i brakujące

historia = [
    {"role": "user", "content": "Co zrobić po zmianie konfiguracji DocAssist?"},
    {"role": "assistant", "content": "Instrukcja opisuje usługę docassist-api."}
]
r = m.resolve_followup_question("Jak sprawdzić jej stan?", historia)
assert r["resolution_status"] == "resolved"
assert "docassist-api" in r["standalone_question"]
assert m.resolve_followup_question(
    "Jak sprawdzić jej stan?", []
)["resolution_status"] == "clarification_needed"

Regułowe podstawienie daje niezręczny szyk „Jak sprawdzić usługi docassist-api stan?”. Obiekt jest jawny, więc pytanie może służyć wyszukiwaniu. Nie przedstawiaj tej funkcji jako pełnej normalizacji języka. Zachowujemy wariant pakietu, ponieważ zmiana pytania zmienia żądanie embeddingowe i klucz checkpointu. Poprawę szyku testuj osobno z nowymi wywołaniami.

3. Źródła należą do konkretnego RAG

Jeżeli tura 3 wyszukała fragmenty, a tura 4 prosi o źródła tej odpowiedzi, odczytaj źródła tury 3. Ponowne wyszukanie mogłoby dać inny ranking niż ten rzeczywiście przekazany modelowi. Lista źródeł jest dowodem konkretnego wejścia, a nie dowolną listą pasujących dokumentów.

Zdarzenie last_result / last_sources turn_log
Nowy RAG Zastąp wynikiem i źródłami, również pustymi. Dodaj turę.
Źródła lub metadane Pozostaw ostatni RAG. Dodaj turę naturalnej wiadomości.
Odmowa lub doprecyzowanie Nie twórz fikcyjnego RAG. Dodaj zakończenie.
/clear Usuń poprzedni stan. Nowa rozmowa; pliki pozostają.

4. Jedno miejsce zakończenia tury

finalize_turn() porządkuje stan i zapis niezależnie od drogi zakończenia. begin_turn() i funkcje tworzące wynik są już w szkielecie. Uzupełnij finalizer teraz; pełne połączenie uruchomisz po R6. Test R3 dostarcza kontrolowany wynik bez modelu i selektora.

Do pliku kod/moj_m5.py: finalize_turn()

def finalize_turn(state, result, turn):
    result['diagnostics']['trace'].append('finalize')
    result['diagnostics']['duration_ms'] = round((time.perf_counter() - turn['started']) * 1000, 3)
    if result['rag']:
        state['last_result'] = deepcopy(result['rag'])
        state['last_sources'] = deepcopy(result['rag'].get('sources') or [])
        state['last_source_turn_id'] = turn['turn_id']
        state['last_source_result_id'] = result['result_id']
    if result['reason'] not in ('invalid_message', 'empty_message', 'message_too_long'):
        state['history'].extend([{'role': 'user', 'content': result['conversation']['original_question']}, {'role': 'assistant', 'content': result['answer']}])
        state['history'] = get_recent_history(state['history'])
    summary = {k: result[k] for k in ('result_id', 'timestamp', 'mode', 'index_id', 'status', 'reason')}
    summary.update(conversation_id=turn['conversation_id'], turn_id=turn['turn_id'], question_sha256=hashlib.sha256((result['conversation']['original_question'] or '').encode()).hexdigest(), history_used=result['conversation']['history_used'], decision_status=result['decision'].get('status'), tool_status=result['tool'].get('status'), rag_status=result['rag'].get('status'), source_ids=[s['chunk_id'] for s in result['sources']], diagnostics=deepcopy(result['diagnostics']), warnings=result['warnings'])
    summary['diagnostics'].pop('selection', None)
    state['turn_log'].append(summary)
    state['next_turn_id'] += 1
    if _APP.get('output_dir'):
        append_jsonl(_APP['output_dir'] / 'assistant_trace.jsonl', summary)
        if _APP.get('save_raw'):
            append_jsonl(_APP['output_dir'] / 'surowe_wyniki.jsonl', result)
    return result

Zewnętrzny result_id identyfikuje turę asystenta. Wewnętrzny rag.result_id identyfikuje wykonanie odziedziczonego RAG. last_source_result_id zachowuje ID tury, aby połączyć źródła z functional_tests.jsonl. Nie mieszaj tych przestrzeni przy łączeniu śladów.

5. Fragment pominięty w pobranym kontekście

DOC-INSTR-dd6055d675-0003

instrukcja.txt | version=1.4 | Kopia konfiguracji

## Kopia konfiguracji Przed edycją zachowaj kopię pliku konfiguracyjnego w katalogu kopii. Nie zapisuj w zgłoszeniu haseł ani kluczy. W razie nieudanego uruchomienia przywróć zatwierdzoną kopię i zgłoś problem administratorowi.

Ćwiczenie 3. Prześledź stan po czterech turach

  1. Uzupełnij funkcje R3 i sprawdź test etapu.
  2. Rozważ dialog: konfiguracja, „Jak sprawdzić jej stan?”, failed, prośba o źródła. Wskaż tury wymagające nowego RAG.
  3. W notebooku R3 porównaj source_origin z ID poprzedniej tury RAG.
  4. Powtórz pytanie z „jej” po /clear. Rozważ też historię wskazującą usługę i bazę danych. Uzasadnij doprecyzowanie.

ENHANCER+ · Politechnika Warszawska · podręcznik do samodzielnej pracy · M5