Bouw je eigen zoekende antwoordgever

import re
import hashlib
from collections import Counter
import numpy as np

ALIASES = {'herstart': 'reset', 'licht': 'lamp', 'accu': 'batterij', 'bijwerken': 'update'}
RECORDS = [
    {'id': 'lamp-reset', 'title': 'Lamp herstellen', 'text': 'Reset de lamp door de knop tien seconden vast te houden.'},
    {'id': 'lamp-garantie', 'title': 'Lampgarantie', 'text': 'De garantie van de lamp duurt twee jaar.'},
    {'id': 'sensor-reset', 'title': 'Sensor herstellen', 'text': 'Reset de sensor door de knop vijf seconden vast te houden.'},
    {'id': 'sensor-batterij', 'title': 'Sensorbatterij', 'text': 'De batterij van de sensor gaat twaalf maanden mee.'},
    {'id': 'hub-reset', 'title': 'Hub herstellen', 'text': 'Reset de hub via de lokale herstelknop.'},
    {'id': 'hub-update', 'title': 'Hub bijwerken', 'text': 'Een update van de hub duurt drie minuten.'},
]

def document_snapshots(records):
    if not isinstance(records, list) or not 1 <= len(records) <= 32:
        raise ValueError('Geef 1 tot 32 documenten')
    seen, result = set(), []
    for row in records:
        if not isinstance(row, dict) or set(row) != {'id', 'title', 'text'}:
            raise ValueError('Document heeft exact id/title/text')
        if not isinstance(row['id'], str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,40}', row['id']) or row['id'] in seen:
            raise ValueError('Unieke begrensde document-ID vereist')
        if not isinstance(row['title'], str) or not 1 <= len(row['title'].strip()) or len(row['title']) > 100:
            raise ValueError('Titel is niet leeg en maximaal 100 tekens')
        if not isinstance(row['text'], str) or not row['text'].strip() or len(row['text']) > 20000:
            raise ValueError('Brontekst is niet leeg en maximaal 20000 tekens')
        try:
            revision = hashlib.sha256(row['text'].encode('utf-8')).hexdigest()
        except UnicodeError as exc:
            raise ValueError('Tekst moet geldige UTF-8 zijn') from exc
        seen.add(row['id'])
        result.append({**row, 'revision': revision})
    return result

def word_chunks(document, window=40, overlap=8):
    if not isinstance(document, dict) or set(document) != {'id', 'title', 'text', 'revision'}:
        raise ValueError('Geef een document-snapshot')
    checked = document_snapshots([{k: document[k] for k in ['id', 'title', 'text']}])[0]
    if document['revision'] != checked['revision']:
        raise ValueError('Documenttekst en versiehash verschillen')
    if type(window) is not int or not 1 <= window <= 256 or type(overlap) is not int or not 0 <= overlap < window:
        raise ValueError('Window 1–256, overlap 0 tot window-1')
    words = list(re.finditer(r'\S+', document['text']))
    result, first = [], 0
    while first < len(words):
        last = min(first + window, len(words))
        start, end = words[first].start(), words[last - 1].end()
        result.append({'id': f'{document["id"]}:{first}', 'document_id': document['id'],
                       'revision': document['revision'], 'word_start': first, 'word_end': last,
                       'start': start, 'end': end, 'text': document['text'][start:end]})
        if last == len(words):
            break
        first += window - overlap
    return result

