🎯 多 Agent：Supervisor 编排 ★

def create_agent(
    model: str | BaseChatModel,
    tools: Sequence[BaseTool | Callable | dict] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware] = (),     # ★ 扩展点：中间件链
    response_format: ... = None,
    state_schema: type[AgentState] | None = None,
    context_schema: type[ContextT] | None = None,
    checkpointer: Checkpointer | None = None,
    store: BaseStore | None = None,
    interrupt_before: list[str] | None = None,
    interrupt_after: list[str] | None = None,
    name: str | None = None,                         # ★ name 用于子图嵌入
    ...
) -> CompiledStateGraph:
    """创建一个循环调用工具直到停止的 agent 图。"""

@dataclass
class Command(Generic[N], ToolOutputMixin):
    """更新图状态并发送消息的指令。"""

    graph: GraphType = None
    """发送到哪个图：
    - None：当前图
    - Command.PARENT：最近的父图  ← 子图向父图发命令"""

    update: ... = None      # 状态更新
    resume: ... = None      # 恢复中断
    goto: ... = None        # 跳转到节点

    PARENT: ClassVar[Literal["__parent__"]] = "__parent__"   # :808 常量定义

# 在 _control_branch 内：
if command.graph == Command.PARENT:
    raise ParentCommand(command)
    # 子图无法自己处理 PARENT 命令，所以【抛异常】
    # 异常会冒泡到父图，由父图的 _get_root (state.py:1780/1791) 接收处理
    # 这就是子→父通信的底层机制

# pip install langgraph langchain-openai
# export OPENAI_API_KEY="sk-..."
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

# ============ ① 定义共享状态 ============
class State(TypedDict):
    messages: Annotated[list, operator.add]   # 消息累积（各 Agent 的输出追加）
    next: str                                  # Supervisor 决定的下一个 Agent

# ============ ② 定义专家 Agent（用 create_react_agent 构建）============
# 每个 Agent 有自己的工具和系统提示，专注一域
research_agent = create_react_agent(
    model, tools=[],  # 实际可加搜索工具
    prompt="你是研究专家。负责调研技术方案、收集信息。简洁回答。",
)
coding_agent = create_react_agent(
    model, tools=[],  # 实际可加文件读写工具
    prompt="你是编码专家。负责写代码实现。只输出关键代码。",
)

def research_node(state: State) -> dict:
    """把专家 Agent 包成图节点。"""
    result = research_agent.invoke({"messages": state["messages"]})
    # 只取最后一条 AI 消息，标注来源
    last = result["messages"][-1]
    last.content = f"[研究专家] {last.content}"
    return {"messages": [last]}

def coding_node(state: State) -> dict:
    result = coding_agent.invoke({"messages": state["messages"]})
    last = result["messages"][-1]
    last.content = f"[编码专家] {last.content}"
    return {"messages": [last]}

# ============ ③ Supervisor 节点（用 LLM 做路由决策）============
from langchain_core.messages import HumanMessage, SystemMessage

def supervisor_node(state: State) -> dict:
    """Supervisor：用 LLM 决定下一步派给谁，或结束。"""
    decision = model.invoke([
        SystemMessage(content=(
            "你是项目主管，协调研究专家和编码专家。"
            "根据当前进度，决定下一步：\n"
            "- 需要调研 → 回复 FINISH 或 RESEARCH 或 CODE\n"
            "- 需要编码 → 回复 CODE\n"
            "- 任务完成 → 回复 FINISH\n"
            "只回复一个词。"
        )),
        *state["messages"],
    ])
    choice = decision.content.strip().upper()
    # 映射到节点名
    next_node = {"RESEARCH": "research", "CODE": "coding", "FINISH": FINISH}.get(choice, FINISH)
    return {"next": next_node}   # 条件边会读这个字段

# ============ ④ 构建图 ============
graph = StateGraph(State)
graph.add_node("supervisor", supervisor_node)
graph.add_node("research", research_node)
graph.add_node("coding", coding_node)

# 入口：先问 Supervisor
graph.add_edge(START, "supervisor")

# 条件边：Supervisor 根据 next 字段路由
def route_from_supervisor(state: State) -> str:
    return state["next"]   # 返回 "research" / "coding" / FINISH

graph.add_conditional_edges(
    "supervisor", route_from_supervisor,
    {"research": "research", "coding": "coding", "FINISH": END},
)
# 专家执行完回到 Supervisor 继续决策（形成循环）
graph.add_edge("research", "supervisor")
graph.add_edge("coding", "supervisor")

app = graph.compile()

# ============ ⑤ 运行 ============
result = app.invoke({
    "messages": [HumanMessage(content="帮我做一个简单的待办事项 CLI 工具")],
    "next": "",
})
print("\n===== 协作过程 =====")
for m in result["messages"]:
    print(f"  {m.content[:80]}")

# pip install langgraph-supervisor  ← 独立包，不在主仓库
from langgraph_supervisor import create_supervisor
# create_supervisor 把上面的手写逻辑封装成一行
# 但理解了手写版，你才知道它在做什么

# 父图状态
class ParentState(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

# 子图状态（可以和父图不同！只暴露必要字段）
class ResearchState(TypedDict):
    messages: Annotated[list, operator.add]
    research_notes: list   # ← 私有字段，不回传父图

# 子图编译时，只共享 schema 重叠的字段（messages）

def supervisor_with_parallel(state):
    """Supervisor 决定并行派给多个专家。"""
    return [
        Send("research", {"messages": state["messages"]}),
        Send("coding", {"messages": state["messages"]}),
    ]   # 两个专家并发执行，结果由 operator.add reducer 聚合