Method Execute

pkg/agent/workflow/parallel.go:69–109

Execute 并行执行所有子 Agent

(ctx context.Context, message string)

// Execute 并行执行所有子 Agent
func (a *ParallelAgent) Execute(ctx context.Context, message string) *stream.Reader[*session.Event] {
	reader, writer := stream.Pipe[*session.Event](len(a.subAgents) * 10)

	go func() {
		defer writer.Close()

		var (
			eg, egCtx = errgroup.WithContext(ctx)
			resultsCh = make(chan result, len(a.subAgents)*10) // 缓冲通道
			doneCh    = make(chan struct{})
		)

		// 启动所有子 Agent
		for i, subAgent := range a.subAgents {
			sa := subAgent
			branch := fmt.Sprintf("%s.%s", a.name, sa.Name())
			index := i

			eg.Go(func() error {
				return a.runSubAgent(egCtx, sa, branch, index, message, resultsCh, doneCh)
			})
		}

		// 等待所有子 Agent 完成
		go func() {
			_ = eg.Wait() // 错误已通过 resultsCh 传递
			close(resultsCh)
		}()

		// 流式返回结果
		defer close(doneCh)

		for res := range resultsCh {
			if writer.Send(res.event, res.err) {
				return // 客户端取消
			}
		}
	}()

	return reader
}

// runSubAgent 运行单个子 Agent
func (a *ParallelAgent) runSubAgent(