Run a search-and-read decision loop

Build the runtime around a search, read and final decision interface. Validate each proposal, retain observations and distinguish valid completion, rejected output, exhausted work and incomplete execution.

Problem

When you need this. A retrieval controller needs explicit next-action and stopping rules.

A policy asks to read a page that belongs to another documentation version. Another policy keeps searching after it has enough source material. A third returns a plausible command and cites a page it never opened. These are different failures. A runtime needs to distinguish them before the application can decide whether to retry, ask for help or return an answer.

You will run a decision loop for the fictional docuctl documentation assistant. The policy proposes search, read or final; Python validates and dispatches those proposals. The default policy is scripted. Its successful trace lets you inspect the runtime without making a model request.

Concepts

A policy returns a proposed decision. The runtime owns the tool allowlist, argument checks, observations and budgets. A tool result supplies an observation for the next decision; it does not authorize every possible final answer.

Stopping and success are separate states. A tool limit can stop a run without a valid answer, and a syntactically valid final result can still fail the task. The default scripted policy makes these transitions repeatable; an optional model-backed policy must also limit the requests it sends to its model provider.

Architecture

The loop asks for one decision, validates it and either dispatches a permitted tool or checks a proposed final result. Search/read observations remain attached to the run. Runtime limits stop repeated work, and the independent outcome check compares the final command and source with the authored task.

Working code

Give the policy a narrow decision interface

The controller comparison established the task and independent answer check. Keep that contract. This lesson changes how decisions reach the tools, rather than changing the retrieval algorithm.

decide(task, observations) receives the task and prior tool results. It returns one dictionary with exactly the fields for its action. A search supplies a query, a read supplies a page ID, and a final answer supplies an answer and citation. Unknown actions, extra fields and invalid values are refused before tool dispatch.

A dictionary that looks like a tool call is still data. validate_decision() checks it, and run() decides whether execution is allowed. There is no shell action in the interface. A policy cannot add one by inventing an action name. The Python callable itself is trusted application code; this dispatcher is not a sandbox for an untrusted Python function.

Download the complete lab and open its agent-engineering/ directory:

python3.12 -m venv .venv
source .venv/bin/activate
python tool_loop.py --case rollback-direct --save my-trace.json
python -m unittest -v test_loop.py

No extra packages or model files are required. my-trace.json is a local record containing the selected case, policy name and runtime result. Open it after the command finishes to inspect each decision alongside the tool observation it produced.

Keep execution limits in the runtime

The default run permits at most four policy turns and three tool calls. These count different work. A final answer uses a policy turn but no tool. A failed read still invokes a tool and consumes its allowance; an invalid decision consumes no tool allowance because dispatch never happens.

run() checks the tool count before dispatch. A policy that repeatedly proposes searches receives tool_limit when the next call would exceed the budget. The for loop separately bounds policy turns and returns turn_limit if no accepted final answer arrives. Neither stop state is treated as a successful answer.

"""Small document tools and a provider-independent decision loop."""
from __future__ import annotations

from dataclasses import dataclass
import json
from pathlib import Path
import re
from typing import Callable

ROOT = Path(__file__).resolve().parent
STOP = frozenset('a an and are can do for how i in is it me my of on the to what with'.split())


def load(name):
    return json.loads((ROOT / name).read_text())


def words(text):
    return set(re.findall(r'[a-z0-9]+', text.casefold())) - STOP


@dataclass(frozen=True)
class Task:
    id: str
    family: str
    question: str
    version: str = 'v2'
    environment: str = 'staging'

    def __post_init__(self):
        if any(not isinstance(v, str) or not v.strip() or len(v) > 100
               for v in (self.id, self.family, self.version, self.environment)):
            raise ValueError('task metadata must contain bounded strings')
        if self.version not in {'v1', 'v2'}:
            raise ValueError('unsupported documentation version')
        if self.environment != 'staging':
            raise ValueError('this assistant only covers staging')
        if not isinstance(self.question, str) or not self.question.strip() or len(self.question) > 500:
            raise ValueError('question must contain 1..500 characters')


class Documents:
    def __init__(self):
        self.pages = load('documents.json')
        self.aliases = load('glossary.json')

    def search(self, query, version, *, expand=False, top_k=1):
        if version not in {'v1', 'v2'} or type(top_k) is not int or top_k not in {1, 2, 3}:
            raise ValueError('invalid search configuration')
        terms = words(query)
        if expand:
            terms |= {self.aliases[t] for t in tuple(terms) if t in self.aliases}
        ranked = []
        for page in self.pages:
            if page['version'] != version:
                continue
            score = len(terms & words(page['title'] + ' ' + page['text']))
            if score:
                ranked.append((-score, page['id'], page['title']))
        return [{'id': id, 'title': title} for _, id, title in sorted(ranked)[:top_k]]

    def read(self, id, version):
        page = next((p for p in self.pages if p['id'] == id), None)
        if page is None or page['version'] != version:
            raise ValueError('page does not exist in the requested version')
        return dict(page)


