Building a Multi‑Agent AI Research Assistant with LangGraph and GPT‑Researcher

Overview

This article describes a multi‑agent AI research assistant built with LangGraph (an extension of LangChain) and the open‑source GPT‑Researcher project. The system automates the end‑to‑end research process: planning, data collection, analysis, drafting, review, revision, and publishing.

Architecture

The workflow consists of the following agents:

Researcher : Retrieves web resources and performs autonomous research.

Editor : Generates the research outline and overall structure.

Reviewer : Checks research results against predefined criteria.

Revisor : Revises the output based on reviewer feedback.

Writer : Produces the final report content.

Publisher : Exports the report in various formats.

Human : Provides optional feedback at key stages.

State Design

LangGraph uses a shared State object to pass data between nodes. The ResearchState TypedDict includes fields for the research task, intermediate findings, section list, collected data, optional human feedback, and the final report structure.

class ResearchState(TypedDict):
    # research task
    task: dict
    initial_research: str
    sections: List[str]
    research_data: List[dict]
    human_feedback: str

    # report structure
    title: str
    headers: dict
    date: str
    table_of_contents: str
    introduction: str
    conclusion: str
    sources: List[str]
    report: str

Main Workflow Construction

The main workflow is assembled by creating agent instances, adding them as graph nodes, defining execution edges, and setting entry/exit points. Conditional edges enable looping back for revisions based on human feedback.

def init_research_team(self):
    writer_agent = WriterAgent(self.websocket, self.stream_output, self.headers)
    editor_agent = EditorAgent(self.websocket, self.stream_output, self.headers)
    research_agent = ResearchAgent(self.websocket, self.stream_output, self.tone, self.headers)
    publisher_agent = PublisherAgent(self.output_dir, self.websocket, self.stream_output, self.headers)
    human_agent = HumanAgent(self.websocket, self.stream_output, self.headers)

    workflow = StateGraph(ResearchState)
    workflow.add_node("browser", research_agent.run_initial_research)
    workflow.add_node("planner", editor_agent.plan_research)
    workflow.add_node("researcher", editor_agent.run_parallel_research)
    workflow.add_node("writer", writer_agent.run)
    workflow.add_node("publisher", publisher_agent.run)
    workflow.add_node("human", human_agent.review_plan)
    workflow.add_edge('browser', 'planner')
    workflow.add_edge('planner', 'human')
    workflow.add_edge('researcher', 'writer')
    workflow.add_edge('writer', 'publisher')
    workflow.set_entry_point('browser')
    workflow.add_edge('publisher', END)
    workflow.add_conditional_edges('human',
        lambda review: "accept" if review['human_feedback'] is None else "revise",
        {"accept": "researcher", "revise": "planner"})
    return workflow

Parallel Sub‑Workflow for Sub‑Topic Research

Each sub‑topic is processed in parallel using an independent sub‑workflow with its own DraftState. The sub‑workflow contains a researcher, reviewer, and reviser, with conditional edges that repeat revision until the reviewer signals acceptance.

async def run_parallel_research(self, research_state: dict):
    research_agent = ResearchAgent(self.websocket, self.stream_output, self.headers)
    reviewer_agent = ReviewerAgent(self.websocket, self.stream_output, self.headers)
    reviser_agent = ReviserAgent(self.websocket, self.stream_output, self.headers)
    workflow = StateGraph(DraftState)
    workflow.add_node("researcher", research_agent.run_depth_research)
    workflow.add_node("reviewer", reviewer_agent.run)
    workflow.add_node("reviser", reviser_agent.run)
    workflow.set_entry_point("researcher")
    workflow.add_edge('researcher', 'reviewer')
    workflow.add_edge('reviser', 'reviewer')
    workflow.add_conditional_edges('reviewer',
        lambda draft: "accept" if draft['review'] is None else "revise",
        {"accept": END, "revise": "reviser"})
    return workflow.compile()

Planner Agent Prompt Example

The planner receives a system prompt defining its role and a user prompt that injects dynamic values (current date, initial research summary, optional human feedback). It must return a JSON object with the research title and a list of section headings.

prompt = [{
    "role": "system",
    "content": "You are a research editor. Your goal is to supervise a research project from start to finish. Your main task is to plan the article's chapter layout based on an initial research summary.
"
}, {
    "role": "user",
    "content": f"Today's date is {datetime.now().strftime('%d/%m/%Y')}.
Research summary: '{initial_research}'
{f'Human feedback: {human_feedback}. You must plan chapters based on this feedback.' if include_human_feedback else ''}
Your task is to generate up to {max_sections} chapter titles relevant to the research topic, without including introduction, conclusion, or references. Return a JSON with fields 'title' and 'sections'."
}]

Human Feedback Handling

The human node simply captures optional feedback from the console and stores it in the state.

async def review_plan(self, research_state: dict):
    layout = research_state.get("sections")
    user_feedback = input(f"Any feedback on this plan? {layout}? If not, reply with 'no'.
> ")
    if user_feedback and "no" in user_feedback.lower():
        user_feedback = None
    return {"human_feedback": user_feedback}

Running the Workflow

After initializing the agents and compiling the workflow, invoke it with a task description. The system will iterate through planning, research, review, revision, writing, and publishing, pausing for human feedback when required.

research_team = init_research_team()
chain = research_team.compile()
result = await chain.ainvoke({"task": "Research task description"})

Further Resources

Source code: https://github.com/assafelovic/gpt-researcher/tree/master/multi_agents