[Chat] Add streaming support & persistence to storage through accumulator callback chaining

src/chat/CHANGELOG.md

@@ -4,6 +4,11 @@ CHANGELOG
 0.1
 ---
 
+* 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()`
  * Introduce the component
  * Add support for external message stores:
    - Doctrine

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 <[email protected]>
  */
-final class DoctrineDbalMessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class DoctrineDbalMessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly string $tableName,

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 <[email protected]>
  */
-final class CacheStore implements ManagedStoreInterface, MessageStoreInterface
+final class CacheStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly CacheItemPoolInterface $cache,

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 <[email protected]>
  */
-final class InMemoryStore implements ManagedStoreInterface, MessageStoreInterface
+final class InMemoryStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     /**
      * @var MessageBag[]

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 <[email protected]>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly HttpClientInterface $httpClient,

src/chat/src/Bridge/MongoDb/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;
@@ -27,7 +28,7 @@
 /**
  * @author Guillaume Loulier <[email protected]>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly Client $client,

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 <[email protected]>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly HttpClientInterface $httpClient,

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 <[email protected]>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     public function __construct(
         private readonly \Redis $redis,

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 <[email protected]>
  */
-final class MessageStore implements ManagedStoreInterface, MessageStoreInterface
+final class MessageStore implements ManagedStoreInterface, MessageStoreInterface, StreamableStoreInterface
 {
     private string $authenticationToken = '';
 

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;

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;
 }

src/chat/src/Result/AccumulatingStreamResult.php

@@ -0,0 +1,81 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * 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 <[email protected]>
+ */
+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 (null !== $this->onComplete) {
+                $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();
+    }
+}

src/chat/src/StreamableStoreInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <[email protected]>
+ *
+ * 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 <[email protected]>
+ */
+interface StreamableStoreInterface
+{
+}