从单兵作战到群智协作：Multi-agent 架构演进与思考

用户 → [售前客服 Agent]
             │
             ├── "我要退货" → 移交给 [售后 Agent]
             ├── "系统报错了" → 移交给 [技术支持 Agent]
             └── "推荐个产品" → 自己处理

// transfer_to_agent 工具：只打标记，不直接执行
func (t *TransferTool) Call(ctx context.Context, jsonArgs []byte) (any, error) {
    var req Request
    json.Unmarshal(jsonArgs, &req)

    // 从上下文获取当前 invocation
    invocation, _ := agent.InvocationFromContext(ctx)

    // 只在 invocation 上打一个标记，标记"要移交给谁"
    invocation.TransferInfo = &agent.TransferInfo{
        TargetAgentName: req.AgentName,
        Message:         req.Message,
    }

    // 立即返回，不在这里启动目标 Agent
    return Response{
        Success:     true,
        Message:     "Transfer initiated to agent: " + req.AgentName,
        TransferType: "agent_handoff",
    }, nil
}

// TransferResponseProcessor：在管线的正确阶段执行移交
func (p *TransferResponseProcessor) ProcessResponse(
    ctx context.Context,
    invocation *agent.Invocation,
    req *model.Request,
    rsp *model.Response,
    ch chan<- *event.Event,
) {
    // 检查是否有待处理的移交标记
    if invocation.TransferInfo == nil {
        return// 没有移交请求，正常继续
    }

    targetAgentName := invocation.TransferInfo.TargetAgentName

    // 查找目标 Agent
    targetAgent := invocation.Agent.FindSubAgent(targetAgentName)

    // 安全校验：通过 TransferController 检查循环移交、次数限制等
    if controller, ok := agent.GetRuntimeStateValue[agent.TransferController](
        &invocation.RunOptions, agent.RuntimeStateKeyTransferController,
    ); ok && controller != nil {
        nodeTimeout, err := controller.OnTransfer(ctx, invocation.AgentName, targetAgentName)
        if err != nil {
            // 移交被拒绝（如循环检测触发）
            return
        }
    }

    // 创建目标 Agent 的 invocation
    targetInvocation := invocation.Clone(
        agent.WithInvocationAgent(targetAgent),
    )

    // 在正确的上下文中启动目标 Agent
    targetEventChan, _ := agent.RunWithPlugins(ctx, targetInvocation, targetAgent)

    // 将目标 Agent 的事件原样转发到最外层
    for targetEvent := range targetEventChan {
        event.EmitEvent(ctx, ch, targetEvent)
    }

    // 结束原 Agent 的 invocation
    invocation.TransferInfo = nil
    invocation.EndInvocation = true
}

用户 → [协调者 Agent]
             │
             ├── 调用 [需求分析师] → "需要处理高并发..."
             ├── 调用 [方案设计师] → "建议用 Redis 预扣减..."
             ├── 调用 [质量审查员] → "风险：Redis 宕机无降级..."
             ├── 再次调用 [方案设计师] → "补充降级方案..."
             │
             └── 协调者综合所有信息 → 最终答案

// 创建 Coordinator Team 时，成员被包装成 Tool
func newMemberToolSet(members []agent.Agent) tool.ToolSet {
    tools := make([]tool.Tool, 0, len(members))
    for _, m := range members {
        // 把每个成员 Agent 包装成一个 AgentTool
        tools = append(tools, agenttool.NewTool(m))
    }
    return &staticToolSet{tools: tools}
}

// 协调者直接委托给 coordinator Agent，它会通过 function calling 调度成员
func (t *Team) runCoordinator(ctx context.Context, invocation *agent.Invocation) (<-chan *event.Event, error) {
    return t.coordinator.Run(ctx, invocation)
}

