DeerFlow：字节跳动的 Deep Research 框架

import asyncio
from src.workflow import run_agent_workflow

async def main():
    # 运行深度研究
    result = await run_agent_workflow(
        user_input="分析 2024-2025 年 AI Agent 框架生态的发展趋势",
        debug=True,
    )
    print(result["final_report"])

asyncio.run(main())

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

class ResearchState(TypedDict):
    user_input: str
    research_plan: List[str]
    search_results: Annotated[List[str], operator.add]
    final_report: str

def coordinator_node(state: ResearchState) -> dict:
    """协调器：判断是直接回答还是需要深度研究"""
    # 简单问题直接回答，复杂问题启动研究流程
    return {}

def planner_node(state: ResearchState) -> dict:
    """规划器：把问题拆解成子研究任务"""
    question = state["user_input"]

    # 调用 LLM 生成研究计划
    plan_response = llm.invoke(f"""
把以下研究问题拆解成 3-5 个独立的子研究方向，每行一个：

问题：{question}

子任务：""")

    tasks = [line.strip() for line in plan_response.content.split("\n") if line.strip()]
    return {"research_plan": tasks}

def researcher_node(state: ResearchState) -> dict:
    """研究员：对一个子任务进行网络搜索和分析"""
    # 每个子任务调用搜索 API
    results = []
    for task in state["research_plan"]:
        search_result = search_web(task)  # 调用 Tavily/Serper
        analyzed = analyze_result(task, search_result)  # LLM 分析
        results.append(analyzed)
    return {"search_results": results}

def writer_node(state: ResearchState) -> dict:
    """写作者：整合所有研究结果，生成最终报告"""
    all_results = "\n\n".join(state["search_results"])

    report = llm.invoke(f"""
基于以下研究材料，为问题"{state['user_input']}"撰写一份深度分析报告：

{all_results}

报告格式：
# 标题
## 执行摘要
## 详细分析
## 结论与建议
""")
    return {"final_report": report.content}

# 组装图
graph = StateGraph(ResearchState)
graph.add_node("coordinator", coordinator_node)
graph.add_node("planner", planner_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)

graph.set_entry_point("coordinator")
graph.add_edge("coordinator", "planner")
graph.add_edge("planner", "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", END)

from src.tools.search import create_search_tool

# 接入自定义搜索源
def custom_search(query: str) -> list[dict]:
    """接入内部知识库或特定数据源"""
    results = your_internal_api.search(query)
    return [{"title": r.title, "content": r.content, "url": r.url} for r in results]

# 注册到 DeerFlow
search_tool = create_search_tool(custom_search)