跳到正文
DEV Community· BotSailor·· 5 小時前AI 評分64

如何為生產環境構建多智能體系統:從單一智能體到團隊的路線圖

From One Agent to a Team: How to Build a Multi-Agent System for Production (A Roadmap)

AI 導讀

將單一 AI 智能體擴展為生產環境多智能體系統,應先量度基線並加固現有智能體,再按可量度的瓶頸決定是否拆分。路線圖涵蓋按職責拆分、設計結構化交接、加入追蹤與預算,以及在 CI/CD 中自動評測和逐步部署。文中提供六至八週的項目計劃,並以成功率或成本優於單智能體基線、符合延遲預算及可端到端追蹤作為完成條件。

正文

From One Agent to a Team: How to Build a Multi-Agent System for Production (A Roadmap)

Your single AI agent works. It answers questions, calls a few tools, and the demo went well. Then production happens: the prompt grows to 3,000 tokens, the agent has 25 tools and picks the wrong one, latency creeps up, and nobody can explain why last Tuesday's run failed.

That is usually the moment someone says, "Let's make it multi-agent." Sometimes that's the right call. Often it isn't, or it's right only after some groundwork.

This article is a practical roadmap for how to build a multi-agent system for production. It covers:

  • How to decide whether you actually need more than one agent
  • A staged path from single agent to coordinated team
  • Code patterns for handoffs, budgets, tracing, and evals
  • The failure modes that tend to appear at each stage
  • A capstone checklist you can turn into a project plan

The examples are framework-agnostic Python, so you can apply them with LangGraph, CrewAI, AG2 (AutoGen), the OpenAI Agents SDK, or plain code.

Table of Contents

Prerequisites

