M08 多智能体系统

M05 单进程编排：                      M08 多智能体系统：
   [主流程]                          [Agent A] ⇄ [MessageBus] ⇄ [Agent B]
    │ 调用                                │           ↕            │
    ├─► worker1()  (函数)                 └────► [Agent C] ◄───────┘
    └─► worker2()  (函数)              (各自独立、并发运行、消息通信)
   控制流在你手里                       控制流分散在 Agent 与拓扑里

package mas // multi-agent system

type Message struct {
	From    string         // 发送者 Agent 名
	To      string         // 收件人；空或 "*" 表示广播
	Content string         // 文本内容
	Meta    map[string]any // 可选的结构化负载
}

package mas

import (
	"context"
	"fmt"
	"sync"
)

type MessageBus struct {
	mu      sync.RWMutex
	inboxes map[string]chan Message
	buffer  int
}

func NewMessageBus(buffer int) *MessageBus {
	return &MessageBus{inboxes: make(map[string]chan Message), buffer: buffer}
}

// Subscribe 为某个 Agent 注册收件箱，返回只读 channel。
func (b *MessageBus) Subscribe(name string) <-chan Message {
	b.mu.Lock()
	defer b.mu.Unlock()
	ch := make(chan Message, b.buffer)
	b.inboxes[name] = ch
	return ch
}

// Publish 把消息路由给收件人；广播则发给除自己外的所有人。
func (b *MessageBus) Publish(ctx context.Context, msg Message) error {
	b.mu.RLock()
	defer b.mu.RUnlock()

	send := func(ch chan Message) error {
		select {
		case ch <- msg:
			return nil
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	if msg.To != "" && msg.To != "*" {
		ch, ok := b.inboxes[msg.To]
		if !ok {
			return fmt.Errorf("收件人不存在: %s", msg.To)
		}
		return send(ch)
	}
	for name, ch := range b.inboxes { // 广播
		if name == msg.From {
			continue
		}
		if err := send(ch); err != nil {
			return err
		}
	}
	return nil
}

func (b *MessageBus) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, ch := range b.inboxes {
		close(ch)
	}
	b.inboxes = make(map[string]chan Message)
}

func RunBusAgent(ctx context.Context, bus *MessageBus, name string, handle func(context.Context, Message) (*Message, error)) {
	inbox := bus.Subscribe(name)
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case msg, ok := <-inbox:
				if !ok {
					return // 总线关闭
				}
				reply, err := handle(ctx, msg)
				if err != nil || reply == nil {
					continue
				}
				_ = bus.Publish(ctx, *reply)
			}
		}
	}()
}

有多 Agent 协作需求
        │
        ├─ 任务结构能画清流程图?
        │     │是 → Pipeline(简单、高吞吐)
        │
        ├─ 大任务可分解为多个独立子任务并行?
        │     │是 → Orchestrator + 隔离子代理(并行最快)
        │
        ├─ 路径动态、需要灵活调度?
        │     │是 → Supervisor(适合大多数复杂任务)
        │
        ├─ 高风险决策、要多方论证?
        │     │是 → Debate(成本高，适合多方复核)
        │
        └─ 多个清晰角色接力转交?
              │是 → Swarm(职责边界清的接力场景)

要解决一个复杂任务
        │
        ├─ 路径完全已知 + 步数 < 5?
        │     │是 → 固定工作流
        │           使用普通代码实现
        │
        ├─ 路径不确定 + 单 Agent 能承担?
        │     │是 → 单 Agent + 工具(M04)
        │           优先采用的方案
        │
        ├─ 单 Agent 出现前述限制?
        │     │是 → 多 Agent
        │           按前面的决策树挑拓扑
        │
        └─ 评估发现多 Agent 比单 Agent 反而差?
              │是 → 退回单 Agent
                    多 Agent 不是默认更优的方案

package mas

import (
	"context"

	"github.com/yourname/llmagent/internal/llm"
)

func chat(ctx context.Context, p llm.Provider, model, system, user string) (string, error) {
	resp, err := p.Chat(ctx, llm.ChatRequest{
		Model: model,
		Messages: []llm.Message{
			{Role: llm.RoleSystem, Content: system},
			{Role: llm.RoleUser, Content: user},
		},
	})
	if err != nil {
		return "", err
	}
	return resp.Content, nil
}

package mas

import (
	"context"
	"fmt"
	"strings"

	"github.com/yourname/llmagent/internal/agent"
	"github.com/yourname/llmagent/internal/llm"
)

type SupervisorDecision struct {
	Next   string `json:"next"`   // 下一个成员名；"FINISH" 表示结束
	Reason string `json:"reason"`
}

type Supervisor struct {
	Provider llm.Provider
	Model    string
	Workers  map[string]*agent.Agent // 成员名 -> Agent
	MaxTurns int
}

