Multi-Agent AI Systems Are Eating Single Agents. Here's How to Build One That Works.

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
 
class ResearchState(TypedDict):
    """Shared state passed between all agents in the graph."""
    query: str                                    # Original user query
    plan: list[str]                               # Decomposed sub-tasks
    research_results: Annotated[list, add_messages]  # Accumulated findings
    draft: str                                    # Written output
    review_feedback: str                          # Reviewer's assessment
    review_passed: bool                           # Gate: did the draft pass?
    revision_count: int                           # Safety valve for loops
    status: str                                   # Current pipeline stage

from anthropic import Anthropic
 
client = Anthropic()
 
def planner_node(state: ResearchState) -> dict:
    """Decompose the query into research sub-tasks."""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="""You are a research planner. Given a query, decompose it
        into 2-4 specific, searchable sub-questions. Return each
        sub-question on its own line. Nothing else.""",
        messages=[{"role": "user", "content": state["query"]}],
    )
    plan = [
        line.strip()
        for line in response.content[0].text.strip().split("\n")
        if line.strip()
    ]
    return {"plan": plan, "status": "planned"}
 
 
def researcher_node(state: ResearchState) -> dict:
    """Research each sub-task and accumulate findings."""
    findings = []
    for sub_task in state["plan"]:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system="""You are a research analyst. Given a specific question,
            provide a thorough, factual analysis with concrete data points.
            Cite your reasoning. Be precise, not verbose.""",
            messages=[{"role": "user", "content": sub_task}],
        )
        findings.append(f"## {sub_task}\n\n{response.content[0].text}")
    return {
        "research_results": findings,
        "status": "researched",
    }
 
 
def writer_node(state: ResearchState) -> dict:
    """Synthesize research into a coherent report."""
    research_context = "\n\n---\n\n".join(state["research_results"])
    feedback_note = ""
    if state.get("review_feedback"):
        feedback_note = (
            f"\n\nPrevious review feedback to address:\n"
            f"{state['review_feedback']}"
        )
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=f"""You are a technical writer. Synthesize the research below
        into a clear, well-structured report that directly answers the
        original query. Use specific data points from the research.
        Do not invent facts.{feedback_note}""",
        messages=[
            {"role": "user", "content": (
                f"Original query: {state['query']}\n\n"
                f"Research findings:\n{research_context}"
            )},
        ],
    )
    return {
        "draft": response.content[0].text,
        "status": "drafted",
    }
 
 
def reviewer_node(state: ResearchState) -> dict:
    """Evaluate the draft for accuracy and completeness."""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="""You are a fact-checking reviewer. Evaluate this report
        against the research findings. Check for:
        1. Claims not supported by the research
        2. Missing key findings from the research
        3. Logical inconsistencies
        Respond with APPROVED if the report passes all checks.
        Otherwise, respond with REVISION NEEDED: followed by specific
        feedback.""",
        messages=[
            {"role": "user", "content": (
                f"Report:\n{state['draft']}\n\n"
                f"Research findings:\n"
                + "\n---\n".join(state["research_results"])
            )},
        ],
    )
    review_text = response.content[0].text.strip()
    passed = review_text.upper().startswith("APPROVED")
    return {
        "review_feedback": review_text,
        "review_passed": passed,
        "revision_count": state.get("revision_count", 0) + 1,
        "status": "approved" if passed else "needs_revision",
    }

def should_revise(state: ResearchState) -> Literal["writer", "end"]:
    """Route based on review outcome and revision count."""
    if state["review_passed"]:
        return "end"
    if state["revision_count"] >= 3:
        # Safety valve: accept the draft after 3 attempts
        return "end"
    return "writer"
 
 
# Assemble the graph
workflow = StateGraph(ResearchState)
 
# Add nodes
workflow.add_node("planner", planner_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)
 
# Wire edges
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges("reviewer", should_revise, {
    "writer": "writer",
    "end": END,
})
 
# Compile
app = workflow.compile()

