Parallel work

Lesson 13
45 min
Module 5: Multiple workers and integrations

Lesson 13 of 18 · AI Agents and Workflow OrchestrationParallel work

Course overview

1. Choose the right orchestration pattern

2. Represent state and tools

3. Reasoning, planning and context

4. Human approval and recovery

5. Multiple workers and integrations

6. Evaluate and operate a bounded agent

7. Case studies

What you will be able to do

Run independent reads at the same time, and merge their results with a rule that you wrote down. Then speed never costs a wrong or a lost result.

Before you startWhat to know and to have readyAbout 45 minutes

What you need to know

Parallel work as one of the five patterns, from Common patterns.
Approval before every change and operation IDs, from Approval boundaries and Side effects and retries.

What you need on your computer

The project after Module 4. python -m pytest prints 60 passed.
If you skipped Module 4, copy the snapshot end-of-m04 from the labs folder and make its virtual environment, as the snapshots README shows.

This lesson adds resolver/parallel.py and its tests. No account and no key: every model decision is a recording. Tested with Python 3.12.13 on macOS; the Windows commands were not run.

Two friends and one shopping list

Two friends shop for a dinner party. One goes to the bakery, the other to the vegetable market, and they meet at the car. They finish in half the time. But the plan only works if three things are clear before they leave.

What each one buys. “You take the bread and the cheese, I take the vegetables.” If both think the wine is theirs, they come back with two bottles, or with none.
What each one brings back. The bags, and a word about anything they could not find. “The bakery was closed” is useful. An empty bag with no word is not.
What happens when the two disagree. One bought red wine and the other white. Who decides? If nobody said, the last one to put a bottle in the car wins by accident.

Some jobs cannot be split at all. You cannot choose the dessert before you know the main dish. Each decision waits for the one before it.

What the project does now

Look at T-90201 again, from Common patterns. The customer asks about two orders in one ticket. In its first response, the chat-small agent asked for two reads at once:

  1  model     chat-small: get_order({"order_id": "LK-640763"}), get_order({"order_id": "LK-640872"})  (2167 in, 55 out, 1.04 s)
  1  tool      get_order({"order_id": "LK-640763"}) -> ok
  1  tool      get_order({"order_id": "LK-640872"}) -> ok

The two reads do not depend on each other. Neither one needs the other’s result to start. The loop still runs them one after the other, in the order that the model gave them.

This is not rare in the recordings. In the three repeats of the chat-small agent, 12 of its 444 model steps asked for two or more reads at once. They were on 5 tickets: T-80007, T-90201, T-90207, T-90209 and T-90602. Each time, the reads were independent.

The problem: speed without rules

Running the reads at the same time looks easy. Start them all, wait for them all, and put the results together. Two things go wrong when “put the results together” has no rule. Both cases below are constructed for this lesson; no recorded run did them.

A wrong result wins. Amira tries a design with one worker per order: a separate model call that looks at one order and proposes what to do with it. On T-90201, one worker reads LK-640763 and proposes to refund the 9.95 delivery fee. That is rule W2, because the order is 13 business days late. A second worker reads the same order and calls it lost. It proposes a refund of the whole order, 45.95, as rule W3 would for more than 15 days. Her first merge code keeps the last answer that arrives, and which one is last depends on which worker finished first. So Grace’s approval screen shows a 45.95 refund one day and a 9.95 refund the next, for the same ticket.

A failed read disappears. The payments service is down, so get_payments fails. A merge that collects only the results that came back has no trace of it. The proposal looks complete, but one of its facts was never read.

Neither problem is about speed. Both come from a merge that nobody wrote down.

The idea: fan out, then merge by a written rule

Parallel work runs parts of a task at the same time, then combines their results. The first half is the fan-out: one input splits into several independent parts, like the ribs of a paper fan. The second half is the merge: the step that puts the results together. When results disagree, the merge must reconcile them: decide, by a rule, what the combined result is. This is not the reconciliation of Side effects and retries, which asks a service whether a write happened. Here, two results are compared with each other.

