Lesson 09 — 毕业项目：多智能体研究系统（简历级）

用户给研究主题
      │
      ▼
┌─────────────────────────────────────────────────┐
│              父图（SystemState）                  │
│                                                   │
│  START → research_team ──→ writer ──→ END        │
│              │                    │               │
│              │  ⭐子图节点          │ glm-4 写报告  │
│              ▼                    │               │
│  ┌───────────────────────┐        │               │
│  │ 并行研究子图            │        │               │
│  │ (ResearchState)        │        │               │
│  │                         │        │               │
│  │  split ──(Send×3)──┐   │        │               │
│  │                    ▼   │        │               │
│  │  researcher   researcher  researcher  │         │
│  │  (glm-4-flash) (flash)   (flash)     │ 并行     │
│  │                    │                 │         │
│  │                    ▼                 │         │
│  │              summarize (glm-4)       │         │
│  └───────────────────────┘        │               │
│              │ findings 回流        │               │
│              └──────────────────────┘               │
└─────────────────────────────────────────────────────┘

class ResearchState(TypedDict):
    findings: Annotated[list[str], operator.add]  # ⭐ reducer 合并并行结果

def route_to_researchers(state):
    # 返回 3 个 Send = 触发 3 个 researcher 并行
    return [Send("researcher", {"subtopic": s}) for s in state["subtopics"]]

builder.add_conditional_edges("split", route_to_researchers)  # 并行 fan-out
builder.add_edge("researcher", "summarize")                   # 所有完成 → 汇总

def research_team(state):
    # 调用编译好的并行子图
    sub_result = research_subgraph.invoke({"topic": topic, ...})
    # 把子图结果回流到父图 State
    return {"findings": sub_result["findings"], "research_summary": sub_result["research_summary"]}

builder.add_node("research_team", research_team)  # 子图作为节点

smart_llm = ChatZhipuAI(model="glm-4", ...)        # 决策/写作
fast_llm = ChatZhipuAI(model="glm-4-flash", ...)   # 并行查询（免费）

# Checkpointer 跨轮记忆
system = builder.compile(checkpointer=InMemorySaver())

# 同 thread_id = 记忆延续
config = {"configurable": {"thread_id": "research-1"}}
result = system.invoke({...}, config=config)

python workflow-lessons/09_capstone/code.py

📘 RAG 手写（9 课）
   embedding → 检索 → 切块 → prompt → 混合检索 → 改写 → 评估 → 工程化 → 毕业
        │
        ▼
🤖 Agent 手写（9 课）
   Function Calling → ReAct → 工具设计 → 记忆 → 规划 → Agentic RAG → 多智能体 → 毕业
        │
        ▼
🔧 框架进阶（9 课）
   LCEL → 三件套 → Document → Retriever → StateGraph → create_agent → HITL → 毕业
        │
        ▼
🔀 多智能体编排（9 课）⭐ 本课程
   supervisor → swarm → 子图 → 并行 → 通信 → 多模型 → CrewAI → AutoGen → 毕业