def chunk_coverage(document, chunks):
    # Reuse the snapshot check without changing the source text.
    word_chunks(document, 256, 0)
    if not isinstance(chunks, list) or len(chunks) > 20000:
        raise ValueError('Geef maximaal 20000 chunks')
    words = list(re.finditer(r'\S+', document['text']))
    covered, seen, occurrences = set(), set(), 0
    for chunk in chunks:
        if not isinstance(chunk, dict) or set(chunk) != {'id', 'document_id', 'revision', 'word_start', 'word_end', 'start', 'end', 'text'}:
            raise ValueError('Ongeldige chunkvelden')
        a, b = chunk['word_start'], chunk['word_end']
        if type(a) is not int or type(b) is not int or not 0 <= a < b <= len(words):
            raise ValueError('Ongeldig woordinterval')
        if chunk['id'] != f'{document["id"]}:{a}' or chunk['id'] in seen or chunk['document_id'] != document['id'] or chunk['revision'] != document['revision']:
            raise ValueError('Chunk-ID, document of versie klopt niet')
        start, end = words[a].start(), words[b - 1].end()
        if type(chunk['start']) is not int or type(chunk['end']) is not int or chunk['start'] != start or chunk['end'] != end or chunk['text'] != document['text'][start:end]:
            raise ValueError('Chunktekst/offsets komen niet overeen met de bron')
        seen.add(chunk['id']);covered.update(range(a, b));occurrences += b - a
    return {'total_words': len(words), 'covered_words': len(covered),
            'coverage': len(covered) / len(words), 'word_occurrences': occurrences,
            'duplication_factor': occurrences / len(covered) if covered else None,
            'missing_word_indices': sorted(set(range(len(words))) - covered)}

def build_rag_index(records):
    documents = document_snapshots(records)
    chunks = [c for d in documents for c in word_chunks(d, 32, 0)]
    if len(chunks) > 256:
        raise ValueError('Dit notebook ondersteunt maximaal 256 chunks')
    tokenized = [re.findall(r'[^\W_]+', c['text'].casefold()) for c in chunks]
    vocabulary = sorted({t for ts in tokenized for t in ts})
    if not 1 <= len(vocabulary) <= 2048:
        raise ValueError('Woordenschat 1 tot 2048 termen')
    counts = [Counter(ts) for ts in tokenized]
    tf = np.array([[count[t] for t in vocabulary] for count in counts], dtype=float)
    lengths, df = tf.sum(axis=1), np.count_nonzero(tf, axis=0)
    idf = np.log1p((len(chunks) - df + .5) / (df + .5))
    weights = idf * tf * 2.2 / (tf + 1.2 * (.25 + .75 * lengths[:, None] / lengths.mean()))
    return {'documents': documents, 'chunks': chunks, 'vocabulary': vocabulary, 'weights': weights,
            'method': 'own-bm25-v1|k1=1.2|b=.75|casefold-unicode-words'}

index=build_rag_index(RECORDS)
print("Bronnen / chunks:",len(index["documents"]),len(index["chunks"]))
print("Indexvorm:",index["weights"].shape)

def query_terms(query, expand):
    if not isinstance(query, str) or not query.strip() or len(query) > 2000 or type(expand) is not bool:
        raise ValueError('Geef een niet-lege query tot 2000 tekens en een bool')
    terms = re.findall(r'[^\W_]+', query.casefold())
    return [ALIASES.get(t, t) if expand else t for t in terms]

def retrieve_rag(index, query, k, expand=False):
    # index is the internal result of build_rag_index, never a deserialized file.
    if type(k) is not int or not 1 <= k <= 20:
        raise ValueError('k tussen 1 en 20')
    terms = set(query_terms(query, expand))
    columns = [i for i, t in enumerate(index['vocabulary']) if t in terms]
    scores = index['weights'][:, columns].sum(axis=1)
    order = sorted((i for i, score in enumerate(scores) if score > 0), key=lambda i: (-scores[i], index['chunks'][i]['id']))[:k]
    return [{'id': index['chunks'][i]['id'], 'score': float(scores[i]), 'chunk': index['chunks'][i]} for i in order]

hits=retrieve_rag(index,"Hoe herstart ik het licht?",3,True)
print("Gevonden:",[h["id"] for h in hits])

