深度研究 Agent

本实战构建 Deep Research 风格的多步研究智能体：给它一个开放式问题，它先拆解成子问题，并行搜索多个子问题，提取要点，判断研究是否充分，不够就再来一轮，最后综合成报告。核心用到 Send API 实现并行搜索 + MapReduce 思想合并结果。

一、需求分析

普通 RAG 是"一问一检索一答"，遇到"对比 A 和 B 的优劣"或"某技术的发展历程"这种需要多步研究的问题就力不从心。Deep Research 风格智能体的流程是：

1. 规划：把大问题拆成几个可独立搜索的子问题。
2. 并行搜索：每个子问题独立检索，互不阻塞。
3. 提取要点：把每个子问题的搜索结果提炼成简短 findings。
4. 判断充分性：综合 findings 看够不够回答原问题，不够就再拆一轮。
5. 综合报告：够了就写出最终报告。

这个流程的难点是并行搜索——LangGraph 的 Send API 正是为此设计。

二、整体流程图

flowchart TD
    S([START]) --> P[planner 规划子问题]
    P --> PAR[并行搜索+提炼]
    PAR --> Q[quality_check 研究充分性判断]
    Q -- 不充分 --> P
    Q -- 充分 --> W[writer 综合报告]
    W --> E([END])

planner 之后用 Send 把每个子问题分发到一个 searcher 节点并行执行，所有 searcher 完成后结果合并进 findings，再走 quality_check。这是典型的 MapReduce 模式。

三、状态设计

状态要承载规划、子问题、并行结果、报告和轮次刹车。

from typing import TypedDict, List, Annotated
from langgraph.graph import add_messages  # 不一定用 messages，但 reducer 思路一致

# 自定义 reducer：并行 searcher 的结果累加，而不是覆盖
def append_list(left: List, right: List) -> List:
    """列表追加 reducer：并行节点返回的列表合并而非覆盖"""
    return (left or []) + (right or [])

class ResearchState(TypedDict):
    question: str                 # 原始大问题
    sub_queries: List[str]        # 当前轮的子问题
    findings: Annotated[List[str], append_list]  # 各子问题的要点（并行累加）
    round: int                    # 研究轮次
    sufficient: bool              # 研究是否充分
    report: str                   # 最终报告

重点看 findings: Annotated[List[str], append_list]——这是 Reducer 的用法。并行 searcher 都会返回 {"findings": [...]}，如果没有自定义 reducer，后返回的会覆盖先返回的；加了 append_list reducer，所有并行结果会累加在一起。这是并行搜索能工作的关键。

四、节点实现

1. 规划节点 planner

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class Plan(BaseModel):
    sub_queries: List[str] = Field(description="用于搜索的子问题列表，3-5个")

plan_llm = llm.with_structured_output(Plan)

def planner(state: ResearchState) -> dict:
    """把大问题拆成可独立搜索的子问题"""
    question = state["question"]
    round_n = state.get("round", 0)
    existing_findings = state.get("findings", [])

    prompt = (
        "你是一个研究规划员。把下面这个研究问题拆成 3-5 个可独立搜索的子问题。\n"
        "子问题要具体、可检索、覆盖问题的不同侧面。\n\n"
        f"研究问题：{question}\n\n"
    )
    if existing_findings:
        prompt += f"已有研究发现：\n{chr(10).join(existing_findings)}\n\n请补充尚未覆盖的角度。\n"
    if round_n > 0:
        prompt += f"这是第 {round_n+1} 轮研究，请聚焦上一轮没查清的部分。\n"

    plan = plan_llm.invoke(prompt)
    return {"sub_queries": plan.sub_queries, "round": round_n + 1}

注意第二轮开始会把已有 findings 喂回去，让 planner 聚焦"还没查清的部分"——这是研究深度的关键。

2. 搜索+提炼节点 searcher

每个 searcher 处理一个子问题：检索 → 提炼成要点。这里用 Mock 搜索，真实项目接 Tavily/向量库。

from langchain_core.documents import Document

