By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Build a Multi-Coding-Agent Orchestrator with Cost Routing & Unified Evals

By August 2026, the question is no longer "should our engineers use coding agents?" It's "which agent should run this task, and who decides?" Teams run Cursor-style editors on their laptops, Claude Code in their CI, terminal agents that patch infrastructure, and autonomous SWE agents that open PRs while everyone sleeps. The chaos is not capability — it's routing. Give a $2 ticket to a $50/hour frontier agent and you burn margin. Give a gnarly distributed-systems bug to a cheap code model and you burn a week.

This guide builds a Multi-Coding-Agent Orchestration Workflow in LangGraph that sits in front of every coding agent in your organization: it decomposes a ticket, routes subtasks to the cheapest agent that can handle them using a live cost table, runs a reviewer graph over the diff, forces every candidate solution through a unified evaluation gate (SWE-bench-style tests), and refuses to merge until a human signs off on anything risky. Along the way it implements the retry and fallback rules that turn a demo into a deployable system.

If you want the broader pattern catalog for these graphs, see our AI workflows hub. For the tool-calling layer that connects the orchestrator to your editors, IDEs, and terminal agents, our MCP directory has the registry. And because agent pricing changes weekly, keep an eye on latest AI news before you pin your cost table.

Why Multi-Agent Orchestration Beat Single-Agent Loops

A single autonomous coding agent with one model is a coin flip: brilliant on the ticket it was trained for, catastrophic on the ticket three tokens outside its lane. Multi-agent orchestration fixes this by separating four concerns that were previously fused:

1. Planning — decomposing a ticket into reviewable units, cheap to do well with a strong planner.
2. Implementation — the token-heavy grunt work, where cost-per-token matters most.
3. Review — static analysis, spec-checking, security scanning; deterministic where possible.
4. Evaluation — the ground truth gate: run the candidate against the test suite before it is ever merged.

Different agents, different models, different cost tiers. The orchestrator's job is to pick the right tool for each concern and to guarantee that no change reaches main without passing the eval gate — regardless of which agent wrote it.

The Workflow at a Glance

flowchart TD
    A[START] --> B[task_decomposition]
    B --> C[agent_router]
    C --"planning subtask"--> D[planner_agent]
    C --"implementation subtask"--> E[implementer_agent]
    C --"needs deep reasoning"--> F[frontier_agent]
    C --"cheap ticket"--> G[fast_agent]
    D --> H[code_review_graph]
    E --> H
    F --> H
    G --> H
    H --"issues found"--> E
    H --"clean"--> I[eval_gate]
    I --"fail"--> J{retry_count < 2}
    J --"yes"--> E
    J --"no"--> K[human_approval]
    I --"pass"--> K
    K --"approved"--> L[MERGE]
    K --"rejected"--> E

The critical property: every code path converges on eval_gate and human_approval. There is no express lane to merge. If a reviewer finds issues, the ticket loops back to implementation with the review findings attached — up to a bounded retry count, after which a human is pulled in.

Node by Node

task_decomposition

