Agent Design Patterns

"""Pattern 0: Basic responder (one model call)

The simplest design: the user's query goes to the model once and the answer
comes straight back. There is no tool use, no loop, and no memory, so the
agency level is the lowest. It is fast and cheap, but the answer is only as
good as the model's built-in knowledge, which may be outdated or wrong.

Illustrative: llm() is a stand-in. In the Colab notebook you can swap it for
a real open model.
"""

def llm(prompt: str) -> str:
    """Stand-in for a language model call (swap in a real model here)."""
    canned = {
        "books": "Here are some books on smart mobility: (1) ... (2) ... (3) ...",
    }
    for key, answer in canned.items():
        if key in prompt.lower():
            return answer
    return "I'm not sure, but here is my best guess based on my training data."


def basic_responder(query: str) -> str:
    prompt = f"You are a helpful assistant.\nUser: {query}\nAssistant:"
    return llm(prompt)


query = "Please suggest recent books on smart mobility"
print("User: ", query)
print("Model:", basic_responder(query))
print("\nNote: nothing checks whether these books exist. Later patterns add tools and verification.")

"""Pattern 1: Sequential workflow (prompt chaining)

A task is split into fixed steps. Each step's output becomes the next step's
input, and a "gate" between steps can stop the chain early if a check fails.
Because the path is fixed in code, the behavior is predictable and easy to debug.

Illustrative: each step would normally be its own LLM call with its own prompt.
"""

def llm(task: str, text: str) -> str:
    """Stand-in for an LLM; each 'task' would be a different prompt."""
    if task == "extract":
        return "smart parking; real-time occupancy sensors; mobile app; reduce search time"
    if task == "outline":
        points = [p.strip() for p in text.split(";")]
        return "\n".join(f"{i}. {p.capitalize()}" for i, p in enumerate(points, 1))
    if task == "write":
        return ("Our campus smart-parking pilot uses real-time occupancy sensors and a "
                "mobile app so drivers spend less time searching for a space.")
    raise ValueError(task)


def gate_has_enough_points(outline: str, minimum: int = 3) -> bool:
    return len(outline.splitlines()) >= minimum


def sequential_pipeline(notes: str) -> str:
    print("Step 1: extract key ideas")
    ideas = llm("extract", notes)
    print("   ->", ideas)

    print("Step 2: turn ideas into an outline")
    outline = llm("outline", ideas)
    print("   ->", outline.replace("\n", "\n      "))

    print("Gate: does the outline have at least 3 points?")
    if not gate_has_enough_points(outline):
        return "Stopped: not enough material. Ask the user for more notes."
    print("   -> yes, continue")

    print("Step 3: write the final paragraph")
    return llm("write", outline)


notes = "Meeting notes: parking is hard near Building 22; sensors could help; app idea..."
print("\nFinal output:", sequential_pipeline(notes))

"""Pattern 2: Routing

A router (often a small, cheap model) classifies each request and sends it to
the most suitable handler: a specialized prompt, a different model, or a tool.
This keeps each handler simple and lets you use expensive models only when needed.

Illustrative: classify() uses keywords; a real router would ask an LLM to
return one label from a fixed list (structured output).
"""

ROUTES = {
    "image": "Image generation model",
    "code": "Coding model",
    "math": "Calculator tool",
    "general": "General chat model",
}


def classify(query: str) -> str:
    """Router: return exactly one label from ROUTES."""
    q = query.lower()
    if any(w in q for w in ("image", "picture", "draw", "generate a photo")):
        return "image"
    if any(w in q for w in ("python", "bug", "function", "code")):
        return "code"
    if any(ch.isdigit() for ch in q) and any(op in q for op in "+-*/"):
        return "math"
    return "general"


def handle(route: str, query: str) -> str:
    if route == "math":
        expr = "".join(ch for ch in query if ch in "0123456789+-*/(). ")
        return f"{expr.strip()} = {eval(expr, {'__builtins__': {}})}"
    if route == "image":
        return "[image request sent to an image model: 'a camel crossing a desert highway']"
    if route == "code":
        return "Here is a fixed version of your function..."
    return "Here is a friendly general answer."


queries = [
    "Generate an image of a camel crossing a desert highway",
    "Why does my Python function return None?",
    "What is 12 * (7 + 3)?",
    "What is KFUPM known for?",
]
for q in queries:
    route = classify(q)
    print(f"Query:   {q}\nRoute:   {route} -> {ROUTES[route]}\nAnswer:  {handle(route, q)}\n")

"""Pattern 3: Parallel execution (fan-out, then aggregate)

Independent subtasks run at the same time and their results are combined.
Two common forms:
  * Sectioning: different workers handle different parts of a task.
  * Voting: several workers do the same task and the answers are compared.

We use asyncio so the three "reviewers" wait concurrently. Each one pretends
to take one second, but the whole run takes about one second, not three.
"""
import asyncio
import time


