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

Laat een agent kiezen, maar begrens de lus

Scheid beleid, tooluitvoering en observaties; stop bij voltooiing, fouten of een uitgeput budget.

75 min3 checkpoints
DIT GA JE ONTDEKKEN
  • Je bouwt een begrensde lus rond een verwisselbare policy en echte lokale functies.
  • Je verklaart waarom een final-bericht, eindige lus en geldige JSON nog geen juist of veilig antwoord bewijzen.

Een assistent die een antwoord moet opzoeken, kan na een zoekresultaat besluiten nog iets te berekenen. Die tweede keuze vraagt een nieuwe ronde. Zonder stopregels kan dezelfde opdracht blijven terugkomen, ook wanneer iedere losse toolaanroep geldig is.

Een workflow volgt vooraf vastgelegde codepaden; bij een agent kiest een model tijdens de uitvoering vervolgstappen. Hier vervangen gewone Python-regels dat model, zodat je de lus precies kunt volgen. De lokale optelling rekent echt; onze policy is een deterministische testdubbel.

1. Maak van een wens een klein voorstel

Denk aan een loket: de aanvrager noemt een handeling, maar bedient zelf geen machines. Onze policy mag een tool voorstellen of met final stoppen. Een syntactisch geldig voorstel geeft nog geen uitvoerrecht en garandeert geen juist antwoord.

Schrijf validate_agent_action met de twee exacte vormen uit de opdracht. Toolargumenten zijn maximaal acht JSON-scalars; containers vallen af en gewone getallen blijven eindig binnen ±1e6. Houd bool apart van int en controleer de grootte vóór math.isfinite.

proposal = {
    'kind': 'tool', 'call_id': 'c1',
    'name': 'add',
    'arguments': {'a': 7, 'b': 8}
}
action = validate_agent_action(proposal)
proposal['arguments']['a'] = 99
print(action['arguments']['a'])
# 7

De kopie bewaart de beoordeelde argumenten wanneer het oorspronkelijke voorstel later verandert. Bij deze vlakke scalarargumenten volstaat een nieuwe argumentendictionary. Ongeldige invoervormen geven ValueError.

01
CHECKPOINTMaak de mogelijke beslissingen klein en expliciet

2. Volg voorstel, uitvoering en observatie

Een observatie beschrijft wat de host werkelijk terugkreeg. De volgende policy vraagt eenmaal de optelling aan en maakt daarna van het resultaat een finaal bericht. Je kunt beide rondes met de hand voorspellen: eerst 15 berekenen, daarna stoppen met tekst 15.

def policy(observations):
    if not observations:
        return action
    data = observations[0]['data']
    return {
        'kind': 'final',
        'text': str(data)
    }

tools = {'add': lambda a, b: a + b}
result = run_bounded_agent(
    policy, tools)
print(result['status'],
      result['answer'])
print(result['tool_attempts'])
# final 15
# 1

Volg in run_bounded_agent steeds dezelfde volgorde: policy en validatie, final, dubbele call-ID, toolbudget, uitvoering en observatie. Final mag dus stoppen als het toolbudget al op is; een herhaalde ID wordt eerder afgewezen dan een nieuwe poging met uitgeput budget. Zonder final eindigt de laatste toegestane ronde met decision-budget als geen andere stopreden optreedt.

Bewaar observaties met call_id, name, ok en data. De tooltrace bevat step, kind, call_id, name, executed en ok; final en rejected hebben hun eigen compacte tracevorm uit de opdracht. Geef de policy een diepe observatiekopie, zodat ook een geneste tooluitkomst het bewaarde log niet kan laten herschrijven.

Stel dat een tool geneste data met count=15 teruggeeft. Verandert de policy haar kopie naar 900, dan moet het eigen log 15 bewaren; alleen de buitenste dictionary kopiëren beschermt die geneste data niet.

Gebruik deze pseudocode voor de lus; de opdracht specificeert de precieze dictionaries:

voor iedere toegestane beslisronde:
  vraag en valideer; fout -> policy-error
  final -> bewaar finalstap en stop
  dubbele ID -> duplicate-call
  geen toolpogingen over -> tool-budget
  tel poging; voer bekende tool uit
  onbekend -> foutobservatie unknown-tool
  exception/te groot -> tool-error
  bewaar observatie en tooltracestap
na laatste ronde: decision-budget
EIGEN TESTPOLICY · LOKALE OPTELLINGEen lus heeft meerdere manieren om te stoppen

Vraag: bereken 7 + 8. Kies het beleid en volg elke beslissing. Ook een mislukte poging verbruikt budget.

  1. 1
    c1 · add

    15

    Functie uitgevoerd
  2. 2
    final

    15

    Geen functie uitgevoerd
StopredenfinalAntwoord: 15
Beslissingen2Toolpogingen1Uitvoeringen1Werkeenheden7

Elke beslissing kost 2, elke uitgevoerde functie 3 fictieve eenheden. Dit is geen tijdmeting. Direct stoppen met 16 is goedkoop én onjuist.

3. Twee budgetten begrenzen verschillend werk

Een onbekende tool of mislukte uitvoering verbruikt wel een toolpoging. Alleen een aangeroepen geregistreerde functie telt als executed, ook wanneer zij een exception gaf. Policyfouten stoppen als policy-error; toolfouten worden de generieke observatie tool-error.

def repeat(observations):
    n = len(observations)
    return {
        'kind': 'tool',
        'call_id': 'c'+str(n),
        'name': 'add',
        'arguments': {'a': 7, 'b': 8}
    }

r = run_bounded_agent(
    repeat, tools, 3, 1)
print(r['status'], len(r['trace']))
print(r['tool_attempts'])
print(run_bounded_agent(
    policy, tools, 1, 4)['status'])
# tool-budget 2
# 1
# decision-budget

Het beslisbudget begrenst het aantal rondes, maar onderbreekt geen hangende callback. Gebruik uitsluitend bekende, begrensde functies; de browserworker heeft daarnaast een eigen uitvoeringstermijn. JSON-conversie maakt willekeurige Python geen veilige sandbox.

De resultaatgrens telt 4096 UTF-8-bytes, niet tekens. Zelfs de JSON-aanhalingstekens tellen mee:

encoded = json.dumps(
    'é'*2200, ensure_ascii=False,
    separators=(',', ':'))
raw = encoded.encode('utf-8')
print(len(encoded), len(raw))
# 2202 4402
02
CHECKPOINTEen voorstel wordt pas na controle een toolaanroep

4. Reken de trace na

Eén optelling plus final kost twee beslissingen en één uitvoering. Met twee werkeenheden per beslissing en drie per optelling is dat 2×2+3=7. Dit zijn gekozen oefenkosten, geen europrijzen, tokens of gemeten rekentijd.

Noem het aantal beslissingen D, hun vaste kosten c en het aantal uitgevoerde tools T. Iedere uitvoering i kost t_i; het totale werk W is:

W=Dc+∑i=1Tti
work = trace_work(
    result['trace'], 2, {'add': 3})
print(work['decisions'],
      work['executed_tools'])
print(work['work_units'])
# 2 1
# 7

Schrijf trace_work voor de opeenvolgende interne tracestappen. Een afgewezen of onbekende toolpoging kost nog steeds een beslissing. Vergelijk bij echte agentproeven ook juistheid en stopreden: minder werk maakt een verkeerd antwoord niet beter.

03
CHECKPOINTMeet de lus voordat je autonomie toevoegt

Wat neem je mee?

  • Voorstel, tooluitvoering en observatie zijn afzonderlijke stappen.
  • Beslisbudget en toolbudget voorkomen verschillende herhalingen.
  • Een final-bericht zegt dat de lus stopt, niet dat het antwoord klopt.