def checked_sources(documents):
    if not isinstance(documents, list) or any(not isinstance(d, dict) or set(d) != {'id', 'title', 'text', 'revision'} for d in documents):
        raise ValueError('Document-snapshots vereist')
    checked = document_snapshots([{k: d[k] for k in ['id', 'title', 'text']} for d in documents])
    if any(d['revision'] != c['revision'] for d, c in zip(documents, checked)):
        raise ValueError('Brontekst en versie verschillen')
    return {d['id']: d for d in checked}

def pack_context(chunks, documents, budget):
    sources = checked_sources(documents)
    if not isinstance(chunks, list) or len(chunks) > 256 or type(budget) is not int or not 0 <= budget <= 60000:
        raise ValueError('Maximaal 256 chunks en budget 0–60000 tekens')
    for chunk in chunks:
        if not isinstance(chunk, dict) or not isinstance(chunk.get('document_id'), str) or chunk['document_id'] not in sources:
            raise ValueError('Onbekende bron')
        chunk_coverage(sources[chunk['document_id']], [chunk])
    text, citations, seen_texts = '', [], set()
    for chunk in chunks:
        if chunk['text'] in seen_texts:
            continue
        label = 'S' + str(len(citations) + 1)
        block = f'[{label}] {chunk["document_id"]}@{chunk["revision"][:12]} [{chunk["start"]},{chunk["end"]})\n{chunk["text"]}'
        candidate = text + ('\n\n' if text else '') + block
        if len(candidate) > budget:
            continue
        text = candidate
        seen_texts.add(chunk['text'])
        citations.append({'source_id': label, 'document_id': chunk['document_id'], 'revision': chunk['revision'],
                          'start': chunk['start'], 'end': chunk['end'], 'quote': chunk['text']})
    return {'text': text, 'used_chars': len(text), 'citations': citations}

def verify_citations(citations, documents):
    sources = checked_sources(documents)
    if not isinstance(citations, list) or len(citations) > 256:
        raise ValueError('Maximaal 256 verwijzingen')
    results, seen = [], set()
    for c in citations:
        if not isinstance(c, dict) or set(c) != {'source_id', 'document_id', 'revision', 'start', 'end', 'quote'}:
            raise ValueError('Ongeldige verwijzingsvelden')
        if not isinstance(c['source_id'], str) or not re.fullmatch(r'S[1-9][0-9]{0,3}', c['source_id']) or c['source_id'] in seen or not isinstance(c['document_id'], str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,40}', c['document_id']):
            raise ValueError('Unieke bronlabels en document-IDs vereist')
        if not isinstance(c['revision'], str) or not re.fullmatch(r'[a-f0-9]{64}', c['revision']) or type(c['start']) is not int or type(c['end']) is not int or not isinstance(c['quote'], str) or not 1 <= len(c['quote']) <= 20000:
            raise ValueError('Versie, posities of citaat ongeldig')
        seen.add(c['source_id']); source = sources.get(c['document_id'])
        status = 'valid'
        if source is None:
            status = 'unknown-source'
        elif c['revision'] != source['revision']:
            status = 'stale-source'
        elif not 0 <= c['start'] < c['end'] <= len(source['text']):
            status = 'invalid-span'
        elif c['quote'] != source['text'][c['start']:c['end']]:
            status = 'quote-mismatch'
        results.append({'source_id': c['source_id'], 'status': status})
    return results

def extract_rag_answer(question, context, expand):
    terms = set(query_terms(question, expand))
    subjects = terms & {'lamp', 'sensor', 'hub'}
    intents = terms & {'reset', 'garantie', 'batterij', 'update'}
    if len(subjects) != 1 or len(intents) != 1:
        return {'answer': None, 'citations': []}
    required = subjects | intents
    candidates = [c for c in context['citations'] if required <= set(query_terms(c['quote'], False))]
    # Different matching passages are ambiguous in this deliberately small policy.
    if not candidates or len({c['quote'] for c in candidates}) != 1:
        return {'answer': None, 'citations': []}
    return {'answer': candidates[0]['quote'], 'citations': [dict(candidates[0])]}