Before you start, you should have:

  • [ ] Working knowledge of Python 3.10+ and async/await
  • [ ] Experience calling an LLM API with tool/function calling
  • [ ] A basic understanding of queues, retries, and idempotency
  • [ ] Familiarity with logging or tracing (OpenTelemetry helps, but isn't required)
  • [ ] An existing single-agent prototype, or a clear use case for one

Install the libraries used in the snippets:

pip install pydantic pytest

Single-agent vs multi-agent: when to split

A multi-agent AI architecture adds coordination cost: more tokens, more latency, more failure surface, and harder debugging. Split only when the evidence says a single agent has hit a ceiling.

Rule of thumb: If you can't point to a measurable failure that a split would fix, don't split yet.

Signals that a split may be justified:

Signal What it looks like Why splitting helps
Tool overload 15+ tools, wrong-tool selection rate rising Smaller toolsets per agent improve selection accuracy
Prompt bloat System prompt mixes unrelated policies Focused prompts are easier to test and version
Different risk levels Read-only research vs. actions that move money or data Separate permissions per agent
Parallelizable work Independent subtasks run serially Workers can run concurrently
Different models needed Cheap model for triage, strong model for reasoning Cost and latency optimization per role

Signals that you should not split:

  • The task is a linear sequence a single prompt chain handles fine
  • Your failures come from bad retrieval or missing evals, not from role confusion
  • You have no tracing yet. Fix that first; you'll need it either way.

Stage 0: Baseline a single agent

Before any architecture change, capture a baseline. You can't claim a multi-agent system is better without numbers.

Track at least:

  • Task success rate on a fixed test set (start with 30 to 50 cases)
  • Median and p95 latency
  • Cost per task (input and output tokens)
  • Tool-call error rate
# baseline.py
import json
import statistics
import time

def run_baseline(agent, cases):
    results = []
    for case in cases:
        start = time.perf_counter()
        output = agent.run(case["input"])
        latency = time.perf_counter() - start
        results.append({
            "id": case["id"],
            "passed": case["check"](output),
            "latency_s": latency,
        })
    latencies = [r["latency_s"] for r in results]
    summary = {
        "success_rate": sum(r["passed"] for r in results) / len(results),
        "p50_latency_s": statistics.median(latencies),
        "p95_latency_s": sorted(latencies)[int(0.95 * len(latencies)) - 1],
    }
    print(json.dumps(summary, indent=2))
    return results

Save this output. Every later stage gets compared against it.

Bar chart comparing baseline single-agent success rate, latency, and cost per task before any architecture changes

Stage 1: Harden the single agent

Many "we need multi-agent" problems disappear after hardening. Do these first:

  1. Constrain tool schemas. Use strict types and enums instead of free-form strings.
  2. Add timeouts and retries around every tool call.
  3. Trim the toolset to what the task needs.
  4. Version your prompts the way you version code.
  5. Return structured errors from tools so the agent can recover.
from pydantic import BaseModel, Field
from typing import Literal

class RefundRequest(BaseModel):
    order_id: str = Field(pattern=r"^ORD-\d{6}$")
    reason: Literal["damaged", "late", "wrong_item", "other"]
    amount_cents: int = Field(gt=0, le=50_000)

A tool that rejects bad input with a clear error is worth more than another paragraph in your prompt.

Stage 2: Split by responsibility

If the baseline and hardening still show a ceiling, split by responsibility, not by "personality." Good boundaries follow real differences in tools, permissions, or knowledge.

The most common production pattern is a supervisor (router) with specialist workers:

            ┌────────────┐
 request ──▶│ Supervisor │
            └─────┬──────┘
        ┌─────────┼──────────┐
        ▼         ▼          ▼
   Research     Action     Review
    agent       agent      agent
  (read-only) (writes)   (checks output)

Here is a minimal supervisor loop with a step cap:

# supervisor.py
from dataclasses import dataclass, field

@dataclass
class RunState:
    task: str
    steps: int = 0
    max_steps: int = 8
    artifacts: dict = field(default_factory=dict)

def supervisor(state: RunState, workers: dict, router) -> RunState:
    while state.steps < state.max_steps:
        decision = router(state)  # returns {"next": "research" | "action" | "review" | "done"}
        if decision["next"] == "done":
            return state
        worker = workers[decision["next"]]
        state.artifacts[decision["next"]] = worker(state)
        state.steps += 1
    raise RuntimeError(f"Step budget exceeded for task: {state.task!r}")

Notice the hard max_steps. An uncapped agent loop is one of the fastest ways to generate a surprise bill.

Other topologies worth knowing

  • Pipeline: fixed order (extract, then validate, then act). Simple and very testable.
  • Parallel fan-out/fan-in: several workers run at once and a reducer merges results.
  • Hierarchical: supervisors of supervisors. Rarely needed; start flat.
  • Peer-to-peer / swarm: agents hand off to each other dynamically. Flexible but hardest to debug.

Start with the least dynamic topology that solves your problem.

Stage 3: Design handoff contracts

Most multi-agent bugs are handoff bugs: one agent passes vague, incomplete, or malformed context to the next. Treat every handoff like an API contract.

# handoff.py
from pydantic import BaseModel
from typing import Literal

class Handoff(BaseModel):
    run_id: str
    from_agent: str
    to_agent: str
    goal: str                      # what the next agent must accomplish
    context: dict                  # only what it needs, not the full history
    constraints: list[str] = []    # e.g. "read-only", "max $50 refund"
    status: Literal["ok", "needs_review", "failed"] = "ok"

def validate_handoff(raw: dict) -> Handoff:
    # Raises a ValidationError instead of letting a bad payload spread downstream
    return Handoff.model_validate(raw)

Good handoff practice:

  • Pass summaries and structured fields, not raw conversation dumps
  • Include a goal and constraints in every handoff
  • Give each run a run_id that survives every hop
  • Fail loudly on invalid payloads

Callout: Shared memory is convenient, but unrestricted shared state means any agent can overwrite any fact. Prefer scoped state with explicit reads and writes.

Stage 4: Add observability

You can't operate what you can't see. For agent observability, trace every step with the same run_id.

# tracing.py
import functools
import json
import logging
import time

log = logging.getLogger("agents")

def traced(agent_name):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(state, *args, **kwargs):
            start = time.perf_counter()
            try:
                result = fn(state, *args, **kwargs)
                status = "ok"
                return result
            except Exception as exc:
                status = f"error:{type(exc).__name__}"
                raise
            finally:
                log.info(json.dumps({
                    "run_id": getattr(state, "run_id", None),
                    "agent": agent_name,
                    "status": status,
                    "latency_ms": round((time.perf_counter() - start) * 1000),
                }))
        return wrapper
    return decorator

Capture per step: input summary, tool calls, output, token counts, latency, and errors. Whether you use OpenTelemetry, a hosted tracing tool, or structured logs, make sure you can answer: "Why did run X take 40 seconds and cost $0.80?"

Trace timeline showing a supervisor delegating to research, action, and review agents with latency per step

Stage 5: Guardrails and budgets

Production agents need limits that the model can't talk its way around. Enforce them in code, not in prompts.

  • Step budget: maximum hops per run
  • Token and cost budget: abort when a run exceeds a threshold
  • Permission scopes: the research agent has no write tools at all
  • Human-in-the-loop gates: require approval for irreversible actions
  • Output validation: schema checks before anything reaches a user or a database
class Budget:
    def __init__(self, max_tokens: int, max_usd: float):
        self.max_tokens, self.max_usd = max_tokens, max_usd
        self.tokens = 0
        self.usd = 0.0

    def charge(self, tokens: int, usd: float):
        self.tokens += tokens
        self.usd += usd
        if self.tokens > self.max_tokens or self.usd > self.max_usd:
            raise RuntimeError("Run budget exceeded; aborting and escalating to a human")

For risky actions, route to approval:

def execute_action(action, approver):
    if action.irreversible and not approver.approved(action):
        return {"status": "needs_review", "action": action.name}
    return action.run()

Stage 6: Evals in CI/CD

This is where the DevOps mindset pays off. Treat agent behavior like code behavior: every change to a prompt, tool, or model should run through an automated eval.

# test_agent_evals.py
import pytest
from my_agents import run_system
from eval_cases import CASES  # list of {"id", "input", "must_include", "must_not_include"}

@pytest.mark.parametrize("case", CASES, ids=lambda c: c["id"])
def test_case(case):
    out = run_system(case["input"])
    for phrase in case["must_include"]:
        assert phrase.lower() in out.lower()
    for phrase in case["must_not_include"]:
        assert phrase.lower() not in out.lower()

A simple GitHub Actions job is enough to start:

name: agent-evals
on: [pull_request]
jobs:
  evals:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements.txt
      - run: pytest test_agent_evals.py -q
        env:
          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}