Parts are independent when no part needs another part’s result to start, and no two parts change the same thing. Reads of two orders are independent. A read and the decision that uses it are not.

So before you run anything in parallel, write down three things:

The inputs of each part. Here: one tool call each, with the customer set by the application, never by a part.
The output of each part. Here: one result per call, ok or failed, and never nothing.
The merge rule. Here: results come back in the order of the calls, and proposals on different orders combine. Two different proposals on one order are a conflict, which goes to a person.

Ticket T-90201, two orders, customer C-50866, fans out into three reads that gather() runs at the same time: get_order LK-640763, get_order LK-640872 and get_payments LK-640763. The results come back in call order, and a failed read stays a failed read. They lead to two parts: for LK-640763, refund 9.95 under W2; for LK-640872, reply only under W1. The merge rule, merge_proposals(), has three exits: a part says hand_to_person, so a person decides; one order has two different actions, so a Conflict goes to a person; different orders or the same actions give one proposal, which goes to approval. — Fan-out and reconciliation in resolver/parallel.py: the reads run at the same time, and the merge follows a written rule.

Look at what the picture does not have: an exit where code picks a winner. Code never guesses between two proposals for one order. A person, who can read both, decides.

Add parallel work to the project

Make a new file, resolver/parallel.py, with this code:

"""Parallel work with explicit inputs, outputs and merge rules.

`gather` runs independent reads at the same time. It is safe because the reads do not depend on each
other and change nothing; the merge rule is explicit: one result per call, in the order of the calls,
a failed read stays a failed read (it is never dropped silently).

`merge_proposals` merges the work of two workers that each looked at part of a ticket (for example one
order each). The rule: actions on different orders are combined; two different proposals for the same
order are a conflict, and a conflict goes to a person. Code never picks a winner by guessing.
"""

import time
from concurrent.futures import ThreadPoolExecutor

from .schema import Resolution
from .systems import World
from .tools import ToolResult, run_read


def gather(calls: list[tuple[str, dict]], customer_id: str, world: World, max_workers: int = 3,
           delay_s: float = 0.0) -> list[ToolResult]:
    """Run the read calls in parallel; the results come back in the order of `calls`.
    `delay_s` simulates a slow service (each read waits that long), to see the time saved."""
    def one(call):
        if delay_s:
            time.sleep(delay_s)
        return run_read(call[0], call[1], customer_id, world)

    with ThreadPoolExecutor(max_workers) as pool:
        return list(pool.map(one, calls))


class Conflict(Exception):
    pass


def merge_proposals(a: Resolution, b: Resolution) -> Resolution:
    """Combine two partial resolutions. Raises Conflict when they disagree about the same order."""
    if "hand_to_person" in (a.outcome, b.outcome):
        keep = a if a.outcome == "hand_to_person" else b
        return keep
    by_order: dict[str, list] = {}
    for source in (a, b):
        for action in source.actions:
            by_order.setdefault(action.order_id, []).append((source, action))
    for order_id, items in by_order.items():
        sources = {id(s) for s, _ in items}
        if len(sources) > 1:
            first = sorted(x.model_dump_json() for s, x in items if s is a)
            second = sorted(x.model_dump_json() for s, x in items if s is b)
            if first != second:
                raise Conflict(f"the two workers propose different actions for {order_id}")
    actions, seen = [], set()
    for source in (a, b):
        for action in source.actions:
            key = action.model_dump_json()
            if key not in seen:
                seen.add(key)
                actions.append(action)
    outcome = "resolve" if actions else ("ask_customer" if "ask_customer" in (a.outcome, b.outcome) else "reply_only")
    return Resolution(outcome=outcome, rule=f"{a.rule}+{b.rule}", actions=actions,
                      reply=f"{a.reply}\n\n{b.reply}".strip()[:2000], reason=f"{a.reason} {b.reason}".strip()[:600])

