Werkplaats / De werkplaats
In jouw eigen tempo
← RAG
MODULE 02 · LES 06

Maak ieder citaat terugvindbaar

Pak een begrensde context, controleer exacte citaten en onderscheid een geldige verwijzing van inhoudelijk bewijs.

75 min3 checkpoints
DIT GA JE ONTDEKKEN
  • Je bewaart de exacte bronversie en posities bij geselecteerde passages.
  • Je controleert verwijzingsgeldigheid apart van ondersteuning van een claim.

Probeer ook echte tekstvectoren in het embedding-lab.

Een bronlabel achter een zin ziet overtuigend uit. Toch kan het naar een andere versie wijzen, een verkeerd stukje tekst citeren of niets over de claim zeggen. We controleren die vragen afzonderlijk.

De vorige les koos een passage. Nu bewaren we haar identiteit, vullen we een begrensde context en meten we verwijzingen. Inhoudelijke ondersteuning komt in deze oefening uit een handmatig vastgelegde supporttabel.

1. Tel de verpakking mee

Een context is meer dan losse brontekst. We geven ieder opgenomen blok een label, document-ID, versieprefix en halfopen tekeninterval. De volledige versiehash blijft daarnaast in de citaatmetadata staan.

Met twee korte bronnen kunnen we ieder teken narekenen. De bron “Een lamp.” heeft negen Python-codepunten; de header en newline maken het eerste blok 35 tekens. Het tweede blok is 38 tekens.

records = [
    dict(id='a', title='Lamp',
         text='Een lamp.'),
    dict(id='b', title='Sensor',
         text='Een sensor.')
]
docs = document_snapshots(records)
chunks = [word_chunks(d)[0]
          for d in docs]
full = pack_context(chunks, docs, 1000)
print(full['text'])
print(full['used_chars'])
# [S1] a@154351ed61a0 [0,9)
# Een lamp.
#
# [S2] b@e5ec903db522 [0,11)
# Een sensor.
# 75

Tussen blokken staan twee newlines: 35+2+38=75. Bij budget 74 past het eerste blok, maar het tweede niet. Een vergeten separator laat het programma ongemerkt over zijn eigen limiet gaan.

Identieke tekst nemen we eenmaal op, ook als twee document-IDs dezelfde tekst bevatten. Alleen opgenomen blokken krijgen een volgend S-label. Een niet-passende passage wordt overgeslagen; een latere, kleinere passage mag nog passen.

small = pack_context(chunks, docs, 74)
print(small['used_chars'])
base = dict(title='T', text='Een lamp.')
copies = document_snapshots(
    [dict(base, id=i)
     for i in ['a', 'b']])
same = [word_chunks(d)[0]
        for d in copies]
one = pack_context(same, copies, 1000)
print(len(one['citations']))
# 35
# 1

Schrijf pack_context met de gegeven snapshot- en chunkhelpers. Valideer alle chunks vooraf, ook degene die later niet passen. Verander de exacte bronuitsneden niet.

Dit greedy beleid zoekt geen optimale combinatie. Overlappende maar verschillende teksten blijven aparte blokken. Ons budget telt codepunten; een echt model vraagt een tokenbudget voor instructies, vraag, bronnen én antwoordreserve.

01
CHECKPOINTEen contextbudget telt ook verpakking en tussenruimte

2. Controleer identiteit vóór inhoud

Een verwijzing kan structureel geldig zijn en toch nergens passen. verify_citations controleert eerst documentbestaan, daarna de volledige versiehash, dan het interval en ten slotte de exacte substring.

Een oude versie krijgt dus stale-source, ook als haar posities óók ongeldig zijn. Zo blijft duidelijk welke controle eerst mislukte. Structurele fouten, zoals een dubbel bronlabel of een boolean als positie, geven ValueError.

c = full['citations'][0]
alterations = [
    {},
    dict(document_id='unknown'),
    dict(revision='0'*64, start=-1),
    dict(start=-1),
    dict(quote='Andere woorden.')
]
for edits in alterations:
    result = verify_citations(
        [{**c, **edits}], docs)
    print(result[0]['status'])
