Neuron 流式步骤

以 Markdown 格式查看在 GitHub 上编辑文档位置 2cc7b2b

此示例展示了清晰的职责划分。Neuron 会逐块流式传输代理响应，而 Queuety 则通过 StreamingStep 持久化保存每个数据块。因此，UI 可以接收实时草稿，同时工作流仍能在工作进程崩溃或重试后继续运行。

这种模式的适用场景

当草稿或响应较长，适合实时流式传输时；当你不希望工作进程退出后丢失部分输出时；以及当最终制品仍应在流结束后写入工作流状态时，都可以使用这种模式。

Neuron 代理

PHP

namespace App\Neuron;

use NeuronAI\Agent\Agent;
use NeuronAI\Providers\AIProviderInterface;
use NeuronAI\Providers\OpenAI\Responses\OpenAIResponses;
use NeuronAI\SystemPrompt;

final class DraftMemoAgent extends Agent
{
    protected function provider(): AIProviderInterface
    {
        return new OpenAIResponses(
            key: $_ENV['OPENAI_API_KEY'],
            model: $_ENV['OPENAI_MODEL'],
        );
    }

    public function instructions(): string
    {
        return (string) new SystemPrompt(
            background: [
                'You write crisp internal strategy memos.',
                'Prefer short sections and operationally useful language.',
            ],
        );
    }
}

Queuety 流式步骤

PHP

namespace App\Workflow\Steps;

use App\Neuron\DraftMemoAgent;
use NeuronAI\Chat\Messages\UserMessage;
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

final class StreamMemoDraftStep implements StreamingStep
{
    public function stream(array $state, array $existing_chunks = []): \Generator
    {
        if (! empty($existing_chunks)) {
            return;
        }

        $handler = (new DraftMemoAgent())->stream(
            new UserMessage(sprintf(
                "Write a strategy memo about %s for account %d.",
                $state['topic'],
                $state['account_id'],
            ))
        );

        foreach ($handler->events() as $i => $chunk) {
            Heartbeat::beat(['chunks' => $i + 1]);

            // This example assumes the agent has no tools attached, so
            // Neuron will stream text/reasoning chunks only.
            yield $chunk->content;
        }
    }

    public function on_complete(array $chunks, array $state): array
    {
        return [
            'draft_markdown' => implode('', $chunks),
        ];
    }

    public function config(): array
    {
        return [
            'max_attempts' => 3,
        ];
    }
}

在工作流中使用

PHP

use Queuety\Queuety;

Queuety::workflow('memo_drafting')
    ->then(BuildDraftBriefStep::class)
    ->then(StreamMemoDraftStep::class)
    ->then(StoreMemoStep::class)
    ->dispatch([
        'account_id' => 10,
        'topic' => 'pricing changes for enterprise accounts',
    ]);

为什么使用 Queuety，而不是只用 Neuron 流适配器

Neuron 的流适配器适用于前端协议和传输。Queuety 提供了不同的能力：每个生成的数据块都会被持久化保存，重试时可以从之前保存的数据块继续，完成的草稿会合并回工作流状态，而且流会成为更大编排中的一个持久化步骤。

因此，如果流是多步骤业务流程的一部分，而不只是请求/响应 UI，这种模式就很合适。

一个注意事项

如果你的 Neuron 代理在流式传输时使用工具，请明确处理流式数据块的类型。Neuron 当前的流式文档介绍了 TextChunk、ReasoningChunk、ToolCallChunk 和 ToolResultChunk。对于后台起草工作流，最简单的做法通常是让流式代理只输出文本。