// AgentTool.Call：协调者通过 function calling 调用成员 Agent
func (at *Tool) Call(ctx context.Context, jsonArgs []byte) (any, error) {
    // LLM 传来的参数作为用户消息发给子 Agent
    message := model.NewUserMessage(string(jsonArgs))

    // 复用父 invocation 的 session，让子 Agent 能看到对话历史
    if parentInv, ok := agent.InvocationFromContext(ctx); ok && parentInv.Session != nil {
        // 克隆父 invocation，设置子 Agent 自己的事件过滤键
        subInv := parentInv.Clone(
            agent.WithInvocationAgent(at.agent),
            agent.WithInvocationMessage(message),
            agent.WithInvocationEventFilterKey(childKey),
        )

        // 运行子 Agent
        subCtx := agent.NewInvocationContext(ctx, subInv)
        evCh, _ := agent.RunWithPlugins(subCtx, subInv, at.agent)

        // 收集子 Agent 的所有响应，拼成一个字符串返回给协调者
        return at.collectResponse(evCh)
    }

    // 没有父上下文时，创建一个隔离的执行环境
    return at.callWithIsolatedRunner(ctx, message)
}

[Agent A] ←→ [Agent B]
    ↕              ↕
[Agent C] ←→ [Agent D]

每个 Agent 都知道其他成员的存在，
自主决定把对话传递给谁。

// 创建 Swarm Team
func NewSwarm(name, entryName string, members []agent.Agent) (*Team, error) {
    // 让每个成员知道其他所有成员——互相设置为 SubAgent
    wireSwarmRoster(members)

    return &Team{
        mode:      ModeSwarm,
        entryName: entryName,
        members:   members,
    }, nil
}

// wireSwarmRoster：为每个成员设置其他所有成员为 SubAgent
func wireSwarmRoster(members []agent.Agent) error {
    for _, m := range members {
        setter := m.(agent.SubAgentSetter)
        roster := make([]agent.Agent, 0, len(members)-1)
        for _, other := range members {
            if other.Info().Name != m.Info().Name {
                roster = append(roster, other) // 除了自己以外的所有成员
            }
        }
        setter.SetSubAgents(roster)
    }
    returnnil
}

// 运行 Swarm：从入口成员开始，后续通过 transfer_to_agent 自由流转
func (t *Team) runSwarm(ctx context.Context, invocation *agent.Invocation) (<-chan *event.Event, error) {
    entry := t.memberByName[t.entryName]

    // 注入 Swarm 运行时控制器（循环检测、次数限制等）
    ensureSwarmRuntime(invocation, t.swarm)

    child := invocation.Clone(agent.WithInvocationAgent(entry))
    return entry.Run(agent.NewInvocationContext(ctx, child), child)
}

// swarmRuntime 实现 TransferController 接口
func (sr *swarmRuntime) OnTransfer(ctx context.Context, fromAgent, toAgent string) (time.Duration, error) {
    sr.count++
    // 检查最大移交次数
    if sr.cfg.MaxHandoffs > 0 && sr.count > sr.cfg.MaxHandoffs {
        return0, errors.New("max handoffs exceeded")
    }
    // 检查循环移交（滑动窗口内的唯一 Agent 数量过低）
    sr.recent = append(sr.recent, toAgent)
    iflen(sr.recent) == sr.cfg.RepetitiveHandoffWindow &&
        uniqueCount(sr.recent) < sr.cfg.RepetitiveHandoffMinUnique {
        return0, errors.New("repetitive handoff detected")
    }
    return sr.cfg.NodeTimeout, nil
}

[规划 Agent] → [研究 Agent] → [写作 Agent] → [审校 Agent]
     ↓              ↓              ↓              ↓
   制定计划      搜集资料        起草文章        校对发布

// Chain 模式核心：按顺序遍历执行子 Agent
func (a *ChainAgent) executeSubAgents(ctx context.Context, invocation *agent.Invocation, ...) {
    for _, subAgent := range a.subAgents {
        // 从基础 invocation 克隆——共享同一个 Session
        // 上一个 Agent 的输出已经作为事件写入了 Session
        subInvocation := invocation.Clone(
            agent.WithInvocationAgent(subAgent),
        )

        // 直接调用子 Agent，不经过任何工具包装
        subEventChan, _ := agent.RunWithPlugins(ctx, subInvocation, subAgent)

        // 转发子 Agent 的所有事件
        for subEvent := range subEventChan {
            // 记录完整响应（下一个 Agent 可以通过 Session 看到）
            if subEvent.Response != nil && !subEvent.Response.IsPartial {
                fullRespEvent = subEvent
            }
            event.EmitEvent(ctx, eventChan, subEvent)
        }
    }
}

[生成 Agent] → [评估 Agent] ─── 不合格 ──→ [生成 Agent]（重来）
                    │
                    └── 合格 → 输出结果

