把三轴、原理、失败模式揉进一个真实项目

# 1 │ 状态：用 operator.add 让并行 worker 的子报告安全合并（D4 · 04 §4.9）
import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

class Finding(TypedDict):
    source_name: str          # 溯源（D4 · 04 §4.11）
    source_url: str           # 每条结论挂得到来源（D3 结构化回传）
    claim: str

class ResearchState(TypedDict):
    topic: str
    sources: list[str]                              # Lead 拆解出的信源清单
    # 关键：reducer 把 N 个并行 worker 的回写合并成一个 list，而非互相覆盖
    findings: Annotated[list[Finding], operator.add]
    report: str

llm = ChatAnthropic(model="claude-sonnet-4-5")

# 每个 worker 是一个独立 create_react_agent —— 自带隔离上下文（D3）
def make_worker(source_name: str, tools: list):
    agent = create_react_agent(llm, tools=tools)
    def worker(state: dict) -> dict:
        # state 里只有这条信源的 task，看不到别的 worker 的检索流水（D3 隔离）
        prompt = f"就主题「{state['topic']}」深挖信源 {source_name}，每条结论附 source_url。"
        try:
            result = agent.invoke({"messages": [("user", prompt)]})
            findings = parse_findings(result, source_name)   # 解析为 Finding 列表
        except Exception:
            # 单 worker 失败降级为占位，不冒泡终止全局（D4 · 04 §4.10）
            findings = [Finding(source_name=source_name,
                                source_url="", claim="（该信源本次不可用）")]
        return {"findings": findings}   # 经 reducer 安全并入共享 state
    return worker

# 2 │ Lead 只拆解 + 路由，不亲自检索（D4 · 04 §4.7 防瓶颈）
def lead(state: ResearchState) -> dict:
    # 真实场景这里用 LLM 把 topic 拆成信源清单；此处写死演示判别落点
    return {"sources": ["官方博客", "GitHub", "媒体评测", "社区论坛"]}

# 条件边：N 个 Send = N 个 worker 真正并行（D1 supervisor + 并行 worker）
def fan_out(state: ResearchState):
    return [Send(s, {"topic": state["topic"]}) for s in state["sources"]]

# Writer 看 N 份结构化子报告，不看原始网页（D3）
def writer(state: ResearchState) -> dict:
    cited = "\n".join(f"- {f['claim']}  [来源]({f['source_url']})"
                      for f in state["findings"] if f["source_url"])
    return {"report": synthesize(state["topic"], cited)}

g = StateGraph(ResearchState)
g.add_node("lead", lead)
g.add_node("官方博客", make_worker("官方博客", [web_search]))
g.add_node("GitHub",  make_worker("GitHub",  [release_api]))
g.add_node("媒体评测", make_worker("媒体评测", [web_search]))
g.add_node("社区论坛", make_worker("社区论坛", [forum_search]))
g.add_node("writer", writer)

g.add_edge(START, "lead")
g.add_conditional_edges("lead", fan_out)          # fan-out：并行派 4 个 worker
for s in ["官方博客", "GitHub", "媒体评测", "社区论坛"]:
    g.add_edge(s, "writer")                       # 4 条汇到 writer，靠 reducer 合并 findings
g.add_edge("writer", END)
app = g.compile()