# valid
# unknown-source
# stale-source
# invalid-span
# quote-mismatch

Schrijf de controle in deze volgorde. Zoek niet opnieuw naar vergelijkbare tekst en normaliseer geen witruimte: daarmee zou je een verkeerde verwijzing verbergen. De offsets zijn Python-stringposities, geen UTF-8-byteposities.

EIGEN BRON · VERSIE EN BEWIJS APARTEen geldig citaat is niet altijd bewijs voor de claim

Verander één onderdeel en volg de technische controle. Ondersteuning is voor deze voorbeelden afzonderlijk met de hand beoordeeld.

Huidige bron: lamp · versie 14a0cc3959ad
Reset duurt tien seconden.
De lamp brandt groen.

Verwijzing S1

Document: lamp
Bronversie: 14a0cc3959ad
Tekeninterval: [0, 26)

Reset duurt tien seconden.

Technische controles

  1. Document bestaatgoed
  2. Volledige versie kloptgoed
  3. Tekeninterval is geldiggoed
  4. Exacte substring kloptgoed

Te controleren claimReset duurt tien seconden.

VerwijzinggeldigOndersteuning claimja

Een hash controleert de vastgelegde tekstversie, niet de waarheid of betrouwbaarheid van de bron. De inhoud wordt nooit als opdracht uitgevoerd.

02
CHECKPOINTEen bestaand label kan nog naar een verkeerd citaat wijzen

3. Scheid geldige verwijzingen van bewijs

Stel dat drie claims samen drie verwijzingen plaatsen. S1 en S2 zijn technisch geldig; S9 is niet vastgelegd. Alleen S1 ondersteunt in onze handlabels de claim dat er een lamp is.

De resetclaim citeert S2, maar “Een sensor.” geeft geen resetinstructie. De kleurclaim citeert niets. Technische geldigheid, passend bewijs en claimdekking krijgen daarom verschillende tellers.

Noem T het totale aantal verwijzingen, V het aantal geldige en S het aantal geldige ondersteunende verwijzingen. N is het aantal claims en G het aantal claims met minstens één passende geldige verwijzing.

R=VTP=STC=GN

Hier zijn T=3, V=2, S=1, N=3 en G=1. Dat geeft reference_validity R=2/3, support_precision P=1/3 en claim_coverage C=1/3.

claims = [
    dict(id='lamp', refs=['S1', 'S9']),
    dict(id='reset', refs=['S2']),
    dict(id='kleur', refs=[])
]
support = {'lamp': ['S1'], 'reset': []}
scores = citation_scores(
    claims, ['S1', 'S2'], support)
for value in scores.values():
    print(round(value, 3))
empty = citation_scores([], [], {})
print(list(empty.values()))
# 0.667
# 0.333
# 0.333
# [None, None, None]

Schrijf citation_scores met de afzonderlijke noemers. Een onbekende of ongeldige verwijzing kan nooit ondersteuning tellen, ook als de supporttabel haar noemt. Bij een nulnoemer geeft de betreffende ratio None.

Onze handlabels nemen aan dat één passende bron voldoende is. Tegenstrijdige bronnen of claims die meerdere bewijsstukken vereisen, vragen een rijker protocol. Een hash bewijst bovendien geen betrouwbare afzender of waarheid.

03
CHECKPOINTMeet verwijzingen én de claims die ze ondersteunen

Wat neem je mee?

  • Contextbudgetten bevatten ook headers, scheidingstekens en exacte brontekst.
  • Een geldig citaat is technisch terugvindbaar; inhoudelijke ondersteuning is een aparte beoordeling.
  • Verwijzingsratio’s tellen links, claimdekking telt claims.

Hierna meten we retrieval en antwoorden samen, met zichtbaar verschillende foutcategorieën.

DE VOLGENDE STAPZoek uit waar de RAG-pijplijn faalt
JOUW EXPERIMENT

Probeer het zelf

Python
main.py

Python start zodra jij code uitvoert.

UITVOER

