Workflow Patterns
Many tasks are best served by a workflow: model calls arranged in a structure that you define in code. Anthropic’s guide “Building effective agents” describes five patterns that cover most needs. They are worth knowing by name.
The examples use a stand-in function for the model, so that they run in your browser and you can see the structure clearly.
def llm(prompt):
"""Stand-in for a model call. It returns canned text chosen by keywords in the prompt."""
p = prompt.lower()
if "classify" in p:
return "billing" if "charged" in p or "refund" in p else "technical"
if "outline" in p:
return "1. Problem 2. Solution 3. Next steps"
if "write" in p:
return "Draft text based on: " + prompt[-40:]
if "review" in p:
return "APPROVED" if "revised" in p else "REVISE: add a clear next step"
return "response to: " + prompt[:40]
print(llm("Classify this ticket: I was charged twice"))
1. Prompt chaining
Break the task into a fixed sequence of steps. The output of each step is the input to the next. You can add checks between steps.
def llm(prompt):
p = prompt.lower()
if "outline" in p: return "1. Problem 2. Solution 3. Next steps"
if "write" in p: return "Draft text following the outline: " + prompt.split("Outline:")[-1].strip()
return "ok"
def write_announcement(topic):
outline = llm(f"Create an outline for an announcement about {topic}")
if "Next steps" not in outline: # a check between steps
raise ValueError("outline is missing a next-steps section")
return llm(f"Write the announcement. Outline: {outline}")
print(write_announcement("the new calendar view"))
Use it when a task divides cleanly into fixed stages, and each stage is easier than the whole.
2. Routing
Classify the input, then send it to a specialised prompt or tool for that kind of input.
def llm(prompt):
p = prompt.lower()
if "classify" in p:
return "billing" if "charged" in p or "refund" in p else "technical"
return "reply from the " + prompt.split(":")[0] + " handler"
HANDLERS = {
"billing": lambda ticket: llm(f"billing: {ticket}"),
"technical": lambda ticket: llm(f"technical: {ticket}"),
}
def handle(ticket):
category = llm(f"Classify this ticket as billing or technical: {ticket}")
handler = HANDLERS.get(category, lambda t: "sent to a human agent")
return category, handler(ticket)
print(handle("I was charged twice this month"))
print(handle("The app crashes on start"))
Use it when inputs fall into distinct types that are better handled separately. Routing also lets you send easy requests to a cheaper model and hard ones to a more capable one.
3. Parallelisation
Run several model calls at the same time and combine the results. There are two forms.
- Sectioning: split the work into independent parts, such as reviewing a document separately for accuracy, tone and legal risk.
- Voting: run the same task several times and combine the answers, for example by taking the majority.
from collections import Counter
def review(aspect, text):
findings = {"accuracy": "no issues", "tone": "too formal", "length": "too long"}
return aspect, findings[aspect]
text = "Draft announcement..."
sections = dict(review(aspect, text) for aspect in ("accuracy", "tone", "length"))
print("sectioning:", sections)
votes = ["safe", "safe", "unsafe"] # three independent checks of the same content
print("voting:", Counter(votes).most_common(1)[0][0])
In a real system the calls run concurrently, so the total time is that of the slowest call, not the sum.
Use it when parts of the work do not depend on each other, or when several independent opinions give a more reliable answer.
4. Orchestrator and workers
A central model call breaks the task into subtasks at run time, hands each to a worker call, and combines the results. Unlike parallelisation, the subtasks are not fixed in advance. The orchestrator decides them from the input.
def orchestrator(task):
# a real orchestrator would ask a model to produce this plan
return ["find pricing of competitor A", "find pricing of competitor B", "summarise differences"]
def worker(subtask):
return f"result of '{subtask}'"
def run(task):
plan = orchestrator(task)
results = [worker(subtask) for subtask in plan]
return "Report:\n" + "\n".join(f"- {r}" for r in results)
print(run("Compare our pricing with competitors"))
Use it when you cannot predict the subtasks, such as a code change that touches an unknown number of files, or research that needs an unknown number of searches.
5. Evaluator and optimiser
One call produces a result. A second call evaluates it against criteria and gives feedback. The loop repeats until the result passes or a limit is reached.
def generate(task, feedback=None):
return "Draft v2 (revised) with a clear next step" if feedback else "Draft v1"
def evaluate(draft):
return ("APPROVED", None) if "revised" in draft else ("REVISE", "add a clear next step")
def improve(task, max_rounds=3):
feedback = None
for round_number in range(1, max_rounds + 1):
draft = generate(task, feedback)
verdict, feedback = evaluate(draft)
print(f"round {round_number}: {draft!r} -> {verdict}")
if verdict == "APPROVED":
return draft
return draft
improve("write an announcement")
Use it when there are clear criteria for quality and revising against feedback measurably improves the result, as in translation or writing to a specification.
Choosing
| Pattern | The steps are | Good for |
|---|---|---|
| Chaining | Fixed and sequential | Tasks with clear stages |
| Routing | One of several fixed branches | Mixed kinds of input |
| Parallelisation | Fixed and independent | Speed, or several viewpoints |
| Orchestrator and workers | Decided at run time | Tasks that cannot be divided in advance |
| Evaluator and optimiser | A loop until good enough | Work that improves with revision |
The patterns combine. A router can send a request to a chain, and a chain can end with an evaluator loop.
A principle
Add a pattern only when it improves results that you have measured. Every extra model call adds cost, delay and another place for things to go wrong.
Practice
Design a workflow for answering customer emails that uses routing and an evaluator loop. Draw it as boxes and arrows, then write the functions using the stand-in llm.