Compare a workflow with a bounded policy

Compare a keyword route, a fixed search workflow and a bounded decision loop on the same documentation tasks. Hold retrieval and tool allowance constant so a control-policy comparison does not hide a changed search system.

Problem

When you need this. You need to decide whether a fixed workflow or an observation-dependent policy fits a task.

A documentation assistant must answer “How do I restore the last release?” with the command for the requested documentation version. A keyword route may miss “restore.” A search workflow may find the right page. An agent loop may choose to search and read that same page. Which change actually helped?

You will run all three policies on the same fictional docuctl tasks. The workflow and loop use identical retrieval settings and a two-tool budget. The loop’s decisions come from a deterministic Python function, so the comparison isolates application control; no language model participates.

Concepts

A fixed workflow follows application-defined steps. A decision loop asks a policy for the next action after observing previous results. The runtime validates and executes that action; a policy proposal is not a tool execution.

This comparison uses scripted decisions to isolate control flow. The same task contract grades all policies, and the search workflow and loop share retrieval settings and tool budgets. A later model policy would introduce interpretation variability and require new recorded trials; it cannot inherit the scripted policy’s task scores.

Architecture

Each policy receives the same versioned task. The keyword baseline routes directly, the fixed workflow performs its prescribed search/read sequence, and the loop observes results before selecting another action. Shared tools and independent outcome checks expose which change affected task handling.

Working code

Define success before choosing the controller

Every case specifies a question, a documentation version, an expected page and the exact staging command that page contains. The task is narrow: return that command with a version-matched citation. A fluent paraphrase is outside this contract.

Task rejects production requests and unknown versions before policy execution. These are application boundaries, not model instructions. A command that happens to exist in another version is still the wrong answer.

The independent outcome() function checks the stop state, citation and command. A policy cannot pass by returning final or saying “Done.” Keeping this checker outside the policy lets each controller fail against the same expectation.

Download the complete lab above and open agent-engineering/:

python3.12 -m venv .venv
source .venv/bin/activate
python task_baseline.py
python -m unittest -v test_baseline.py

The lab uses only the standard library. Its README lists the other workshop experiments; this lesson uses task_baseline.py, core.py and the shared JSON fixtures.

Hold retrieval constant when comparing control policies

The keyword route looks for a known topic word and reads that topic’s page in the requested version. A question with no single recognized topic returns needs_review. This baseline is deliberately simple, but it is a real runnable alternative.

The workflow always searches and then reads the first hit. Its search expands known aliases and returns at most two candidates. The scripted loop takes the same search and read steps through the runtime’s decision interface. It receives observations after each tool and returns a final answer after reading a page.

"""Compare a fixed workflow and a scripted decision loop on the same cases."""
from core import Documents, load, outcome, run, scripted, task_from, words


def keyword_route(task):
    """Fixed keyword route, followed by one version-scoped read."""
    routes = {'deploy': 'deploy', 'rollback': 'rollback', 'tokens': 'tokens',
              'config': 'config', 'migrate': 'migrate', 'releases': 'releases',
              'logs': 'logs', 'health': 'health'}
    topics = [topic for term, topic in routes.items() if term in words(task.question)]
    if len(topics) != 1:
        return {'stop': 'needs_review', 'tool_calls': 0}
    id = f'{topics[0]}-{task.version}'
    try:
        page = Documents().read(id, task.version)
    except ValueError:
        return {'stop': 'needs_review', 'tool_calls': 1}
    return {'stop': 'final', 'answer': page['command'], 'citation': id, 'tool_calls': 1}


def workflow(task):
    """Fixed search/read sequence, using the same search settings as the loop."""
    docs = Documents()
    hits = docs.search(task.question, task.version, expand=True, top_k=2)
    if not hits:
        return {'stop': 'needs_review', 'tool_calls': 1}
    page = docs.read(hits[0]['id'], task.version)
    return {'stop': 'final', 'answer': page['command'], 'citation': page['id'], 'tool_calls': 2}


def compare(cases):
    report = {}
    for name, execute in [('keyword route', keyword_route), ('search/read workflow', workflow),
                          ('scripted loop', lambda t: run(t, scripted, max_turns=3, max_tools=2))]:
        results = [execute(task_from(case)) for case in cases]
        report[name] = {'passed': sum(outcome(c, r) for c, r in zip(cases, results)),
                        'total': len(cases), 'tool_calls': sum(r['tool_calls'] for r in results)}
    return report


