Home / ai / llm / langgraph

LangGraph 最佳实践完整指南与实战项目

Python 3.12 · LangGraph ≥ 0.2 · LangChain ≥ 0.3 涵盖：State + Checkpointer、提示词工程、串联、并联、动态节点、ReAct、回流（反思/重试）形式：一个可运行的完整项目 —「多智能体研究写作系统」（Research-Writer-Critic）

核心概念速览
项目结构
环境准备
State 设计最佳实践
Checkpointer（持久化 / 断点续跑）
提示词工程
串联（Sequential）
并联（Fan-out / Fan-in）
动态节点（Send API / Map-Reduce）
ReAct Agent（工具调用循环）
回流（反思-修正循环 / 重试）
完整项目整合
运行方式
常见坑与最佳实践清单

1. 核心概念速览

概念	说明
`State`	图在节点间传递的共享数据结构，通常是 `TypedDict` 或 `pydantic.BaseModel`
`Node`	一个函数 `(state) -> dict`，返回对 State 的局部更新
`Edge`	节点之间的连接，分为固定边（`add_edge`）和条件边（`add_conditional_edges`）
`Reducer`	用 `Annotated[list, operator.add]` 等方式定义某字段如何"合并"多个节点的写入（而不是覆盖）
`Checkpointer`	每一步都把 State 快照写入存储（内存/SQLite/Postgres），支持中断恢复、时间旅行、多轮对话记忆
`Send`	在运行时动态生成任意数量的子任务节点调用，实现 map-reduce / 动态并行
`interrupt`	在某节点前暂停，等待人工审核（Human-in-the-loop）

心智模型：LangGraph = 一个带状态机语义的有向图执行引擎，节点是纯函数，边决定路由，State 是唯一的"真相来源"，Checkpointer 让这个状态机可持久化、可恢复、可回放。

2. 项目结构

research_writer_agent/
├── pyproject.toml
├── .env.example
├── src/
│   └── research_writer/
│       ├── __init__.py
│       ├── state.py            # State 定义（Step 4）
│       ├── prompts.py          # 提示词工程（Step 6）
│       ├── tools.py            # ReAct 工具集（Step 10）
│       ├── nodes/
│       │   ├── __init__.py
│       │   ├── planner.py      # 串联：拆解大纲
│       │   ├── researcher.py   # 并联 + 动态节点：并行调研
│       │   ├── react_agent.py  # ReAct：单个调研 agent
│       │   ├── writer.py       # 串联：汇总成稿
│       │   └── critic.py       # 回流：审稿 + 打回重写
│       ├── graph.py            # 组图（Step 12）
│       └── checkpoint.py       # Checkpointer 配置（Step 5）
└── run.py                      # 入口：CLI 调用示例

设计原则： - 节点函数不依赖全局变量，所有依赖（LLM、工具）通过闭包或 functools.partial 注入，方便单测。 - State、Prompt、Node 三者分离：State 是数据契约，Prompt 是行为契约，Node 是两者的胶水。 - 一个节点只做一件事：便于并行、便于测试、便于在图里重排。

3. 环境准备

# pyproject.toml 关键依赖
pip install "langgraph>=0.2.60" "langchain>=0.3.0" "langchain-anthropic>=0.3.0" \
            "langgraph-checkpoint-sqlite" "pydantic>=2.7" python-dotenv

# pyproject.toml
[project]
name = "research-writer-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "langgraph>=0.2.60",
    "langchain>=0.3.0",
    "langchain-anthropic>=0.3.0",
    "langgraph-checkpoint-sqlite>=2.0.0",
    "pydantic>=2.7",
    "python-dotenv>=1.0",
]

# .env.example
ANTHROPIC_API_KEY=sk-ant-xxxx

4. State 设计最佳实践

要点： - 用 Annotated[..., reducer] 明确"多个节点同时写入同一字段时如何合并"，否则默认是覆盖（后写覆盖先写），并行节点写同一字段极易出 bug。 - 用 pydantic.BaseModel 而不是裸 TypedDict，可以获得运行期校验；但 LangGraph 内部合并机制对 TypedDict 支持更原生，二者都可用，团队约定一种即可。 - 拆分「输入型字段」「过程型字段」「输出型字段」，避免一个大字典污染。