func (s *Supervisor) decide(ctx context.Context, task, progress string) (SupervisorDecision, error) {
	names := make([]string, 0, len(s.Workers))
	for n := range s.Workers {
		names = append(names, n)
	}
	system := fmt.Sprintf(
		"你是团队主管。可调度的成员有：%s。根据任务和当前进展，决定下一步该谁来做；"+
			"若任务已完成则输出 FINISH。严格只输出 JSON：{\"next\":\"成员名或FINISH\",\"reason\":\"理由\"}",
		strings.Join(names, ", "))
	out, err := chat(ctx, s.Provider, s.Model, system, "任务："+task+"\n\n当前进展：\n"+progress)
	if err != nil {
		return SupervisorDecision{}, err
	}
	return llm.ParseInto[SupervisorDecision](out)
}

func (s *Supervisor) Run(ctx context.Context, task string) (string, error) {
	var progress strings.Builder
	for turn := 0; turn < s.MaxTurns; turn++ {
		d, err := s.decide(ctx, task, progress.String())
		if err != nil {
			return "", err
		}
		if d.Next == "FINISH" {
			return progress.String(), nil
		}
		w, ok := s.Workers[d.Next]
		if !ok {
			return "", fmt.Errorf("主管选了不存在的成员: %q", d.Next)
		}
		result, err := w.Run(ctx, task+"\n\n已有进展：\n"+progress.String())
		if err != nil {
			return "", err
		}
		fmt.Fprintf(&progress, "【%s】\n%s\n\n", d.Next, result)
	}
	return progress.String(), fmt.Errorf("达到最大轮次 %d 仍未 FINISH", s.MaxTurns)
}

package mas

import (
	"context"
	"fmt"
	"strings"

	"github.com/yourname/llmagent/internal/agent"
	"github.com/yourname/llmagent/internal/llm"
	"github.com/yourname/llmagent/internal/patterns" // 复用 M05 的 Sectioning
)

type IsolatedOrchestrator struct {
	Provider llm.Provider
	Model    string
	// NewSubagent 为某个角色现造一个全新 Agent —— 每次调用都是干净、隔离的上下文。
	NewSubagent func(role string) *agent.Agent
}

type subtask struct {
	Role string `json:"role"`
	Task string `json:"task"`
}
type plan struct {
	Subtasks []subtask `json:"subtasks"`
}

func (o *IsolatedOrchestrator) Run(ctx context.Context, goal string) (string, error) {
	// 1) 编排者分解任务（结构化输出）
	sys := "你是编排者。把任务拆成可独立完成的子任务，为每个子任务指定一个角色。" +
		"严格只输出 JSON：{\"subtasks\":[{\"role\":\"...\",\"task\":\"...\"}]}"
	raw, err := chat(ctx, o.Provider, o.Model, sys, goal)
	if err != nil {
		return "", err
	}
	p, err := llm.ParseInto[plan](raw)
	if err != nil {
		return "", fmt.Errorf("分解结果无法解析: %w", err)
	}

	// 2) 并行跑各子代理。每个子代理上下文隔离，只交回浓缩结论。
	jobs := make([]func(context.Context) (string, error), len(p.Subtasks))
	for i, st := range p.Subtasks {
		st := st
		jobs[i] = func(ctx context.Context) (string, error) {
			sub := o.NewSubagent(st.Role) // 全新、隔离的上下文
			result, err := sub.Run(ctx, st.Task)
			if err != nil {
				return "", fmt.Errorf("子代理[%s]失败: %w", st.Role, err)
			}
			return fmt.Sprintf("【%s 的结论】%s", st.Role, result), nil
		}
	}
	results, err := patterns.Sectioning(ctx, jobs)
	if err != nil {
		return "", err
	}

	// 3) 编排者只基于“浓缩结论”综合 —— 它的上下文从未被子代理内部细节污染
	return chat(ctx, o.Provider, o.Model,
		"下面是各子代理的结论，请综合成对原始目标的完整回答。",
		"目标："+goal+"\n\n各子代理结论：\n"+strings.Join(results, "\n\n"))
}

package mas

import (
	"context"
	"sync"
)

