研究助手：多 Agent 自动化研究
并行 · 子图 · Map-Reduce 的终极综合

1 整体架构：这套系统在干什么

研究是一项耗时耗力的工作——搜集资料、分析不同角度、综合形成见解、写成报告，往往需要一整个团队协作完成。这节课的目标是用 LangGraph 构建一个轻量级多 Agent 研究系统，自动化完成上述全流程。

系统的五大能力设计：

• 1

  来源选择（Source Selection）

  用户可自定义使用哪些信息来源——网页搜索（Tavily）、维基百科、特定网站、本地索引文档等。

• 2

  规划（Planning）

  用户输入研究主题，系统自动生成一个 AI 分析师团队，每人专注一个子议题。Human-in-the-loop 允许用户在研究开始前审查并调整分析师阵容。

• 3

  多轮访谈（LLM Utilization）

  每位分析师都会与一位"专家 AI"进行深度多轮对话。对话过程封装在子图（sub-graph）中，拥有独立的内部状态。

• 4

  并行研究（Parallelization + Map-Reduce）

  专家在回答问题时同时调用网络搜索和维基百科（并行），所有分析师的访谈也同时进行（Map-Reduce）。

• 5

  报告输出（Output Format）

  汇总所有访谈洞见，写出引言、正文、结论，形成一份结构完整的最终报告。

系统中的 Agent 角色

2 第一阶段：生成分析师团队（Human-in-the-loop）

研究开始前，系统先生成一批 AI 分析师，并通过 Human-in-the-loop 让用户审查、调整阵容，直到满意为止。

数据模型：Analyst 和 Perspectives

每位分析师用 Pydantic BaseModel 描述，包含四个核心属性，以及一个用于生成提示词的 persona 属性：

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 节点：LLM 生成分析师

# 提示词模板：要求 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 节点与循环控制

human_feedback 是一个空操作节点（no-op），它的存在只是为了让图在此处暂停，等待用户更新状态。

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 的操作流程

3 第二阶段：访谈子图——并行信息收集

每一位分析师的访谈过程被封装在一个独立的访谈子图（interview sub-graph）中。子图有自己的状态 InterviewState，独立运行，互不干扰。

InterviewState：子图的内部状态

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                              # 写好的备忘录（传递给外层主图）

访谈子图内的节点

子图包含六个节点，形成一个"提问 → 并行搜索 → 回答 → 再提问"的循环：

专家在回答时如何并行搜索

这是 Module 4 Lesson 1 并行化技术的直接应用。分析师提问后，search_web 和 search_wikipedia 同时启动：

# 两个搜索节点都连接到 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")

访谈终止逻辑：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"

4 Tavily 搜索工具：如何联网获取信息

Tavily 是专门为 AI Agent 设计的搜索 API，相比 Google 搜索，它返回的是清洁的、直接可用的文本摘要，而不是一堆 HTML。

搜索节点的完整实现

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="{doc["url"]}"/>\n{doc["content"]}\n</Document>'
        for doc in search_docs
    ])

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

维基百科搜索节点

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]}

专家回答节点：带引用的回复生成

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]}

5 第三阶段：Map-Reduce 启动所有访谈

分析师团队确认后，主图使用 Send() API 把所有访谈同时分发出去（Map），每个访谈子图并行运行，完成后结果自动聚合到主图状态（Reduce）。

ResearchGraphState：主图状态

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 函数

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"]
    ]

执行输出示例

实际运行时，可以看到三个 conduct_interview 并行完成，随后写作节点同步触发：

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

6 第四阶段：最终报告的写作与汇总

所有访谈子图完成后，主图触发三个并行的写作节点：write_report、write_introduction、write_conclusion，最后由 finalize_report 将它们组合成完整报告。

write_report：综合所有备忘录

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 和 write_conclusion

引言和结论使用同一套提示词模板 intro_conclusion_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：组合最终报告

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}

7 完整图拓扑与数据流详解

下面是主图（ResearchGraph）的完整构建代码，包含所有节点和边的定义：

# 把访谈子图编译后作为一个节点挂入主图
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)

完整数据流可视化

数据在主图与子图之间如何传递

8 Module 4 知识融合：三大技术如何协同

这节课是 Module 4 的终章，将前三节课的技术完整融合进一个真实场景。让我们梳理每项技术在系统中的具体体现：

技术一：并行化（Parallelization）— Lesson 1

技术二：子图（Sub-graphs）— Lesson 2

技术三：Map-Reduce — Lesson 3

本节独有的新技术

如何在自己的项目中扩展这个系统

• 替换信息来源：把 search_web 换成对自己的 RAG 向量数据库的检索，或者使用 WebBaseLoader 抓取特定网站
• 调整分析师数量：修改 max_analysts 参数控制并行访谈的规模——注意 API 并发限制和成本
• 定制报告格式：修改 section_writer_instructions 和 report_writer_instructions 中的提示词，可以输出 JSON、幻灯片大纲、邮件摘要等任意格式
• 添加更多写作节点：可以并行生成执行摘要、风险清单、行动建议等多个子报告，最后统一汇总
• 接入 LangSmith：设置 LANGSMITH_TRACING=true 可以在 LangSmith 平台追踪每次访谈的完整调用链，便于调试和优化