def run_rag(index, question, config):
    if not isinstance(config, dict) or set(config) != {'k', 'budget', 'expand'}:
        raise ValueError('Config bevat exact k, budget en expand')
    hits = retrieve_rag(index, question, config['k'], config['expand'])
    context = pack_context([h['chunk'] for h in hits], index['documents'], config['budget'])
    result = extract_rag_answer(question, context, config['expand'])
    verified = verify_citations(result['citations'], index['documents'])
    if any(v['status'] != 'valid' for v in verified):
        raise ValueError('Generator leverde een ongeldig citaat')
    return {**result, 'ranking': [h['id'] for h in hits], 'context': context,
            'quote_valid': bool(verified) and result['answer'] == result['citations'][0]['quote']}

answer=run_rag(index,"Hoe herstart ik het licht?",{"k":3,"budget":700,"expand":True})
print("Antwoord:",answer["answer"])
print("Citaat:",answer["citations"][0]["document_id"])
print("Citaat exact:",answer["quote_valid"])

def retrieval_metrics(ranking, relevant, k):
    for items in [ranking, relevant]:
        if not isinstance(items, list) or len(items) > 1000 or any(not isinstance(i, str) or not 1 <= len(i) <= 100 for i in items) or len(set(items)) != len(items):
            raise ValueError('Unieke begrensde IDs vereist')
    if type(k) is not int or not 1 <= k <= 1000:
        raise ValueError('k tussen 1 en 1000')
    selected, gold = ranking[:k], set(relevant)
    positions = [rank for rank, identifier in enumerate(selected, 1) if identifier in gold]
    return {'precision': len(positions) / k, 'recall': len(positions) / len(gold) if gold else None,
            'rr': (1 / positions[0] if positions else 0.) if gold else None}

def validate_rag_rows(rows):
    if not isinstance(rows, list) or not 1 <= len(rows) <= 1000:
        raise ValueError('Geef 1 tot 1000 meetrijen')
    seen = set()
    for row in rows:
        if not isinstance(row, dict) or set(row) != {'id', 'ranking', 'relevant', 'answer', 'expected', 'supported'} or not isinstance(row['id'], str) or not 1 <= len(row['id']) <= 100 or row['id'] in seen:
            raise ValueError('Unieke meet-IDs en vaste velden vereist')
        seen.add(row['id']); retrieval_metrics(row['ranking'], row['relevant'], 1)
        if any(v is not None and (not isinstance(v, str) or not v.strip() or len(v) > 2000) for v in [row['answer'], row['expected']]) or type(row['supported']) is not bool:
            raise ValueError('Antwoorden zijn tekst of None; supported is bool')
        if (row['expected'] is None) != (not row['relevant']) or (row['answer'] is None and row['supported']):
            raise ValueError('Deze oefening vereist bronlabels voor antwoordbare vragen en geen support bij onthouding')

def evaluate_rag(rows, k):
    validate_rag_rows(rows)
    metrics = [retrieval_metrics(r['ranking'], r['relevant'], k) for r in rows]
    answerable = [m for m in metrics if m['recall'] is not None]
    answered = [r for r in rows if r['answer'] is not None]
    unknown = [r for r in rows if r['expected'] is None]
    return {'n': len(rows), 'answerable': len(answerable), 'answered': len(answered),
            'recall': sum(m['recall'] for m in answerable) / len(answerable) if answerable else None,
            'mrr': sum(m['rr'] for m in answerable) / len(answerable) if answerable else None,
            'accuracy': sum(r['answer'] == r['expected'] for r in rows) / len(rows),
            'coverage': len(answered) / len(rows),
            'faithfulness': sum(r['supported'] for r in answered) / len(answered) if answered else None,
            'unknown_abstention': sum(r['answer'] is None for r in unknown) / len(unknown) if unknown else None}

