上手实操：从「能讲」到「能跑」

# Python 3.11+ ；演示用，未本机执行
python -m venv .venv && source .venv/bin/activate

pip install \
  "langgraph>=1.0,<1.1" \
  "langchain-anthropic>=0.3" \
  "langchain-core>=0.3"

# LLM provider 凭证（换 OpenAI 则设 OPENAI_API_KEY 并装 langchain-openai）
export ANTHROPIC_API_KEY="sk-ant-..."

# 自检：导入不报错 = 装对了
python -c "import langgraph, langchain_anthropic; print('ok')"

# 演示用，未本机执行 —— LangGraph 1.0 / langchain-anthropic 0.3
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

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

# ── 共享 state：worker 各写不同字段，互不覆盖 ──
class ResearchState(TypedDict):
    topic: str            # 用户输入
    search_result: str    # search_worker 写
    outline: str          # summarize_worker 写
    answer: str           # 汇总节点写

@tool
def web_search(query: str) -> str:
    """查资料（演示桩；生产接 Tavily / Bing 等）。"""
    return f"[资料] 关于 '{query}' 的三条要点 ..."

# ── worker 1：search_worker（节点即 agent）──
def search_worker(state: ResearchState) -> dict:
    hits = web_search.invoke({"query": state["topic"]})
    note = llm.invoke(
        f"用三句话提炼这份资料的关键事实：\n{hits}"
    ).content
    return {"search_result": note}          # 只写自己那格

# ── worker 2：summarize_worker（与上一个并行）──
def summarize_worker(state: ResearchState) -> dict:
    outline = llm.invoke(
        f"为主题「{state['topic']}」列一个 3 点提纲，不要展开。"
    ).content
    return {"outline": outline}             # 写另一格，不碰 search_result

# ── supervisor 的派发：一次性 fan-out 两个 worker（并行）──
def dispatch(state: ResearchState):
    return [
        Send("search_worker", state),
        Send("summarize_worker", state),
    ]

# ── 汇总节点：两个 worker 都到齐后才跑，读两格、合成回答 ──
def aggregate(state: ResearchState) -> dict:
    merged = llm.invoke(
        "把以下资料与提纲合成一段连贯回答：\n\n"
        f"资料：{state['search_result']}\n\n提纲：{state['outline']}"
    ).content
    return {"answer": merged}

# ── 装配 graph ──
g = StateGraph(ResearchState)
g.add_node("search_worker", search_worker)
g.add_node("summarize_worker", summarize_worker)
g.add_node("aggregate", aggregate)

g.add_conditional_edges(START, dispatch, ["search_worker", "summarize_worker"])
g.add_edge("search_worker", "aggregate")      # 两个 worker 都连向汇总
g.add_edge("summarize_worker", "aggregate")   # aggregate 等两边都到齐
g.add_edge("aggregate", END)

app = g.compile()

result = app.invoke({"topic": "MCP 协议是什么"})
print(result["answer"])

# 演示用，未本机执行 —— 在 3.2 基础上抠掉 3 个决策
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain_anthropic import ChatAnthropic

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

# ─────────────────────────────────────────────
# 决策 (a) · worker 间走 shared-state 还是消息传递？
#   现在两个 worker 写 state 不同字段。若改成「summarize 要先读
#   search 的结果再出提纲」，state 该怎么改？字段怎么聚合？
# ─────────────────────────────────────────────
class ResearchState(TypedDict):
    topic: str
    # findings: Annotated[list[str], ???]   # 多 worker 写同一字段时填什么 reducer？
    answer: str

# ─────────────────────────────────────────────
# 决策 (b) · worker 要不要隔离上下文？
#   现在每个 worker 只拿到 state、各调各的 LLM（隔离）。
#   若让两个 worker 共享同一段对话历史，会省 token 还是更贵？质量呢？
# ─────────────────────────────────────────────
def search_worker(state: ResearchState) -> dict:
    # context_passed_in = ???   # 只传 topic，还是把别的 worker 的中间推理也塞进来？
    note = llm.invoke(f"提炼关于「{state['topic']}」的三条关键事实").content
    return {"findings": [note]}   # 配合 (a) 的 reducer

def summarize_worker(state: ResearchState) -> dict:
    outline = llm.invoke(f"为「{state['topic']}」列 3 点提纲").content
    return {"findings": [outline]}

def dispatch(state: ResearchState):
    return [Send("search_worker", state), Send("summarize_worker", state)]

# ─────────────────────────────────────────────
# 决策 (c) · 终止条件怎么定？
#   两个 worker 都写完才汇总——「都写完」用什么判定？
#   靠 graph 结构汇合（结构信号），还是数 findings 长度（内容信号）？
# ─────────────────────────────────────────────
def aggregate(state: ResearchState) -> dict:
    # if ???:   # 收齐了吗？
    merged = llm.invoke(
        "把以下片段合成一段连贯回答：\n" + "\n\n".join(state["findings"])
    ).content
    return {"answer": merged}

g = StateGraph(ResearchState)
g.add_node("search_worker", search_worker)
g.add_node("summarize_worker", summarize_worker)
g.add_node("aggregate", aggregate)
g.add_conditional_edges(START, dispatch, ["search_worker", "summarize_worker"])
g.add_edge("search_worker", "aggregate")
g.add_edge("summarize_worker", "aggregate")
g.add_edge("aggregate", END)
app = g.compile()