Breaking a Monolithic AI Research Agent into a Multi-Agent System with LangGraph

Review

DeepRearchSystem already has the HITL mechanism. Now, the basic pipeline is essentially running through. What other optimizations can we make?

Previously, I was a mobile developer. Before writing code, the design always considered some programming design patterns and principles in advance, like the six major design principles:

Single Responsibility Principle
Open-Closed Principle
Liskov Substitution Principle
Interface Segregation
Dependency Inversion
Law of Demeter

Here, the Single Responsibility Principle and the Law of Demeter seem reusable. Let's review the overall process of DeepRearchSystem:

User Input → Research Plan → Query Planning → Parallel Search → Reflection & Evaluation → Iterative Deepening → Report Generation

But all our functionality is concentrated in a single Graph, which clearly violates the Single Responsibility Principle and the Law of Demeter. Can we, like traditional class objects, distribute different functionalities into different Graphs?

Solution

MAS

In fact, MAS (Multi-Agent System) is exactly what solves the above problem. It is a team composed of multiple autonomous AI agents with specialized capabilities, working collaboratively to solve complex problems. Each agent plays a specific role and works together towards a common goal.

Besides the separation of duties from design principles, what other reasons does a complete Agent orchestration system need MAS for?

1. Context Window and Attention Bottleneck

When a task requires continuous coordination across roles and tools, a single Agent easily loses coherence during long interactions, making it difficult to stably coordinate multi-step actions. Splitting different responsibilities into different Agents, each maintaining its own local context, and then aggregating them at the orchestration layer can significantly alleviate this problem.

2. Security and Compliance Boundary Isolation

Industries like finance often require "one Agent to prepare a transaction, another Agent to verify the transaction," enforcing separation of duties through the architecture itself. This "principle of least privilege" design can limit the blast radius of a security incident to the boundary of a single Agent—something a single Agent cannot do.

3. Knowledge Boundaries Across Multiple Teams and Domains

When different teams manage their own knowledge domains and require independent development cycles and data sources, a decoupled multi-Agent architecture allows teams to develop in parallel and deploy updates independently, with explicit interfaces reducing integration risks.

DeepResearch-MAS Implementation

Now, let's implement MAS for DeepResearch.

Role Splitting: Three Core Agents

Main Agent (graph): Responsible for planning
Research Agent (reasearch_graph): Responsible for the Query-Search-Evaluate Loop
Report Writer Agent (writer_graph): Responsible for Outline → Draft → Final Draft

Orchestration Pattern: Supervisor + SubGraph Supervisor Pattern

Supervisor: The main graph (graph.py), does not directly handle complex business logic:
- State Routing: Decides whether to flow to Research or Replan based on plan_status.
- Subgraph Scheduling: Calls research_agent_graph and writer_agent_graph.
- Lifecycle Management: Manages HITL interruptions and recovery.

State Management: Centralized Sharing + Local Isolation
- Centralized: OverallState runs throughout, storing global data like plan, web_search_result, sources_gathered.
- Local Isolation: Subgraphs have their own refined internal states (e.g., QueryGenerationState, DraftModel). The main graph does not care about the internal implementation of the subgraph, only consuming its results.

Comparison with Single Agent

Dimension	MAS Architecture (Current Design)	Original `graph` (Single Agent Architecture)
Code Organization	High cohesion, low coupling. Split into subgraphs by business domain, physically isolated files.	Strong coupling. All node logic piled into one file, code expands quickly. The single `graph` file reached nearly 500 lines.
Testing Strategy	Unit-test friendly. `research_graph.py` can be independently Mock tested without starting the main graph.	Primarily integration testing. Deep dependencies between nodes, high Mock cost, often requires E2E testing.
Reusability	Extremely high. ResearchAgent can be reused for customer service Q&A, competitive analysis, etc.; WriterAgent can be reused for email writing.	Extremely low. Logic is bound within the DeepResearch process, difficult to extract.
Debugging Difficulty	Visualization-friendly. Clear Subgraph hierarchy visible in LangSmith, Trace structure is clear.	Flat Trace. All nodes unfold on one plane, difficult to locate problems in long chains.

Coding

Talk is cheap, show you the code...

graph

The main graph here mainly involves process changes, using two subgraphs to replace the original specific subgraph functions.

# Subgraph nodes
builder.add_node(RESEARCH_AGENT_NODE, research_agent_graph)
builder.add_node(WRITER_AGENT_NODE, writer_agent_graph)


