[Agent] Streaming ResultInterface parts

[franzwilding]

I'm trying to use symfony/ai as a backend for https://ai-sdk.dev. ai-sdk requires the backend to send the content in a structured way (also when streaming). This allows the frontend to display status information, internal tool calling, reasoning etc. (like in le chat or chat-gpt).

Basic content streaming of one text part works well, by wrapping each chunk of the content response in a small envelope:

$result = $defaultAgent->call($messages, ['stream' => true]);
$response = new StreamedResponse(function () use ($result) {

        $send = function (string $type, array $data = []) {
            echo 'data: ' . json_encode(array_merge(['type' => $type], $data), JSON_UNESCAPED_SLASHES) . "\n\n";
            @ob_flush(); @flush();
        };

        $send('start');
        $textId = (string) new Ulid();
        $send('text-start', ['id' => $textId]);
        
        foreach ($result->getContent() as $part) {
            $send('text-delta', ['id' => $textId, 'delta' => $part]);
        }

        $send('text-end', ['id' => $textId]);
        $send('finish');
        echo "data: [DONE]\n\n";
        @ob_flush();
        @flush();
}

However when i want to stream tool calling status or internal reasoning, things are getting more complicated.

I tried to implement a custom OutputProcessor, however the internal agentic flow breaks, when replacing tool_calling result with a custom envenlope.

The best result I got, was writing to the output-stream directly from an output-processor, however this is not very elegant or maintanable.

Basically what I need would be:

• a better $result->getResultPart()s generator, that will not return simple content, but instances of ResultInterface or mybe a new FinalResultInterface?
• A new TextChunkResult to not get a mixture of string and ResultInterface objects. This would allow me to add my envelopes in the $result->getResultParts() loop depending on the part class.

I tink this would be a very nice feature for any UI that wants to display whats going on inside one agent call.

[franzwilding]

Another approach would be to have more (and consistent events). I managed to stream tool calling status to the client using a small toolbox wrapper to send events before and after each single tool-call:

<?php

namespace App\Service\Ai;

use App\Event\ToolCallEnded;
use App\Event\ToolCallStarted;
use Symfony\AI\Agent\Toolbox\Exception\ToolExecutionExceptionInterface;
use Symfony\AI\Agent\Toolbox\Exception\ToolNotFoundException;
use Symfony\AI\Agent\Toolbox\ToolboxInterface;
use Symfony\AI\Platform\Result\ToolCall;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

readonly class DispatcherToolbox implements ToolboxInterface
{
    public function __construct(
        private ToolboxInterface $innerToolbox,
        private EventDispatcherInterface $eventDispatcher,
    ) {}

    public function getTools(): array
    {
        return $this->innerToolbox->getTools();
    }

    public function execute(ToolCall $toolCall): mixed
    {
        try {
            $this->eventDispatcher->dispatch(new ToolCallStarted($toolCall));
            $result = $this->innerToolbox->execute($toolCall);
            $this->eventDispatcher->dispatch(new ToolCallEnded($toolCall, true, $result));
            return $result;
        } catch (ToolExecutionExceptionInterface $e) {
            $this->eventDispatcher->dispatch(new ToolCallEnded($toolCall, false, $e->getToolCallResult()));
            throw $e;
        } catch (ToolNotFoundException $e) {
            $this->eventDispatcher->dispatch(new ToolCallEnded($toolCall, false, $e->getMessage()));
            throw $e;
        }
    }
}

With this, I can now use the default $result->getContent() loop for text response together with event listeners for internal agent status update:

$this->send('start');

            // Register listeners so we can stream tool I/O and bracket text correctly
            $this->eventDispatcher->addListener(ToolCallStarted::class, fn(ToolCallStarted $event) => $this->onToolCallStarted($event) );
            $this->eventDispatcher->addListener(ToolCallEnded::class, fn(ToolCallEnded $event) => $this->onToolCallEnded($event));

            try {
                // --- Stream the content -------------------------------------------------
                foreach ($result->getContent() as $part) {
                    if ($part === '' || $part === null) {
                        continue;
                    }

                    // If no text block is open (e.g., right after a tool call), open a new one.
                    if ($this->openTextId === null) {
                        $this->openText();
                    }

                    $this->send('text-delta', [
                        'id'    => $this->openTextId,
                        'delta' => (string) $part,
                    ]);
                }

                // Close any open text block and finish the SSE stream
                $this->closeText();
                $this->send('finish');
                echo "data: [DONE]\n\n";
                @ob_flush();
                @flush();
            } finally {
                // Deregister listeners to avoid leaks
                $this->eventDispatcher->removeListener(ToolCallArgumentsResolved::class, [$this, 'onToolCallStarted']);
                $this->eventDispatcher->removeListener(ToolCallsExecuted::class, [$this, 'onToolCallEnded']);
            }

So maybe dispatching more events would be a more generic and easier way to get internal agent status.

[OskarStark]

- With this, I can no use the default $result->getContent() loop for text response together with event listeners for internal agent status update:
+ With this, I can now use the default $result->getContent() loop for text response together with event listeners for internal agent status update:

?

[franzwilding]

• With this, I can now use the default $result->getContent() loop for text response together with event listeners for internal agent status update.

[OskarStark]

Are you willing to provide a PR throwing the events?

[franzwilding]

Yes, I can do this, but I’m not sure where it should go—inside Toolbox or AgentProcessor? The ToolCallsExecuted event is currently dispatched in AgentProcessor, but I think it would be cleaner to dispatch all events from the Toolbox. What do you think?

And do you have any preferences about the event layout? What do you think about:ToolCallStarted, ToolCallSucceeded and ToolCallFailed?

[chr-hertel]

I get the concern about centralizing the event dispatching, but currently the toolbox is not aware of the entire collection of tool calls needed after one run. it's not that common in my reality, but some models respond with multiple tool calls in parallel. at least that's why it's in the processor.

i don't have an issue tho to have events dispatched in both - so go for the toolbox 👍

@valtzu - do you recall why you added the ToolCallArgumentsResolved event? it doesn't have any capabilities than notifying.

[franzwilding]

Okay!

About the ToolCallArgumentsResolved event @valtzu @chr-hertel: It would be great to allow to modify the arguments!

[valtzu]

ToolCallArgumentsResolved was added for checking IsGrantedTool attribute which needs the denormalized arguments.

[franzwilding]

@valtzu Alright, I didn't touch ToolCallArgumentsResolved.
@chr-hertel Thanks for the feedback. I created a PR.