# src/research_writer/state.py
from __future__ import annotations
import operator
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage


class ResearchTask(TypedDict):
    """并联/动态节点要处理的最小任务单元"""
    topic: str
    question: str


class ResearchFinding(TypedDict):
    question: str
    summary: str
    sources: list[str]


class OverallState(TypedDict):
    # ---- 输入 ----
    topic: str

    # ---- 过程：会被多个并行分支写入，必须用 reducer 累加，而不是覆盖 ----
    subtasks: list[ResearchTask]
    findings: Annotated[list[ResearchFinding], operator.add]

    # ---- 对话型历史，用 LangGraph 内置的 add_messages reducer 自动去重/合并 ----
    messages: Annotated[list[AnyMessage], add_messages]

    # ---- 串联主线 ----
    outline: str
    draft: str
    critique: str
    revision_count: int

    # ---- 输出 ----
    final_report: str

⚠️ 常见坑：findings 字段会被并行的多个 researcher 节点同时返回，如果不加 Annotated[list, operator.add]，最终只会保留"最后一个完成"的那一份，其余全部丢失。这是并行图里最典型的 State 设计错误。

5. Checkpointer（持久化 / 断点续跑）

Checkpointer 让图在每一步之后自动保存快照，具备三大能力： 1. 多轮对话记忆：同一个 thread_id 下次调用自动带上历史 State。 2. 中断恢复：进程崩溃/人工中断后，从最后一个 checkpoint 继续跑，不用重头开始。 3. 时间旅行 / 调试：可以拉出某一步的历史 State 做对比、回放。

# src/research_writer/checkpoint.py
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.memory import MemorySaver


def get_dev_checkpointer() -> MemorySaver:
    """本地调试用：进程内内存，重启即丢失"""
    return MemorySaver()


def get_prod_checkpointer(db_path: str = "checkpoints.sqlite") -> SqliteSaver:
    """生产/持久化：SQLite 文件，重启后可续跑"""
    conn = sqlite3.connect(db_path, check_same_thread=False)
    return SqliteSaver(conn)


# 若需要多进程/多机共享，替换为 Postgres：
#   pip install langgraph-checkpoint-postgres
#   from langgraph.checkpoint.postgres import PostgresSaver
#   checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
#   checkpointer.setup()  # 首次需要建表

编译图时传入：

graph = builder.compile(checkpointer=get_prod_checkpointer())

config = {"configurable": {"thread_id": "user-42-session-1"}}
graph.invoke({"topic": "..."}, config=config)          # 第一次
graph.invoke({"messages": [("user", "继续补充第二段")]}, config=config)  # 自动带上历史

Human-in-the-loop（结合 checkpointer）：

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["critic"],   # 在 critic 节点执行前暂停，等人工确认
)
# 恢复执行：
graph.invoke(None, config=config)

6. 提示词工程

原则： - System Prompt 与角色强绑定，每个节点/agent 一个独立、职责单一的 system prompt，不要共用一个"万能提示词"。 - 用 结构化输出（with_structured_output + pydantic）替代"让模型输出 JSON 然后自己 parse"，减少解析失败率。 - Few-shot 示例放在 prompt 模板里而不是硬编码在代码字符串拼接中，便于统一管理和 A/B 测试。 - 明确输出边界：长度、语言、禁止事项，减少模型"跑题"。

# src/research_writer/prompts.py
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field


# ---------- 结构化输出 schema ----------
class Outline(BaseModel):
    sections: list[str] = Field(description="报告的章节标题列表，3-6 个")
    key_questions: list[str] = Field(description="需要调研澄清的关键问题列表")


class Critique(BaseModel):
    approved: bool = Field(description="草稿是否达到发布标准")
    issues: list[str] = Field(description="需要修正的具体问题，approved=True 时为空列表")
    severity: str = Field(description="none | minor | major")


