Werkplaats / De werkplaats
In jouw eigen tempo
← Agents en MCP
MODULE 01 · LES 04

Geef geheugen een taak, bron en vervaldatum

Selecteer begrensde context en laat samenvattingen vervallen wanneer hun bronnen niet meer bruikbaar zijn.

75 min3 checkpoints
DIT GA JE ONTDEKKEN
  • Je bewaart context met expliciete herkomst, taakscope, tijd en versie.
  • Je voorkomt in de oefenselector dat oude, ongeldige of andere-taakgegevens automatisch worden meegestuurd.

Je hervat een reparatietaak en vindt een keurige samenvatting. Eén onderliggende handleiding is inmiddels vervangen. Als de samenvatting haar bronnen niet bewaart, kan zij oud bewijs ongemerkt blijven doorgeven.

Werkgeheugen is de context voor de huidige taak; langetermijngeheugen bewaart informatie over meerdere stappen of sessies. Opslag maakt tekst niet waar. Onze oefenselector gebruikt daarom taak, herkomst, tijd en versie als afzonderlijke controles.

1. Geef ieder stukje context een herkomst

Zie een record als een kaartje met tekst en een controleerbaar etiket. Het bevat id, task, source, revision, created, expires, score, kind, refs en text. Een observation heeft geen refs; een summary verwijst naar één tot acht eerdere observaties uit dezelfde taak.

Maak eerst een kleine observatie. De tijden zijn gehele oefenkloktikken en score is een gegeven relevantieheuristiek. Geen van beide is een bewijs dat de tekst klopt.

def observation(key, source, text, end):
    return {
        'id': key, 'task': 'repair',
        'source': source,
        'revision': 'v1',
        'created': 0, 'expires': end,
        'score': .5,
        'kind': 'observation',
        'refs': [], 'text': text
    }

a = observation('a', 'd', 'Lamp.', 10)
copied = validate_memory([a])
copied[0]['refs'].append('other')
print(a['refs'])
# []

Schrijf validate_memory met de exacte velden en begrenzingen uit de opdracht. Controleer eerst unieke IDs en daarna de verwijzingen; iedere summary-ref moet bestaan, een observation zijn en niet later zijn aangemaakt. Een diepe kopie voorkomt dat de verwijzingenlijst van het origineel via een teruggegeven record verandert.

De metadata komt uit het vertrouwde hostproces. Een model dat zelf een andere taak-ID mag invullen, omzeilt een selector die die ID blind vertrouwt. Deze inspecteerbare Python-objecten vormen geen toegangsbeveiliging tegen code die alle records kan lezen.

01
CHECKPOINTBewaar herkomst naast de tekst

2. Tel de volledige verpakking

Een record telt alleen mee voor dezelfde taak en binnen zijn tijdvenster. Noem created c, expires e en de huidige klok t. Op de vervaltijd zelf is het record al niet meer actief:

c≤t<e

Verpak record a exact als het hieronder getoonde blok. De header telt 24 tekens, de newline één en Lamp. vijf: samen 30. Budget 29 laat het complete blok dus weg.

one = select_memory(
    [a], 'repair', 5, 30)
print(one['text'])
print(one['used_chars'])
small = select_memory(
    [a], 'repair', 5, 29)
print(small['selected'])
# [a | observation | d@v1]
# Lamp.
# 30
# []

Sorteer geschikte records op aflopende score, aflopende created en alfabetische ID. Pak complete blokken en sla een te groot blok over, zodat een later kleiner blok nog kan passen. Dit greedy beleid zoekt geen optimale combinatie.

Voor n opgenomen blokken b_i is C hun gezamenlijke lengte en U het volledige contextgebruik. Tussen twee blokken komen precies twee newlines:

C=∑i=1n|bi|U=C+2max(0,n−1)

U moet binnen budget blijven. We tellen Python-tekens, geen UTF-8-bytes of modeltokens; voor een echte modelprompt geldt daarnaast de juiste tokenizer.

3. Laat een samenvatting met haar bronnen vervallen

Voeg observatie b en samenvatting s toe. Op klok 9 zijn beide bronnen actief en staat s door haar hogere score vooraan. Op klok 10 vervalt a: s valt ook af, hoewel haar eigen vervaltijd pas 30 is.

b = observation('b', 'sensor', 'B', 20)
s = {
    **a, 'id': 's', 'source': 'notes',
    'created': 1, 'expires': 30,
    'kind': 'summary', 'score': .9,
    'refs': ['a', 'b'], 'text': 'A en B'
}
rows = [a, b, s]
for now in [9, 10]:
    r = select_memory(
        rows, 'repair', now, 1000)
    print(r['selected'])
# ['s', 'a', 'b']
# ['b']