// Cycle 模式核心：带退出条件的循环执行
func (a *CycleAgent) Run(ctx context.Context, invocation *agent.Invocation) (<-chan *event.Event, error) {
    // ...
    var timesLooped int

    // 主循环：直到达到最大迭代次数或触发退出条件
    for a.maxIterations == nil || timesLooped < *a.maxIterations {
        // 一轮循环 = 按顺序执行所有子 Agent
        if a.runSubAgentsLoop(ctx, invocation, eventChan, &fullRespEvent) {
            break// 升级条件触发，退出循环
        }
        timesLooped++
    }
}

// 升级条件判断：检查事件是否表明应该退出循环
func (a *CycleAgent) shouldEscalate(evt *event.Event) bool {
    // 支持自定义升级函数
    if a.escalationFunc != nil {
        return a.escalationFunc(evt)
    }
    // 默认：错误事件触发退出
    return evt.Error != nil
}

// 单个子 Agent 执行：边运行边检查退出条件
func (a *CycleAgent) runSubAgent(ctx context.Context, subAgent agent.Agent, ...) bool {
    subInvocation := invocation.Clone(agent.WithInvocationAgent(subAgent))
    subEventChan, _ := agent.RunWithPlugins(ctx, subInvocation, subAgent)

    for subEvent := range subEventChan {
        event.EmitEvent(ctx, eventChan, subEvent)
        // 每收到一个事件就检查：是否该退出了？
        if a.shouldEscalate(subEvent) {
            returntrue// 退出循环
        }
    }
    returnfalse// 继续下一轮
}

[开始] → [分类器] ─── 简单问题 ───→ [快速回答] → [结束]
              │
              └── 复杂问题 ──→ [研究] → [分析] ──┐
                                 ↑                │
                                 └── 需要更多信息 ─┘
                                              └── 完成 → [结束]

// Graph 中的 Agent 节点：直接调用 agent.Run()
func NewAgentNodeFunc(agentName string, opts ...Option) NodeFunc {
    returnfunc(ctx context.Context, state State) (any, error) {
        // 从全局 State 中查找目标 Agent
        targetAgent := findAgentFromState(state, agentName)

        // 可选：把上一个节点的 last_response 映射为当前节点的 user_input
        parentForInput := mapParentInputFromLastResponse(state, cfg.inputFromLast)

        // 构建子 invocation，注入当前状态
        invocation := buildAgentInvocation(ctx, parentForInput, childState, targetAgent)

        // 直接调用 Agent（不是工具！）
        agentEventChan, _ := targetAgent.Run(subCtx, invocation)

        // 处理事件流，收集最终输出
        streamRes := processAgentEventStream(ctx, agentEventChan, ...)

        // 返回状态更新——写入 last_response 和 node_responses
        return State{
            "last_response":  streamRes.lastResponse,
            "node_responses": map[string]any{nodeID: streamRes.lastResponse},
            "user_input":     "", // 清空，防止下游节点重复消费
        }, nil
    }
}

// 上一个节点的输出 → 当前节点的输入
func mapParentInputFromLastResponse(state State, enabled bool) State {
    if !enabled {
        return state
    }
    lastResponse := state["last_response"].(string) // 上一个节点的输出
    cloned := state.Clone()
    cloned["user_input"] = lastResponse // 变成当前节点的输入
    return cloned
}

// 添加条件边：根据 State 内容动态路由
graph.AddConditionalEdges("classifier",
    // 条件函数：读取 State，返回下一个节点
    func(ctx context.Context, state State) (ConditionResult, error) {
        msgs := state["messages"].([]model.Message)
        lastMsg := msgs[len(msgs)-1]
        if lastMsg.Content == "simple" {
            return ConditionResult{NextNodes: []string{"quick_answer"}}, nil
        }
        return ConditionResult{NextNodes: []string{"research"}}, nil
    },
    map[string]string{
        "quick_answer": "quick_answer",
        "research":     "research",
    },
)

         [CEO Agent]
        /           \
  [Tech Lead]    [PM Agent]
  /        \         |
[Coder] [Tester] [Designer]

type Agent interface {
    // 统一的执行契约：接收 invocation，返回事件流
    Run(ctx context.Context, invocation *Invocation) (<-chan *event.Event, error)
    Tools() []tool.Tool
    Info() Info
    SubAgents() []Agent
    FindSubAgent(name string) Agent
}