The 3 Questions Every AI Agent Architect Must Answer About Execution Patterns

1Agent self-decides
2  Query → agent reasons about the pattern → picks → executes
3  Costs nothing extra, and varies run to run
4
5Orchestrator classifies
6  Query → LLM classifier → pattern type → executes with it
7  Reliable, and pays a model call on every request
8
9Heuristics
10  Query → regex and keyword check → matched pattern
11                                  → no match, default sequential
12  Reliable, and effectively free

1import re
2from enum import Enum
3
4class ExecutionPattern(Enum):
5    SEQUENTIAL = "sequential"          # 80% of queries
6    PARALLEL_TOOLS = "parallel_tools"  # 15% of queries
7    MULTI_AGENT = "multi_agent"        # 5% of queries
8
9def classify_pattern(user_message: str) -> ExecutionPattern:
10    """
11    Zero-LLM-call pattern classifier. Runs in under 5ms.
12    """
13    message_lower = user_message.lower()
14
15    # Comparison queries → parallel tools
16    comparison_patterns = [
17        r"compare\s+.+\s+(?:and|vs|versus|with)\s+",
18        r"difference(?:s)?\s+between",
19        r"(?:which|what)\s+is\s+better",
20    ]
21    if any(re.search(p, message_lower) for p in comparison_patterns):
22        return ExecutionPattern.PARALLEL_TOOLS
23
24    # Multi-item queries → parallel tools
25    multi_item_patterns = [
26        r"(?:all|each|every)\s+(?:of\s+)?(?:the|my|our)\s+\w+",
27        r"(\d+)\s+(?:different|separate|distinct)",
28        r"list\s+(?:all|every)",
29    ]
30    if any(re.search(p, message_lower) for p in multi_item_patterns):
31        return ExecutionPattern.PARALLEL_TOOLS
32
33    # Heavy research → multi-agent
34    heavy_research_patterns = [
35        r"(?:deep|comprehensive|thorough)\s+(?:research|analysis|audit)",
36        r"audit\s+\d+\s+(?:competitors|companies|websites)",
37        r"research\s+(?:multiple|several|all)\s+",
38    ]
39    if any(re.search(p, message_lower) for p in heavy_research_patterns):
40        return ExecutionPattern.MULTI_AGENT
41
42    # Default
43    return ExecutionPattern.SEQUENTIAL
44
45
46assert classify_pattern("What is our pricing?") == ExecutionPattern.SEQUENTIAL
47assert classify_pattern("Compare product A and B") == ExecutionPattern.PARALLEL_TOOLS
48assert classify_pattern("Audit 5 competitors") == ExecutionPattern.MULTI_AGENT

1Tier 1: parallel tools           Tier 2: parallel agents
2(reach for this first)           (reach for this rarely)
3
4  Single agent                     Orchestrator
5      │                             ├──► Sub-agent 1 ─┐
6  asyncio.gather()                  ├──► Sub-agent 2 ─┼──► Synthesis
7      ├──► Tool 1 ─┐                └──► Sub-agent 3 ─┘        agent
8      ├──► Tool 2 ─┼──► Combine
9      └──► Tool 3 ─┘

1import asyncio
2from typing import Any
3
4async def parallel_search(
5    queries: list[str],
6    search_fn: callable,
7    max_concurrent: int = 5
8) -> list[Any]:
9    """
10    Run multiple searches in parallel with rate limiting.
11    Three KB searches that took 3s sequentially finish in about 1s.
12    """
13    semaphore = asyncio.Semaphore(max_concurrent)
14
15    async def limited_search(query: str) -> Any:
16        async with semaphore:
17            return await search_fn(query)
18
19    results = await asyncio.gather(*[
20        limited_search(q) for q in queries
21    ])
22    return results
23
24
25async def compare_products(
26    ctx: RunContext,
27    product_names: list[str]
28) -> str:
29    """Compare multiple products side by side."""
30
31    results = await parallel_search(
32        queries=[f"product info: {name}" for name in product_names],
33        search_fn=ctx.deps.search_kb,
34        max_concurrent=3
35    )
36
37    comparison_table = format_comparison(results)
38    return comparison_table

1from langgraph.prebuilt import create_react_agent
2
3async def parallel_competitor_audit(
4    competitors: list[str]
5) -> str:
6    """
7    Heavy research requiring parallel LLM reasoning.
8    Only for 5+ complex independent tasks.
9    """
10
11    async def audit_one(competitor: str) -> dict:
12        agent = create_react_agent(
13            model=get_model(),
14            tools=[web_search, crawl_url, analyze_page],
15            prompt=f"Audit {competitor}'s pricing, features, and positioning."
16        )
17        result = await agent.invoke({"messages": [...]})
18        return {"competitor": competitor, "audit": result}
19
20    audits = await asyncio.gather(*[
21        audit_one(c) for c in competitors
22    ])
23
24    synthesis = await synthesize_audits(audits)
25    return synthesis

1How many parallel operations?
2  1      → sequential, no parallelism needed
3  2-5    → simple lookups?      yes → Tier 1 (asyncio.gather)
4                                no  → ask the next question
5  5+     → separate reasoning?  yes → Tier 2 (parallel agents)
6                                no  → Tier 1

1Phase 1 (MVP)   Fully automatic. The system detects, users neither know nor care.
2Phase 2         Natural-language hints. "do a thorough comparison", "quick summary".
3Phase 3         Explicit controls. Mode selection, API parameters. If ever.

1SPEED_HINTS = {
2    "quick": ExecutionPattern.SEQUENTIAL,
3    "fast": ExecutionPattern.SEQUENTIAL,
4    "brief": ExecutionPattern.SEQUENTIAL,
5}
6
7QUALITY_HINTS = {
8    "thorough": ExecutionPattern.PARALLEL_TOOLS,
9    "comprehensive": ExecutionPattern.PARALLEL_TOOLS,
10    "detailed": ExecutionPattern.PARALLEL_TOOLS,
11    "deep": ExecutionPattern.MULTI_AGENT,
12}
13
14def detect_user_hint(message: str) -> ExecutionPattern | None:
15    """Detect explicit user preferences from natural language."""
16    message_lower = message.lower()
17
18    for hint, pattern in {**SPEED_HINTS, **QUALITY_HINTS}.items():
19        if hint in message_lower:
20            return pattern
21
22    return None  # no hint, fall through to auto-classification