# ---------- Planner ----------
PLANNER_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你是一名资深研究报告策划编辑。给定一个主题，输出报告大纲和需要调研的关键问题。\n"
     "要求：\n"
     "1. 章节标题简洁、覆盖面互不重叠；\n"
     "2. 关键问题必须是可通过检索/调研回答的具体问题，不要空泛；\n"
     "3. 只输出结构化字段，不要额外解释。"),
    ("human", "主题：{topic}"),
])

# ---------- Writer ----------
WRITER_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你是一名专业的中文技术报告撰写者。基于给定的大纲和调研结果撰写完整报告正文。\n"
     "要求：\n"
     "- 使用 Markdown 格式，每个大纲章节对应一个二级标题；\n"
     "- 严格基于提供的调研结果撰写，不要编造未提供的数据或来源；\n"
     "- 语言简洁、专业，避免空话套话；\n"
     "- 如果收到了'修改意见'，必须逐条对照修正，不能忽略任何一条。"),
    ("human",
     "主题：{topic}\n\n大纲：\n{outline}\n\n调研结果：\n{findings}\n\n"
     "上一版修改意见（如无则为空）：\n{critique}"),
])

# ---------- Critic（回流判断依据） ----------
CRITIC_PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "你是严格的报告审校编辑。检查草稿是否满足：\n"
     "1. 覆盖大纲全部章节；2. 论述有调研依据支撑；3. 没有明显事实矛盾；4. 篇幅合理。\n"
     "只有完全满足才能 approved=True。有任何一条不满足都要指出具体问题。"),
    ("human", "大纲：\n{outline}\n\n草稿：\n{draft}"),
])

# ---------- ReAct 调研 Agent 的 system prompt ----------
REACT_RESEARCHER_SYSTEM = (
    "你是一名调研助手，可以使用 web_search 和 calculator 工具。\n"
    "流程：思考需要什么信息 -> 调用工具 -> 根据结果继续思考，直到能够回答问题为止。\n"
    "最终必须给出一段简明的调研总结（summary）以及引用来源（sources）。\n"
    "不要在没有工具支持的情况下编造具体数据。"
)

绑定结构化输出：

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
planner_llm = llm.with_structured_output(Outline)
critic_llm = llm.with_structured_output(Critique)

7. 串联（Sequential）

最基础的模式：节点按固定顺序执行，上一节点的输出是下一节点的输入。

# src/research_writer/nodes/planner.py
from research_writer.state import OverallState
from research_writer.prompts import PLANNER_PROMPT
from langchain_anthropic import ChatAnthropic
from research_writer.prompts import Outline

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0).with_structured_output(Outline)


def planner_node(state: OverallState) -> dict:
    chain = PLANNER_PROMPT | llm
    result: Outline = chain.invoke({"topic": state["topic"]})
    outline_text = "\n".join(f"- {s}" for s in result.sections)
    subtasks = [{"topic": state["topic"], "question": q} for q in result.key_questions]
    return {"outline": outline_text, "subtasks": subtasks}

串联在图里就是最普通的 add_edge：

builder.add_node("planner", planner_node)
builder.add_node("writer", writer_node)
builder.add_edge("planner", "researcher_fanout")   # 计划 -> 调研
builder.add_edge("writer", "critic")               # 成稿 -> 审校

8. 并联（Fan-out / Fan-in）

当子任务数量固定已知时，可以直接用多条并行边指向下游同一个节点，LangGraph 会自动等待所有分支完成后再执行下游（superstep 语义）。

# 固定并联示例：3 路固定并行检索不同来源
builder.add_node("search_web", search_web_node)
builder.add_node("search_arxiv", search_arxiv_node)
builder.add_node("search_news", search_news_node)
builder.add_node("merge", merge_node)

for src in ["search_web", "search_arxiv", "search_news"]:
    builder.add_edge("planner", src)   # fan-out：一个节点触发多个下游
    builder.add_edge(src, "merge")     # fan-in：多个上游汇聚到一个节点

merge 节点只有在 search_web、search_arxiv、search_news 全部执行完毕后才会被调度一次（而不是被调用三次）—— 这就是 fan-in 的语义，前提是它们写入 State 的字段都配置了 reducer（如第 4 节的 operator.add）。