Hierna voegen we expliciete toestemming en herkenning van herhaalde schrijfacties toe.

Verder lezen & bronnen

Inhoud gecontroleerd op 2026-10-29.
DE VOLGENDE STAPBind toestemming aan de exacte handeling
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
Maak de mogelijke beslissingen klein en expliciet

Schrijf validate_agent_action(value). final heeft exact kind/text: niet-lege tekst na strip, maximaal 2000 tekens. tool heeft exact kind/call_id/name/arguments. Call-ID en naam voldoen aan [A-Za-z][A-Za-z0-9_-]{0,39}; maximaal acht argumenten met sleutels [A-Za-z][A-Za-z0-9_]{0,31}. Waarden: str tot 2000 tekens, gewone int/float eindig binnen ±1e6, bool of None; geen containers. Geef een kopie met gekopieerde arguments; anders ValueError.

Een hint

Gebruik type(x) voor getallen zodat bool apart blijft. Controleer eerst de magnitude; math.isfinite op een extreem groot Python-int kan zelf falen.

02
Een voorstel wordt pas na controle een toolaanroep

Schrijf run_bounded_agent(policy,tools,max_decisions=8,max_tool_calls=4). Contract en pseudocode staan hierboven; beslisbudget is gewone int 1..32, toolbudget 0..16, tools is een dictionary met maximaal acht geldige namen en callables. Retourneer status,answer,observations,trace,tool_attempts. Statussen: final,policy-error,duplicate-call,tool-budget,decision-budget. Trace per ronde: step vanaf 0, kind final; of kind rejected met reason invalid-policy-action/duplicate-call-id/tool-budget; of kind tool met call_id,name,executed,ok. Observaties hebben call_id,name,ok,data. Zie de tests voor precieze velden. Valideer elk voorstel met validate_agent_action.

Een hint

Volg de volgorde: policy en validatie → final → dubbele ID → toolbudget → uitvoering → observatie. Voeg bij uitgeput beslisbudget geen extra ronde toe. executed betekent dat de geregistreerde functie is aangeroepen, ook als zij een fout gaf.

03
Meet de lus voordat je autonomie toevoegt

Schrijf trace_work(trace,decision_units,tool_units). Accepteer maximaal 32 rijen met opeenvolgende gehele step vanaf 0 en kind tool/final/rejected. Iedere rij kost decision_units (gewone int 0..1000). Een toolrij heeft string name en bool executed; tel bij True een uitvoering plus de bijbehorende gehele toolkosten 0..1000. tool_units is een dictionary met maximaal acht stringnamen. Ontbrekende kosten bij een uitgevoerde tool zijn ValueError. Geef decisions,executed_tools,work_units. Verwerk alleen het eigen interne traceformaat; dit is geen algemene logvalidator.

Een hint

Onbekende, niet-uitgevoerde toolpogingen kosten wel een beslissing. Een functie die een exception geeft is wel uitgevoerd.

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