async def reviewer(aspect: str, text: str, delay: float = 1.0) -> dict:
    """Stand-in for an LLM call that reviews one aspect of the text."""
    await asyncio.sleep(delay)                       # simulated network latency
    findings = {
        "clarity": "Clear overall; define 'agentic' in the first sentence.",
        "accuracy": "No factual problems found in this short draft.",
        "safety": "Mentions sensor data: add a note about privacy.",
    }
    return {"aspect": aspect, "finding": findings[aspect]}


async def main():
    draft = "Agentic systems can monitor campus parking sensors and guide drivers."
    start = time.perf_counter()

    # Fan-out: start all reviewers at once.
    results = await asyncio.gather(
        reviewer("clarity", draft),
        reviewer("accuracy", draft),
        reviewer("safety", draft),
    )

    # Aggregate: combine the independent results into one report.
    print("Combined review:")
    for r in results:
        print(f"  - {r['aspect']:<9} {r['finding']}")
    print(f"\nElapsed: {time.perf_counter() - start:.1f} s for 3 reviews (run in parallel)")


asyncio.run(main())

"""Pattern 4: Tool calling with human confirmation

The model decides which tools to call and with what arguments, observes the
results, and continues until it can answer. Actions that change the world
(like sending email) pause for human confirmation first.

Scenario from the course slides: "Filter my emails by importance and send an
out-of-office (OOO) notification to the top 5 most important emails."

Illustrative: plan_tool_calls() stands in for the model's tool-call decisions.
"""

INBOX = [
    {"id": 1, "sender": "dean@university.edu", "subject": "Budget approval needed", "urgent": True},
    {"id": 2, "sender": "newsletter@shop.com", "subject": "Weekend sale", "urgent": False},
    {"id": 3, "sender": "student@university.edu", "subject": "Thesis draft feedback", "urgent": True},
    {"id": 4, "sender": "colleague@university.edu", "subject": "Paper review deadline", "urgent": True},
    {"id": 5, "sender": "it@university.edu", "subject": "Password expires soon", "urgent": False},
    {"id": 6, "sender": "partner@industry.com", "subject": "Project kickoff", "urgent": True},
    {"id": 7, "sender": "events@university.edu", "subject": "Seminar tomorrow", "urgent": False},
]

# ---- Tools ----------------------------------------------------------------
def read_emails() -> list:
    return INBOX

def rank_by_importance(emails: list, top_k: int) -> list:
    def score(e):
        return (2 if e["urgent"] else 0) + (1 if e["sender"].endswith("university.edu") else 0)
    return sorted(emails, key=score, reverse=True)[:top_k]

def send_ooo(email_ids: list) -> str:
    return f"Sent OOO replies to emails {email_ids}"

TOOLS = {"read_emails": read_emails, "rank_by_importance": rank_by_importance, "send_ooo": send_ooo}
NEEDS_APPROVAL = {"send_ooo"}                     # tools with side effects

# ---- Agent loop -----------------------------------------------------------
def human_approves(tool: str, args: dict) -> bool:
    print(f"   [approval needed] {tool}({args}) -> auto-approved for this demo")
    return True                                   # in an app, show a confirm button

def agent(task: str):
    print("Task:", task)
    emails = call("read_emails", {})
    top = call("rank_by_importance", {"emails": emails, "top_k": 5})
    for e in top:
        print(f"   #{e['id']} {e['subject']} ({e['sender']})")
    return call("send_ooo", {"email_ids": [e["id"] for e in top]})

def call(tool: str, args: dict):
    shown = {k: (f"<{len(v)} items>" if isinstance(v, list) and v and isinstance(v[0], dict) else v) for k, v in args.items()}
    print(f"-> tool call: {tool}({shown})")
    if tool in NEEDS_APPROVAL and not human_approves(tool, args):
        return "Canceled by the user"
    return TOOLS[tool](**args)

print("\nResult:", agent("Filter my emails by importance and send OOO to the top 5"))

"""Pattern 5: Reflection / evaluator-optimizer loop

One model call generates a draft; another call (or a set of checks) evaluates
it and gives feedback; the generator revises. The loop stops when the
evaluator is satisfied or after a maximum number of rounds.

Illustrative: generate() and evaluate() are stand-ins. In practice the
evaluator is often an LLM with a rubric, unit tests, or both.
"""

def generate(task: str, feedback: list[str], round_no: int) -> str:
    """Stand-in generator that improves when it receives feedback."""
    drafts = [
        "def average(xs): return sum(xs) / len(xs)",
        "def average(xs):\n    if not xs:\n        return 0.0\n    return sum(xs) / len(xs)",
        'def average(xs: list[float]) -> float:\n    """Return the mean of xs, or 0.0 for an empty list."""\n'
        "    if not xs:\n        return 0.0\n    return sum(xs) / len(xs)",
    ]
    return drafts[min(round_no, len(drafts) - 1)]


