拆解字节开源 DeerFlow：查资料写报告这件事，Agent 真能全自动了

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
import operator


class Plan(TypedDict):
    """研究计划：由规划者生成，人工可编辑"""
    locale: str                      # 输出语言
    has_enough_context: bool         # 信息是否足够，不够就继续规划
    thought: str                     # 规划思路
    title: str                       # 报告标题
    steps: list                      # 步骤列表，每步有执行说明

class Step(TypedDict):
    """单个研究步骤"""
    need_search: bool                # 是否需要联网搜索
    title: str                       # 步骤标题
    description: str                 # 执行说明
    step_type: str                   # research / processing
    execution_res: str               # 执行结果（研究员/程序员回填）


class State(TypedDict):
    """全局状态：五个角色共享的工作台"""
    messages: Annotated[Sequence[BaseMessage], add_messages]

    # 流程控制
    goto: str                        # 下一跳节点
    locale: str

    # 研究主题相关
    research_topic: str              # 用户的研究主题
    background_investigation_results: str   # 背景调查结果
    plan: Plan                       # 研究计划
    observations: Annotated[list, operator.add]  # 观察记录，只增不减

from langgraph.graph import StateGraph, START, END
from src.graph.types import State
from src.graph.nodes import (
    coordinator_node,
    background_investigation_node,
    planner_node,
    human_feedback_node,
    research_team_node,
    researcher_node,
    coder_node,
    reporter_node,
)


def _build_graph() -> StateGraph:
    builder = StateGraph(State)

    # 注册节点：五个角色 + 背景调查 + 人工确认
    builder.add_node("coordinator", coordinator_node)
    builder.add_node("background_investigator", background_investigation_node)
    builder.add_node("planner", planner_node)
    builder.add_node("human_feedback", human_feedback_node)
    builder.add_node("research_team", research_team_node)
    builder.add_node("researcher", researcher_node)
    builder.add_node("coder", coder_node)
    builder.add_node("reporter", reporter_node)

    # 连线
    builder.add_edge(START, "coordinator")

    # 协调者分流：闲聊直接结束，研究任务进背景调查
    builder.add_conditional_edges(
        "coordinator",
        lambda state: state["goto"],
        ["background_investigator", END],
    )

    # 背景调查 → 规划
    builder.add_edge("background_investigator", "planner")

    # 规划 → 人工确认（interrupt_before 是关键！）
    builder.add_edge("planner", "human_feedback")

    # 人工确认 → 研究调度器
    builder.add_edge("human_feedback", "research_team")

    # 核心循环：调度器根据步骤类型，派给研究员或程序员
    builder.add_conditional_edges(
        "research_team",
        lambda state: state["goto"],   # research / coder / reporter
        ["researcher", "coder", "reporter"],
    )

    # 研究员/程序员干完活，回到调度器看还有没有下一步
    builder.add_edge("researcher", "research_team")
    builder.add_edge("coder", "research_team")

    # 全部完成 → 报告员 → 结束
    builder.add_edge("reporter", END)

    return builder


# 编译图。interrupt_before 让流程在 human_feedback 前暂停，
# 等用户确认计划后再继续——这就是"人机协同"的实现方式
graph = _build_graph().compile(checkpointer=memory, interrupt_before=["human_feedback"])

# 研究员的工具配置（简化）
from langgraph.prebuilt import create_react_agent
from src.tools.tavily_search import tavily_search
from src.tools.crawler import crawl_tool


def create_researcher_agent():
    prompt = """你是专业研究员。基于给定步骤执行研究：
    1. 需要信息时调用 tavily_search 搜索
    2. 搜索结果不够细，用 crawl_tool 抓取完整网页
    3. 综合多来源信息，产出带来源标注的研究发现
    4. 引用格式：使用标题和 URL 列表"""

    return create_react_agent(
        model=llm,
        tools=[tavily_search, crawl_tool],   # 搜索 + 爬虫，就这俩
        prompt=prompt,
    )

# src/tools/tavily_search.py（简化）
from langchain_core.tools import tool
from tavily import TavilyClient

@tool
def tavily_search(query: str, search_depth: str = "advanced",
                  max_results: int = 5, topic: str = "general") -> str:
    """联网搜索工具：查询最新信息、行业动态、技术资料时使用。

    Args:
        query: 搜索关键词，尽量具体
        search_depth: basic 或 advanced
        max_results: 返回条数
        topic: general / news，新闻类查询用 news 效果更好
    """
    client = TavilyClient(api_key=TAVILY_API_KEY)
    results = client.search(
        query=query,
        search_depth=search_depth,
        max_results=max_results,
        topic=topic,
    )

    # 把结果压成模型友好的格式：标题 + 摘要 + URL
    formatted = []
    for r in results.get("results", []):
        formatted.append(
            f"标题: {r['title']}\n"
            f"摘要: {r.get('content', '')[:500]}\n"
            f"URL: {r['url']}\n"
            f"相关度: {r.get('score', 0):.2f}\n---"
        )
    return "\n".join(formatted)

