流式传输

在工作流执行期间实时流式传输更新

工作流可能很复杂，它们旨在处理复杂的、分支式的、可迭代的逻辑，这意味着它们可能需要一些时间才能完全执行。为了给用户提供良好的体验，您可能希望通过在事件发生时流式传输事件来显示进度。工作流内置支持可直接在节点内部完成这一点。

从节点发出事件

让我们设置一个新的事件来处理在执行过程中流式输出我们的进度：

namespace App\Neuron;

class ProgressEvent implements Event 
{
    public function __construct(protected string $msg){}
}

我们将采用前一教程中的、包含多个节点的示例 MyWorkflow，并修改这些节点，使其流式输出更新，而不是直接 echo 输出。

注意：要从节点流式传输事件，你需要添加 \Generator 作为 __invoke 方法时。

namespace App\Neuron;

use NeuronAI\Workflow\Node;
use NeuronAI\Workflow\StartEvent;
use NeuronAI\Workflow\StopEvent;

class InitialNode extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): \Generator|FirstEvent
    {
        yield new ProgressEvent("处理 StartEvent");
        
        return new FirstEvent("InitialNode 完成");
    }
}

class NodeOne extends Node
{
    public function __invoke(FirstEvent $event, WorkflowState $state): \Generator|SecondEvent
    {
        yield new ProgressEvent($event->firstMsg);
        
        return new SecondEvent("NodeOne 完成");
    }
}

class NodeTwo extends Node
{
    public function __invoke(SecondEvent$event, WorkflowState $state): \Generator|StopEvent
    {
        yield new ProgressEvent($event->secondMsg);
        
        yield new ProgressEvent("NodeTwo 完成");
        
        $state->set('message', '流式传输结束');
        
        return new StopEvent();
    }
}

要实际获得此输出，我们需要启动工作流并监听事件，就像这样：

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

$stream = $handler->events();

foreach ($stream as $event) {
    if ($event instanceof ProgressEvent) {
        echo "\n- ".$event->message;
    }
}

$finalState = $stream->getResult();

// 它会打印 "流式传输结束"
echo "\n- ".$finalState->get('message');

完整输出将是：

- 处理 StartEvent
- InitialNode 完成
- NodeOne 完成
- NodeTwo 完成
- 流式传输结束

流式代理输出

在节点中运行代理是使用工作流时最常见的用例之一。您可能希望将内部代理的输出直接流式传输给客户端，以便为底层生成过程提供实时反馈。您只需在节点内部流式传输代理的输出即可。

class InitialNode extends Node
{
    public function __invoke(StartEvent $event, WorkflowState $state): \Generator|FirstEvent
    {
        // 运行带流式输出的代理
        yield from Agent::make()
            ->stream(new UserMessage($state->get('prompt')))
            ->events();
        
        return new FirstEvent("InitialNode 完成");
    }
}

要获取此输出，您可以像往常一样监听工作流事件：

$handler = MyWorkflow::make()->init();

foreach ($handler->events() as $event) {
    echo match($event::class) {
        TextChunk::class => "\n- ".$event->content,
        ...
    }
}

监控与调试

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

Neuron AI | Inspectordocs.inspector.dev