def validate_decision(value):
    """Reject unknown fields and actions before dispatching a tool."""
    if not isinstance(value, dict):
        raise ValueError('decision must be an object')
    action = value.get('action')
    keys = {'search': {'action', 'query'}, 'read': {'action', 'id'},
            'final': {'action', 'answer', 'citation'}}
    if not isinstance(action, str) or action not in keys or set(value) != keys[action]:
        raise ValueError('unknown action or unexpected fields')
    if any(not isinstance(v, str) or len(v) > 2000 for v in value.values()):
        raise ValueError('decision values must be bounded strings')
    if action == 'final' and not value['answer'].strip():
        raise ValueError('answer must not be empty')
    if action in {'search', 'read'} and not value['query' if action == 'search' else 'id'].strip():
        raise ValueError('tool argument must not be empty')
    return value


def run(task: Task, decide: Callable, *, max_turns=4, max_tools=3):
    """The policy chooses; this runtime validates, executes and stops."""
    if type(max_turns) is not int or not 1 <= max_turns <= 8:
        raise ValueError('max_turns must be 1..8')
    if type(max_tools) is not int or not 0 <= max_tools <= 6:
        raise ValueError('max_tools must be 0..6')
    docs, trace, seen, observations = Documents(), [], {}, []
    tools = 0
    for turn in range(1, max_turns + 1):
        try:
            decision = validate_decision(decide(task, list(observations)))
        except (ValueError, TypeError, KeyError) as error:
            return {'stop': 'invalid_decision', 'detail': str(error), 'trace': trace,
                    'turns': turn, 'tool_calls': tools}
        except OSError as error:
            return {'stop': 'provider_error', 'detail': type(error).__name__, 'trace': trace,
                    'turns': turn, 'tool_calls': tools}
        action = decision['action']
        if action == 'final':
            citation = decision['citation']
            # A citation is accepted only after that page was read in this run.
            if citation and citation not in seen:
                return {'stop': 'unread_citation', 'trace': trace, 'turns': turn,
                        'tool_calls': tools}
            trace.append({'turn': turn, 'decision': decision})
            return {'stop': 'final', 'answer': decision['answer'], 'citation': citation,
                    'trace': trace, 'turns': turn, 'tool_calls': tools}
        if tools == max_tools:
            return {'stop': 'tool_limit', 'trace': trace, 'turns': turn,
                    'tool_calls': tools}
        try:
            if action == 'search':
                result = docs.search(decision['query'], task.version, expand=True, top_k=2)
            else:
                result = docs.read(decision['id'], task.version)
                seen[result['id']] = result
        except ValueError as error:
            result = {'error': str(error)}
        tools += 1
        observations.append({'action': action, 'result': result})
        trace.append({'turn': turn, 'decision': decision, 'result': result})
    return {'stop': 'turn_limit', 'trace': trace, 'turns': max_turns,
            'tool_calls': tools}


def scripted(task, observations):
    """A transparent offline policy for testing the loop, not a model."""
    if not observations:
        return {'action': 'search', 'query': task.question}
    last = observations[-1]
    if last['action'] == 'search':
        hits = last['result']
        if not hits:
            return {'action': 'final', 'answer': 'No matching page was found.', 'citation': ''}
        return {'action': 'read', 'id': hits[0]['id']}
    page = last['result']
    if 'error' in page:
        return {'action': 'final', 'answer': page['error'], 'citation': ''}
    return {'action': 'final', 'answer': page['command'], 'citation': page['id']}


def task_from(case):
    return Task(**{k: case[k] for k in ('id', 'family', 'question', 'version')})


def outcome(case, result):
    """Independent task expectation; a final stop alone never passes."""
    return (result.get('stop') == 'final'
            and result.get('citation') == case['expected_doc']
            and result.get('answer') == case['expected_command'])

The bounds here cover iteration and local tool work. They are not a wall-clock deadline for an arbitrary decide() implementation. A callable that never returns can still block this synchronous loop. A remote policy needs transport timeouts, and a longer-running process needs a separate deadline and cancellation design.

test_repeated_search_stops_at_tool_limit uses a policy that always searches. test_turn_limit_is_distinct_from_final stops the normal scripted policy after its first turn. Together they show why “the loop ended” is too vague a result for a caller.

Follow the successful trace from query to source

