Research Assistant: Multi-Agent Automation Research
Parallel · Subgraph · The ultimate synthesis of Map-Reduce

1Overall architecture: What is this system doing?

Research is a time-consuming and labor-intensive task - collecting information, analyzing different perspectives, synthesizing insights, and writing reports often requires the collaboration of an entire team. The goal of this lesson is to build aLightweight multi-agent research system, automatically completing the entire process above.

Five major capabilities design of the system:

• 1

  Source Selection

  Users can customize which information sources are used - web search (Tavily), Wikipedia, specific websites, locally indexed documents, etc.

• 2

  Planning

  The user enters a research topic, and the system automatically generates a team of AI analysts, each focusing on a sub-topic.Human-in-the-loopAllows users to review and adjust analyst lineups before research begins.

• 3

  Multiple rounds of interviews (LLM Utilization)

  Each analyst engages in in-depth, multiple rounds of conversations with an "expert AI." The conversation process is encapsulated inSub-graph, has independent internal state.

• 4

  Parallel Research (Parallelization + Map-Reduce)

  Experts call on both web searches and Wikipedia when answering questions (Parallel), all analyst interviews were also conducted simultaneously (Map-Reduce）。

• 5

  Report output (Output Format)

  Summarize all interview insights and write an introduction, main body, and conclusion to form a complete final report.

Agent role in the system

2Phase 1: Generating an Analyst Team (Human-in-the-loop)

Before the study begins, the system first generates a group of AI analysts and allows users to review and adjust the lineup through Human-in-the-loop until they are satisfied.

Data Model: Analyst and Perspectives

Every analyst uses PydanticBaseModelDescription, including four core attributes and a prompt word used to generatepersonaProperties:

from pydantic import BaseModel, Field

class Analyst(BaseModel):
    affiliation: str = Field(description="分析师所属机构")
    name: str = Field(description="分析师姓名")
    role: str = Field(description="分析师在本课题中的角色")
    description: str = Field(description="分析师的关注重点、担忧和动机")

    @property
    def persona(self) -> str:
        # 合并成提示词片段，传给 LLM 作为角色扮演指令
        return f"Name: {self.name}\nRole: {self.role}\nAffiliation: {self.affiliation}\nDescription: {self.description}\n"

class Perspectives(BaseModel):
    # LLM 的结构化输出包装器：一次性返回多个分析师
    analysts: List[Analyst] = Field(description="分析师列表")

class GenerateAnalystsState(TypedDict):
    topic: str                    # 研究主题
    max_analysts: int              # 分析师数量上限
    human_analyst_feedback: str   # 用户反馈（用于迭代修改）
    analysts: List[Analyst]       # 当前分析师列表

create_analysts node: LLM Build Analysts

# 提示词模板：要求 LLM 根据主题和用户反馈生成分析师
analyst_instructions = """You are tasked with creating a set of AI analyst personas...
1. First, review the research topic: {topic}
2. Examine any editorial feedback: {human_analyst_feedback}
3. Determine the most interesting themes.
4. Pick the top {max_analysts} themes.
5. Assign one analyst to each theme."""

def create_analysts(state: GenerateAnalystsState):
    topic = state['topic']
    max_analysts = state['max_analysts']
    human_analyst_feedback = state.get('human_analyst_feedback', '')

    # 强制结构化输出，返回 Perspectives 对象
    structured_llm = llm.with_structured_output(Perspectives)

    system_message = analyst_instructions.format(
        topic=topic,
        human_analyst_feedback=human_analyst_feedback,
        max_analysts=max_analysts
    )
    analysts = structured_llm.invoke(
        [SystemMessage(content=system_message)]
        + [HumanMessage(content="Generate the set of analysts.")]
    )
    return {"analysts": analysts.analysts}

human_feedback node and loop control

human_feedbackIs a no-op node (no-op), which exists only to pause the graph here and wait for the user to update the status.

def human_feedback(state: GenerateAnalystsState):
    """ No-op node that should be interrupted on """
    pass  # 什么都不做，但图会在这里暂停

def should_continue(state: GenerateAnalystsState):
    # 有用户反馈 → 重新生成分析师（循环）
    human_analyst_feedback = state.get('human_analyst_feedback', None)
    if human_analyst_feedback:
        return "create_analysts"
    # 无反馈（用户满意）→ 结束
    return END

# 编译时声明在 human_feedback 节点前中断
graph = builder.compile(interrupt_before=['human_feedback'], checkpointer=memory)

Human-in-the-loop operation process

3Phase 2: Interview subgraph—parallel information collection

Each analyst's interview process is encapsulated in a separateinterview sub-graphin. Subgraphs have their own stateInterviewState, operates independently without interfering with each other.

InterviewState: the internal state of the subgraph

import operator
from langgraph.graph import MessagesState

class InterviewState(MessagesState):  # 继承 MessagesState 获得消息历史管理
    max_num_turns: int                        # 最大对话轮次
    context: Annotated[list, operator.add]     # 搜索结果累积（追加式 Reducer）
    analyst: Analyst                           # 当前分析师的 persona
    interview: str                             # 完整访谈记录（字符串）
    sections: list                              # 写好的备忘录（传递给外层主图）

Nodes within the interview subgraph

The subgraph contains six nodes, forming a cycle of "Ask → Parallel Search → Answer → Ask Again":

How experts search in parallel while answering

This is a direct application of Module 4 Lesson 1 parallelization techniques. After the analyst asked questions,search_webandsearch_wikipediaStart simultaneously:

# 两个搜索节点都连接到 ask_question → 并行执行
interview_builder.add_edge("ask_question", "search_web")
interview_builder.add_edge("ask_question", "search_wikipedia")

# 两者都完成后，才进入 answer_question
interview_builder.add_edge("search_web", "answer_question")
interview_builder.add_edge("search_wikipedia", "answer_question")

Interview termination logic: route_messages

def route_messages(state: InterviewState, name: str = "expert"):
    messages = state["messages"]
    max_num_turns = state.get('max_num_turns', 2)

    # 统计专家已回答的次数
    num_responses = len(
        [m for m in messages if isinstance(m, AIMessage) and m.name == name]
    )

    # 条件1：专家已回答次数超过上限 → 结束访谈
    if num_responses >= max_num_turns:
        return 'save_interview'

    # 条件2：分析师说出"Thank you so much for your help" → 结束访谈
    last_question = messages[-2]  # 最后一次分析师提问（倒数第二条消息）
    if "Thank you so much for your help" in last_question.content:
        return 'save_interview'

    # 否则 → 分析师继续提问
    return "ask_question"

4Tavily Search Tool: How to get information online

Tavily is a search API designed specifically for AI agents. Compared to Google search, it returns clean, directly usable text summaries instead of a bunch of HTML.

Complete implementation of search nodes

from langchain_tavily import TavilySearch

# 初始化：每次最多返回 3 条结果
tavily_search = TavilySearch(max_results=3)

# 生成搜索查询的提示词
search_instructions = SystemMessage(content="""You will be given a conversation between an analyst and an expert.
Your goal is to generate a well-structured query for use in retrieval and/or web-search
related to the conversation. Pay particular attention to the final question posed by the analyst.
Convert this final question into a well-structured web search query.""")

def search_web(state: InterviewState):
    # Step 1: 让 LLM 把对话内容转化成搜索关键词
    structured_llm = llm.with_structured_output(SearchQuery)
    search_query = structured_llm.invoke([search_instructions] + state['messages'])

    # Step 2: 调用 Tavily 执行搜索
    data = tavily_search.invoke({"query": search_query.search_query})
    search_docs = data.get("results", data)

    # Step 3: 格式化为带 URL 标签的文档字符串
    formatted_search_docs = "\n\n---\n\n".join([
        f'<Document href="../../../langGraph/module-4/{doc["url"]}"/>\n{doc["content"]}\n</Document>'
        for doc in search_docs
    ])

    # 追加到 context（operator.add Reducer）
    return {"context": [formatted_search_docs]}

Wikipedia search node

from langchain_community.document_loaders import WikipediaLoader

def search_wikipedia(state: InterviewState):
    # 同样先把对话转化为搜索查询
    structured_llm = llm.with_structured_output(SearchQuery)
    search_query = structured_llm.invoke([search_instructions] + state['messages'])

    # 加载最多 2 篇维基百科文章
    search_docs = WikipediaLoader(
        query=search_query.search_query,
        load_max_docs=2
    ).load()

    formatted_search_docs = "\n\n---\n\n".join([
        f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
        for doc in search_docs
    ])
    return {"context": [formatted_search_docs]}

Expert Answer Node: Reply Generation with Quotes

answer_instructions = """You are an expert being interviewed by an analyst.
Here is analyst area of focus: {goals}.

To answer question, use this context: {context}

Guidelines:
1. Use only the information provided in the context.
2. Include sources next to relevant statements. For example: [1].
5. List your sources at the bottom: [1] Source 1, [2] Source 2, etc."""

def generate_answer(state: InterviewState):
    analyst = state["analyst"]
    messages = state["messages"]
    context = state["context"]  # 网络搜索 + 维基百科的合并结果

    system_message = answer_instructions.format(
        goals=analyst.persona, context=context
    )
    answer = llm.invoke([SystemMessage(content=system_message)] + messages)

    # 给回答打上 "expert" 标签，供 route_messages 统计轮次
    answer.name = "expert"
    return {"messages": [answer]}

5Phase 3: Map-Reduce starts all interviews

After confirmation by the analyst team, the main image is usedSend()API puts all interviewsAt the same timeDistributed (Map), each interview subgraph runs in parallel, and after completion, the results are automatically aggregated to the main graph state (Reduce).

ResearchGraphState: main graph state

class ResearchGraphState(TypedDict):
    topic: str                                      # 研究主题
    max_analysts: int                                # 分析师数量
    human_analyst_feedback: str                     # 人工反馈
    analysts: List[Analyst]                         # 已确认的分析师列表
    sections: Annotated[list, operator.add]         # 所有备忘录（Send() 的汇聚点）
    introduction: str                               # 报告引言
    content: str                                    # 报告正文
    conclusion: str                                 # 报告结论
    final_report: str                               # 最终完整报告

initiate_all_interviews: Map function

from langgraph.types import Send

def initiate_all_interviews(state: ResearchGraphState):
    """ 这是 'map' 步骤：用 Send API 并行启动所有访谈子图 """

    # 检查是否还有人工反馈未处理
    human_analyst_feedback = state.get('human_analyst_feedback')
    if human_analyst_feedback:
        return "create_analysts"  # 回到分析师生成节点

    # 为每个分析师创建一个 Send 对象 → 并行启动所有子图
    topic = state["topic"]
    return [
        Send(
            "conduct_interview",  # 目标节点（子图）
            {
                "analyst": analyst,
                "messages": [
                    HumanMessage(content=f"So you said you were writing an article on {topic}?")
                ]
            }
        )
        for analyst in state["analysts"]
    ]

Execution output example

When actually running, you can see threeconduct_interviewCompleted in parallel, then the writing node is triggered synchronously:

# 执行日志
--Node-- conduct_interview  # 分析师1的访谈（并行）
--Node-- conduct_interview  # 分析师2的访谈（并行）
--Node-- conduct_interview  # 分析师3的访谈（并行）
--Node-- write_conclusion   # 三个写作任务同时触发
--Node-- write_introduction
--Node-- write_report
--Node-- finalize_report    # 最终汇总

6Phase Four: Writing and Summary of Final Report

After all interview subgraphs are completed, the main graph triggers three parallel writing nodes:write_report、write_introduction、write_conclusion, finally byfinalize_reportCombine them into a complete report.

write_report: Comprehensive all memos

report_writer_instructions = """You are a technical writer creating a report on: {topic}

You have a team of analysts. Each analyst has:
1. Conducted an interview with an expert on a specific sub-topic.
2. Written up their findings into a memo.

Your task:
1. You will be given a collection of memos from your analysts.
2. Consolidate these into a crisp overall summary.
3. Start with: ## Insights
4. Create a final consolidated list of sources under: ## Sources"""

def write_report(state: ResearchGraphState):
    sections = state["sections"]   # 所有分析师的备忘录
    topic = state["topic"]

    # 把所有备忘录拼接在一起
    formatted_str_sections = "\n\n".join([f"{section}" for section in sections])

    system_message = report_writer_instructions.format(
        topic=topic, context=formatted_str_sections
    )
    report = llm.invoke(
        [SystemMessage(content=system_message)]
        + [HumanMessage(content="Write a report based upon these memos.")]
    )
    return {"content": report.content}

write_introduction and write_conclusion

Use the same prompt word template for the introduction and conclusionintro_conclusion_instructions, distinguished by different user instructions ("Write the report introduction" vs "Write the report conclusion"):

intro_conclusion_instructions = """You are a technical writer finishing a report on {topic}.
You will be given all sections of the report.
Write a crisp and compelling introduction or conclusion.
Target around 100 words.
For your introduction: create a compelling title with # header, use ## Introduction as section header.
For your conclusion: use ## Conclusion as section header."""

def write_introduction(state: ResearchGraphState):
    sections = state["sections"]
    topic = state["topic"]
    formatted_str_sections = "\n\n".join(sections)
    instructions = intro_conclusion_instructions.format(
        topic=topic, formatted_str_sections=formatted_str_sections
    )
    intro = llm.invoke([instructions] + [HumanMessage(content="Write the report introduction")])
    return {"introduction": intro.content}

# write_conclusion 结构相同，只改用户指令

finalize_report: combined final report

def finalize_report(state: ResearchGraphState):
    content = state["content"]

    # 处理 "## Insights" 标题和 "## Sources" 来源分离
    if content.startswith("## Insights"):
        content = content.strip("## Insights")
    if "## Sources" in content:
        try:
            content, sources = content.split("\n## Sources\n")
        except:
            sources = None
    else:
        sources = None

    # 拼接：引言 + 分隔线 + 正文 + 分隔线 + 结论 + 来源
    final_report = (
        state["introduction"] + "\n\n---\n\n"
        + content + "\n\n---\n\n"
        + state["conclusion"]
    )
    if sources is not None:
        final_report += "\n\n## Sources\n" + sources

    return {"final_report": final_report}

7Detailed explanation of complete graph topology and data flow

The following is the complete construction code of the main graph (ResearchGraph), including the definition of all nodes and edges:

# 把访谈子图编译后作为一个节点挂入主图
builder = StateGraph(ResearchGraphState)

# ─── 注册节点 ───
builder.add_node("create_analysts", create_analysts)
builder.add_node("human_feedback", human_feedback)
builder.add_node("conduct_interview", interview_builder.compile())  # 子图作为节点
builder.add_node("write_report", write_report)
builder.add_node("write_introduction", write_introduction)
builder.add_node("write_conclusion", write_conclusion)
builder.add_node("finalize_report", finalize_report)

# ─── 定义边 ───
builder.add_edge(START, "create_analysts")
builder.add_edge("create_analysts", "human_feedback")

# 条件边：人工反馈后的路由（返回修改 or 启动访谈）
builder.add_conditional_edges(
    "human_feedback",
    initiate_all_interviews,              # 路由函数 / Map 函数
    ["create_analysts", "conduct_interview"]
)

# 所有访谈完成后，同时触发三个写作节点（Fan-Out）
builder.add_edge("conduct_interview", "write_report")
builder.add_edge("conduct_interview", "write_introduction")
builder.add_edge("conduct_interview", "write_conclusion")

# 三个写作节点全部完成后，才进入 finalize_report（Fan-In）
builder.add_edge(
    ["write_conclusion", "write_report", "write_introduction"],
    "finalize_report"
)
builder.add_edge("finalize_report", END)

# 编译：在 human_feedback 处中断，启用 MemorySaver 保存状态
memory = MemorySaver()
graph = builder.compile(interrupt_before=['human_feedback'], checkpointer=memory)

Complete data flow visualization

How to transfer data between main graph and subgraph

8Module 4 Knowledge Fusion: How the three major technologies work together

This lesson is the final chapter of Module 4. It completely integrates the techniques of the first three lessons into a real scene. Let us sort out the specific embodiment of each technology in the system:

Technology 1: Parallelization — Lesson 1

Technology 2: Sub-graphs — Lesson 2

Technology 3: Map-Reduce — Lesson 3

New technology unique to this section

How to extend this system in your own projects

• Alternate sources of information:putsearch_webInstead search your own RAG vector database, or useWebBaseLoaderCrawl a specific website
• Adjust the number of analysts:Modifymax_analystsParameter controls the size of parallel interviews - be aware of API concurrency limits and costs
• Customized report format:Modifysection_writer_instructionsandreport_writer_instructionsThe prompt words in can be output in any format such as JSON, slide outline, email summary, etc.
• Add more writing nodes:Multiple sub-reports such as executive summaries, risk lists, and action recommendations can be generated in parallel and finally summarized in a unified manner
• Access LangSmith:settingsLANGSMITH_TRACING=trueThe complete call chain of each interview can be tracked on the LangSmith platform to facilitate debugging and optimization.