def diagnose_rag(rows, k):
    validate_rag_rows(rows)
    result = []
    for row in rows:
        metrics = retrieval_metrics(row['ranking'], row['relevant'], k)
        if row['expected'] is None:
            reason = 'appropriate-abstention' if row['answer'] is None else 'unanswerable-answered'
        elif metrics['rr'] == 0:
            reason = 'retrieval-miss'
        elif row['answer'] is None:
            reason = 'unnecessary-abstention'
        elif row['answer'] != row['expected']:
            reason = 'wrong-answer'
        elif not row['supported']:
            reason = 'unsupported-answer'
        else:
            reason = 'supported-correct'
        result.append({'id': row['id'], 'reason': reason})
    return result

def make_rag_cases():
    # Own questions and explicit ground truth. These labels never enter run_rag.
    definitions = [
        ('dev', 'direct', 'Hoe reset ik de lamp?', 'lamp-reset'),
        ('dev', 'direct', 'Hoe lang duurt de garantie van de lamp?', 'lamp-garantie'),
        ('dev', 'direct', 'Hoe lang gaat de batterij van de sensor mee?', 'sensor-batterij'),
        ('dev', 'direct', 'Hoe reset ik de sensor?', 'sensor-reset'),
        ('dev', 'alias', 'Hoe herstart ik het licht?', 'lamp-reset'),
        ('dev', 'alias', 'Hoe lang gaat de accu van de sensor mee?', 'sensor-batterij'),
        ('dev', 'unknown', 'Welke kleur heeft de hub?', None),
        ('dev', 'unknown', 'Hoe zwaar is de maan?', None),
        ('test', 'direct', 'Hoe voer ik een reset van de hub uit?', 'hub-reset'),
        ('test', 'direct', 'Hoe lang duurt een update van de hub?', 'hub-update'),
        ('test', 'direct', 'Wat is de garantie op de lamp?', 'lamp-garantie'),
        ('test', 'alias', 'Hoe herstart ik de sensor?', 'sensor-reset'),
        ('test', 'unseen', 'Hoe start ik de lamp opnieuw?', 'lamp-reset'),
        ('test', 'unseen', 'Wanneer vervang ik de stroomcel van de sensor?', 'sensor-batterij'),
        ('test', 'unknown', 'Welke kleur heeft de lamp?', None),
        ('test', 'unknown', 'Wat is het weer morgen?', None),
    ]
    texts = {r['id']: r['text'] for r in RECORDS}
    return [{'id': f'q{i:02}', 'split': split, 'group': group, 'question': q,
             'relevant': [doc+':0'] if doc else [], 'expected': texts[doc] if doc else None}
            for i, (split, group, q, doc) in enumerate(definitions)]

cases=make_rag_cases()
variants={'literal':{'k':3,'budget':700,'expand':False},'aliases':{'k':3,'budget':700,'expand':True}}

def evaluate_rag_pipeline(index, cases, config, split):
    if split not in ('dev', 'test') or not isinstance(cases, list) or not 1 <= len(cases) <= 1000:
        raise ValueError('Begrensde cases en split dev/test vereist')
    ids, questions = set(), set()
    for c in cases:
        if not isinstance(c, dict) or set(c) != {'id', 'split', 'group', 'question', 'relevant', 'expected'} or any(not isinstance(c[k], str) or not c[k].strip() or len(c[k]) > 2000 for k in ['id', 'split', 'group', 'question']):
            raise ValueError('Ongeldig meetgeval')
        normalized = c['question'].strip().casefold()
        if c['id'] in ids or normalized in questions or c['split'] not in ('dev', 'test'):
            raise ValueError('Dubbele ID/vraag of onbekende split')
        ids.add(c['id']); questions.add(normalized)
        # Use the same ground-truth and shape validator as the evaluation lesson.
        validate_rag_rows([{'id': c['id'], 'ranking': [], 'relevant': c['relevant'],
                                      'answer': None, 'expected': c['expected'], 'supported': False}])
    rows, groups = [], {}
    for c in cases:
        if c['split'] != split:
            continue
        result = run_rag(index, c['question'], config)
        row = {'id': c['id'], 'ranking': result['ranking'], 'relevant': c['relevant'],
               'answer': result['answer'], 'expected': c['expected'], 'supported': result['quote_valid']}
        rows.append(row); groups.setdefault(c['group'], []).append(row)
    return {'split': split, 'rows': rows, 'summary': evaluate_rag(rows, config['k']),
            'groups': {g: evaluate_rag(rs, config['k']) for g, rs in sorted(groups.items())},
            'diagnoses': diagnose_rag(rows, config['k'])}