固定并联的局限：分支数量必须在建图时写死。当分支数量取决于运行时数据（比如"根据 planner 生成了几个调研问题就并行几个"），需要用第 9 节的动态节点。

9. 动态节点（Send API / Map-Reduce）

Send 是 LangGraph 实现"运行时动态扇出"的核心机制：一个节点可以在运行时根据 State 生成任意数量的子任务，每个子任务被独立调度到同一个目标节点，各自拥有一份局部 State，执行完后再通过 reducer 汇总回主 State。这是动态节点（也叫 map-reduce 模式）的标准做法。

# src/research_writer/nodes/researcher.py
from langgraph.types import Send
from research_writer.state import OverallState, ResearchTask


def fanout_to_researchers(state: OverallState) -> list[Send]:
    """条件边函数：不返回节点名，而是返回一组 Send，
    每个 Send 携带一份独立的局部 state 派发给 react_researcher 节点。
    分支数量 = len(state['subtasks'])，运行时才确定。"""
    return [
        Send("react_researcher", {"task": task})
        for task in state["subtasks"]
    ]


def collect_node(state: OverallState) -> dict:
    """fan-in 节点：所有动态分支通过 findings 的 operator.add reducer
    已经自动汇总完毕，这里只做后处理（例如去重、排序）。"""
    seen = set()
    deduped = []
    for f in state["findings"]:
        if f["question"] not in seen:
            seen.add(f["question"])
            deduped.append(f)
    return {"findings": deduped}

注册为条件边（注意：Send 用在 add_conditional_edges 里，源节点是 planner，目标是"虚拟的"，由 Send 决定）：

builder.add_conditional_edges(
    "planner",
    fanout_to_researchers,      # 返回 list[Send]
    ["react_researcher"],       # 声明可能的目标节点（供图校验/可视化）
)
builder.add_edge("react_researcher", "collect")

react_researcher 节点接收的是 {"task": ResearchTask}（而不是完整 OverallState），返回值仍然通过 reducer 合并回主图的 findings 字段——这正是第 10 节 ReAct agent 要处理的输入形状。

10. ReAct Agent（工具调用循环）

ReAct = 推理（Reason）→ 行动（Act，调用工具）→ 观察（Observe） 的循环，直到模型认为不再需要工具、给出最终答案。在 LangGraph 里，这本质上是一个"LLM 节点 ⇄ 工具节点"之间的回流子图。

# src/research_writer/tools.py
from langchain_core.tools import tool


@tool
def web_search(query: str) -> str:
    """搜索网络获取与 query 相关的信息，返回摘要文本。"""
    # 实际项目中替换为真实检索（Tavily / SerpAPI / 内部知识库等）
    return f"[mock] 关于「{query}」的检索结果摘要……"


