Added streaming support to Chat API & persistence to storage through accumulator callback chaining

Marco van Angeren · Marco van Angeren · commit 72ad069faa75 · 2025-12-01T09:27:22.000+01:00
diff --git a/src/agent/src/Toolbox/StreamResult.php b/src/agent/src/Toolbox/StreamResult.php
@@ -33,10 +33,7 @@ public function getContent(): \Generator
             if ($value instanceof ToolCallResult) {
                 $innerResult = ($this->handleToolCallsCallback)($value, Message::ofAssistant($streamedResult));
 
-                // Propagate metadata from inner result to this result
-                foreach ($innerResult->getMetadata()->all() as $key => $metadataValue) {
-                    $this->getMetadata()->add($key, $metadataValue);
-                }
+                $this->getMetadata()->set($innerResult->getMetadata()->all());
 
                 $content = $innerResult->getContent();
                 // Strings are iterable in PHP but yield from would iterate character-by-character.
diff --git a/src/chat/CHANGELOG.md b/src/chat/CHANGELOG.md
@@ -1,5 +1,11 @@
 CHANGELOG
 =========
+0.2 
+  * Add streaming support to `ChatInterface::submit()`
+    - Add `StreamableStoreInterface` which indicates `StoreInterface` implementation can be configured with streaming 
+    - Add `AccumulatingStreamResult` wrapper class which adds accumulation logic & callback chaining to `StreamResult` implementations (can wrap both `Agent` and `Platform` variants) to return the full message once `Generator` is exhausted
+    - Streamed responses now also create `AssistantMessage` & are added to `Store` in `Chat::submit()`
+    - Bugfixed loss of metadata in `Chat::submit()`
 
 0.1
 ---
diff --git a/src/chat/src/Bridge/Doctrine/DoctrineDbalMessageStore.php b/src/chat/src/Bridge/Doctrine/DoctrineDbalMessageStore.php
@@ -24,6 +24,7 @@
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageNormalizer;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\MessageInterface;
 use Symfony\Component\Serializer\Encoder\JsonEncoder;
@@ -34,7 +35,7 @@
 /**
  * @author Guillaume Loulier <personal@guillaumeloulier.fr>
  */
-final class DoctrineDbalMessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class DoctrineDbalMessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly string $tableName,
diff --git a/src/chat/src/Bridge/Local/CacheStore.php b/src/chat/src/Bridge/Local/CacheStore.php
@@ -15,12 +15,13 @@
 use Symfony\AI\Agent\Exception\RuntimeException;
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 
 /**
  * @author Christopher Hertel <mail@christopher-hertel.de>
  */
-final class CacheStore implements ManagedStoreInterface, MessageStoreInterface
+final class CacheStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly CacheItemPoolInterface $cache,
diff --git a/src/chat/src/Bridge/Local/InMemoryStore.php b/src/chat/src/Bridge/Local/InMemoryStore.php
@@ -13,12 +13,13 @@
 
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 
 /**
  * @author Christopher Hertel <mail@christopher-hertel.de>
  */
-final class InMemoryStore implements ManagedStoreInterface, MessageStoreInterface
+final class InMemoryStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     /**
      * @var MessageBag[]
diff --git a/src/chat/src/Bridge/Meilisearch/MessageStore.php b/src/chat/src/Bridge/Meilisearch/MessageStore.php
@@ -16,6 +16,7 @@
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageNormalizer;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\MessageInterface;
 use Symfony\Component\Clock\ClockInterface;
@@ -31,7 +32,7 @@
 /**
  * @author Guillaume Loulier <personal@guillaumeloulier.fr>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly HttpClientInterface $httpClient,
diff --git a/src/chat/src/Bridge/Pogocache/MessageStore.php b/src/chat/src/Bridge/Pogocache/MessageStore.php
@@ -15,6 +15,7 @@
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageNormalizer;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\MessageInterface;
 use Symfony\Component\Serializer\Encoder\JsonEncoder;
@@ -28,7 +29,7 @@
 /**
  * @author Guillaume Loulier <personal@guillaumeloulier.fr>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly HttpClientInterface $httpClient,
diff --git a/src/chat/src/Bridge/Redis/MessageStore.php b/src/chat/src/Bridge/Redis/MessageStore.php
@@ -14,6 +14,7 @@
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageNormalizer;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\MessageInterface;
 use Symfony\Component\Serializer\Encoder\JsonEncoder;
@@ -24,7 +25,7 @@
 /**
  * @author Guillaume Loulier <personal@guillaumeloulier.fr>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly \Redis $redis,
diff --git a/src/chat/src/Bridge/SurrealDb/MessageStore.php b/src/chat/src/Bridge/SurrealDb/MessageStore.php
@@ -16,6 +16,7 @@
 use Symfony\AI\Chat\ManagedStoreInterface;
 use Symfony\AI\Chat\MessageNormalizer;
 use Symfony\AI\Chat\MessageStoreInterface;
+use Symfony\AI\Chat\StreamableStoreInterface;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\MessageInterface;
 use Symfony\Component\Serializer\Encoder\JsonEncoder;
@@ -29,7 +30,7 @@
 /**
  * @author Guillaume Loulier <personal@guillaumeloulier.fr>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     private string $authenticationToken = '';
 
diff --git a/src/chat/src/Chat.php b/src/chat/src/Chat.php
@@ -12,10 +12,14 @@
 namespace Symfony\AI\Chat;
 
 use Symfony\AI\Agent\AgentInterface;
+use Symfony\AI\Agent\Exception\RuntimeException;
+use Symfony\AI\Agent\Toolbox\StreamResult as ToolboxStreamResult;
+use Symfony\AI\Chat\Result\AccumulatingStreamResult;
 use Symfony\AI\Platform\Message\AssistantMessage;
 use Symfony\AI\Platform\Message\Message;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\UserMessage;
+use Symfony\AI\Platform\Result\StreamResult;
 use Symfony\AI\Platform\Result\TextResult;
 
 /**
@@ -35,18 +39,31 @@ public function initiate(MessageBag $messages): void
         $this->store->save($messages);
     }
 
-    public function submit(UserMessage $message): AssistantMessage
+    public function submit(UserMessage $message): AssistantMessage|AccumulatingStreamResult
     {
         $messages = $this->store->load();
 
         $messages->add($message);
         $result = $this->agent->call($messages);
 
+        if ($result instanceof StreamResult || $result instanceof ToolboxStreamResult) {
+            if (!$this->store instanceof StreamableStoreInterface) {
+                throw new RuntimeException($this->store::class . ' does not support streaming.');
+            }
+
+            return new AccumulatingStreamResult($result, function (AssistantMessage $assistantMessage) use ($messages) {
+                $messages->add($assistantMessage);
+                $this->store->save($messages);
+            });
+        }
+
         \assert($result instanceof TextResult);
 
         $assistantMessage = Message::ofAssistant($result->getContent());
-        $messages->add($assistantMessage);
 
+        $assistantMessage->getMetadata()->set($result->getMetadata()->all());
+
+        $messages->add($assistantMessage);
         $this->store->save($messages);
 
         return $assistantMessage;
diff --git a/src/chat/src/ChatInterface.php b/src/chat/src/ChatInterface.php
@@ -12,6 +12,7 @@
 namespace Symfony\AI\Chat;
 
 use Symfony\AI\Agent\Exception\ExceptionInterface;
+use Symfony\AI\Chat\Result\AccumulatingStreamResult;
 use Symfony\AI\Platform\Message\AssistantMessage;
 use Symfony\AI\Platform\Message\MessageBag;
 use Symfony\AI\Platform\Message\UserMessage;
@@ -26,5 +27,5 @@ public function initiate(MessageBag $messages): void;
     /**
      * @throws ExceptionInterface When the chat submission fails due to agent errors
      */
-    public function submit(UserMessage $message): AssistantMessage;
+    public function submit(UserMessage $message): AssistantMessage|AccumulatingStreamResult;
 }
