循环与分支

得益于其事件驱动设计，Workflow 让分支和循环逻辑很容易实现。一旦你理解了节点如何归属于事件，就很容易开始设想如何创建循环和分支，而这只是决定在“if 条件”或其他逻辑中应该返回哪个事件。

循环

要创建一个循环，只需将前一个节点的入口事件作为当前节点的退出事件返回。你也可以将当前节点的同一个入口事件作为退出事件，从而让当前节点自身形成循环。

看看下面的示例。这个 NodeOne 可以有两个事件作为返回类型， FirstEvent 和 SecondEvent。如果节点返回 FirstEvent，它将导致同一节点再次执行，因为 FirstEvent 由其自身处理，从而形成一个循环。

如果节点返回 SecondEvent 它最终会将执行推进到另一个节点。

class NodeOne extends Node
{
    public function __invoke(FirstEvent $event, WorkflowState $state): FirstEvent|SecondEvent
    {
        echo "\n- ".$event->firstMsg;
        
        if (rand(0, 1) === 1) {
            // 返回 FirstEvent 将触发 NodeOne 的另一次执行
            return new FirstEvent("在 NodeOne 上运行一个循环");
        }
        
        return new SecondEvent("NodeOne 完成，继续前进");
    }
}

注意，节点现在在以下方面有两个返回类型用于 __invoke 方法： FirstEvent 和 SecondEvent。你必须在方法签名中声明所有可能的返回事件，以便 Workflow 构建执行链。

返回 FirstEvent 将触发 NodeOne的另一次执行。所以最终输出可能是：

$state = Workflow::make()
    ->addNodes([
        new InitialNode(),
        new NodeOne(),
        new NodeTwo()
    ])
    ->init()
    ->run();

/*
- 处理 StartEvent
- InitialNode 完成
- 在 NodeOne 上运行一个循环
- 在 NodeOne 上运行一个循环
- NodeOne 完成，继续前进
- NodeTwo 完成
*/

你可以通过为 invoke 方法定义适当的输入事件和返回事件，在工作流中的任意节点之间创建循环。

该 NodeOne 节点甚至可以返回一个 StartEvent，直接跳转到 Workflow 的第一个节点。事件驱动架构允许你直接将工作流中的任何节点指向前后两个方向。

分支

正如你已经看到的，你可以根据条件从节点返回不同的事件来定义自定义执行流程。本节我们将看到一个分叉为两条不同路径的工作流示例。

首先让我们创建一些自定义事件：

namespace App\Neuron;

class BrancheA1Event implements Event 
{
    public function __construct(protected string $firstMsg){}
}

class BrancheA2Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

class BrancheB1Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

class BrancheB2Event implements Event 
{
    public function __construct(protected string $secondMsg){}
}

在工作流的初始节点中，我们决定要走哪条分支。请记住始终在 __invoke 方法签名中定义适当的返回类型：

class InitialNode extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): BrancheA1Event|BrancheB1Event
    {
        if (rand(0, 1) === 1) {
            // 返回 FirstEvent 将触发 NodeOne 的另一次执行
            return new BrancheA1Event();
        }
        
        return new BrancheB1Event();
    }
}

其他节点将按顺序向前执行。

$state = Workflow::make()
    ->addNodes([
        new InitialNode(),
        new A1Node(),
        new A2Node(),
        new B1Node(),
        new B2Node(),
    ])
    ->init()
    ->run();

当然，你可以按任意顺序组合分支和循环，以满足你的应用需求。

并行分支

当你想并行调用多个分支的执行时，你需要返回特殊事件 ParallelEvent 来自你的节点。

use NeuronAI\\Workflow\\Events\\ParallelEvent;

class DocumentProcessing extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): ParallelEvent
    {
        // 这里是节点逻辑...
	
        // 最后返回一个 ParallelEvent
        return new ParallelEvent([
            'text' => new TextProcessEvent(),
            'image' => new ImageProcessEvent(),
        ]);
    }
}

该 ParallelEvent 必须使用一个数组来构造 <branch_name> => <FirstInputEvent>:

new ParallelEvent([
	<branch_name> => <FirstInputEvent>
	...
])

处理你为每个分支声明的事件的节点必须注册到工作流中：

class MyWorkflow extends Workflow
{
	protected function nodes(): array
	{
		return [
			new DocumentProcessing(),
			
			// “text” 分支
			new DescriptionGenerationNode(), // 处理 TextProcessEvent
			new TextRefactorNode(),
			
			// “image” 分支
			new ImageProcessNode(), // 处理 ImageProcessEvent
			new AddWatermarkNode(),
			
			new MergeNode(),
		];
	}
}

一个分支可以只有一个节点，也可以是始终通过事件连接的多个节点列表。

处理分支的结束

你分支中的最后一个节点必须返回框架内置的 StopEvent.

在上面的示例中 TextRefactorNode 和 AddWatermarkNode 将通过返回 StopEvent 来声明其分支结束：

class AddWatermarkNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
        // 这里是节点代码...
		
        // 返回 StopEvent，分支结束
        return new StopEvent(result: 'Hello World!');
    }
}

注意：StopEvent 也可以携带一些结果。

获取分支结果

该 ParallelEvent 由 DocumentProcessing 节点基本上是在等待分支执行结束后，再被传递到下一个节点。

在上面的示例中， MergeNode 负责最终处理 ParallelEvent:

class MergeNode extends Node
{
    public function __invoke(ParallelEvent $event, WorkflowState $state): StopEvent
    {
        $textBranchResult = $event->getResult('text');
        $imageBranchResult = $event->getResult('image');
        
        return new StopEvent();
    }
}

这个节点可以通过 getResult() 方法传入 <branch_name>.

和往常一样，合并节点可以停止工作流，或者返回其他事件推动工作流继续向前。

分支状态隔离

有一点值得注意： 每个分支都会获得一份隔离的工作流状态副本。它们从相同的快照开始，但分支内部的变更不会传播到兄弟分支或主工作流。传回数据的唯一方式是通过 StopEvent 结果。

这样设计是有意为之，它避免了一整类并发 bug，即分支相互踩踏彼此的状态。

异步执行器

我们还提供了内部工作流执行器的一个实现，使你能够并发运行多个分支。要使用 异步执行器 你需要安装 Amp 软件包：

composer require amphp/amp

class MyAgent extends Workflow 
{
    /**
     * 使用异步执行器
     */
    protected function executor(): WorkflowExecutorInterface
    {
        return new AsyncExecutor();
    }

    protected function nodes(): array
    {
        return [...];
    }
}

如果你想并行运行多个 agentic 任务，这尤其有用，因为 Neuron AI 已经提供了 AmpHttpClient 你可以将其注入到所有组件中。

use NeuronAI\\HttpClient\\AmpHttpClient;

class DescriptionGenerationNode extends Node
{
    public function __invoke(TextProcessEvent $event, WorkflowState $state): StopEvent
    {
        $input = new UserMessage('描述这张图片');
        $input->addContent(
            new ImageContent(...)
        );

        $response = AsyncAgent::make()
            ->chat($input)
            ->getMessage();

        return new StopEvent(result: $response);
    }
}

监控与调试

你使用 Neuron 构建的许多应用都会包含多个步骤，以及对 LLM 调用的多次执行。随着这些应用变得越来越复杂，能够检查你的智能体系统内部究竟发生了什么就变得至关重要。实现这一点的最佳方式是使用 Inspector.

Neuron AI | Inspectordocs.inspector.dev