Make a new file, tests/test_parallel.py, with these three tests:

"""Parallel reads with an explicit merge rule; conflicting workers go to a person."""

import time

import pytest

from resolver.parallel import Conflict, gather, merge_proposals
from resolver.schema import parse_resolution


def test_parallel_reads_keep_the_order_and_save_time(world):
    calls = [("get_order", {"order_id": "LK-640763"}), ("get_order", {"order_id": "LK-640872"}),
             ("get_payments", {"order_id": "LK-640763"})]
    start = time.monotonic()
    results = gather(calls, "C-50866", world("T-90201"), delay_s=0.2)
    took = time.monotonic() - start
    assert [r.order_ids for r in results] == [["LK-640763"], ["LK-640872"], ["LK-640763"]]
    assert took < 0.5          # three simulated 0.2 s reads, at the same time


def res(actions, outcome="resolve"):
    flat = [dict(a, reason=a.get("reason") or a.get("reason_code")) for a in actions]
    return parse_resolution({"outcome": outcome, "rule": "W", "actions": flat, "reply": "r", "reason": "x"})


def test_workers_on_different_orders_merge():
    a = res([{"tool": "request_refund", "order_id": "LK-640763", "amount": 9.95, "reason_code": "RFD-LATE"}])
    b = res([], "reply_only")
    assert merge_proposals(a, b).actions == a.actions


def test_workers_that_disagree_about_one_order_conflict():
    a = res([{"tool": "request_refund", "order_id": "LK-640763", "amount": 9.95, "reason_code": "RFD-LATE"}])
    b = res([{"tool": "request_refund", "order_id": "LK-640763", "amount": 45.95, "reason_code": "RFD-LATE"}])
    with pytest.raises(Conflict):
        merge_proposals(a, b)

Run the tests:
Terminal window
```
python -m pytest
```
The last line is 63 passed.

Read the two functions against the three things you wrote down.

Inputs. gather takes a list of calls and one customer_id, which the application passes in. No part can choose another customer. Each part runs run_read, the same checked read as the loop, with every check from Tool contracts.
Outputs. pool.map returns one result per call, in the order of the calls, whichever finished first. run_read never raises: a failure is a ToolResult with ok false. So a failed read stays in its place in the list.
The merge rule. merge_proposals sends the ticket to a person if either part says so. It compares the actions per order. Two different action lists for one order raise Conflict. Otherwise, it keeps each action once.

ThreadPoolExecutor runs each call in its own thread: a second line of work inside the same Python program. Threads suit reads well, because a read spends its time waiting for a service, and Python can wait for several at once. Its max_workers is the number of threads. These thread workers are not the model workers of the next lesson.

A worked example with every value visible

Time the three reads of T-90201. A real read of the practice database takes less than a millisecond, which is too fast to see. So the example adds a simulated delay of 0.2 seconds to each read, as a slow order service would have. Start Python in the project folder with python, then paste these lines. Press Enter until the >>> prompt comes back:

import time
from resolver.parallel import gather
from resolver.systems import World
calls = [("get_order", {"order_id": "LK-640763"}),
         ("get_order", {"order_id": "LK-640872"}),
         ("get_payments", {"order_id": "LK-640763"})]
for workers in (1, 2, 3):
    start = time.monotonic()
    results = gather(calls, "C-50866", World(), max_workers=workers, delay_s=0.2)
    print(f"{workers} worker(s): {time.monotonic() - start:.2f} s", [r.order_ids for r in results])

1 worker(s): 0.62 s [['LK-640763'], ['LK-640872'], ['LK-640763']]
2 worker(s): 0.41 s [['LK-640763'], ['LK-640872'], ['LK-640763']]
3 worker(s): 0.21 s [['LK-640763'], ['LK-640872'], ['LK-640763']]

