多智能体系统：什么时候值得用，如何可靠落地

请求 -> Manager -> [Worker A, Worker B, Worker C] -> Verifier -> 响应

分诊 -> 账单
     -> 账号恢复
     -> 技术支持

控制面
  任务分解、路由、预算、鉴权、重试、取消、
  Checkpoint、人工批准、审计

Worker 数据面
  模型推理、受限检索、已批准的 Tool Call、
  类型化 Result 与 Provenance

{
  "schema_version": "1",
  "run_id": "run-42",
  "task_id": "research-security",
  "attempt": 1,
  "sender": "manager",
  "recipient": "security-researcher",
  "deadline": "2026-10-04T12:00:00Z",
  "input_refs": ["artifact://brief/17"],
  "required_output": "evidence_bundle.v1"
}

package main

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"time"
)

type Assignment struct {
	ID            string
	Brief         string
	RequiredTools []string
}

type Evidence struct {
	ID      string
	Source  string
	Summary string
}

type Result struct {
	TaskID   string
	AgentID  string
	Evidence []Evidence
	Err      error
}

type Agent struct {
	ID           string
	AllowedTools map[string]bool
	Run          func(context.Context, Assignment) ([]Evidence, error)
}

type WorkItem struct {
	Agent      Agent
	Assignment Assignment
}

func (a Agent) Execute(
	ctx context.Context,
	assignment Assignment,
) Result {
	for _, tool := range assignment.RequiredTools {
		if !a.AllowedTools[tool] {
			return Result{
				TaskID:  assignment.ID,
				AgentID: a.ID,
				Err:     fmt.Errorf("tool %q is not allowed", tool),
			}
		}
	}

	evidence, err := a.Run(ctx, assignment)
	return Result{
		TaskID:   assignment.ID,
		AgentID:  a.ID,
		Evidence: evidence,
		Err:      err,
	}
}

func runParallel(
	ctx context.Context,
	work []WorkItem,
) ([]Result, error) {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	results := make(chan Result, len(work))
	var waitGroup sync.WaitGroup
	for _, item := range work {
		waitGroup.Add(1)
		go func(agent Agent, assignment Assignment) {
			defer waitGroup.Done()
			results <- agent.Execute(ctx, assignment)
		}(item.Agent, item.Assignment)
	}

	go func() {
		waitGroup.Wait()
		close(results)
	}()

	collected := make([]Result, 0, len(work))
	for result := range results {
		if result.Err != nil {
			cancel()
			return nil, fmt.Errorf(
				"%s failed: %w",
				result.AgentID,
				result.Err,
			)
		}
		if result.TaskID == "" || len(result.Evidence) == 0 {
			cancel()
			return nil, errors.New("invalid worker result")
		}
		collected = append(collected, result)
	}

	sort.Slice(collected, func(i, j int) bool {
		return collected[i].TaskID < collected[j].TaskID
	})
	return collected, nil
}

func mockRunner(
	agentID string,
	source string,
) func(context.Context, Assignment) ([]Evidence, error) {
	return func(ctx context.Context, assignment Assignment) ([]Evidence, error) {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}
		return []Evidence{{
			ID:      assignment.ID + ":1",
			Source:  source,
			Summary: agentID + " completed: " + assignment.Brief,
		}}, nil
	}
}

func main() {
	marketAgent := Agent{
		ID:           "market-researcher",
		AllowedTools: map[string]bool{"public_search": true},
		Run:          mockRunner("market-researcher", "public://market"),
	}
	riskAgent := Agent{
		ID:           "risk-reviewer",
		AllowedTools: map[string]bool{"policy_read": true},
		Run:          mockRunner("risk-reviewer", "policy://controls"),
	}

	work := []WorkItem{
		{
			Agent: marketAgent,
			Assignment: Assignment{
				ID:            "market",
				Brief:         "collect demand evidence",
				RequiredTools: []string{"public_search"},
			},
		},
		{
			Agent: riskAgent,
			Assignment: Assignment{
				ID:            "risk",
				Brief:         "identify launch controls",
				RequiredTools: []string{"policy_read"},
			},
		},
	}

	results, err := runParallel(context.Background(), work)
	if err != nil {
		panic(err)
	}
	for _, result := range results {
		fmt.Printf("%s: %s\n", result.TaskID, result.Evidence[0].Summary)
	}
}

run -> assignment -> model call -> tool call -> artifact
    -> handoff/retry -> review -> final answer