Method executeTaskNodeWithActor

pkg/workflow/actor_engine.go:381–436

executeTaskNodeWithActor 使用 Actor 执行任务节点

(execution *WorkflowExecution, state *workflowState, node *NodeDef)

// executeTaskNodeWithActor 使用 Actor 执行任务节点
func (c *CoordinatorActor) executeTaskNodeWithActor(execution *WorkflowExecution, state *workflowState, node *NodeDef) error {
	if node.Agent == nil {
		return errors.New("task node requires agent configuration")
	}

	// 获取或创建 Agent PID
	pid, ok := c.engine.GetAgentPID(node.ID)
	if !ok {
		// 需要创建 Agent - 这里简化处理，实际应该从 AgentFactory 创建
		return fmt.Errorf("agent not found for node: %s", node.ID)
	}

	// 准备输入
	inputMessage, err := c.prepareInputMessage(execution, node)
	if err != nil {
		return err
	}

	// 通过 Actor 消息执行
	replyCh := make(chan *pkgagent.ChatResultMsg, 1)
	pid.Tell(&pkgagent.ChatMsg{
		Text:    inputMessage,
		Ctx:     execution.Context.Context,
		ReplyTo: replyCh,
	})

	// 等待结果
	timeout := node.Timeout
	if timeout == 0 {
		timeout = c.engine.actorConfig.DefaultTimeout
	}

	select {
	case result := <-replyCh:
		if result.Error != nil {
			return result.Error
		}
		// 保存结果
		execution.NodeResults[node.ID] = &NodeResult{
			NodeID:   node.ID,
			NodeName: node.Name,
			NodeType: node.Type,
			Status:   StatusCompleted,
			Outputs: map[string]any{
				"content": result.Result.Text,
			},
		}
		return nil

	case <-time.After(timeout):
		return fmt.Errorf("task node %s timed out", node.ID)

	case <-execution.Context.Context.Done():
		return execution.Context.Context.Err()
	}
}

// executeParallelNodeWithActors 使用 Actor 执行并行节点