Agent patterns

react.py# ReAct: interleave Thought -> Action -> Observation until we can answer.
# The 'model' is a deterministic fake so this runs offline; swap it for a real
# LLM call and the control flow is IDENTICAL.

TOOLS = {
    'population': lambda city: {'Paris': 2_100_000, 'Lyon': 500_000}[city],
}

def fake_model(trace):
    """Return the next 'Thought' + 'Action' given the trace so far.
    A real LLM replaces this; it reads the same trace and emits the same shape."""
    obs = [line for line in trace if line.startswith('Observation:')]
    if len(obs) == 0:
        return ('Thought: I need the population of Paris first.',
                ('population', 'Paris'))
    if len(obs) == 1:
        return ('Thought: Now I need the population of Lyon.',
                ('population', 'Lyon'))
    return ('Thought: I have both; I can answer now.', ('finish', None))

def react(goal, max_steps=6):
    trace = [f'Goal: {goal}']
    facts = []
    for step in range(max_steps):
        thought, (tool, arg) = fake_model(trace)
        trace.append(thought)
        if tool == 'finish':
            total = sum(facts)
            trace.append(f'Answer: combined population = {total:,}')
            break
        trace.append(f'Action: {tool}({arg!r})')
        result = TOOLS[tool](arg)
        facts.append(result)
        trace.append(f'Observation: {result:,}')
    return trace

for line in react('total population of Paris and Lyon'):
    print(line)

plan_execute.py# Plan-and-execute: model writes an ordered plan, then we run each step.
# fake_planner() is deterministic; a real LLM would generate the plan text.

STEP_TOOLS = {
    'fetch_orders': lambda: [12, 7, 31],
    'sum_orders':   lambda xs: sum(xs),
    'format':       lambda n: f'Total orders this week: {n}',
}

def fake_planner(goal):
    """Return an ordered plan (list of step names). A real LLM writes this."""
    return ['fetch_orders', 'sum_orders', 'format']

def execute(goal):
    plan = fake_planner(goal)
    print('PLAN:')
    for i, step in enumerate(plan, 1):
        print(f'  {i}. {step}')
    print('EXECUTE:')
    state = None
    for step in plan:
        if step == 'fetch_orders':
            state = STEP_TOOLS[step]()
        elif step == 'sum_orders':
            state = STEP_TOOLS[step](state)
        elif step == 'format':
            state = STEP_TOOLS[step](state)
        print(f'  ran {step} -> {state!r}')
    return state

print('RESULT:', execute('report this week\'s order total'))

reflection.py# Reflection: draft -> critique -> revise. All three 'model' calls are faked
# deterministically here; a real LLM replaces each, control flow unchanged.

REQUIRED = ['cause', 'fix']   # a good incident summary must mention both

def fake_draft(task):
    return 'The checkout service went down.'   # incomplete on purpose

def fake_critique(text):
    """Return a list of missing requirements. A real LLM judges the draft."""
    missing = []
    if 'queue' not in text.lower():
        missing.append('cause')
    if 'restart' not in text.lower() and 'scale' not in text.lower():
        missing.append('fix')
    return missing

def fake_revise(text, missing):
    add = {'cause': ' The queue backed up.',
           'fix': ' Drain the queue and scale the consumer group to fix it.'}
    for m in missing:
        text += add[m]
    return text

def reflect(task, max_rounds=3):
    draft = fake_draft(task)
    print(f'draft: {draft}')
    for r in range(max_rounds):
        missing = fake_critique(draft)
        if not missing:
            print(f'round {r}: looks complete, stopping.')
            break
        print(f'round {r}: missing {missing} -> revising')
        draft = fake_revise(draft, missing)
    return draft

print('final:', reflect('summarize the checkout incident'))

router.py# Routing: a classifier picks the handler, then we dispatch to it.
# fake_route() is deterministic; a real LLM (or a small classifier) replaces it.

def fake_route(request):
    """Classify the request into an intent. A real model does this step."""
    r = request.lower()
    if any(w in r for w in ('refund', 'charge', 'invoice')):
        return 'billing'
    if any(w in r for w in ('error', 'crash', 'broken', 'down')):
        return 'bug'
    return 'general'

HANDLERS = {
    'billing': lambda q: f'[billing] looking up your account for: {q}',
    'bug':     lambda q: f'[bug] opening a ticket for: {q}',
    'general': lambda q: f'[general] answering: {q}',
}

def agent(request):
    intent = fake_route(request)
    return intent, HANDLERS[intent](request)

for req in ['I want a refund on invoice 88',
            'the checkout page is down',
            'what are your hours?']:
    intent, reply = agent(req)
    print(f'{intent:8} <- {req!r}')
    print(f'         {reply}')

retries.py# Capped retries: catch the failure, feed the error back, try again up to a limit.
# The 'tool' here fails the first two times, then succeeds -- deterministic, no
# randomness or clock. A real agent feeds the error text back to the LLM.

class ToolError(Exception):
    pass

def flaky_tool(state):
    """Fails until it has been nudged twice; then returns a value."""
    if state['attempts'] < 2:
        raise ToolError(f"transient failure #{state['attempts'] + 1}")
    return 'OK: fetched 42 records'

def call_with_retries(tool, max_attempts=3):
    state = {'attempts': 0}
    last_error = None
    for attempt in range(1, max_attempts + 1):
        try:
            result = tool(state)
            print(f'attempt {attempt}: success -> {result}')
            return result
        except ToolError as e:
            last_error = str(e)
            print(f'attempt {attempt}: caught {last_error!r}; feeding back + retrying')
            state['attempts'] += 1        # what the model would learn from the observation
    print(f'giving up after {max_attempts} attempts; last error: {last_error!r}')
    return None

call_with_retries(flaky_tool)