def evaluate(code: str) -> list[str]:
    """Evaluator: run tests and style checks, return a list of problems."""
    problems = []
    scope = {}
    exec(code, scope)                       # fine for this demo; sandbox untrusted code in real systems
    try:
        scope["average"]([])
    except ZeroDivisionError:
        problems.append("Crashes on an empty list.")
    if scope["average"]([2, 4]) != 3:
        problems.append("Wrong result for [2, 4].")
    if '"""' not in code:
        problems.append("Missing docstring.")
    return problems


task = "Write a Python function average(xs)."
feedback: list[str] = []
for round_no in range(5):
    draft = generate(task, feedback, round_no)
    feedback = evaluate(draft)
    print(f"--- Round {round_no + 1} ---\n{draft}")
    print("Evaluator:", "PASS" if not feedback else "; ".join(feedback), "\n")
    if not feedback:
        break

"""Pattern 6: Plan-and-execute

A planner first writes the whole plan as structured steps. An executor then
runs each step with a tool and records the result. If a step fails, the
planner is asked to revise the remaining plan (replanning).

Illustrative: make_plan() and replan() stand in for LLM calls that return JSON.
"""
import json

def make_plan(goal: str) -> list[dict]:
    plan = [
        {"step": 1, "tool": "search_papers", "input": "agentic AI survey"},
        {"step": 2, "tool": "summarize", "input": "results of step 1"},
        {"step": 3, "tool": "email", "input": "send summary to the study group"},
    ]
    print("Planner output:\n" + json.dumps(plan, indent=2))
    return plan

def replan(failed_step: dict) -> list[dict]:
    print(f"   Replanning after failure in step {failed_step['step']}...")
    return [{"step": failed_step["step"], "tool": "post_to_forum", "input": "share summary on the course forum"}]

# ---- Tools (one fails on purpose to show replanning) ----------------------
def search_papers(q):  return ["A Survey on LLM-based Autonomous Agents", "ReAct"]
def summarize(_):      return "Two key ideas: agents combine LLM reasoning with tools; ReAct interleaves thoughts and actions."
def email(_):          raise ConnectionError("mail server unavailable")
def post_to_forum(_):  return "Posted to forum"
TOOLS = {f.__name__: f for f in (search_papers, summarize, email, post_to_forum)}

def execute(goal: str):
    queue = make_plan(goal)
    results = {}
    while queue:
        step = queue.pop(0)
        try:
            results[step["step"]] = TOOLS[step["tool"]](step["input"])
            print(f"Executed step {step['step']} [{step['tool']}] -> {results[step['step']]}")
        except Exception as err:
            print(f"Step {step['step']} [{step['tool']}] failed: {err}")
            queue = replan(step) + queue
    return results

execute("Find and share a short summary of recent work on agentic AI")

"""Pattern 7: Multi-agent collaboration (manager and sub-agents)

A manager agent splits the task, assigns parts to specialized sub-agents, and
combines their work. The course slides name three cooperation styles:
  * Augmentative: each agent adds a different piece (division of labor).
  * Integrative: agents' outputs are merged into one consistent result.
  * Debative: agents critique each other until they converge.

Illustrative: every agent here is a small Python class with canned behavior.
"""

class Agent:
    def __init__(self, name: str, role: str):
        self.name, self.role = name, role

    def work(self, task: str) -> str:
        raise NotImplementedError


class Researcher(Agent):
    def work(self, task):
        return "Facts: Riyadh Metro opened in 2024; ride-hailing demand is growing."

class Analyst(Agent):
    def work(self, task):
        return "Insight: first/last-mile links to metro stations are the main gap."

class Critic(Agent):
    def work(self, draft):
        return "Missing: cite data sources and consider summer heat on walking distance." if "heat" not in draft else "OK"


class Manager(Agent):
    def __init__(self, team):
        super().__init__("Manager", "coordinates the team")
        self.team = team

    def work(self, task):
        print(f"[Manager] Task: {task}")
        # Augmentative: split the work between specialists.
        parts = {a.name: a.work(task) for a in self.team if not isinstance(a, Critic)}
        for name, out in parts.items():
            print(f"[{name}] {out}")
        # Integrative: merge the pieces into one draft.
        draft = " ".join(parts.values())
        # Debative: a critic challenges the draft, and the manager revises.
        critic = next(a for a in self.team if isinstance(a, Critic))
        for round_no in range(1, 3):
            verdict = critic.work(draft)
            print(f"[{critic.name}] round {round_no}: {verdict}")
            if verdict == "OK":
                break
            draft += " Recommendation: shaded walkways and e-scooter docks within 500 m, given summer heat; sources to be cited."
        return draft


team = [Researcher("Researcher", "finds facts"), Analyst("Analyst", "draws insights"), Critic("Critic", "reviews")]
print("\nFinal answer:\n" + Manager(team).work("Brief on improving access to metro stations"))