def choose_rag_variant(reports):
    if not isinstance(reports, dict) or not 1 <= len(reports) <= 8:
        raise ValueError('Geef 1 tot 8 devrapporten')
    expected_ids, ranked = None, []
    for name, report in reports.items():
        if not isinstance(name, str) or not name.strip() or len(name) > 100 or not isinstance(report, dict) or report.get('split') != 'dev':
            raise ValueError('Kies alleen benoemde devvarianten')
        rows = report.get('rows')
        # Selection uses answer accuracy and support, not a supplied summary.
        summary = evaluate_rag(rows, 1)
        ids = {row['id'] for row in rows}
        if expected_ids is not None and ids != expected_ids:
            raise ValueError('Alle varianten moeten dezelfde devvragen hebben')
        expected_ids = ids
        support = summary['faithfulness'] if summary['faithfulness'] is not None else -1.
        ranked.append((-summary['accuracy'], -support, name))
    return sorted(ranked)[0][2]

dev_reports={name:evaluate_rag_pipeline(index,cases,config,"dev") for name,config in variants.items()}
selected=choose_rag_variant(dev_reports)
print("Dev literal / aliases:",dev_reports["literal"]["summary"]["accuracy"],dev_reports["aliases"]["summary"]["accuracy"])
print("Gekozen:",selected)

import matplotlib.pyplot as plt

def final_rag_evaluation(index, cases, chosen, registry):
    import matplotlib.pyplot as plt
    if not isinstance(registry, dict) or not isinstance(chosen, str) or chosen not in registry:
        raise ValueError('Kies een bestaande variantnaam')
    report = evaluate_rag_pipeline(index, cases, registry[chosen], 'test')
    groups = list(report['groups'])
    scores = [report['groups'][g]['accuracy'] for g in groups]
    fig, ax = plt.subplots(figsize=(4.2, 4.4))
    bars = ax.bar(groups, scores, color='#236b4f')
    ax.set_ylim(0, 1.25); ax.set_ylabel('Exact goed\nof terecht onthouden', fontsize=12)
    ax.tick_params(labelsize=12)
    ax.set_title('Eigen RAG: bevroren variant\nop testvragen', fontsize=12)
    for group, bar, score in zip(groups, bars, scores):
        ax.text(bar.get_x() + bar.get_width()/2, score+.03,
                f'{score:.0%}\n(n={report["groups"][group]["n"]})', ha='center', fontsize=12)
    fig.tight_layout()
    return report

test_report=final_rag_evaluation(index,cases,selected,variants)
print('Test accuracy:',test_report['summary']['accuracy'])
print('Ongeziene formuleringen:',test_report['groups']['unseen']['accuracy'])
print('Diagnoses:',[d['reason'] for d in test_report['diagnoses']])
experiment_prompt='Beantwoord de vraag op basis van de broncontext. Noem het bronlabel. Onthoud je bij ontbrekend bewijs.\nVraag: Hoe herstart ik het licht?\nBroncontext:\n'+answer['context']['text']
assert len(experiment_prompt)<=1000
print('Prompt voor vrij modelonderzoek:\n'+experiment_prompt)