# Edge definitions
builder.add_edge(START, GENERATE_PLAN_NODE)
# Conditional edge: Does human intervention approve the research plan?
builder.add_conditional_edges(GENERATE_PLAN_NODE, evaluate_plan, [RESEARCH_AGENT_NODE, SEARCH_REPLAN, AWAITING_PLAN_CONFIRMATION])
# Regenerate plan
builder.add_edge(SEARCH_REPLAN, GENERATE_PLAN_NODE)
# Parallel search for sub-topics
builder.add_edge(RESEARCH_AGENT_NODE, WRITER_AGENT_NODE)
builder.add_edge(WRITER_AGENT_NODE, END)

reasearch_graph

For the research subgraph, the research-related code from the original graph is similarly extracted here. In fact, the code was already written before; it's just reorganized into a new Graph.

_builder = StateGraph(OverallState, context_schema=Configuration)

_builder.add_node(GENERATE_SEARCH_NODE, _generate_search)
_builder.add_node(WEB_SEARCH_NODE, _web_search)
_builder.add_node(CRITIQUE_NODE, _critique)

_builder.add_edge(START, GENERATE_SEARCH_NODE)
_builder.add_conditional_edges(GENERATE_SEARCH_NODE, _fan_out_to_web_search, [WEB_SEARCH_NODE])
_builder.add_edge(WEB_SEARCH_NODE, CRITIQUE_NODE)
_builder.add_conditional_edges(CRITIQUE_NODE, _route_after_critique, [WEB_SEARCH_NODE, END])

research_agent_graph = _builder.compile(name=SUB_RESEARCH_AGENT)

writer_graph

For report generation, we've further refined it based on the original:

Original process: Search content → Report generation
Optimized process: Search content → Outline writing → Draft writing → Draft review → Report generation This makes our report writing phase more professional.

outline

def _outline(state: OverallState, config: RunnableConfig) -> dict:
    """
    Generate an outline based on the research topic and plan.
    :param state:
    :param config:
    :return:
    """

    configuration = Configuration.runnable_config(config)
    reasoning_model = state.get("reasoning_model") or configuration.answer_model
    logger.info(f"[WriterAgent] outline using model={reasoning_model}")

    agent = Agent(model_id=reasoning_model)
    agent.step_prompt(outline_instructions)
    raw = agent.step(
        research_topic=get_research_topic(state["messages"]),
        research_proposal=state.get("plan", ""),
        summaries="\n---\n\n".join(state["web_search_result"])
    )
    outline =- JsonUtils.extract_pattern(raw, pattern="markdown")
    logger.info(f"[WriterAgent] outline generated ({len(outline)} chars)")

    return {
        "report_outline": outline,
        "revision_count": 0,
        "max_revisions": DEFAULT_MAX_REVISIONS,
    }

draft

def _draft(state: OverallState, config: RunnableConfig) -> DraftModel:
    """
    Write the main body draft based on the outline.
    """

    configurable = Configuration.from_runnable_config(config)
    reasoning_model = state.get("reasoning_model") or configurable.answer_model
    logger.info(f"[WriterAgent] drafting using model={reasoning_model}")

    feedback = state.get("critic_feedback", "")
    outline = state.get("report_outline", "")
    is_revision = bool(feedback)
    revision_count = state.get("revision_count", 0) + (1 if is_revision else 0)

    if is_revision:
        logger.info(f"[WriterAgent] revision draft (revision {revision_count})")
        revision_context = (
            f"\n# Revision Notes (Revision {revision_count})\n"
            f"Please modify the draft according to the following review suggestions:\n\n"
            f"{feedback}\n\n"
            f"Please address the above issues point by point, prioritizing critical and major level issues."
            f"Please retain content from the previous draft that the reviewer did not object to.\n"
        )
        return_update = {"revision_count": revision_count, "critic_feedback": ""}
    else:
        logger.info(f"[WriterAgent] writing draft from scratch")
        revision_context = ""
        return_update = {}

    agent = Agent(model_id=reasoning_model)
    agent.set_step_prompt(draft_instructions)
    raw = agent.step(
        current_date=get_current_date(),
        research_topic=get_research_topic(state["messages"]),
        research_proposal=state.get("plan", ""),
        outline=outline,
        summaries="\n---\n\n".join(state["web_search_result"]),
        revision_context=revision_context,
    )
    draft = JsonUtils.extract_pattern(raw, pattern="markdown")
    logger.info(f"[WriterAgent] draft generated ({len(draft)} chars)")

    return {**return_update, "report_draft": draft}

review

