Method Execute

pkg/workflow/workflow.go:309–575

===== 执行 ===== Execute 执行 Workflow

(ctx context.Context, input *WorkflowInput)

// Execute 执行 Workflow
func (w *Workflow) Execute(ctx context.Context, input *WorkflowInput) *stream.Reader[*RunEvent] {
	reader, writer := stream.Pipe[*RunEvent](10)

	go func() {
		defer writer.Close()
		// 验证输入
		if err := w.ValidateInput(input.Input); err != nil {
			writer.Send(nil, fmt.Errorf("input validation failed: %w", err))
			return
		}

		// 初始化会话
		sessionID, userID := w.InitializeSession(input.SessionID, input.UserID)
		session := w.GetOrCreateSession(sessionID, userID)

		// 生成 RunID
		runID := uuid.New().String()
		startTime := time.Now()

		// 创建运行记录
		run := &WorkflowRun{
			RunID:       runID,
			SessionID:   sessionID,
			WorkflowID:  w.ID,
			Input:       input.Input,
			StepOutputs: make(map[string]*StepOutput),
			Status:      RunStatusRunning,
			StartTime:   startTime,
			Metrics:     &RunMetrics{TotalSteps: len(w.Steps), StepMetrics: make(map[string]*StepMetrics)},
		}

		// 发送开始事件
		if writer.Send(&RunEvent{
			Type:         EventWorkflowStarted,
			EventID:      uuid.New().String(),
			WorkflowID:   w.ID,
			WorkflowName: w.Name,
			RunID:        runID,
			Timestamp:    startTime,
			Data: map[string]any{
				"input":      input.Input,
				"session_id": sessionID,
				"user_id":    userID,
			},
		}, nil) {
			return
		}

		// 合并会话状态
		sessionState := make(map[string]any)
		if session.State != nil {
			maps.Copy(sessionState, session.State)
		}
		if input.SessionState != nil {
			maps.Copy(sessionState, input.SessionState)
		}

		stepOutputs := make(map[string]*StepOutput)