if __name__ == '__main__':
    cases = load('cases.json')
    print('Task: return the exact staging command and a version-matched source.')
    for name, row in compare(cases).items():
        print(f"{name}: {row['passed']}/{row['total']} outcomes; {row['tool_calls']} tool calls")
    print('The loop uses scripted decisions here; these are not model-quality scores.')

This distinction prevents an unfair comparison. If the loop used alias expansion while the workflow used literal matching, a higher score could come entirely from retrieval. It would not show that choosing the next action dynamically improved the task.

test_workflow_and_loop_share_retrieval_and_tool_budget compares the complete summary rows for those two controllers. The expected result is a tie. The test would fail if a future edit silently gave one policy more successful lookups or counted its tools differently.

Read the result as a diagnosis

Task: return the exact staging command and a version-matched source.
keyword route: 9/18 outcomes; 9 tool calls
search/read workflow: 17/18 outcomes; 36 tool calls
scripted loop: 17/18 outcomes; 36 tool calls
The loop uses scripted decisions here; these are not model-quality scores.

The keyword route performs less tool work and misses more tasks. The search/read workflow and scripted loop produce the same outcome and tool totals. For these cases, changing from keyword routing to retrieval helps; wrapping that retrieval in a decision loop adds no measured task benefit.

This does not mean an agent can never help. The fixed workflow already knows the useful sequence: search, read, answer. To test adaptive control, add cases where an observation changes the next useful action—for example, an empty search that calls for a revised query or conflicting pages that require a second read. Define their expected outcomes before implementing that policy.

The tool counts are actual local calls. They are not tokens, latency or price. A model-driven controller would add provider requests and could make different decisions; it needs its own run and evaluation.

Test the refusal boundary as well as the happy path

An ambiguous keyword question such as “deploy and rollback staging” is refused by the keyword route rather than assigned to whichever topic appears first in a dictionary. test_ambiguous_workflow_route_stops checks that refusal and confirms that no tool ran.

A version mismatch is checked again inside Documents.read(). This matters even after validating the task: a policy can still propose a page ID from another version. The tool checks its own boundary instead of trusting the controller’s earlier decision.

The two safeguards protect different things. Task validation limits the supported request. Tool validation limits what a particular operation may read. Neither makes the answer correct; the outcome check still compares it with the case expectation.

"""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'])

Choose the next experiment from the remaining failures

Start with the simplest controller that meets the task requirement. Inspect the failing case rather than adding another worker or a larger model automatically. If search never retrieves the source, fix retrieval. If the source is present but the next action is wrong, investigate the policy. If the cited command is wrong, inspect the answer check and source selection.

The decision-loop lesson adds runtime limits and checks for an unread citation. It reuses these tools and cases rather than rebuilding search. The separate Building AI Systems course uses this result as an input to an application architecture decision; it does not repeat the controller experiment.

Design exercise: write one new case that the fixed search/read workflow cannot solve but a bounded adaptive policy could. Name the observation that should change the next action, the maximum tools allowed, and the outcome checker. If the proposed loop still follows the same two actions on every case, explain what its additional control is buying.

Failure cases

An ambiguous keyword route must stop rather than guess. A loop that says it is done must still supply the expected command and source. Keeping the same cases and retrieval settings is part of the comparison contract, not a formatting preference.

Deployment

Run this experiment before choosing an application controller. Keep the task and tool contracts when replacing the scripted policy with a model adapter, and evaluate that replacement separately. Production identity and permission checks must come from the actual service context rather than the fixture request.

Scaling

A loop can add requests and tool work that a fixed workflow avoids. Compare steps and observations per completed task before adding workers or increasing budgets. These local results do not measure model latency, provider cost or the reliability of a model-driven policy.

Key takeaways

• A controller change and a retrieval change are different experiments.
• A policy proposes actions; the runtime enforces tool and stopping rules.
• The same independent task contract must grade every compared policy.

Best practices

• Freeze retrieval and tool budgets for a control-policy comparison.
• Keep unsupported requests and ambiguous routes in the case set.
• Inspect the trace that explains a score difference.

References

• Anthropic, Building effective agents: workflows follow predefined code paths; agents direct their own process and tool use.
• Google, Rules of ML: begin with a simple baseline and choose metrics tied to the actual task.

The comparison above comes from the bundled local scripts and synthetic cases. The scripted controller is a test fixture for the runtime.