def _review(state: OverallState, config: RunnableConfig) -> ReviewModel:
    """
    Critic reviews the draft and provides structured feedback.
    Uses JsonAgent and CritiqueResult schema to generate structured output.
    """

    configurable = Configuration.from_runnable_config(config)
    reasoning_model = state.get("reasoning_model") or configurable.answer_model
    logger.info(f"[WriterAgent] critic reviewing draft using model={reasoning_model}")

    draft = state.get("report_draft", "")
    agent = JsonAgent(model_id=reasoning_model, keys=ReviewModel)
    agent.set_step_prompt(review_instructions)
    result: ReviewModel = agent.step(
        research_topic=get_research_topic(state["messages"]),
        research_proposal=state.get("plan", ""),
        summaries="\n---\n\n".join(state["web_search_result"]),
        draft=draft,
    )

    # Feedback for revised drafts
    if result.issues:
        issues_text = "\n".join(
            f"- [{iss.severity.upper()}] {iss.location}: {iss.problem}\n"
            f"  Suggestion: {iss.suggestion}"
            for iss in result.issues
        )
    else:
        issues_text = "No significant issues."

    feedback = (
        f"## Review Score: {result.overall_rating}/10\n"
        f"## Overall Assessment: {result.summary}\n\n"
        f"## Specific Issues:\n{issues_text}"
    )

    logger.info(
        f"[WriterAgent] review score={result.overall_rating}/10, "
        f"issues={len(result.issues)} "
        f"(critical={sum(1 for i in result.issues if i.severity == 'critical')}, "
        f"major={sum(1 for i in result.issues if i.severity == 'major')}, "
        f"minor={sum(1 for i in result.issues if i.severity == 'minor')}), "
        f"ready_for_polish={result.ready_for_polish}"
    )

    return {
        "critic_feedback": feedback,
        "critic_score": result.overall_rating,
        "ready_for_polish": result.ready_for_polish,
    }

after_review

def _route_after_review(state: OverallState, config: RunnableConfig) -> str:
    """
    Decide: continue revising or proceed to final polish.
    Conditions for entering polish:
      - Critic explicitly marks ready_for_polish, or
      - revision_count >= max_revisions (safety fallback)
    Otherwise, return to draft for further revision.
    """

    # Get current state
    revision = state.get("revision_count", 0)
    max_rev = state.get("max_revisions", DEFAULT_MAX_REVISIONS)
    ready = state.get("ready_for_polish", False)

    if ready:
        logger.info(f"[WriterAgent] Critic ready_for_polish → polish")
        return _POLISHNODE

    if revision >= max_rev:
        logger.info(f"[WriterAgent] Max revisions reached ({revision}/{max_rev}) → polish")
        return _POLISHNODE

    logger.info(f"[WriterAgent] Need to rewrite draft (rev={revision}/{max_rev}) → draft")
    return _DRAFTNODE

polish

def _polish(state: OverallState, config: RunnableConfig) -> PolishModel:
    """
    Replace short links with real links, remove duplicate sources, polish language.
    """
    configurable = Configuration.from_runnable_config(config)
    reasoning_model = state.get("reasoning_model") or configurable.answer_model
    logger.info(f"[WriterAgent] polishing using model={reasoning_model}")

    draft = state.get("report_draft", "")

    # Step A — LLM polish pass
    agent = Agent(model_id=reasoning_model)
    agent.set_step_prompt(polish_instructions)
    raw = agent.step(
        research_topic=get_research_topic(state["messages"]),
        draft=draft,
        summaries="\n---\n\n".join(state["web_search_result"]),
    )
    polished = JsonUtils.extract_pattern(raw, pattern="markdown")

    unique_sources = []
    for source in state.get("sources_gathered", []):
        if source["short_url"] in polished:
            polished = polished.replace(source["short_url"], source["value"])
            unique_sources.append(source)

    logger.info(f"[WriterAgent] polished ({len(polished)} chars), {len(unique_sources)} cited sources")
    return {
        "messages": [AIMessage(content=polished)],
        "sources_gathered": unique_sources,
    }

compile

# Write outline
_builder.add_node(_OUTLINENODE, _outline)
# Write draft
_builder.add_node(_DRAFTNODE, _draft)
# Draft review
_builder.add_node(_REVIEWNODE, _review)
# Polish to final draft
_builder.add_node(_POLISHNODE, _polish)

_builder.add_edge(START, _OUTLINENODE)
_builder.add_edge(_OUTLINENODE, _DRAFTNODE)
_builder.add_edge(_DRAFTNODE, _REVIEWNODE)
_builder.add_conditional_edges(_REVIEWNODE, _route_after_review, [_DRAFTNODE, _POLISHNODE])
_builder.add_edge(_POLISHNODE, END)

writer_agent_graph = _builder.compile(name=SUB_WRITER_AGENT)

Prompts