result = app.invoke({
    "query": "What are the cost and latency tradeoffs of running "
             "open-weight LLMs on-device vs. cloud API inference "
             "for mobile applications in 2026?",
    "plan": [],
    "research_results": [],
    "draft": "",
    "review_feedback": "",
    "review_passed": False,
    "revision_count": 0,
    "status": "started",
})
 
print(result["draft"])
print(f"\nRevisions required: {result['revision_count']}")
print(f"Final status: {result['status']}")

from crewai import Agent, Task, Crew, Process
 
planner = Agent(
    role="Research Planner",
    goal="Decompose complex queries into specific research sub-questions",
    backstory="You are a senior research strategist who excels at "
              "breaking down complex topics into focused investigations.",
    verbose=False,
)
 
researcher = Agent(
    role="Research Analyst",
    goal="Gather thorough, factual information with concrete data points",
    backstory="You are a meticulous analyst who prioritizes accuracy "
              "and always supports claims with evidence.",
    verbose=False,
)
 
writer = Agent(
    role="Technical Writer",
    goal="Synthesize research into clear, well-structured reports",
    backstory="You are a technical writer known for turning complex "
              "research into accessible, authoritative content.",
    verbose=False,
)
 
reviewer = Agent(
    role="Fact-Checking Reviewer",
    goal="Validate report accuracy against source research",
    backstory="You are an editor who catches unsupported claims, "
              "missing context, and logical gaps.",
    verbose=False,
)
 
# Define tasks with explicit dependencies
plan_task = Task(
    description="Decompose this query into 2-4 research sub-questions: "
                "{query}",
    expected_output="A numbered list of specific research sub-questions",
    agent=planner,
)
 
research_task = Task(
    description="Research each sub-question from the plan thoroughly. "
                "Provide factual analysis with data points.",
    expected_output="Detailed findings for each sub-question",
    agent=researcher,
)
 
write_task = Task(
    description="Write a comprehensive report synthesizing all research "
                "findings. Address the original query directly.",
    expected_output="A well-structured technical report",
    agent=writer,
)
 
review_task = Task(
    description="Fact-check the report against the research. Verify all "
                "claims are supported. Flag any gaps or errors.",
    expected_output="APPROVED or specific revision feedback",
    agent=reviewer,
)
 
crew = Crew(
    agents=[planner, researcher, writer, reviewer],
    tasks=[plan_task, research_task, write_task, review_task],
    process=Process.sequential,
    verbose=False,
)
 
result = crew.kickoff(inputs={"query": "Cost and latency tradeoffs..."})

# Add token tracking to every node
import tiktoken
 
class CostTracker:
    def __init__(self, max_budget_usd: float = 0.50):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.max_budget = max_budget_usd
 
    def track(self, response) -> None:
        self.total_input_tokens += response.usage.input_tokens
        self.total_output_tokens += response.usage.output_tokens
 
    @property
    def estimated_cost(self) -> float:
        # Claude Sonnet 4.6 pricing
        input_cost = (self.total_input_tokens / 1_000_000) * 3.00
        output_cost = (self.total_output_tokens / 1_000_000) * 15.00
        return input_cost + output_cost
 
    def check_budget(self) -> None:
        if self.estimated_cost > self.max_budget:
            raise BudgetExceededError(
                f"Request cost ${self.estimated_cost:.4f} "
                f"exceeds budget ${self.max_budget:.2f}"
            )

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
)
async def resilient_llm_call(client, **kwargs):
    """LLM call with exponential backoff."""
    return await asyncio.wait_for(
        client.messages.create(**kwargs),
        timeout=60.0,  # Per-call timeout
    )

import structlog
 
logger = structlog.get_logger()
 
def researcher_node(state: ResearchState) -> dict:
    logger.info(
        "researcher.start",
        sub_tasks=len(state["plan"]),
        revision_count=state.get("revision_count", 0),
    )
    # ... LLM call ...
    logger.info(
        "researcher.complete",
        findings_count=len(findings),
        total_chars=sum(len(f) for f in findings),
    )
    return {"research_results": findings, "status": "researched"}