# Mock 搜索：真实项目替换为 TavilySearch 或向量库检索
def mock_search(query: str) -> List[Document]:
    """模拟搜索引擎，返回与 query 相关的伪结果"""
    return [
        Document(page_content=f"关于「{query}」的要点一：核心概念是 X，特点包括稳定性和扩展性。"),
        Document(page_content=f"关于「{query}」的要点二：实际应用中需注意成本和延迟的权衡。"),
        Document(page_content=f"关于「{query}」的要点三：最新趋势是结合多智能体协作提升效果。"),
    ]

def searcher(state: ResearchState) -> dict:
    """处理单个子问题：检索 + 提炼要点。
    注意：在并行模式下，这个节点接收的是 Send 注入的子状态。"""
    # Send 注入时，sub_query 单独传进来（见下面 Send 用法）
    sub_query = state["sub_query"]
    docs = mock_search(sub_query)
    context = "\n".join(d.page_content for d in docs)
    # 让 LLM 把检索内容提炼成 1-2 条要点
    summary = llm.invoke(
        f"把下面资料提炼成 1-2 条与「{sub_query}」相关的简明要点，每条一行：\n{context}"
    ).content
    # 拆成列表返回，由 append_list reducer 累加到 findings
    points = [p.strip() for p in summary.split("\n") if p.strip()]
    return {"findings": points}

3. 研究充分性判断节点 quality_check

class Quality(BaseModel):
    sufficient: bool = Field(description="现有发现是否足以回答原问题")
    missing: str = Field(description="若不充分，缺什么角度（简短）")

quality_llm = llm.with_structured_output(Quality)

def quality_check(state: ResearchState) -> dict:
    """判断现有 findings 是否足够回答原问题"""
    question = state["question"]
    findings = state.get("findings", [])
    round_n = state.get("round", 1)

    # 硬刹车：最多 3 轮，防止无限研究
    if round_n >= 3:
        return {"sufficient": True}

    prompt = (
        "判断现有研究发现是否足以回答原研究问题。\n\n"
        f"原问题：{question}\n\n已有发现：\n{chr(10).join(findings)}\n\n"
        "如果能写出有依据的完整回答，sufficient=true；否则 false 并指出缺什么。"
    )
    q = quality_llm.invoke(prompt)
    return {"sufficient": q.sufficient}

4. 综合报告节点 writer

def writer(state: ResearchState) -> dict:
    """根据所有 findings 写综合报告"""
    question = state["question"]
    findings = state.get("findings", [])
    findings_text = "\n".join(f"- {f}" for f in findings)
    report = llm.invoke(
        f"你是研究报告撰写者。根据下面研究发现，写一份回答原问题的结构化报告，"
        f"分'结论''依据''建议'三部分，引用研究发现：\n\n"
        f"原问题：{question}\n\n研究发现：\n{findings_text}"
    ).content
    return {"report": report}

五、用 Send 实现并行搜索

Send 是 LangGraph 的并行原语：在条件边里返回多个 Send(node, state)，图会同时为每个 Send 启动一个该节点实例。这是把"一个子问题列表"扇出成"多个并行 searcher"的关键。

from langgraph.types import Send

def fan_out_searches(state: ResearchState) -> List[Send]:
    """planner 之后：为每个子问题发一个并行 searcher"""
    return [
        Send("searcher", {"sub_query": q, "question": state["question"]})
        for q in state["sub_queries"]
    ]

注意 Send 的第二个参数是该 searcher 实例独享的子状态——这里只传 sub_query 和 question，不传 findings（因为 findings 是输出，不是输入）。并行 searcher 各自返回 {"findings": [...]}，靠 append_list reducer 自动合并。

小贴士：searcher 节点读的是 Send 注入的子状态，没有 findings 字段，所以它只写不读，避免并行写冲突。这是并行图设计的常见模式。

六、条件边与路由

def route_after_quality(state: ResearchState) -> str:
    if state.get("sufficient"):
        return "writer"
    return "planner"   # 不充分 → 再规划一轮