Several new nodes have been added, and corresponding prompts need to be written. You can follow the previous method for writing Prompts to complete each prompt yourself. Here is an example using the outline writing (outline) prompt.

outline_instructions = """# Role Definition
You are a research report architect.
# Task Description
Based on the research topic and collected materials, you need to design a clear chapter structure for the final report.

# Instruction
- Analyze the research topic and existing materials to determine which core chapters the report should include
- Each chapter is a second-level heading (##). If the content is complex, plan down to third-level headings (###)
- The chapter order should have logical progression: Overview → Dimensional Analysis → Comprehensive Discussion → Conclusion/Outlook
- Control within 3-6 main chapters, list 2-3 sub-chapters under each main chapter
- Think about the key information points each chapter should cover

# Output Format
Please output a pure markdown format outline between ```markdown and ```. For example:

\```markdown
## 1. Market Overview
### 1.1 Industry Background and Current Status
### 1.2 Market Size and Growth Trends

## 2. Key Player Analysis
### 2.1 Leading Enterprise Comparison
### 2.2 Emerging Competitors
...
\```

# Research Topic
{research_topic}

# Research Plan
{research_proposal}

# Collected Material Summaries
{summaries}

# Output"""

Expansion🤔🤔🤔

MAS History

Actually, MAS is not a new concept. Why hasn't it been widely adopted before? Anyone working with LLMs, especially on Agents, is certainly familiar with the paper: 《Why Do Multi-Agent LLM Systems Fail?》. The paper identifies 14 failure modes and 3 fatal pitfalls of MAS.

Category	Typical Failure Modes
System Design Issues	Unclear agent role definitions, orchestration topology mismatch with tasks, lack of exit conditions
Inter-Agent Misalignment	Duplicate work, conflicting changes, information transmission distortion
Task Verification Issues	Lack of termination judgment, error propagation in collaboration, no effective rollback

MAS is beautiful, but the paper's conclusion is heartbreaking😭😭😭: The efficiency gains brought by multiple Agents compared to a Single Agent are often offset by the coordination overhead between multiple Agents. Moreover, simply piling up the number of Agents leads to performance improvement quickly saturating. See more MAS defect analysis here

Can MAS Still Be Used?

The answer is definitely yes, otherwise today's work would have been in vain. My personal understanding is that it can certainly be used, but some means are needed to overcome the problems discussed in the paper above. And a large part is not from a technical perspective, but from engineering practices.

Coordination and Conflict → A unified orchestration layer (central "commander" agent or structured hierarchy) is responsible for routing, enforcing shared context, and resolving conflicts. Adopt patterns like Supervisor + SubGraph, Orchestrator + Worker.
Observability Black Hole → Implement monitoring, tracing, and evaluation for every operational step of the Agent, introduce guardrail agents to monitor intermediate states, and log end-to-end decisions.
Security Governance → Dedicated security and policy agents + sandboxing + a unified governance framework.

How to Choose

Actually, my understanding boils down to one sentence: If you can use a Single Agent, use a Single Agent; if you want to use MAS, you must clearly justify why. Below is a selection consideration summary based on Microsoft Learn - Single agent or multiple agents.

Single Agent

Low Complexity Tasks: Low task complexity, limited decomposability, unified domain knowledge.
Coordination Overhead Sensitive: Low coordination overhead, resource-constrained.
MVP Validation Phase: First use a single Agent to establish a performance baseline, measure exactly where it fails, then decide whether to split.

MAS

Task Itself Requires It: Decomposable into independent sub-tasks, requires diverse domain expertise, benefits from parallel processing.
Future Growth Planning: The solution diversifies into multiple functions, data sources, or business units, expected to exceed 3-5.
Involves Multiple Modules/Teams: Different modules/teams manage independent knowledge domains, requiring decoupled, independent development cycles.
Crosses Security and Compliance Boundaries: Regulations or policies require strict data isolation, which a single Agent cannot provide an independent processing environment for.

SOP

In actual engineering practice, it's often a step of "Single first, then split; from one to many," just like our DeepRearchSystem, which started as a single Agent and was only split at a certain stage of development.

Establish Baseline: Implement based on requirements, deploy a single Agent to handle core use cases, establish baselines for accuracy, latency, and cost.
Identify Shortcomings: Is the single Agent bottleneck due to context limitations? Insufficient specialized knowledge? Or the need for parallelism?
Multi-Agent Splitting: Targeted expansion, add one Agent for one bottleneck, incrementally build the orchestration.
MAS Observability Optimization: Monitor performance across Agents, merge what should be merged, expand what should be expanded.

For more AI technical insights, please subscribe to the AI Tech Notes Column.