Building a multi-agent research assistant

If you want to build multi-agent research assistant systems that actually work in production, you need to understand how to orchestrate specialized agents, manage shared state, and handle tool failures gracefully. This tutorial walks through a complete implementation using LangGraph, with each agent responsible for a distinct research phase: planning, searching, synthesizing, and critiquing.

Prerequisites

You’ll need Python 3.10+ and the following packages:

pip install langgraph langchain-openai langchain-community tavily-python pydantic python-dotenv

You’ll also need API keys for:

• OpenAI (for GPT-4o-mini)
• Tavily (for web search)

Create a .env file:

OPENAI_API_KEY=sk-...
TAVILY_API_KEY=tvly-...

Architecture overview

The system uses four agents connected in a directed graph:

1. Planner — breaks the user query into a research plan with specific sub-questions
2. Searcher — executes searches for each sub-question using Tavily
3. Synthesizer — combines search results into a coherent answer
4. Critic — evaluates the answer for completeness and accuracy, triggers revision if needed

The graph loops between synthesizer and critic until the critic approves or a max-revision limit is hit.

Step 1: Define the shared state

# state.py
from typing import List, Optional, Literal
from pydantic import BaseModel, Field

class SubQuestion(BaseModel):
    question: str
    rationale: str

class ResearchPlan(BaseModel):
    sub_questions: List[SubQuestion] = Field(default_factory=list)
    max_revisions: int = 2

class SearchResult(BaseModel):
    sub_question: str
    sources: List[dict]
    summary: str

class AgentState(BaseModel):
    user_query: str
    plan: Optional[ResearchPlan] = None
    search_results: List[SearchResult] = Field(default_factory=list)
    draft_answer: Optional[str] = None
    critique: Optional[str] = None
    revision_count: int = 0
    status: Literal["planning", "searching", "synthesizing", "critiquing", "done", "failed"] = "planning"

This state flows through every node. Using Pydantic gives us validation and makes debugging easier — you can print the state at any checkpoint and see exactly what each agent produced.

Step 2: Build the planner agent

# agents/planner.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from state import AgentState, ResearchPlan, SubQuestion