diff --git a/src/chat/src/Result/AccumulatingStreamResult.php b/src/chat/src/Result/AccumulatingStreamResult.php
@@ -0,0 +1,82 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\AI\Chat\Result;
+
+use Symfony\AI\Agent\Toolbox\StreamResult as ToolboxStreamResult;
+use Symfony\AI\Platform\Message\AssistantMessage;
+use Symfony\AI\Platform\Message\Message;
+use Symfony\AI\Platform\Metadata\Metadata;
+use Symfony\AI\Platform\Result\StreamResult;
+use Symfony\AI\Platform\Result\ToolCallResult;
+
+/**
+ * @author Marco van Angeren <marco@jouwweb.nl>
+ */
+final class AccumulatingStreamResult
+{
+    private ?\Closure $onComplete = null;
+
+    public function __construct(
+        private readonly StreamResult|ToolboxStreamResult $innerResult,
+        ?\Closure $onComplete = null,
+    ) {
+        $this->onComplete = $onComplete;
+    }
+
+    public function addOnComplete(\Closure $callback): void
+    {
+        $existingCallback = $this->onComplete;
+
+        $this->onComplete = $existingCallback
+            ? function (AssistantMessage $message) use ($existingCallback, $callback) {
+                $existingCallback($message);
+                $callback($message);
+            }
+            : $callback;
+    }
+
+    public function getContent(): \Generator
+    {
+        $accumulatedContent = '';
+        $toolCalls = [];
+
+        try {
+            foreach ($this->innerResult->getContent() as $value) {
+                if ($value instanceof ToolCallResult) {
+                    array_push($toolCalls, ...$value->getContent());
+                    yield $value;
+                    continue;
+                }
+
+                $accumulatedContent .= $value;
+                yield $value;
+            }
+        } finally {
+            if ($this->onComplete !== null) {
+                $assistantMessage = Message::ofAssistant(
+                    $accumulatedContent === '' ? null : $accumulatedContent,
+                    $toolCalls ?: null
+                );
+
+                $assistantMessage->getMetadata()->set($this->innerResult->getMetadata()->all());
+
+                ($this->onComplete)($assistantMessage);
+            }
+        }
+    }
+
+    public function getMetadata(): Metadata
+    {
+        return $this->innerResult->getMetadata();
+    }
+}
+
diff --git a/src/chat/src/StreamableStoreInterface.php b/src/chat/src/StreamableStoreInterface.php
@@ -0,0 +1,20 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\AI\Chat;
+
+/**
+ * @author Marco van Angeren <marco@jouwweb.nl>
+ */
+interface StreamableStoreInterface
+{
+}
+
diff --git a/src/chat/tests/ChatTest.php b/src/chat/tests/ChatTest.php
diff --git a/src/platform/src/Result/StreamResult.php b/src/platform/src/Result/StreamResult.php
diff --git a/src/platform/tests/Result/StreamResultTest.php b/src/platform/tests/Result/StreamResultTest.php