# src/tools/python_repl.py（简化思路）
from langchain_experimental.tools.python.tool import PythonAstREPLTool

@tool
def python_repl_tool(code: str) -> str:
    """执行 Python 代码做数据分析。可以处理前面步骤产出的
    结构化数据（表格、数值对比），返回计算结果或生成图表。

    注意：代码在受限环境中执行，无网络访问权限。"""
    # DeerFlow 的实际实现是一个持久化的 REPL 会话，
    # 变量在多次调用之间保留，方便分步处理数据
    ...

# 1. 克隆并安装（官方推荐 uv，依赖装得快）
git clone https://github.com/bytedance/deer-flow.git
cd deer-flow
uv sync

# 2. 配置 LLM：复制示例配置后编辑
cp conf.yaml.example conf.yaml
# conf.yaml 关键项：
#   BASIC_MODEL:      便宜快的模型，干协调者、规划者的活
#   REASONING_MODEL:  推理强的模型，干深度研究和报告成稿

# 3. 配置搜索工具的 Key（.env）
echo "TAVILY_API_KEY=tvly-xxxxx" >> .env

# 4. 启动后端（默认 8000 端口）
uv run python server.py --port 8000

# 5. 前端（另开一个终端）
cd web && pnpm install && pnpm dev

# conf.yaml 关键部分
BASIC_MODEL:
  base_url: "https://api.deepseek.com/v1"
  model: "deepseek-chat"          # 快、便宜，规划协调够用

REASONING_MODEL:
  base_url: "https://api.deepseek.com/v1"
  model: "deepseek-reasoner"      # 推理强，研究报告质量关键

SEARCH_API: tavily                 # 可选 tavily / brave / duckduckgo / arxiv
MAX_PLAN_STEPS: 5                  # 限制计划步数，防止规划者上头

curl -N -X POST http://localhost:8000/api/chat/stream \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "调研人形机器人主要厂商的技术路线差异和最新融资情况，输出中文报告"}
    ],
    "recursion_limit": 50,
    "max_plan_iterations": 2
  }'

event: agent
data: {"node": "planner", "content": {"title": "人形机器人行业研究报告", "steps": [
  {"title": "梳理主要厂商及产品矩阵", "step_type": "research", "need_search": true},
  {"title": "对比技术路线：全尺寸/轮式/仿生关节方案", "step_type": "research", "need_search": true},
  {"title": "汇总 2024-2026 年融资事件与金额", "step_type": "research", "need_search": true},
  {"title": "量化对比：融资规模 vs 发布机型数量", "step_type": "processing", "need_search": false},
  {"title": "撰写综合报告", "step_type": "processing", "need_search": false}
]}}

event: agent
data: {"node": "researcher", "content": "步骤1执行完毕：识别出 8 家主要厂商...[含 12 个引用来源]"}

event: agent
data: {"node": "coder", "content": "已完成融资规模与机型数量的交叉对比，TOP3 为..."}

event: agent
data: {"node": "reporter", "content": "# 人形机器人行业研究报告\n\n## 一、行业概况\n..."}

# conf.yaml 追加 MCP 配置
MCP_SERVERS:
  - name: internal-wiki
    transport: sse
    url: "http://internal-mcp.example.com/sse"
    description: "公司内部知识库搜索，包含历史项目文档、技术方案、流程规范"

# src/tools/internal_wiki_search.py
import os
import requests
from langchain_core.tools import tool


@tool
def internal_wiki_search(query: str, max_results: int = 5) -> str:
    """搜索公司内部知识库。当研究涉及公司内部项目、
    历史技术方案、内部流程规范时调用此工具。

    Args:
        query: 搜索关键词
        max_results: 最多返回条数
    """
    resp = requests.get(
        "http://internal-wiki.example.com/api/v1/search",
        params={"q": query, "limit": max_results},
        headers={
            "Authorization": f"Bearer {os.environ['WIKI_TOKEN']}",
            "X-User-Id": os.environ.get("CURRENT_USER", "system"),
        },
        timeout=10,
    )
    resp.raise_for_status()
    docs = resp.json().get("results", [])

    if not docs:
        return "内部知识库未找到相关内容，建议仅依赖外部公开资料。"

    formatted = []
    for d in docs:
        formatted.append(
            f"[内部文档] {d['title']} (更新于 {d['updated_at'][:10]})\n"
            f"摘要: {d['snippet'][:300]}\n"
            f"来源: {d['url']}\n---"
        )
    return "\n".join(formatted)

# src/agents/agents.py 中研究员的工具注入（简化）
from src.tools.internal_wiki_search import internal_wiki_search

research_agent = create_react_agent(
    model=llm,
    tools=[tavily_search, crawl_tool, internal_wiki_search],  # 挂上内部搜索
    prompt=researcher_prompt,
)