"""Run one bounded document lookup with an offline or installed-model policy."""
import argparse
import json
from core import load, outcome, run, scripted, task_from


def main():
    p = argparse.ArgumentParser()
    p.add_argument('--case', default='rollback-direct')
    p.add_argument('--provider', choices=['scripted', 'ollama'], default='scripted')
    p.add_argument('--save', help='Optional local trace JSON path')
    args = p.parse_args()
    case = next((c for c in load('cases.json') if c['id'] == args.case), None)
    if not case:
        p.error('unknown case')
    policy = scripted
    if args.provider == 'ollama':
        from provider import OllamaPolicy
        policy = OllamaPolicy()
    result = run(task_from(case), policy)
    for event in result['trace']:
        d = event['decision']
        print(f"turn {event['turn']}: {d['action']}")
    print('answer:', result.get('answer', 'none'))
    print('source:', result.get('citation', 'none'))
    print('stop:', result['stop'], '| independent outcome:', outcome(case, result))
    print('policy:', args.provider)
    if args.save:
        from pathlib import Path
        Path(args.save).write_text(json.dumps({'case': case['id'], 'policy': args.provider,
                                              'result': result}, indent=2) + '\n')


if __name__ == '__main__':
    main()

turn 1: search
turn 2: read
turn 3: final
answer: docuctl rollback --env staging --release LAST_GOOD
source: rollback-v2
stop: final | independent outcome: True
policy: scripted

The first observation is a search result containing page IDs and titles. The policy then chooses a page to read. Only the read result contains the command it returns. The runtime records that page in seen, a map local to this invocation.

The final decision cites rollback-v2. Before accepting it, the runtime checks that the page is in seen. A search hit alone is insufficient: finding a title does not mean the policy inspected the source text. test_unread_citation_cannot_finish exercises a policy that jumps straight to a cited final answer and receives unread_citation.

That provenance check has a limited purpose. A policy can read the correct page and still invent a command. outcome() independently compares the returned command and citation with the case expectation. test_invented_command_fails_the_independent_outcome changes the answer while leaving the final state and source in place; the outcome must fail.

Return failures that a caller can interpret

An unknown shell action returns invalid_decision without running a tool. A read from the wrong version returns an error observation, allowing the policy to inspect that failed operation within its remaining budget. An OSError from the policy adapter returns provider_error rather than a fabricated answer.

Those cases should lead to different application behavior. A malformed decision may require correcting the policy’s response format. An unavailable provider may justify a later retry under a caller-owned retry budget. A missing source may require a narrower question. This lab records the stop state and performs no automatic retries.

The README contains an optional adapter for an already installed Ollama model. That adapter has an offline request-limit test, but it has not been executed against a model for this lesson. The recorded output above comes entirely from scripted(). Replacing that callable would require a new trace and the same independent checks.

Change one policy while keeping the checks fixed

The useful extension is an adaptive action, not a larger loop limit by itself. Write a scripted policy that revises one empty search, then stops if the revised query also finds nothing. Keep the tool schema and outcome checker unchanged.

Design exercise: predict the stop state and maximum dispatched tools for three runs: an invalid first action, an endless search policy, and a final answer citing an unread page. Then express each prediction as a test against run(). Explain which boundary belongs to the runtime and which belongs to the independent task checker.

The next lesson delegates independent requests. Each worker will reuse this runtime with its own observations and source map.

Failure cases

The unread-citation case distinguishes an observed source from a plausible identifier. The invented-command case goes further: even a final result with an observed citation can fail the task. Transport failure and exhausted budgets must remain visible instead of becoming successful final answers.

Deployment

Use the narrow decision interface when integrating another policy implementation. Keep validation, dispatch and request accounting in the runtime, and bind run observations to the request being served. Add authenticated scope and dependency cancellation at their actual boundaries before exposing a remote service.

Scaling

Count policy decisions, provider requests and tool executions separately. Raising one limit can increase cost without improving source selection. Before increasing concurrency, measure queueing and downstream capacity while preserving each run’s independent observations and budgets.

Key takeaways

• Validating a tool proposal precedes tool execution.
• A run can stop without succeeding.
• Observed sources and a correct task result need separate checks.

Best practices

• Check the allowlist before resolving a handler.
• Keep transport errors distinct from policy refusal.
• Record the reason and work consumed at every stop.

References

• Anthropic, Building effective agents: the distinction between predefined workflows and model-directed control.
• Python, dataclasses: generated data-class methods; the lab adds explicit value checks in Task.__post_init__() rather than treating annotations as validation.

References checked October 24, 2026. The runtime behavior and output described here come from the bundled local implementation and its failure tests.