// Stage 是流水线的一个阶段：从 in 读取，用 workers 个 goroutine 并行处理，结果写入返回的 channel。
func Stage[I, O any](ctx context.Context, in <-chan I, workers int, fn func(context.Context, I) (O, error)) <-chan O {
	out := make(chan O)
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for item := range in {
				o, err := fn(ctx, item)
				if err != nil {
					continue // 教学简化：实际应把错误送往专门的错误 channel
				}
				select {
				case out <- o:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() { wg.Wait(); close(out) }() // 所有 worker 退出后关闭输出
	return out
}

// 把工单流依次经过：分类 → 检索 → 起草 → 质检
classified := Stage(ctx, tickets, 4, classifyAgent)   // 4 个并行做分类
retrieved := Stage(ctx, classified, 2, retrieveAgent)
drafted := Stage(ctx, retrieved, 4, draftAgent)
reviewed := Stage(ctx, drafted, 2, reviewAgent)

for r := range reviewed { // 消费最终结果
	save(r)
}

package mas

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"github.com/yourname/llmagent/internal/llm"
)

type Debater struct {
	Name     string
	Provider llm.Provider
	Model    string
	Persona  string // 该辩手的立场/视角，作为 system 提示词
}

// Debate 进行 rounds 轮辩论，返回每位辩手的最终答案。
func Debate(ctx context.Context, debaters []Debater, question string, rounds int) (map[string]string, error) {
	answers := make(map[string]string)
	var mu sync.Mutex

	for round := 0; round < rounds; round++ {
		// 快照本轮开始时的答案，供各辩手参考（避免并发读写同一 map）
		mu.Lock()
		prev := make(map[string]string, len(answers))
		for k, v := range answers {
			prev[k] = v
		}
		mu.Unlock()

		var wg sync.WaitGroup
		errCh := make(chan error, len(debaters))
		for _, d := range debaters {
			d := d
			wg.Add(1)
			go func() {
				defer wg.Done()
				ans, err := chat(ctx, d.Provider, d.Model, d.Persona, debatePrompt(question, d.Name, prev))
				if err != nil {
					errCh <- err
					return
				}
				mu.Lock()
				answers[d.Name] = ans
				mu.Unlock()
			}()
		}
		wg.Wait()
		close(errCh)
		if err := <-errCh; err != nil { // 取第一个错误（channel 空且关闭时返回 nil）
			return nil, err
		}
	}
	return answers, nil
}

func debatePrompt(question, self string, prev map[string]string) string {
	if len(prev) == 0 {
		return "问题：" + question + "\n请给出你的回答和理由。"
	}
	var sb strings.Builder
	sb.WriteString("问题：" + question + "\n\n其他成员上一轮的观点：\n")
	for name, ans := range prev {
		if name == self {
			continue
		}
		fmt.Fprintf(&sb, "- %s：%s\n", name, ans)
	}
	sb.WriteString("\n请批判性地参考他们的观点，修订并强化你的回答。")
	return sb.String()
}

// Judge 综合各辩手的最终答案，给出定稿。
func Judge(ctx context.Context, p llm.Provider, model, question string, answers map[string]string) (string, error) {
	var sb strings.Builder
	for name, ans := range answers {
		fmt.Fprintf(&sb, "【%s】%s\n\n", name, ans)
	}
	return chat(ctx, p, model,
		"你是评审。综合下面各位专家的最终回答，给出最准确、全面、平衡的定稿。",
		"问题："+question+"\n\n各方回答：\n"+sb.String())
}

package mas

import (
	"context"
	"fmt"
)

// SwarmResult：Agent 要么给出答案，要么转交给下一个。
type SwarmResult struct {
	Answer    string // 非空表示给出最终答案
	HandoffTo string // 非空表示转交给该 Agent
}

type SwarmAgent struct {
	Name string
	Run  func(ctx context.Context, input string) (SwarmResult, error)
}

type Swarm struct {
	Agents  map[string]SwarmAgent
	MaxHops int // 最大转交次数，防止无限转交
}

func (s *Swarm) Run(ctx context.Context, start, input string) (string, error) {
	cur := start
	for hop := 0; hop < s.MaxHops; hop++ {
		a, ok := s.Agents[cur]
		if !ok {
			return "", fmt.Errorf("不存在的 Agent: %q", cur)
		}
		res, err := a.Run(ctx, input)
		if err != nil {
			return "", err
		}
		if res.HandoffTo == "" {
			return res.Answer, nil // 有人给出了答案，结束
		}
		// 转交：把当前的处理结果作为上下文带给下一个 Agent
		cur = res.HandoffTo
		if res.Answer != "" {
			input = input + "\n\n[" + a.Name + " 的处理]：" + res.Answer
		}
	}
	return "", fmt.Errorf("转交超过 %d 次仍无最终答案", s.MaxHops)
}

func main() {
	ctx := context.Background()
	question := "我们准备用单体架构起步、后期再拆微服务，这个决策有哪些风险？"

	debaters := []mas.Debater{
		{Name: "务实派", Provider: p, Model: m, Persona: "你务实，关注落地速度与团队现状。"},
		{Name: "谨慎派", Provider: p, Model: m, Persona: "你谨慎，关注长期可维护性与技术债。"},
		{Name: "数据派", Provider: p, Model: m, Persona: "你重证据，倾向用事实和案例说话。"},
	}

	answers, err := mas.Debate(ctx, debaters, question, 3)
	if err != nil { /* ... */ }

	final, err := mas.Judge(ctx, p, m, question, answers)
	if err != nil { /* ... */ }
	fmt.Println("=== 定稿 ===\n", final)
}