@tool
def calculator(expression: str) -> str:
    """计算一个数学表达式，例如 '3 * (4 + 5)'。"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算出错: {e}"


TOOLS = [web_search, calculator]

# src/research_writer/nodes/react_agent.py
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict
from research_writer.tools import TOOLS
from research_writer.prompts import REACT_RESEARCHER_SYSTEM

llm_with_tools = ChatAnthropic(model="claude-sonnet-4-6", temperature=0).bind_tools(TOOLS)


class ReactState(TypedDict):
    task: dict                       # {"topic":..., "question":...}
    messages: Annotated[list, add_messages]


def agent_reason_node(state: ReactState) -> dict:
    msgs = state["messages"]
    if not msgs:
        msgs = [
            SystemMessage(content=REACT_RESEARCHER_SYSTEM),
            HumanMessage(content=f"请调研并回答：{state['task']['question']}"),
        ]
    response: AIMessage = llm_with_tools.invoke(msgs)
    return {"messages": [response]}


def finalize_node(state: ReactState) -> dict:
    """ReAct 循环结束后，把最后一条 AI 消息整理成标准的 ResearchFinding。"""
    last = state["messages"][-1]
    sources = [
        m.content for m in state["messages"]
        if getattr(m, "type", "") == "tool"
    ]
    return {
        "findings": [{
            "question": state["task"]["question"],
            "summary": last.content,
            "sources": sources[:3],
        }]
    }


# ---- 子图：独立的 ReAct 循环 ----
react_builder = StateGraph(ReactState)
react_builder.add_node("agent", agent_reason_node)
react_builder.add_node("tools", ToolNode(TOOLS))
react_builder.add_node("finalize", finalize_node)

react_builder.add_edge(START, "agent")
react_builder.add_conditional_edges(
    "agent",
    tools_condition,           # 内置：AI 消息里有 tool_calls -> "tools"，否则 -> END 对应分支
    {"tools": "tools", END: "finalize"},
)
react_builder.add_edge("tools", "agent")   # 回流：工具结果送回 agent 继续推理
react_builder.add_edge("finalize", END)

react_researcher_subgraph = react_builder.compile()

tools_condition -> "tools" -> agent -> tools_condition -> ... 就是 ReAct 的回流边：只要模型还在请求工具，就一直在 agent 与 tools 之间打转，直到某一轮不再产生 tool_calls 才跳出循环。这个子图会被作为主图中 react_researcher 节点的实现（LangGraph 原生支持"节点=已编译子图"）。

11. 回流（反思-修正循环 / 重试）

回流（cycle）是 LangGraph 相比纯 DAG 编排框架的核心优势：审校不通过就打回重写，直到通过或达到最大重试次数，这是典型的"反思型 agent"模式。

# src/research_writer/nodes/critic.py
from research_writer.state import OverallState
from research_writer.prompts import CRITIC_PROMPT, Critique
from langchain_anthropic import ChatAnthropic

MAX_REVISIONS = 3
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0).with_structured_output(Critique)


def critic_node(state: OverallState) -> dict:
    chain = CRITIC_PROMPT | llm
    result: Critique = chain.invoke({
        "outline": state["outline"],
        "draft": state["draft"],
    })
    critique_text = "; ".join(result.issues) if not result.approved else ""
    return {
        "critique": critique_text,
        "revision_count": state.get("revision_count", 0) + (0 if result.approved else 1),
        "final_report": state["draft"] if result.approved else state.get("final_report", ""),
        "_approved": result.approved,   # 临时字段，仅供路由函数读取
    }


def route_after_critic(state: OverallState) -> str:
    """回流控制：未通过且未超过重试上限 -> 打回 writer 重写；
    否则（通过 / 超过上限）-> 结束。"""
    if state.get("_approved"):
        return "end"
    if state.get("revision_count", 0) >= MAX_REVISIONS:
        # 兜底：达到重试上限，强制采用当前草稿，避免死循环
        return "end"
    return "revise"

builder.add_node("critic", critic_node)
builder.add_conditional_edges(
    "critic",
    route_after_critic,
    {"revise": "writer", "end": END},   # revise -> 回流到 writer；end -> 结束
)

⚠️ 回流的两条铁律： 1. 必须有终止条件（如 MAX_REVISIONS），否则可能死循环，graph.invoke 会因为达到 LangGraph 默认的 recursion_limit（默认 25）而报错，而不是无限跑下去——但更好的做法是自己显式兜底，而不是依赖框架报错。 2. 回流携带的信息必须递增/变化（如把 critique 写回 State 供 writer 下一轮读取），否则模型每次拿到完全相同的输入，大概率重复同样的错误。

12. 完整项目整合

# src/research_writer/graph.py
from langgraph.graph import StateGraph, START, END
from research_writer.state import OverallState
from research_writer.nodes.planner import planner_node
from research_writer.nodes.researcher import fanout_to_researchers, collect_node
from research_writer.nodes.react_agent import react_researcher_subgraph
from research_writer.nodes.writer import writer_node
from research_writer.nodes.critic import critic_node, route_after_critic
from research_writer.checkpoint import get_prod_checkpointer


def build_graph():
    builder = StateGraph(OverallState)

    # ---- 节点注册 ----
    builder.add_node("planner", planner_node)                        # 串联起点
    builder.add_node("react_researcher", react_researcher_subgraph)  # 动态并联 + ReAct 子图
    builder.add_node("collect", collect_node)                        # fan-in 汇总
    builder.add_node("writer", writer_node)                          # 串联
    builder.add_node("critic", critic_node)                          # 回流判定

    # ---- 边 ----
    builder.add_edge(START, "planner")
    builder.add_conditional_edges(
        "planner", fanout_to_researchers, ["react_researcher"]        # 动态节点：Send
    )
    builder.add_edge("react_researcher", "collect")
    builder.add_edge("collect", "writer")
    builder.add_edge("writer", "critic")
    builder.add_conditional_edges(
        "critic", route_after_critic, {"revise": "writer", "end": END}  # 回流
    )

    return builder.compile(checkpointer=get_prod_checkpointer())


graph = build_graph()

# src/research_writer/nodes/writer.py
from research_writer.state import OverallState
from research_writer.prompts import WRITER_PROMPT
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.3)


def writer_node(state: OverallState) -> dict:
    findings_text = "\n\n".join(
        f"Q: {f['question']}\nA: {f['summary']}" for f in state["findings"]
    )
    chain = WRITER_PROMPT | llm
    result = chain.invoke({
        "topic": state["topic"],
        "outline": state["outline"],
        "findings": findings_text,
        "critique": state.get("critique", ""),
    })
    return {"draft": result.content}

# run.py
from dotenv import load_dotenv
load_dotenv()

from src.research_writer.graph import graph

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "demo-thread-1"}}
    result = graph.invoke({"topic": "生成式 AI 在企业软件研发中的落地路径"}, config=config)
    print(result["final_report"])

流程图（文字版）：

START
  └─▶ planner  ──(Send: 动态生成 N 个调研问题)──▶ react_researcher × N（并行）
                                                        │  ReAct 内部回流：agent ⇄ tools
                                                        ▼
                                                     collect（fan-in）
                                                        │
                                                        ▼
                                                     writer  ◀──────┐
                                                        │           │ revise（回流）
                                                        ▼           │
                                                     critic ────────┘
                                                        │ approved / 超过重试上限
                                                        ▼
                                                       END

13. 运行方式

uv venv --python 3.12
source .venv/bin/activate
uv pip install -e .

cp .env.example .env   # 填入 ANTHROPIC_API_KEY

python run.py

调试建议：用 LangGraph 内置的可视化确认图结构是否符合预期，尤其是动态节点和回流边容易连错：

from src.research_writer.graph import graph
graph.get_graph().draw_mermaid_png(output_file_path="graph.png")

14. 常见坑与最佳实践清单

[ ] 并行写同一 State 字段必须加 reducer（operator.add / 自定义合并函数），否则数据互相覆盖。
[ ] 动态节点用 Send，不要试图用 for 循环在建图阶段模拟"运行时才知道数量"的并行，那样只能处理固定数量。
[ ] ReAct / 回流循环一定要设终止条件（最大轮数、最大重试数），并显式兜底，不要依赖框架的 recursion_limit 报错来"兜底"。
[ ] 提示词按角色拆分独立管理，不要多个节点共用一个 system prompt。
[ ] 结构化输出优先用 with_structured_output(PydanticModel)，比"要求模型输出 JSON + 手工 json.loads"更稳。
[ ] thread_id 是多轮对话/断点续跑的关键，同一会话务必复用同一个 thread_id。
[ ] 生产环境用 SqliteSaver / PostgresSaver，MemorySaver 仅用于本地开发和单测，进程重启即丢失。
[ ] 子图（如 ReAct agent）的局部 State 与主图的 OverallState 是两套 schema，注意字段命名不要"看起来一样但语义不同"。
[ ] 给每个可能产生高成本调用（LLM/工具）的节点加超时和异常兜底，避免一个分支失败拖垮整个 fan-in。
[ ] 用 graph.get_graph().draw_mermaid_png() 在开发阶段可视化，比脑内推演边的连接靠谱得多。

以上覆盖了：State 设计 + reducer、Checkpointer（内存/SQLite/Postgres）、提示词工程（角色化 + 结构化输出）、串联（planner→writer）、并联（固定 fan-out/fan-in）、动态节点（Send/map-reduce）、ReAct（agent⇄tools 子图）、回流（critic→writer 反思修正循环），是一个可以直接作为脚手架扩展的完整项目。

Page Source