Tips:

  • Include regression cases from every real production failure
  • Test per-agent (unit) and end-to-end (integration)
  • Track pass rate over time, because LLM outputs are non-deterministic and you want trends, not single runs
  • Use LLM-as-judge sparingly and spot-check it against human review

Stage 7: Deploy and operate

Multi-agent systems are distributed systems wearing a trench coat. Apply the usual disciplines:

  • Run agents as queued jobs rather than long synchronous requests
  • Make tool calls idempotent so retries don't duplicate actions
  • Use dead-letter queues for runs that fail repeatedly
  • Version everything: prompts, tool schemas, model IDs, routing logic
  • Roll out gradually: canary or shadow-mode new agent versions against live traffic
  • Set alerts on cost per run, failure rate, and p95 latency
def call_tool_idempotent(tool, args, run_id, step):
    key = f"{run_id}:{step}:{tool.name}"
    if store.exists(key):
        return store.get(key)
    result = tool.run(**args)
    store.set(key, result)
    return result

Common failure modes

Here are the failures that show up most often when teams scale from one agent to many, and what to do about them.

Failure mode Symptom Mitigation
Infinite delegation loops Agents keep handing work back and forth Step caps, loop detection, a terminal "done" state
Context loss at handoff Downstream agent asks for info already collected Structured handoff contracts
Error compounding A small upstream mistake becomes a confident wrong answer Validation between stages, a review agent, or human gates
Cost explosion Token usage multiplies per hop Budgets, smaller models for simple roles, summaries over transcripts
Conflicting actions Two agents write to the same record Single writer per resource, locking, or idempotency keys
Silent degradation Quality drops after a model or prompt change Eval suite in CI, canary rollouts
Unclear ownership Nobody knows which agent caused the bug Trace IDs and per-agent logs