七、组装图

from langgraph.graph import StateGraph, START, END

gb = StateGraph(ResearchState, input=ResearchState, output=ResearchState)
gb.add_node("planner", planner)
gb.add_node("searcher", searcher)
gb.add_node("quality_check", quality_check)
gb.add_node("writer", writer)

gb.add_edge(START, "planner")
# planner 之后用 Send 扇出并行 searcher
gb.add_conditional_edges("planner", fan_out_searches, ["searcher"])
# 所有 searcher 完成后进 quality_check
gb.add_edge("searcher", "quality_check")
gb.add_conditional_edges("quality_check", route_after_quality, {
    "writer": "writer", "planner": "planner"
})
gb.add_edge("writer", END)

research_app = gb.compile()

add_conditional_edges("planner", fan_out_searches, ["searcher"]) 第三个参数是路径映射，这里用列表形式表示"返回的 Send 都指向 searcher 节点"。

八、完整可运行示例

import os
from typing import TypedDict, List, Annotated
from pydantic import BaseModel, Field

from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ---- 状态 + reducer ----
def append_list(left, right): return (left or []) + (right or [])

class ResearchState(TypedDict):
    question: str
    sub_queries: List[str]
    findings: Annotated[List[str], append_list]
    round: int
    sufficient: bool
    report: str

# ---- 结构化输出 ----
class Plan(BaseModel):
    sub_queries: List[str]
class Quality(BaseModel):
    sufficient: bool
    missing: str

# ---- Mock 搜索 ----
def mock_search(query):
    return [
        Document(page_content=f"「{query}」要点1：核心是稳定性与扩展性的平衡。"),
        Document(page_content=f"「{query}」要点2：需注意成本与延迟的权衡。"),
        Document(page_content=f"「{query}」要点3：趋势是多智能体协作。"),
    ]

# ---- 节点 ----
def planner(state):
    q = state["question"]
    findings = state.get("findings", [])
    p = llm.with_structured_output(Plan).invoke(
        f"把研究问题拆成3-5个可独立搜索的子问题：\n{q}\n"
        + (f"已有发现：{findings}\n请补充未覆盖角度。" if findings else ""))
    return {"sub_queries": p.sub_queries, "round": state.get("round", 0) + 1}

def searcher(state):
    sq = state["sub_query"]
    ctx = "\n".join(d.page_content for d in mock_search(sq))
    s = llm.invoke(f"把资料提炼成1-2条与「{sq}」相关的简明要点，每条一行：\n{ctx}").content
    return {"findings": [p.strip() for p in s.split("\n") if p.strip()]}

def quality_check(state):
    if state.get("round", 1) >= 3:
        return {"sufficient": True}
    q = llm.with_structured_output(Quality).invoke(
        f"判断发现是否足以回答原问题：\n原问题：{state['question']}\n"
        f"发现：{state.get('findings', [])}\n")
    return {"sufficient": q.sufficient}

def writer(state):
    f = "\n".join(f"- {x}" for x in state.get("findings", []))
    r = llm.invoke(f"根据研究发现写结构化报告(结论/依据/建议)：\n"
                   f"原问题：{state['question']}\n发现：\n{f}").content
    return {"report": r}

# ---- 扇出 + 路由 ----
def fan_out(state):
    return [Send("searcher", {"sub_query": sq, "question": state["question"]})
            for sq in state["sub_queries"]]

def route(state):
    return "writer" if state.get("sufficient") else "planner"

# ---- 建图 ----
gb = StateGraph(ResearchState)
gb.add_node("planner", planner)
gb.add_node("searcher", searcher)
gb.add_node("quality_check", quality_check)
gb.add_node("writer", writer)
gb.add_edge(START, "planner")
gb.add_conditional_edges("planner", fan_out, ["searcher"])
gb.add_edge("searcher", "quality_check")
gb.add_conditional_edges("quality_check", route, {"writer":"writer","planner":"planner"})
gb.add_edge("writer", END)
app = gb.compile()