Je uitvoer verschijnt hier.
Alles rekent in je eigen browser.

Jouw checkpoints0 / 3
01
Een contextbudget telt ook verpakking en tussenruimte

Schrijf pack_context(chunks,documents,budget). De gegeven helpers valideren snapshots en exacte chunks. Maximaal 256 chunks; budget echte int 0–60000. Valideer alle chunks vooraf, ook als ze niet passen. Blokformaat: [S1] doc@eerste12hash [start,end) gevolgd door newline en exacte tekst. Verbind blokken met twee newlines. Geef text, used_chars en citations met source_id/document_id/revision/start/end/quote. Nummer alleen opgenomen blokken. Ongeldig: ValueError.

Een hint

Vergelijk len(candidate) met budget. Bewaar geen strip-versie van de bron. De geleverde helpers zijn eerder gebouwde code; je schrijft nu alleen het inpakken.

02
Een bestaand label kan nog naar een verkeerd citaat wijzen

Schrijf verify_citations(citations,documents), maximaal 256 exacte dictionaries met de zes velden uit pack_context. source_id uniek en S[1-9][0-9]{0,3}; document_id [A-Za-z0-9_-]{1,40}; revision 64 lowercase hex; start/end echte ints; quote string 1–20000 tekens. Structureel ongeldig: ValueError. Geef per citaat source_id/status met, in volgorde: unknown-source, stale-source, invalid-span, quote-mismatch of valid. Gebruik checked_sources voor de snapshots.

Een hint

Stale-source gaat vóór de positiecontrole. Gebruik source.text[start:end] zonder normaliseren of opnieuw zoeken; anders verberg je een foutieve verwijzing.

03
Meet verwijzingen én de claims die ze ondersteunen

Schrijf citation_scores(claims,valid_ids,support). Maximaal 128 unieke claims exact id/refs. support maximaal 128 claim-ID→lijst-bronlabels. IDs niet-lege strings ≤100; iedere lijst maximaal 256 unieke IDs. valid_ids bevat technisch geldige labels. Onbekende claim in support betekent geen ondersteuning. Geef reference_validity, support_precision, claim_coverage; None bij nulnoemer. Anders ValueError.

Een hint

Neem per claim refs ∩ valid_ids ∩ support[id]. Een bronlabel dat niet bestaat kan nooit als ondersteund meetellen, ook niet als de supporttabel het noemt.

Vastgelopen? Bekijk de voorbeeldoplossing
import re
import hashlib
import numpy as np

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 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 citation_scores(claims, valid_ids, support):
    def identifiers(items):
        return isinstance(items, list) and len(items) <= 256 and all(isinstance(i, str) and 1 <= len(i) <= 100 for i in items) and len(set(items)) == len(items)
    if not isinstance(claims, list) or len(claims) > 128 or not identifiers(valid_ids) or not isinstance(support, dict) or len(support) > 128:
        raise ValueError('Begrensde claims, geldige bronlabels en supporttabel vereist')
    if any(not isinstance(k, str) or not 1 <= len(k) <= 100 or not identifiers(v) for k, v in support.items()):
        raise ValueError('Supporttabel gebruikt claim-IDs en unieke bronlabels')
    seen, total, valid, supporting, covered = set(), 0, 0, 0, 0
    for claim in claims:
        if not isinstance(claim, dict) or set(claim) != {'id', 'refs'} or not isinstance(claim['id'], str) or not 1 <= len(claim['id']) <= 100 or claim['id'] in seen or not identifiers(claim['refs']):
            raise ValueError('Elke claim heeft een unieke ID en unieke refs')
        seen.add(claim['id'])
        refs, allowed = set(claim['refs']), set(support.get(claim['id'], []))
        good = refs & set(valid_ids)
        total += len(refs); valid += len(good); supporting += len(good & allowed); covered += bool(good & allowed)
    return {'reference_validity': valid / total if total else None,
            'support_precision': supporting / total if total else None,
            'claim_coverage': covered / len(claims) if claims else None}

Voer de code uit en verklaar voor jezelf waarom elke stap werkt.