PLANNER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a research planner. Break the user's query into 3-5 specific sub-questions.
Each sub-question should be answerable via web search and together they should comprehensively address the original query.
Return a ResearchPlan with sub_questions (question + rationale) and max_revisions (default 2)."""),
    ("human", "Query: {query}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
planner_chain = PLANNER_PROMPT | llm.with_structured_output(ResearchPlan)

def planner_node(state: AgentState) -> AgentState:
    plan = planner_chain.invoke({"query": state.user_query})
    state.plan = plan
    state.status = "searching"
    return state

Checkpoint — run the planner in isolation:

# test_planner.py
from agents.planner import planner_node
from state import AgentState

state = AgentState(user_query="What are the trade-offs between RAG and fine-tuning for domain adaptation?")
result = planner_node(state)
print(result.plan.model_dump_json(indent=2))

Expected output:

{
  "sub_questions": [
    {
      "question": "What is RAG and how does it work for domain adaptation?",
      "rationale": "Establishes baseline understanding of RAG approach"
    },
    {
      "question": "What is fine-tuning and how does it work for domain adaptation?",
      "rationale": "Establishes baseline understanding of fine-tuning approach"
    },
    {
      "question": "What are the cost and latency trade-offs between RAG and fine-tuning?",
      "rationale": "Directly addresses the trade-off aspect of the query"
    },
    {
      "question": "When should you choose RAG over fine-tuning and vice versa?",
      "rationale": "Provides practical decision guidance"
    }
  ],
  "max_revisions": 2
}

Step 3: Build the searcher agent

# agents/searcher.py
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from state import AgentState, SearchResult

SEARCH_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a research searcher. Given a sub-question and raw search results,
produce a concise summary that directly answers the sub-question. Cite sources by their index."""),
    ("human", "Sub-question: {question}\n\nSearch results:\n{results}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
search_tool = TavilySearchResults(max_results=5)
summarizer_chain = SEARCH_PROMPT | llm

def searcher_node(state: AgentState) -> AgentState:
    if not state.plan:
        state.status = "failed"
        return state

    results = []
    for sq in state.plan.sub_questions:
        raw = search_tool.invoke({"query": sq.question})
        formatted = "\n".join([f"[{i}] {r['content']} (source: {r['url']})" for i, r in enumerate(raw)])
        summary = summarizer_chain.invoke({
            "question": sq.question,
            "results": formatted
        }).content

        results.append(SearchResult(
            sub_question=sq.question,
            sources=raw,
            summary=summary
        ))

    state.search_results = results
    state.status = "synthesizing"
    return state

Checkpoint — test the searcher:

# test_searcher.py
from agents.planner import planner_node
from agents.searcher import searcher_node
from state import AgentState

state = AgentState(user_query="What are the trade-offs between RAG and fine-tuning for domain adaptation?")
state = planner_node(state)
state = searcher_node(state)

for r in state.search_results:
    print(f"Q: {r.sub_question}")
    print(f"Summary: {r.summary[:200]}...")
    print(f"Sources: {len(r.sources)}")
    print("---")

You should see 4 summaries with 5 sources each.

Step 4: Build the synthesizer agent

# agents/synthesizer.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from state import AgentState

SYNTHESIZER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a research synthesizer. Combine the search summaries into a comprehensive,
well-structured answer to the original query. Use clear sections. Cite sources inline like [1], [2]
referring to the source indices from the search results. Be thorough but concise."""),
    ("human", "Original query: {query}\n\nSearch summaries:\n{summaries}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
synthesizer_chain = SYNTHESIZER_PROMPT | llm

def format_summaries(results) -> str:
    out = []
    for i, r in enumerate(results):
        out.append(f"--- Sub-question {i+1}: {r.sub_question} ---")
        out.append(r.summary)
        out.append("Sources:")
        for j, s in enumerate(r.sources):
            out.append(f"  [{i*5 + j}] {s['url']}")
    return "\n".join(out)

def synthesizer_node(state: AgentState) -> AgentState:
    if not state.search_results:
        state.status = "failed"
        return state

    summaries = format_summaries(state.search_results)
    answer = synthesizer_chain.invoke({
        "query": state.user_query,
        "summaries": summaries
    }).content

    state.draft_answer = answer
    state.status = "critiquing"
    return state

Step 5: Build the critic agent

# agents/critic.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from state import AgentState

class Critique(BaseModel):
    approved: bool
    feedback: str
    missing_aspects: list[str] = Field(default_factory=list)

CRITIC_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a research critic. Evaluate the draft answer for:
1. Completeness — does it address all aspects of the original query?
2. Accuracy — are claims well-supported by the cited sources?
3. Clarity — is the structure logical and easy to follow?

If approved=false, provide specific feedback on what's missing or wrong.
Return a Critique object."""),
    ("human", "Original query: {query}\n\nDraft answer:\n{answer}\n\nSearch summaries:\n{summaries}")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
critic_chain = CRITIC_PROMPT | llm.with_structured_output(Critique)

def critic_node(state: AgentState) -> AgentState:
    if not state.draft_answer or not state.search_results:
        state.status = "failed"
        return state

    summaries = format_summaries(state.search_results)
    critique = critic_chain.invoke({
        "query": state.user_query,
        "answer": state.draft_answer,
        "summaries": summaries
    })

    state.critique = critique.feedback

    if critique.approved or state.revision_count >= (state.plan.max_revisions if state.plan else 2):
        state.status = "done"
    else:
        state.revision_count += 1
        state.status = "synthesizing"  # loop back

    return state

Step 6: Wire the graph

# graph.py
from langgraph.graph import StateGraph, END
from state import AgentState
from agents.planner import planner_node
from agents.searcher import searcher_node
from agents.synthesizer import synthesizer_node
from agents.critic import critic_node

def route_after_critic(state: AgentState) -> str:
    return "synthesizer" if state.status == "synthesizing" else "done"

workflow = StateGraph(AgentState)

