🌳 多 Agent：层级树 + Handoff

// task 工具执行时，创建一个【子会话】，绑定到当前会话
const nextSession = session ?? (yield* sessions.create({
  parentID: ctx.sessionID,    // ★ 父子关系建立点！parentID 指向当前会话
  title: params.description + ` (@${next.name} subagent)`,
  agent: next.name,           // 用哪个子 agent（explore/general/...）
  permission: [
    ...childPermission,       // 派生的子权限（下面讲）
    ...childToolDenies,       // 额外禁止的工具
  ],
}))
// 每个 task 调用 = 创建一个子会话节点，parentID 指向父
// 多次 task 调用 = 形成一棵会话树

export function deriveSubagentSessionPermission(input) {
  // 合并父会话的 deny 规则 + external_directory 规则
  // 并【额外禁止】：
  return Permission.merge(parentRuleset, {
    todowrite: "deny",   // 子 agent 不改待办
    task: "deny",        // ★ 关键：禁止子 agent 再派生子 agent！防递归爆炸
  })
}
// canTask/canTodo 检测：仅当子 agent 自己显式允许 task 才放行（默认 deny）

// 模型继承：子 agent 没指定模型就用父会话的
const model = next.model ?? {
  modelID: msg.info.modelID,
  providerID: msg.info.providerID,
}

// 同步回流：子 agent 跑完，取最后一条 text part 作为 task 工具返回值
const runTask = Effect.fn("TaskTool.runTask")(function* () {
  const parts = yield* ops.resolvePromptParts(params.prompt)
  const result = yield* ops.prompt({ messageID: ..., ... })
  return result.parts.findLast((p) => p.type === "text").text   // ★ 同步回流
})
// 主 agent 收到这段文本，作为 task 工具的结果，继续推理

# pip install langgraph langchain-openai
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

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

class State(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

# ============ 底层：工人节点（实际干活）============
def researcher(state: State) -> dict:
    resp = model.invoke([SystemMessage(content="你是研究员，简洁回答。"), *state["messages"]])
    return {"messages": [AIMessage(content=f"[研究] {resp.content}")]}

def coder(state: State) -> dict:
    resp = model.invoke([SystemMessage(content="你是程序员，简洁回答。"), *state["messages"]])
    return {"messages": [AIMessage(content=f"[编码] {resp.content}")]}

def tester(state: State) -> dict:
    resp = model.invoke([SystemMessage(content="你是测试员，简洁回答。"), *state["messages"]])
    return {"messages": [AIMessage(content=f"[测试] {resp.content}")]}

# ============ 中层：组长（管一组工人）============
def make_team_lead(members: dict, lead_name: str):
    """工厂：创建一个组长节点（mini-supervisor）。"""
    def team_lead(state: State) -> dict:
        resp = model.invoke([
            SystemMessage(content=(
                f"你是{lead_name}组长。根据进度决定下一步："
                f"可选 {list(members.keys())} 或 DONE。只回一个词。"
            )),
            *state["messages"],
        ])
        choice = resp.content.strip()
        next_node = members.get(choice, "DONE")
        return {"next": next_node}
    return team_lead

# 研究组：组长管 researcher
research_members = {"RESEARCH": "researcher", "DONE": "DONE"}
# 开发组：组长管 coder + tester
dev_members = {"CODE": "coder", "TEST": "tester", "DONE": "DONE"}

# ============ 顶层：主管（管组长）============
def top_supervisor(state: State) -> dict:
    resp = model.invoke([
        SystemMessage(content=(
            "你是项目主管，协调【研究组长】和【开发组长】。"
            "决定下一步：RESEARCH_LEAD（研究组）/ DEV_LEAD（开发组）/ DONE。只回一个词。"
        )),
        *state["messages"],
    ])
    mapping = {"RESEARCH_LEAD": "research_lead", "DEV_LEAD": "dev_lead", "DONE": "DONE"}
    return {"next": mapping.get(resp.content.strip(), "DONE")}

# ============ 构建三层图 ============
graph = StateGraph(State)
# 顶层
graph.add_node("top_supervisor", top_supervisor)
# 中层（组长）
graph.add_node("research_lead", make_team_lead(research_members, "研究"))
graph.add_node("dev_lead", make_team_lead(dev_members, "开发"))
# 底层（工人）
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("tester", tester)

# 顶层路由
graph.add_edge(START, "top_supervisor")
graph.add_conditional_edges("top_supervisor", lambda s: s["next"],
    {"research_lead": "research_lead", "dev_lead": "dev_lead", "DONE": END})
# 组长执行后回到顶层（形成层级循环）
graph.add_edge("research_lead", "top_supervisor")
graph.add_edge("dev_lead", "top_supervisor")
# 底层工人执行后回到各自的组长
graph.add_edge("researcher", "research_lead")
graph.add_edge("coder", "dev_lead")
graph.add_edge("tester", "dev_lead")

app = graph.compile()
# 注意：这个简化版组长没做条件边路由到工人（完整版需加）
# 实际生产中，每个组长内部也是一个 mini-supervisor 图（AD1 的结构）

result = app.invoke({"messages": [HumanMessage(content="做一个待办CLI工具")], "next": ""})

def safe_subtask(state):
    """子任务失败时，返回错误信息而非抛异常。"""
    try:
        result = sub_agent.invoke({"messages": state["messages"]})
        return {"messages": [result["messages"][-1]]}
    except Exception as e:
        # 隔离错误：父 Agent 收到"子任务失败"消息，可决定重试或换策略
        return {"messages": [AIMessage(content=f"[子任务失败: {e}]，请重试或调整")]}