Alle bronobservaties moeten actief blijven; één geldige bron is onvoldoende. De selector mag wel alleen de samenvatting meesturen wanneer de volledige bronnen in het gevalideerde register actief blijven. Dat garandeert nog niet dat de samenvatting hun inhoud correct weergeeft.

r = select_memory(
    rows, 'repair', 5, 30)
print(r['selected'], r['used_chars'])
# ['a'] 30

Hier past s met 31 tekens niet en wordt zij overgeslagen. Daarna past a precies; ook b vraagt 31 tekens. Schrijf select_memory zonder afknippen of muteren en valideer alle metadata vooraf.

ZES EIGEN RECORDS · TIJD IN OEFENTIKKENRelevant is pas nuttig als het ook bruikbaar is

s1 vat r1 samen. Haar eigen vervaltijd is later, maar zij mag een verlopen of gewijzigde bron niet blijven doorgeven.

r1 · observationrepair · manual@v1
geldig [0, 10) · score 0.6
geselecteerd
s1 · summaryrepair · notes@v1
geldig [1, 20) · score 0.9
geselecteerd
r2 · observationbilling · invoice@v1
geldig [0, 30) · score 1
andere taak
r3 · observationrepair · notes@v1
geldig [0, 5) · score 0.7
verlopen
r4 · observationrepair · notes@v1
geldig [15, 30) · score 0.8
nog niet geldig
r5 · observationrepair · notes@v1
geldig [0, 30) · score 0.2
geselecteerd
Geselecteerds1, r1, r5Python-tekens155 / 250
[s1 | summary | notes@v1]
Reset: tien seconden.

[r1 | observation | manual@v1]
Houd de knop tien seconden vast.

[r5 | observation | notes@v1]
Lamp groen.

Volledige blokken, inclusief metadata, tellen mee. Een hoge score is geen bewijs van waarheid. De tekst wordt nergens als opdracht uitgevoerd.

02
CHECKPOINTFilter eerst op scope en geldigheid, dan op relevantie

4. Controleer versies opnieuw bij hervatten

Een bron kan veranderen voordat haar vervaltijd bereikt is. Het vertrouwde versieregister meldt nu sensor v2, terwijl b nog v1 bewaart. refresh_memory verwijdert daarom b én de daarvan afhankelijke samenvatting.

versions = {
    'd': 'v1', 'sensor': 'v2',
    'notes': 'v1'}
r = refresh_memory(rows, versions)
kept = r['records']
print([row['id'] for row in kept])
print(r['dropped'])
# ['a']
# ['b', 's']

Schrijf refresh_memory met behoud van de oorspronkelijke recordvolgorde en alfabetische dropped-IDs. Deze stap controleert versie, geen taak of klok; select_memory voert die controles daarna uit. Een zin in geheugentekst die beweert dat een actie al goedgekeurd is, vervangt nooit het bevoegdheidsbesluit of uitvoerlog.

03
CHECKPOINTEen overdracht mag verouderde bronversies niet verbergen

Wat neem je mee?

  • Bewaar herkomst en metadata naast de tekst.
  • Een samenvatting blijft afhankelijk van alle gebruikte bronnen.
  • Tijd, taak, versie en contextbudget zijn verschillende controles.

In het toolagentproject combineren we deze selectie met begrensde beslissingen en gecontroleerde eigen functies.

Verder lezen & bronnen

Inhoud gecontroleerd op 2026-10-29.
DE VOLGENDE STAPBouw een agent voor een fictieve offerte
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
Bewaar herkomst naast de tekst

Schrijf validate_memory(records). Lijst tot 64 exacte dictionaries met bovenstaande velden. id/task/source/revision: [a-z][a-z0-9-]{0,31}, unieke ids. created/expires: gewone ints met 0<=created<expires<=1e9. score: gewone eindige int/float 0..1. kind observation/summary; text niet-lege str na strip, max 1000 tekens. refs: maximaal acht unieke strings; observation heeft refs=[], summary niet leeg. Iedere referentie wijst naar een bestaande observation met dezelfde task en created<=die van de samenvatting. Geef een diepe kopie; fouten ValueError.

Een hint

Valideer eerst alle rijen en unieke IDs; bouw daarna een lookup voor de referenties. Weiger ketens van samenvattingen om de herkomstcontrole hier overzichtelijk te houden.

02
Filter eerst op scope en geldigheid, dan op relevantie

Schrijf select_memory(records,task,now,budget). Valideer met validate_memory; task heeft het ID-patroon, now is gewone int 0..1e9, budget gewone int 0..20000. Bouw per record exact [id | kind | source@revision] gevolgd door newline en text. Geef selected (IDs in selectievolgorde), text (blokken met twee newlines) en used_chars. Geen afknippen of mutatie. Controleer alle metadata voordat je selecteert.