def validate_agent_action(value):
    if not isinstance(value, dict):
        raise ValueError('Actie is een dictionary')
    if value.get('kind') == 'final':
        if set(value) != {'kind', 'text'} or not isinstance(value['text'], str) or not value['text'].strip() or len(value['text']) > 2000:
            raise ValueError('Final heeft uitsluitend niet-lege tekst tot 2000 tekens')
        return dict(value)
    if value.get('kind') != 'tool' or set(value) != {'kind', 'call_id', 'name', 'arguments'}:
        raise ValueError('Toolactie heeft kind/call_id/name/arguments')
    if any(not isinstance(value[k], str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', value[k]) for k in ['call_id', 'name']):
        raise ValueError('Begrensde ASCII-call-ID en toolnaam vereist')
    args = value['arguments']
    if not isinstance(args, dict) or len(args) > 8:
        raise ValueError('Maximaal acht argumenten')
    for key, item in args.items():
        if not isinstance(key, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,31}', key):
            raise ValueError('Ongeldige argumentnaam')
        if type(item) is str:
            if len(item) > 2000:
                raise ValueError('Argumenttekst te lang')
        elif type(item) in (int, float):
            if abs(item) > 1e6 or not math.isfinite(item):
                raise ValueError('Getal buiten bereik')
        elif item is not None and type(item) is not bool:
            raise ValueError('Alleen JSON-scalarargumenten in deze oefening')
    return {**value, 'arguments': dict(args)}

def run_bounded_agent(policy, tools, max_decisions=8, max_tool_calls=4):
    if not callable(policy) or not isinstance(tools, dict) or len(tools) > 8 or any(not isinstance(name, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,39}', name) or not callable(tool) for name, tool in tools.items()):
        raise ValueError('Policy en maximaal acht benoemde tools vereist')
    if type(max_decisions) is not int or not 1 <= max_decisions <= 32 or type(max_tool_calls) is not int or not 0 <= max_tool_calls <= 16:
        raise ValueError('Beslisbudget 1–32, toolbudget 0–16')
    observations, trace, seen, attempts = [], [], set(), 0
    def finish(status, text=None):
        return {'status': status, 'answer': text, 'observations': observations, 'trace': trace, 'tool_attempts': attempts}
    for step in range(max_decisions):
        try:
            action = validate_agent_action(policy(copy.deepcopy(observations)))
        except Exception:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'invalid-policy-action'})
            return finish('policy-error')
        if action['kind'] == 'final':
            trace.append({'step': step, 'kind': 'final'})
            return finish('final', action['text'])
        if action['call_id'] in seen:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'duplicate-call-id'})
            return finish('duplicate-call')
        if attempts >= max_tool_calls:
            trace.append({'step': step, 'kind': 'rejected', 'reason': 'tool-budget'})
            return finish('tool-budget')
        seen.add(action['call_id']); attempts += 1
        tool = tools.get(action['name'])
        if tool is None:
            observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': False, 'data': 'unknown-tool'}
        else:
            try:
                value = tool(**action['arguments'])
                encoded = json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(',', ':'))
                if len(encoded.encode('utf-8')) > 4096:
                    raise ValueError('Toolresultaat te groot')
                observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': True, 'data': json.loads(encoded)}
            except Exception:
                observation = {'call_id': action['call_id'], 'name': action['name'], 'ok': False, 'data': 'tool-error'}
        observations.append(observation)
        trace.append({'step': step, 'kind': 'tool', 'call_id': action['call_id'], 'name': action['name'], 'executed': tool is not None, 'ok': observation['ok']})
    return finish('decision-budget')

def trace_work(trace, decision_units, tool_units):
    if not isinstance(trace, list) or len(trace) > 32 or type(decision_units) is not int or not 0 <= decision_units <= 1000 or not isinstance(tool_units, dict) or len(tool_units) > 8 or any(not isinstance(k, str) or type(v) is not int or not 0 <= v <= 1000 for k, v in tool_units.items()):
        raise ValueError('Begrensde trace en gehele oefenkosten vereist')
    decisions, calls, units = 0, 0, 0
    for i, row in enumerate(trace):
        if not isinstance(row, dict) or type(row.get('step')) is not int or row['step'] != i or row.get('kind') not in ['tool', 'final', 'rejected']:
            raise ValueError('Ongeldige opeenvolgende tracestappen')
        decisions += 1; units += decision_units
        if row['kind'] == 'tool':
            if type(row.get('executed')) is not bool or not isinstance(row.get('name'), str):
                raise ValueError('Tooltrace mist naam/uitvoering')
            if row['executed']:
                if row['name'] not in tool_units:
                    raise ValueError('Onbekende toolkosten')
                calls += 1; units += tool_units[row['name']]
    return {'decisions': decisions, 'executed_tools': calls, 'work_units': units}

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