if __name__ == "__main__":
    question = "对比 LangGraph 和 LangChain 在构建智能体时的优劣，给出选型建议。"
    result = app.invoke({"question": question, "round": 0})
    print("=== 研究报告 ===")
    print(result["report"])
    print("\n（共研究", result.get("round", 0), "轮，收集",
          len(result.get("findings", [])), "条发现）")

运行后你会看到：planner 拆出 3-5 个子问题（如"LangGraph 的核心特性""LangChain 的智能体支持""两者学习曲线"等），并行搜索器各提炼要点，quality_check 判断后写出报告。如果第一轮不够，会自动进入第二轮补充研究。

九、用 stream 观察并行执行

for chunk in app.stream({"question": question, "round": 0}, stream_mode="updates"):
    for node, update in chunk.items():
        if node == "searcher":
            # 并行 searcher 会连续出现多个 update（每个子问题一个）
            print(f"[searcher] 产出: {update.get('findings', [])}")
        elif node == "planner":
            print(f"[planner] 子问题: {update.get('sub_queries', [])}")
        elif node == "quality_check":
            print(f"[quality] sufficient={update.get('sufficient')}")

你会看到多个 [searcher] 连续输出——这就是并行的体现，它们在图里是同时执行的。

十、常见踩坑

1. 并行结果合并丢失

最经典的坑：并行 searcher 都返回 {"findings": [...]}，但状态里 findings 没加 reducer，结果只保留最后一个 searcher 的输出，前面的全丢。

根因：默认行为是"后写覆盖先写"，并行场景下谁最后写完谁赢。

解决：必须给 findings 加 Annotated[List[str], append_list] 自定义 reducer，让多个并行返回值累加。这是并行图的硬性要求，详见 Reducer 归约器 和 并行与 MapReduce。

2. 研究深度控制

Deep Research 容易"刹不住车"——quality_check 一直说不充分，planner 一直拆新子问题，token 飙升。控制手段：

• 硬性轮次上限：本例 round >= 3 强制 sufficient=true，是最可靠的刹车。
• findings 数量上限：findings 超过 20 条就强制结束。
• 成本预算：累计 token 超阈值就停。
• quality_check prompt 调严：让它更倾向于判 sufficient，避免无意义补轮。

3. 子问题质量差导致研究跑偏

planner 拆出的子问题如果太宽泛（如"LangGraph 是什么"）或重复，并行搜索结果会冗余且浅。优化：

• planner prompt 里要求"子问题互不重叠、各自具体可检索"。
• 给 planner 几个 few-shot 示例。
• 第二轮起把已有 findings 喂回去（本例已做），让 planner 聚焦缺口。

4. 成本问题

并行搜索 = 多倍 LLM 调用。3 个子问题 × 2 轮 = 6 次 searcher + 多次 planner/quality_check/writer。控制：

• searcher 用小模型（gpt-4o-mini）做提炼，writer 用大模型写报告。
• 限制子问题数量上限（3-5 个）。
• mock_search 在开发期用，避免真实搜索 API 费用。

5. Send 子状态字段不全

Send("searcher", {"sub_query": ...}) 只传了 sub_query，searcher 读 state["question"] 会 KeyError。本例 searcher 只读 sub_query，所以没问题；如果 searcher 需要 question，记得在 Send 的子状态里也带上（如本例 {"sub_query": sq, "question": state["question"]}）。

十一、小结

• 深度研究 Agent = planner（拆问题）+ searcher（并行搜索提炼）+ quality_check（充分性）+ writer（综合报告）。
• Send API 实现 planner → 多个 searcher 的扇出并行，是本实战的核心。
• 并行图的命门是 reducer：findings 必须加 append_list reducer，否则并行结果互相覆盖。
• 研究深度控制三件套：轮次上限 + findings 数量上限 + 成本预算。
• searcher 用小模型，writer 用大模型，平衡质量和成本。

下一篇 多智能体协作平台 把多个智能体组合成"团队"，用 Supervisor 调度协作。