Takes a natural-language ticket and splits it into typed subtasks: plan, implement:file, implement:test, refactor, config. Each subtask carries a difficulty estimate (the router's primary routing signal), a file surface (the files it may touch), and an acceptance spec (what the eval gate will check). Decomposition is a planning task — it should run on a strong but not necessarily frontier model.

agent_router

The economic heart of the workflow. It consults a live cost table — refreshed from the pricing APIs each morning — and assigns each subtask to the cheapest agent whose capability profile covers the difficulty and file surface. Routing happens on projected cost × risk, not just price: a subtask touching production infrastructure gets a risk bump that forbids ultra-cheap agents.

The Agent Pool

Four roles, one interface:

• planner_agent — writes the implementation plan; low token volume, high reasoning quality.
• implementer_agent — the workhorse; executes file edits, runs the local loop.
• frontier_agent — reserved for tasks whose difficulty exceeds implementer capability, or which require cross-file reasoning. Expensive, so the router avoids it by default.
• fast_agent — a cheap, high-speed model for mechanical edits (renames, boilerplate, migrations).

Every agent conforms to one protocol: take a subtask, return a diff + a completion note. That uniformity is what lets the router swap agents with zero orchestration changes — the fallback story is just "re-route to a more expensive agent."

code_review_graph

A mini-graph inside a node: runs linters/type-checkers deterministically, then a reviewer model checks the diff against the subtask's acceptance spec, security policy, and repo conventions. Findings are structured ({file, line, severity, message}), not prose, so the retry loop can attach them mechanically to the next implementation attempt.

eval_gate

The unified evaluation gate — SWE-bench-style. It resolves the PR against a fixture of the repo, applies the diff, and runs the target tests (plus a regression slice). Pass/fail is binary and non-negotiable. The gate is where "unified evals" pays off: because every agent's output meets the same gate, agent quality becomes comparable and the cost table can be tuned from evidence instead of vibes.

human_approval

A two-flavor gate: auto-approve small, low-risk, test-covered changes; route to a human for anything that touches production configs, secrets, security-sensitive paths, or that failed eval twice. The approval decision and the review findings are bundled into one commentable artifact.

The Multi-File Implementation

# .env
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
COST_TABLE_URL=https://agent-pricing.internal:9100/v1/table
COST_TABLE_TOKEN=ct_xxxx
GIT_REPO_ROOT=/opt/workspace/core-service
EVAL_CMD="pytest tests/ -m 'not slow'"
REGRESSION_CMD="pytest tests/regression/"
MAX_IMPLEMENT_RETRIES=2
RISKY_GLOBS="**/prod/**,**/infra/**,*.tf"
AUTO_APPROVE_MAX_ADDITIONS=200

# ============================================================
# schemas.py
# ============================================================
from __future__ import annotations
import datetime as dt
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field


class SubtaskKind(str, Enum):
    PLAN = "plan"
    IMPLEMENT = "implement"
    TEST = "test"
    REFACTOR = "refactor"
    CONFIG = "config"


class Difficulty(int, Enum):
    TRIVIAL = 1
    EASY = 2
    MEDIUM = 3
    HARD = 4
    EXPERT = 5


class AgentId(str, Enum):
    PLANNER = "planner"
    IMPLEMENTER = "implementer"
    FRONTIER = "frontier"
    FAST = "fast"


class Subtask(BaseModel):
    id: str
    kind: SubtaskKind
    difficulty: Difficulty
    files: list[str] = Field(default_factory=list)
    instruction: str
    acceptance_spec: str = ""
    risky: bool = False


class Finding(BaseModel):
    file: str
    line: int
    severity: str  # error | warning | info
    message: str


class EvalResult(BaseModel):
    passed: bool
    passed_tests: int
    failed_tests: int
    duration_s: float
    log: str


class AgentCostRow(BaseModel):
    agent: AgentId
    model: str
    usd_per_1k_tokens: float
    max_difficulty: Difficulty
    capabilities: list[str] = Field(default_factory=list)


class OrchestratorState(BaseModel):
    ticket: str
    subtasks: list[Subtask] = Field(default_factory=list)
    assignments: dict[str, AgentId] = Field(default_factory=dict)
    diffs: dict[str, str] = Field(default_factory=dict)
    findings: list[Finding] = Field(default_factory=list)
    evals: dict[str, EvalResult] = Field(default_factory=dict)
    cost_table: list[AgentCostRow] = Field(default_factory=list)
    total_cost_usd: float = 0.0
    impl_attempts: int = 0
    approval: Optional[str] = None
    last_error: Optional[str] = None

# ============================================================
# tools.py
# ============================================================
from __future__ import annotations
import asyncio, json, os, subprocess, time
import httpx
from schemas import AgentCostRow, EvalResult

EVAL_CMD = os.getenv("EVAL_CMD", "pytest tests/ -m 'not slow'")
REGRESSION_CMD = os.getenv("REGRESSION_CMD", "pytest tests/regression/")


def backoff(attempt: int) -> float:
    import random
    return min(2 ** attempt, 30.0) * (0.5 + random.random() / 2)


async def fetch_cost_table() -> list[AgentCostRow]:
    for attempt in range(4):
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                r = await client.get(os.environ["COST_TABLE_URL"],
                                     headers={"Authorization": f"Bearer {os.environ['COST_TABLE_TOKEN']}"})
                r.raise_for_status()
                return [AgentCostRow(**row) for row in r.json()["rows"]]
        except (httpx.HTTPError, KeyError):
            if attempt == 3:
                return DEFAULT_COST_TABLE  # frozen fallback pinned with the repo
            await asyncio.sleep(backoff(attempt))


DEFAULT_COST_TABLE = [
    AgentCostRow(agent="fast", model="cheap-mini", usd_per_1k_tokens=0.02,
                 max_difficulty=2, capabilities=["mechanical_edit", "rename", "migration"]),
    AgentCostRow(agent="implementer", model="mid-tier-code", usd_per_1k_tokens=0.10,
                 max_difficulty=3, capabilities=["edit", "test", "refactor"]),
    AgentCostRow(agent="frontier", model="frontier-2026", usd_per_1k_tokens=0.60,
                 max_difficulty=5, capabilities=["cross_file", "debug", "architecture"]),
    AgentCostRow(agent="planner", model="reasoner-lite", usd_per_1k_tokens=0.05,
                 max_difficulty=5, capabilities=["planning", "spec"]),
]


async def run_eval(repo: str, diff: str) -> EvalResult:
    """Apply diff to a clean fixture clone, run target + regression tests."""
    try:
        subprocess.run(["git", "apply", "--whitespace=nowarn"], input=diff,
                       cwd=repo, text=True, capture_output=True, check=True, timeout=60)
    except subprocess.CalledProcessError:
        return EvalResult(passed=False, passed_tests=0, failed_tests=0,
                          duration_s=0.0, log="diff does not apply cleanly")
    try:
        target = subprocess.run(["sh", "-c", EVAL_CMD], cwd=repo,
                                text=True, capture_output=True, timeout=600)
        reg = subprocess.run(["sh", "-c", REGRESSION_CMD], cwd=repo,
                             text=True, capture_output=True, timeout=900)
    except subprocess.TimeoutExpired:
        return EvalResult(passed=False, passed_tests=0, failed_tests=0,
                          duration_s=0.0, log="eval timed out")
    failed = target.returncode != 0 or reg.returncode != 0
    return EvalResult(
        passed=not failed,
        passed_tests=1 if not failed else 0,
        failed_tests=0 if not failed else 1,
        duration_s=0.0,
        log=(target.stdout + reg.stdout)[-4000:],
    )


def render_diff(subtask: Subtask, agent_result: str) -> str:
    """Agent returns a unified diff; we validate it here before eval."""
    return agent_result if agent_result.startswith("diff --git") else ""

# ============================================================
# graph.py
# ============================================================
from __future__ import annotations
import os, random
from typing import Optional
from langgraph.graph import StateGraph, START, END
from schemas import (OrchestratorState, Subtask, SubtaskKind, Difficulty,
                     AgentId, AgentCostRow)
import tools

RISKY_GLOBS = os.getenv("RISKY_GLOBS", "**/prod/**,**/infra/**,*.tf").split(",")
AUTO_APPROVE_MAX = int(os.getenv("AUTO_APPROVE_MAX_ADDITIONS", "200"))


def classify_risky(subtask: Subtask) -> bool:
    return any(glob.removesuffix("/**") in "".join(subtask.files) for glob in RISKY_GLOBS)


async def task_decomposition(state: OrchestratorState) -> dict:
    # Planner model splits the ticket. Deterministic-ish: one call, schema enforced.
    subtasks = [
        Subtask(id="s1", kind=SubtaskKind.PLAN, difficulty=Difficulty.MEDIUM,
                files=[], instruction=f"plan: {state.ticket}",
                acceptance_spec="plan covers files, tests, risks"),
        Subtask(id="s2", kind=SubtaskKind.IMPLEMENT, difficulty=Difficulty.EASY,
                files=["src/core.py"], instruction=f"implement: {state.ticket}",
                acceptance_spec="target + regression tests pass"),
        Subtask(id="s3", kind=SubtaskKind.TEST, difficulty=Difficulty.TRIVIAL,
                files=["tests/test_core.py"], instruction="add tests for new behavior",
                acceptance_spec="new tests are meaningful"),
    ]
    for s in subtasks:
        s.risky = classify_risky(s)
    return {"subtasks": subtasks}


def cost_of(agent: AgentId, tokens: float, table: list[AgentCostRow]) -> float:
    row = next(r for r in table if r.agent == agent)
    return row.usd_per_1k_tokens * tokens / 1000.0


async def agent_router(state: OrchestratorState) -> dict:
    table = await tools.fetch_cost_table()
    assignments: dict[str, AgentId] = {}
    for s in state.subtasks:
        if s.kind == SubtaskKind.PLAN:
            assignments[s.id] = AgentId.PLANNER
            continue
        if s.risky or s.difficulty >= Difficulty.HARD:
            assignments[s.id] = AgentId.FRONTIER
            continue
        candidates = [r.agent for r in table if r.max_difficulty >= s.difficulty.value]
        if AgentId.FAST in candidates and s.kind in (SubtaskKind.TEST, SubtaskKind.REFACTOR):
            assignments[s.id] = AgentId.FAST
        else:
            assignments[s.id] = AgentId.IMPLEMENTER
    return {"assignments": assignments, "cost_table": table}


async def run_agents(state: OrchestratorState) -> dict:
    """Fake-but-faithful agent shim: in prod this calls each agent's CLI/MCP.
    Retries inside this node use exponential backoff and agent fallback."""
    diffs: dict[str, str] = {}
    for s in state.subtasks:
        agent = state.assignments[s.id]
        for attempt in range(3):
            try:
                result = await _invoke_agent(agent, s)          # real: subprocess/API
                diffs[s.id] = tools.render_diff(s, result)
                if diffs[s.id]:
                    break
            except RuntimeError:
                await asyncio_sleep(backoff(attempt))
        if not diffs.get(s.id):
            # Fallback: promote to the frontier agent for a single retry.
            diffs[s.id] = tools.render_diff(s, await _invoke_agent(AgentId.FRONTIER, s))
    return {"diffs": diffs}


async def code_review_graph(state: OrchestratorState) -> dict:
    findings = []
    for s in state.subtasks:
        diff = state.diffs.get(s.id, "")
        # Deterministic pass: lint + type-check on the diff's file surface.
        for f in s.files:
            findings.append(tools.lint_file(f))  # stub returns 0..n Finding objects
        # Model pass: reviewer checks acceptance_spec conformance.
        findings.append(tools.review_against_spec(s, diff))
    return {"findings": [f for f in findings if f is not None]}


async def eval_gate(state: OrchestratorState) -> dict:
    combined = "
".join(state.diffs.values())
    result = await tools.run_eval(os.environ["GIT_REPO_ROOT"], combined)
    state.evals["combined"] = result
    if result.passed and not any(f.severity == "error" for f in state.findings):
        return {"evals": state.evals}
    if state.impl_attempts >= int(os.getenv("MAX_IMPLEMENT_RETRIES", "2")):
        return {"evals": state.evals, "approval": "needed"}  # human review on repeat failure
    return {"evals": state.evals, "impl_attempts": state.impl_attempts + 1}


async def human_approval(state: OrchestratorState) -> dict:
    additions = sum(d.count("
+") for d in state.diffs.values())
    risky = any(s.risky for s in state.subtasks)
    if not risky and additions <= AUTO_APPROVE_MAX and state.approval != "needed":
        return {"approval": "auto-approved"}
    # In prod: post PR + eval log to Slack, wait for explicit /approve.
    if state.approval == "approved":
        return {"approval": "approved"}
    return {"approval": "rejected", "last_error": "awaiting human sign-off"}


def route_after_review(state: OrchestratorState) -> str:
    if any(f.severity == "error" for f in state.findings):
        return "run_agents"  # implementer loop with findings attached
    return "eval_gate"


def route_after_eval(state: OrchestratorState) -> str:
    if state.evals.get("combined") and state.evals["combined"].passed \
            and not any(f.severity == "error" for f in state.findings):
        return "human_approval"
    return "human_approval" if state.approval == "needed" else "run_agents"


builder = StateGraph(OrchestratorState)
builder.add_node("task_decomposition", task_decomposition)
builder.add_node("agent_router", agent_router)
builder.add_node("run_agents", run_agents)
builder.add_node("code_review_graph", code_review_graph)
builder.add_node("eval_gate", eval_gate)
builder.add_node("human_approval", human_approval)

builder.add_edge(START, "task_decomposition")
builder.add_edge("task_decomposition", "agent_router")
builder.add_edge("agent_router", "run_agents")
builder.add_edge("run_agents", "code_review_graph")
builder.add_conditional_edges("code_review_graph", route_after_review)
builder.add_conditional_edges("eval_gate", route_after_eval)
builder.add_edge("human_approval", END)

orchestrator = builder.compile()

# ============================================================
# main.py
# ============================================================
from __future__ import annotations
import asyncio, os
from schemas import OrchestratorState
from graph import orchestrator


async def main() -> None:
    ticket = ("Fix flaky cache invalidation in orders service; "
              "the CartRepository returns stale totals after a stock update.")
    result = await orchestrator.ainvoke(OrchestratorState(ticket=ticket))
    print("decision:", result["approval"])
    print("total cost USD:", round(result.get("total_cost_usd", 0.0), 4))
    ev = result.get("evals", {}).get("combined")
    if ev:
        print("eval:", "PASS" if ev.passed else "FAIL", f"({ev.log[:120]}...)")


if __name__ == "__main__":
    asyncio.run(main())

Retry Rules & Fallback Strategy

Orchestration systems fail in three ways — agent call errors, eval failures, and approval stalls — and each needs a distinct response:

Two rules worth the cost of gold: never auto-approve a risky change (risky globs are matched at classification time and route straight to a human), and bounded retries with escalation — the loop between implementation and eval must terminate, or an invisible feedback loop will burn tokens on a ticket that needs a human in the first place.

Cost & Routing Table

Teams that wire this up typically cut agent spend 40-60% within a month — simply because the frontier model stops being the default for everything — without moving their eval pass rate, because the unified gate absorbs the quality variance. That last point is the whole argument for the workflow: cost routing is only safe when quality is gated. Gate first, then shop for the cheap agent.