Your times differ a little. With one worker, the three reads run one after the other: three times 0.2 seconds. With two, two reads run together and the third waits. With three, all run together, and the whole set takes as long as the slowest read. The order of the results never changes.

one after the other
at the same time

Three reads of T-90201 with a simulated 0.2 s delay each, measured once with gather(). One worker runs them one after the other.

Text description of the chart

A column chart of the time for three reads. One worker: 0.62 seconds. Two workers: 0.41 seconds. Three workers: 0.21 seconds. The time falls to about the time of one read when every read runs at the same time.

Now the output rule: a failure stays a failure. Make the payments service fail once (a simulated fault), and run the same three calls:

from resolver.systems import Faults
down = World(faults=Faults([{"tool": "get_payments", "mode": "unavailable"}]))
for (name, args), r in zip(calls, gather(calls, "C-50866", down)):
    print(name, args["order_id"], "->", "ok" if r.ok else f"FAILED: {r.error}")

get_order LK-640763 -> ok
get_order LK-640872 -> ok
get_payments LK-640763 -> FAILED: The service failed: get_payments: HTTP 503 Service Unavailable (simulated). You may try again once.

Three calls, three results. The third says what failed and why, in its own place. Whatever reads this list next, a model or code, can see that the payments are unknown.

Last, the merge rule. Build the two parts of T-90201 by hand, then the constructed conflict from the start of the lesson:

from resolver.parallel import Conflict, merge_proposals
from resolver.schema import Resolution
late = Resolution(outcome="resolve", rule="W2", reply="We will refund the 9.95 delivery fee.", reason="13 days late",
                  actions=[{"tool": "request_refund", "order_id": "LK-640763", "amount": 9.95, "reason_code": "RFD-LATE"}])
status = Resolution(outcome="reply_only", rule="W1", reply="LK-640872 is being prepared.", reason="processing")
merged = merge_proposals(late, status)
print(merged.outcome, merged.rule, [(a.tool, a.order_id, a.amount) for a in merged.actions])
lost = Resolution(outcome="resolve", rule="W3", reply="We will refund your order.", reason="read as lost",
                  actions=[{"tool": "request_refund", "order_id": "LK-640763", "amount": 45.95, "reason_code": "RFD-LATE"}])
try:
    merge_proposals(late, lost)
except Conflict as e:
    print("Conflict:", e)

resolve W2+W1 [('request_refund', 'LK-640763', 9.95)]
Conflict: the two workers propose different actions for LK-640763

The first merge combines a refund on one order with a reply about the other. The result is one proposal, with the 9.95 refund, that goes to approval. The second merge has two different refunds for one order, so it raises Conflict. No refund reaches Grace’s screen by accident.

What parallel reads save at Larkfield

In this project, very little. Two real reads of the practice database took about 0.6 milliseconds on a laptop, together or one after the other. The model calls of T-90201 took 2.814 seconds. The time of a task is in the model calls, not in the reads.

And the model calls of one loop cannot run in parallel. Each call needs the results of the one before: the model chooses its next read after it sees the last result. That is the dessert that waits for the main dish.

So the project keeps the loop as it is: it runs the reads of one step one after the other. gather is ready for the day a read is slow. Mei’s real order service would answer over a network, and a payments service can take a second. Then three reads at once save real time, and the merge rule is already written and tested.

When to use parallel work, and what it costs

Use it when the parts are truly independent and each one is slow: reads from slow services, several orders, several policy searches. Measure first. If the parts take milliseconds, the threads only add code.

Do not use it for:

Writes. Two writes on one order at the same time can both succeed, or one can undo the other. Every change in this project is a proposal, approved by a person, and run once with an operation ID, as in Side effects and retries. Keep writes one after the other.
Steps that depend on each other. A decision needs its evidence first.

What it costs:

A merge rule, written and tested. Without it, the order of arrival decides.
Failure handling per part. One part can fail while the others succeed; the result must say so.
Load on the services. Three calls at once hit a service three times as hard, and model providers limit this. While the course’s runs were recorded, the shared Azure deployments refused 887 requests with HTTP 429, “too many requests”. Nobody provoked those: another job used the same deployments. Keep the number of calls at once small, and use a retry that waits.

Common mistake

The merge keeps the last answer

Symptom: the same ticket gives a different proposal from run to run, with no change in the code or the recordings. On T-90201, the approval screen shows a 45.95 refund on one run and a 9.95 refund on the next. (Constructed: Amira’s first merge, before merge_proposals.)

Cause: the merge keeps one result per order, and it keeps whichever arrives last. Threads finish in a different order each time, so chance decides.

Diagnose: run the merge with the parts in both orders. If merge(a, b) and merge(b, a) give different proposals, the merge depends on arrival. Then search the merge code for a rule about two results for one order. If there is none, that is the bug.

Fix: write the rule down and test it. In merge_proposals, two different action lists for one order raise Conflict, and the ticket goes to a person with both proposals. test_workers_that_disagree_about_one_order_conflict keeps it that way.

Your turn: predict the merge

The customer of T-90201 adds one sentence: “Also, the parcel I got from you last week smells of gas.” A part that reads this applies rule H1, a safety case, and proposes hand_to_person. The other part still proposes the 9.95 refund. (Constructed: the real ticket has no such sentence.)

Predict first: what does merge_proposals(late, safety) return, and does the 9.95 refund survive? Then build safety in your Python session and check:

safety = Resolution(outcome="hand_to_person", rule="H1", reply="A colleague will contact you.", reason="smell of gas")

Show the answer

print(merge_proposals(late, safety).outcome, merge_proposals(late, safety).actions)

hand_to_person []

The merge returns the hand_to_person part, and the refund is gone. That is the rule: if any part needs a person, the whole ticket goes to a person. The person can still decide on the refund, with the full ticket in front of them. A merge that kept the refund and dropped the safety case would let a cheap, automatic action hide the important one.

Knowledge check

Check your understanding

Which of these can run in parallel, and why? (a) get_order for two orders of one ticket (b) get_order, then search_policy for the rule that the order needs (c) two request_refund actions on the same order

Show the answer

Only (a). The two reads do not need each other’s results and change nothing, so they are independent. In (b), the policy search depends on what the order shows, so it must wait. (c) are writes on one order: they go through approval and run one after the other with operation IDs. Two different refunds on one order are a conflict for a person.

Check your understanding

gather() with three workers took 0.21 s for three reads with a simulated 0.2 s delay. Omar asks: 'So parallel reads make Larkfield three times faster?' What do you answer?

Show the answer

No. The 0.2 s delay is simulated. Two real reads of the practice database take about 0.6 ms, and the model calls of T-90201 took 2.814 s. The time of a task is in the model calls, which depend on each other and cannot run in parallel. Parallel reads help only with slow real services, and then you measure the saving.

Check your understanding

Practical: in your Python session, call merge_proposals(lost, late), with the two parts in the other order. What happens, and why does that matter?

Show the answer

It raises the same Conflict: “the two workers propose different actions for LK-640763”. The rule compares the actions per order, so it does not depend on which part came first. A merge whose result depends on the order of arrival would make the same ticket give different proposals from run to run.

Recap

Parallel work fans a task out into independent parts, then merges their results. Parts are independent when none needs another’s result and no two change the same thing.
Before you run anything in parallel, write down three things: the inputs of each part, its output, and the merge rule. A failure stays a failure.
gather runs reads at the same time and returns one result per call, in call order. merge_proposals combines parts on different orders and raises Conflict for two different proposals on one order: a person decides, never chance.
In this project, parallel reads save almost nothing: the reads take milliseconds and the model calls seconds. Never run writes in parallel.
python -m pytest prints 63 passed.

Next: Multi-agent trade-offs