Method ExecuteStream

pkg/workflow/router.go:196–392

ExecuteStream 流式执行 Router - 支持实时事件流

(ctx context.Context, input *StepInput, streamEvents bool)

// ExecuteStream 流式执行 Router - 支持实时事件流
func (r *Router) ExecuteStream(ctx context.Context, input *StepInput, streamEvents bool) *stream.Reader[any] {
	reader, writer := stream.Pipe[any](10)

	go func() {
		defer writer.Close()
		startTime := time.Now()
		routerID := r.id

		// 调用选择器函数
		stepsToExecute := r.selector(input)

		// 发送 Router 开始事件
		if streamEvents {
			startEvent := &RouterEvent{
				Type:          RouterEventStarted,
				RouterName:    r.name,
				SelectedSteps: getStepNames(stepsToExecute),
				ExecutedSteps: 0,
				Timestamp:     startTime,
			}
			if writer.Send(startEvent, nil) {
				return
			}
		}

		if len(stepsToExecute) == 0 {
			// 没有选中任何步骤
			if streamEvents {
				completeEvent := &RouterEvent{
					Type:          RouterEventCompleted,
					RouterName:    r.name,
					SelectedSteps: []string{},
					ExecutedSteps: 0,
					Timestamp:     time.Now(),
				}
				writer.Send(completeEvent, nil)
			}

			output := &StepOutput{
				StepID:      routerID,
				StepName:    r.name,
				StepType:    StepTypeRouter,
				Content:     fmt.Sprintf("Router %s: no steps selected", r.name),
				StartTime:   startTime,
				EndTime:     time.Now(),
				NestedSteps: []*StepOutput{},
				Metadata: map[string]any{
					"selected_steps": 0,
				},
				Metrics: &StepMetrics{ExecutionTime: time.Since(startTime).Seconds()},
			}
			output.Duration = output.EndTime.Sub(output.StartTime).Seconds()
			writer.Send(output, nil)
			return
		}

		// 收集所有执行结果
		allResults := make([]*StepOutput, 0, len(stepsToExecute))