Framework options and trade-offs

There's no single right choice. Pick based on how much control you need.

Option Good for Watch out for
LangGraph Explicit graph/state-machine control, durable workflows More upfront modeling
CrewAI Fast role-based prototypes Less fine-grained control as complexity grows
AG2 (AutoGen) Conversational multi-agent patterns, research-style setups Harder to keep deterministic
OpenAI Agents SDK Simple handoffs and tracing within that ecosystem Ecosystem coupling
Plain Python + your own queue Maximum control, minimal dependencies You build the plumbing yourself

Framework features change quickly, so check each project's current docs before committing. Whatever you pick, keep the handoff contracts, tracing, budgets, and evals framework-independent so you can swap the orchestration layer later.

Capstone roadmap

Use this as a project plan to take one real use case from single agent to production team. A reasonable timeline is six to eight weeks for a small team.

Week 1: Baseline

  • [ ] Define the use case and success criteria
  • [ ] Build a 30 to 50 case test set
  • [ ] Record baseline success, latency, and cost

Week 2: Harden

  • [ ] Tighten tool schemas, timeouts, and retries
  • [ ] Trim tools and version prompts
  • [ ] Add basic tracing with a run_id

Week 3: Split

  • [ ] Identify 2 to 3 responsibility boundaries with evidence from failures
  • [ ] Implement a supervisor plus workers
  • [ ] Add step caps

Week 4: Contracts and state

  • [ ] Define typed handoff payloads
  • [ ] Scope shared state and permissions per agent

Week 5: Safety

  • [ ] Add token/cost budgets
  • [ ] Add human-approval gates for irreversible actions
  • [ ] Add output validation

Week 6: Evals and CI

  • [ ] Automate evals on every pull request
  • [ ] Add regression cases from failures found so far
  • [ ] Compare against the Week 1 baseline

Week 7 to 8: Ship and operate

  • [ ] Run in shadow mode, then a canary rollout
  • [ ] Set alerts and dashboards
  • [ ] Write a runbook for common failures

Definition of done: the multi-agent system beats the single-agent baseline on success rate or cost, stays within your latency budget, and any failed run can be traced end to end.

FAQ

How do I build a multi-agent system for production?

Start with a measured single-agent baseline, harden it, split only where evidence shows a ceiling, define typed handoffs, add tracing, budgets, and approval gates, then automate evals in CI/CD before a gradual rollout.

When should I use multiple agents instead of one?

When you see tool overload, mixed policies in one prompt, different permission needs, parallelizable work, or a need for different models per role, and you can measure the improvement.

Are multi-agent systems more expensive?

Often, yes. Each hop adds tokens and latency. Budgets, smaller models for simple roles, and passing summaries instead of full transcripts keep cost under control.

What is the biggest cause of multi-agent failures?

Poorly defined handoffs. Vague context, missing constraints, and unvalidated payloads cause more production incidents than model quality does.

Conclusion

Moving from one agent to a team isn't a leap, it's a sequence. The teams that succeed measure first, harden before they split, treat handoffs as contracts, and bring ordinary engineering practice (tracing, budgets, CI evals, gradual rollouts) to a non-deterministic system.

Key takeaways:

  • Earn the split. Use a baseline and real failures to justify going multi-agent.
  • Split by responsibility, with the simplest topology that works.
  • Contracts over conversations. Typed handoffs prevent most bugs.
  • Enforce limits in code, not in prompts.
  • Eval like you deploy. Put agent tests in CI/CD.

If you're building multi-channel automation and AI agent workflows, platforms like Botsailor are one example of tooling that can handle the messaging layer while you focus on agent logic. Evaluate any option against your own requirements.

What did your single-to-multi-agent transition look like? What broke first? Share it in the comments, and follow for more on production agentic systems.

來源:DEV Community · dev.to