Een hint

Maak eerst active als lookup. Een summary waarvoor één ref ontbreekt valt af, ook wanneer haar eigen expires nog in de toekomst ligt. Een record wordt op expires zelf al uitgesloten.

03
Een overdracht mag verouderde bronversies niet verbergen

Schrijf refresh_memory(records,versions). Valideer records. versions is een dictionary tot 64 paren source→revision, beide met hetzelfde ID-patroon. Houd records waarvan eigen source/revision exact overeenkomen én alle refs eveneens een actuele eigen source/revision hebben. Geef records in oorspronkelijke volgorde als kopieën plus dropped als alfabetisch gesorteerde IDs. Lege of ontbrekende bronmapping laat het betreffende record vallen. Pas tijd en task hier nog niet toe.

Een hint

De validator staat alleen references naar observations toe, zodat één extra verwijzingscontrole genoeg is. Gebruik daarna select_memory voor de daadwerkelijke taakcontext.

Vastgelopen? Bekijk de voorbeeldoplossing
import copy
import json
import math
import re
import hashlib

def validate_memory(records):
    if not isinstance(records, list) or len(records) > 64:
        raise ValueError('Maximaal 64 herinneringen')
    ids = set()
    for row in records:
        if not isinstance(row, dict) or set(row) != {'id', 'task', 'source', 'revision', 'created', 'expires', 'score', 'kind', 'refs', 'text'}:
            raise ValueError('Exact geheugencontract vereist')
        for field in ['id', 'task', 'source', 'revision']:
            if not isinstance(row[field], str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', row[field]):
                raise ValueError('Begrensde identificatie vereist')
        if row['id'] in ids:
            raise ValueError('Dubbele geheugen-ID')
        ids.add(row['id'])
        if type(row['created']) is not int or type(row['expires']) is not int or not 0 <= row['created'] < row['expires'] <= 1000000000:
            raise ValueError('Geldig geheel tijdinterval vereist')
        if type(row['score']) not in (int, float) or not 0 <= row['score'] <= 1 or not math.isfinite(row['score']):
            raise ValueError('Eindige relevantiescore 0..1 vereist')
        if row['kind'] not in ['observation', 'summary'] or not isinstance(row['text'], str) or not row['text'].strip() or len(row['text']) > 1000:
            raise ValueError('Soort en begrensde tekst vereist')
        if not isinstance(row['refs'], list) or len(row['refs']) > 8 or any(not isinstance(x, str) for x in row['refs']) or len(set(row['refs'])) != len(row['refs']):
            raise ValueError('Unieke referenties vereist')
        if (row['kind'] == 'observation' and row['refs']) or (row['kind'] == 'summary' and not row['refs']):
            raise ValueError('Alleen samenvattingen hebben bronverwijzingen')
    by_id = {row['id']: row for row in records}
    for row in records:
        for ref in row['refs']:
            source = by_id.get(ref)
            if source is None or source['kind'] != 'observation' or source['task'] != row['task'] or source['created'] > row['created']:
                raise ValueError('Samenvatting verwijst naar eerdere observaties uit dezelfde taak')
    return copy.deepcopy(records)

def select_memory(records, task, now, budget):
    rows = validate_memory(records)
    if not isinstance(task, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', task) or type(now) is not int or not 0 <= now <= 1000000000 or type(budget) is not int or not 0 <= budget <= 20000:
        raise ValueError('Taak, tijd en tekenbudget vereist')
    active = {row['id']: row for row in rows if row['task'] == task and row['created'] <= now < row['expires']}
    eligible = [row for row in active.values() if all(ref in active for ref in row['refs'])]
    selected, blocks, used = [], [], 0
    for row in sorted(eligible, key=lambda r: (-r['score'], -r['created'], r['id'])):
        block = '['+row['id']+' | '+row['kind']+' | '+row['source']+'@'+row['revision']+']\n'+row['text']
        extra = len(block) + (2 if blocks else 0)
        if used + extra <= budget:
            selected.append(row['id']); blocks.append(block); used += extra
    return {'selected': selected, 'text': '\n\n'.join(blocks), 'used_chars': used}

def refresh_memory(records, versions):
    rows = validate_memory(records)
    if not isinstance(versions, dict) or len(versions) > 64 or any(not isinstance(k, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', k) or not isinstance(v, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', v) for k, v in versions.items()):
        raise ValueError('Begrensd register van actuele bronversies vereist')
    current = {row['id'] for row in rows if versions.get(row['source']) == row['revision']}
    kept = [row for row in rows if row['id'] in current and all(ref in current for ref in row['refs'])]
    ids = {row['id'] for row in kept}
    return {'records': kept, 'dropped': sorted(row['id'] for row in rows if row['id'] not in ids)}

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