workflow.add_node("planner", planner_node)
workflow.add_node("searcher", searcher_node)
workflow.add_node("synthesizer", synthesizer_node)
workflow.add_node("critic", critic_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "searcher")
workflow.add_edge("searcher", "synthesizer")
workflow.add_edge("synthesizer", "critic")
workflow.add_conditional_edges("critic", route_after_critic, {
    "synthesizer": "synthesizer",
    "done": END
})

app = workflow.compile()

Step 7: Run the full pipeline

# main.py
from graph import app
from state import AgentState

def run_research(query: str) -> AgentState:
    initial = AgentState(user_query=query)
    final = app.invoke(initial)
    return AgentState(**final)

if __name__ == "__main__":
    query = "What are the trade-offs between RAG and fine-tuning for domain adaptation?"
    result = run_research(query)

    print(f"Status: {result.status}")
    print(f"Revisions: {result.revision_count}")
    print(f"\n=== FINAL ANSWER ===\n")
    print(result.draft_answer)
    print(f"\n=== CRITIQUE ===\n")
    print(result.critique)

Expected output (truncated):

Status: done
Revisions: 1

=== FINAL ANSWER ===

# Trade-offs Between RAG and Fine-tuning for Domain Adaptation

## Overview
Both Retrieval-Augmented Generation (RAG) and fine-tuning adapt LLMs to domain-specific tasks, but they operate on fundamentally different principles...

## When to Choose RAG
- **Rapid iteration** — Update knowledge base without retraining [1, 2]
- **Auditability** — Trace answers to specific source documents [3]
- **Lower upfront cost** — No GPU training infrastructure needed [4]

## When to Choose Fine-tuning
- **Style and format adherence** — Learns domain-specific phrasing, templates [5]
- **Latency-critical paths** — No retrieval step at inference time [6]
- **Implicit knowledge** — Captures patterns not easily documented [7]

## Hybrid Approaches
Many production systems combine both: fine-tune for style/format, use RAG for factual grounding [8, 9].

=== CRITIQUE ===
Approved. The answer covers cost, latency, auditability, and decision criteria with specific citations. Revision 1 added the hybrid section which was missing initially.

Handling failures and edge cases

Real systems need guardrails. Here are three patterns worth adding:

1. Search failure retry

# agents/searcher.py (excerpt)
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3))
def safe_search(query: str):
    return search_tool.invoke({"query": query})

2. Token budget enforcement

# agents/synthesizer.py (add to imports)
import tiktoken

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

# In synthesizer_node, before invoking chain:
MAX_CONTEXT = 12000
summaries = format_summaries(state.search_results)
if count_tokens(summaries) > MAX_CONTEXT:
    # Truncate oldest summaries or summarize them further
    summaries = truncate_to_budget(summaries, MAX_CONTEXT)

3. Structured logging for observability

# graph.py (add to each node)
import structlog

logger = structlog.get_logger()

def planner_node(state: AgentState) -> AgentState:
    logger.info("planner_start", query=state.user_query[:100])
    # ... existing logic ...
    logger.info("planner_complete", sub_question_count=len(state.plan.sub_questions))
    return state

Extending the system

Once the core loop works, common extensions include:

For production workloads where you’re routing across multiple model providers, an inference gateway like n4n.ai can simplify failover and usage metering without changing your agent code — just point the OpenAI-compatible client at the gateway endpoint.

Full file structure

research_assistant/
├── .env
├── main.py
├── graph.py
├── state.py
└── agents/
    ├── __init__.py
    ├── planner.py
    ├── searcher.py
    ├── synthesizer.py
    └── critic.py

Run it with python main.py. The graph executes deterministically — same query produces same plan, same search results (modulo search index updates), same revision loop behavior. That reproducibility matters when you’re debugging why the critic rejected a draft on revision 2 but approved it on revision 1.

Next steps

• Add a fact-checker agent that verifies specific claims against primary sources
• Implement incremental research — cache search results, only re-search stale sub-questions
• Build a web UI with Server-Sent Events to stream the agent’s reasoning in real time
• Add evaluation harness — run a benchmark suite of queries, score answers with LLM-as-judge

The pattern here — plan, execute, synthesize, critique, revise — generalizes beyond research. Any task that benefits from decomposition, external tool use, and iterative refinement fits this architecture.