feat: support streaming of responses

php-llm · Oct 3, 2024 · c15e9dd · c15e9dd
1 parent ddf5110
commit c15e9dd
Show file tree

Hide file tree

Showing 17 changed files with 250 additions and 31 deletions.
diff --git a/examples/stream-claude-anthropic.php b/examples/stream-claude-anthropic.php
@@ -0,0 +1,36 @@
+<?php
+
+use PhpLlm\LlmChain\Anthropic\Model\Claude;
+use PhpLlm\LlmChain\Anthropic\Platform\Anthropic;
+use PhpLlm\LlmChain\Chain;
+use PhpLlm\LlmChain\Message\Message;
+use PhpLlm\LlmChain\Message\MessageBag;
+use Symfony\Component\Dotenv\Dotenv;
+use Symfony\Component\HttpClient\HttpClient;
+
+require_once dirname(__DIR__).'/vendor/autoload.php';
+(new Dotenv())->loadEnv(dirname(__DIR__).'/.env');
+
+if (empty($_ENV['ANTHROPIC_API_KEY'])) {
+    echo 'Please set the ANTHROPIC_API_KEY environment variable.'.PHP_EOL;
+    exit(1);
+}
+
+$platform = new Anthropic(HttpClient::create(), $_ENV['ANTHROPIC_API_KEY']);
+$llm = new Claude($platform);
+
+$chain = new Chain($llm);
+$messages = new MessageBag(
+    Message::forSystem('You are a thoughtful philosopher.'),
+    Message::ofUser('What is the purpose of an ant?'),
+);
+$response = $chain->call($messages, [
+    'stream' => true, // enable streaming of response text
+]);
+
+assert($response instanceof Generator);
+
+foreach ($response as $word) {
+    echo $word;
+}
+echo PHP_EOL;
diff --git a/examples/stream-gpt-openai.php b/examples/stream-gpt-openai.php
@@ -0,0 +1,37 @@
+<?php
+
+use PhpLlm\LlmChain\Chain;
+use PhpLlm\LlmChain\Message\Message;
+use PhpLlm\LlmChain\Message\MessageBag;
+use PhpLlm\LlmChain\OpenAI\Model\Gpt;
+use PhpLlm\LlmChain\OpenAI\Model\Gpt\Version;
+use PhpLlm\LlmChain\OpenAI\Platform\OpenAI;
+use Symfony\Component\Dotenv\Dotenv;
+use Symfony\Component\HttpClient\EventSourceHttpClient;
+
+require_once dirname(__DIR__).'/vendor/autoload.php';
+(new Dotenv())->loadEnv(dirname(__DIR__).'/.env');
+
+if (empty($_ENV['OPENAI_API_KEY'])) {
+    echo 'Please set the OPENAI_API_KEY environment variable.'.PHP_EOL;
+    exit(1);
+}
+
+$platform = new OpenAI(new EventSourceHttpClient(), $_ENV['OPENAI_API_KEY']);
+$llm = new Gpt($platform, Version::gpt4oMini());
+
+$chain = new Chain($llm);
+$messages = new MessageBag(
+    Message::forSystem('You are a thoughtful philosopher.'),
+    Message::ofUser('What is the purpose of an ant?'),
+);
+$response = $chain->call($messages, [
+    'stream' => true, // enable streaming of response text
+]);
+
+assert($response instanceof Generator);
+
+foreach ($response as $word) {
+    echo $word;
+}
+echo PHP_EOL;
diff --git a/examples/toolbox-youtube.php b/examples/toolbox-youtube.php
@@ -31,6 +31,13 @@
 $chain = new Chain($llm, [$processor], [$processor]);
 
 $messages = new MessageBag(Message::ofUser('Please summarize this video for me: https://www.youtube.com/watch?v=6uXW-ulpj0s'));
-$response = $chain->call($messages);
+$response = $chain->call($messages, [
+    'stream' => true, // enable streaming of response text
+]);
 
-echo $response.PHP_EOL;
+assert($response instanceof Generator);
+
+foreach ($response as $word) {
+    echo $word;
+}
+echo PHP_EOL;
diff --git a/src/Anthropic/Model/Claude.php b/src/Anthropic/Model/Claude.php
@@ -10,6 +10,8 @@
 use PhpLlm\LlmChain\Message\MessageBag;
 use PhpLlm\LlmChain\Response\Choice;
 use PhpLlm\LlmChain\Response\Response;
+use PhpLlm\LlmChain\Response\ResponseInterface;
+use PhpLlm\LlmChain\Response\StreamResponse;
 
 final class Claude implements LanguageModel
 {
@@ -28,7 +30,7 @@ public function __construct(
      * @param array<string, mixed> $options The options to be used for this specific call.
      *                                      Can overwrite default options.
      */
-    public function call(MessageBag $messages, array $options = []): Response
+    public function call(MessageBag $messages, array $options = []): ResponseInterface
     {
         $system = $messages->getSystemMessage();
         $body = array_merge($this->options, $options, [
@@ -39,6 +41,10 @@ public function call(MessageBag $messages, array $options = []): Response
 
         $response = $this->platform->request($body);
 
+        if ($response instanceof \Generator) {
+            return new StreamResponse($this->convertStream($response));
+        }
+
         return new Response(new Choice($response['content'][0]['text']));
     }
 
@@ -56,4 +62,15 @@ public function supportsStructuredOutput(): bool
     {
         return false;
     }
+
+    private function convertStream(\Generator $generator): \Generator
+    {
+        foreach ($generator as $data) {
+            if ('content_block_delta' != $data['type'] || !isset($data['delta']['text'])) {
+                continue;
+            }
+
+            yield $data['delta']['text'];
+        }
+    }
 }
diff --git a/src/Anthropic/Platform.php b/src/Anthropic/Platform.php
@@ -11,5 +11,5 @@ interface Platform
      *
      * @return array<string, mixed>
      */
-    public function request(array $body): array;
+    public function request(array $body): iterable;
 }
diff --git a/src/Anthropic/Platform/Anthropic.php b/src/Anthropic/Platform/Anthropic.php
@@ -5,22 +5,28 @@
 namespace PhpLlm\LlmChain\Anthropic\Platform;
 
 use PhpLlm\LlmChain\Anthropic\Platform;
+use Symfony\Component\HttpClient\Chunk\ServerSentEvent;
+use Symfony\Component\HttpClient\EventSourceHttpClient;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
 
 final readonly class Anthropic implements Platform
 {
+    private EventSourceHttpClient $httpClient;
+
     public function __construct(
-        private HttpClientInterface $httpClient,
+        HttpClientInterface $httpClient,
         private string $apiKey,
     ) {
+        $this->httpClient = $httpClient instanceof EventSourceHttpClient ? $httpClient : new EventSourceHttpClient($httpClient);
     }
 
     /**
      * @param array<string, mixed> $body
      *
      * @return array<string, mixed>
      */
-    public function request(array $body): array
+    public function request(array $body): iterable
     {
         $response = $this->httpClient->request('POST', 'https://api.anthropic.com/v1/messages', [
             'headers' => [
@@ -31,6 +37,21 @@ public function request(array $body): array
             'json' => $body,
         ]);
 
+        if ($body['stream'] ?? false) {
+            return $this->stream($response);
+        }
+
         return $response->toArray();
     }
+
+    private function stream(ResponseInterface $response): \Generator
+    {
+        foreach ((new EventSourceHttpClient())->stream($response) as $chunk) {
+            if (!$chunk instanceof ServerSentEvent || '[DONE]' === $chunk->getData()) {
+                continue;
+            }
+
+            yield $chunk->getArrayData();
+        }
+    }
 }
diff --git a/src/Chain/Output.php b/src/Chain/Output.php
@@ -6,7 +6,7 @@
 
 use PhpLlm\LlmChain\LanguageModel;
 use PhpLlm\LlmChain\Message\MessageBag;
-use PhpLlm\LlmChain\Response\Response;
+use PhpLlm\LlmChain\Response\ResponseInterface;
 
 final readonly class Output
 {
@@ -15,7 +15,7 @@
      */
     public function __construct(
         public LanguageModel $llm,
-        public Response $response,
+        public ResponseInterface $response,
         public MessageBag $messages,
         public array $options,
     ) {

diff --git a/src/LanguageModel.php b/src/LanguageModel.php
@@ -5,14 +5,14 @@
 namespace PhpLlm\LlmChain;
 
 use PhpLlm\LlmChain\Message\MessageBag;
-use PhpLlm\LlmChain\Response\Response;
+use PhpLlm\LlmChain\Response\ResponseInterface;
 
 interface LanguageModel
 {
     /**
      * @param array<string, mixed> $options
      */
-    public function call(MessageBag $messages, array $options = []): Response;
+    public function call(MessageBag $messages, array $options = []): ResponseInterface;
 
     public function supportsToolCalling(): bool;
 

diff --git a/src/OpenAI/Model/Gpt.php b/src/OpenAI/Model/Gpt.php
@@ -11,6 +11,8 @@
 use PhpLlm\LlmChain\OpenAI\Platform;
 use PhpLlm\LlmChain\Response\Choice;
 use PhpLlm\LlmChain\Response\Response;
+use PhpLlm\LlmChain\Response\ResponseInterface;
+use PhpLlm\LlmChain\Response\StreamResponse;
 use PhpLlm\LlmChain\Response\ToolCall;
 
 final class Gpt implements LanguageModel
@@ -30,7 +32,7 @@ public function __construct(
      * @param array<mixed> $options The options to be used for this specific call.
      *                              Can overwrite default options.
      */
-    public function call(MessageBag $messages, array $options = []): Response
+    public function call(MessageBag $messages, array $options = []): ResponseInterface
     {
         $body = array_merge($this->options, $options, [
             'model' => $this->version->name,
@@ -39,6 +41,10 @@ public function call(MessageBag $messages, array $options = []): Response
 
         $response = $this->platform->request('chat/completions', $body);
 
+        if ($response instanceof \Generator) {
+            return new StreamResponse($this->convertStream($response));
+        }
+
         return new Response(...array_map([$this, 'convertChoice'], $response['choices']));
     }
 
@@ -57,6 +63,17 @@ public function supportsStructuredOutput(): bool
         return $this->version->supportStructuredOutput;
     }
 
+    private function convertStream(\Generator $generator): \Generator
+    {
+        foreach ($generator as $data) {
+            if (!isset($data['choices'][0]['delta']['content'])) {
+                continue;
+            }
+
+            yield $data['choices'][0]['delta']['content'];
+        }
+    }
+
     /**
      * @param array{
      *     index: integer,

diff --git a/src/OpenAI/Platform.php b/src/OpenAI/Platform.php
@@ -11,7 +11,7 @@ interface Platform
      *
      * @return array<string, mixed>
      */
-    public function request(string $endpoint, array $body): array;
+    public function request(string $endpoint, array $body): iterable;
 
     /**
      * @param array<array<string, mixed>> $bodies

diff --git a/src/OpenAI/Platform/AbstractPlatform.php b/src/OpenAI/Platform/AbstractPlatform.php
@@ -6,15 +6,23 @@
 
 use PhpLlm\LlmChain\Exception\RuntimeException;
 use PhpLlm\LlmChain\OpenAI\Platform;
+use Symfony\Component\HttpClient\Chunk\ServerSentEvent;
+use Symfony\Component\HttpClient\EventSourceHttpClient;
 use Symfony\Component\HttpClient\Exception\ClientException;
 use Symfony\Contracts\HttpClient\ResponseInterface;
 
-abstract class AbstractPlatform implements Platform
+abstract readonly class AbstractPlatform implements Platform
 {
-    public function request(string $endpoint, array $body): array
+    public function request(string $endpoint, array $body): iterable
     {
+        $response = $this->rawRequest($endpoint, $body);
+
+        if ($body['stream'] ?? false) {
+            return $this->stream($response);
+        }
+
         try {
-            return $this->rawRequest($endpoint, $body)->toArray();
+            return $response->toArray();
         } catch (ClientException $e) {
             dump($e->getResponse()->getContent(false));
             throw new RuntimeException('Failed to make request', 0, $e);
@@ -33,6 +41,17 @@ public function multiRequest(string $endpoint, array $bodies): \Generator
         }
     }
 
+    private function stream(ResponseInterface $response): \Generator
+    {
+        foreach ((new EventSourceHttpClient())->stream($response) as $chunk) {
+            if (!$chunk instanceof ServerSentEvent || '[DONE]' === $chunk->getData()) {
+                continue;
+            }
+
+            yield $chunk->getArrayData();
+        }
+    }
+
     /**
      * @param array<string, mixed> $body
      */

diff --git a/src/OpenAI/Platform/Azure.php b/src/OpenAI/Platform/Azure.php
@@ -5,18 +5,22 @@
 namespace PhpLlm\LlmChain\OpenAI\Platform;
 
 use PhpLlm\LlmChain\OpenAI\Platform;
+use Symfony\Component\HttpClient\EventSourceHttpClient;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
 use Symfony\Contracts\HttpClient\ResponseInterface;
 
-final class Azure extends AbstractPlatform implements Platform
+final readonly class Azure extends AbstractPlatform implements Platform
 {
+    private EventSourceHttpClient $httpClient;
+
     public function __construct(
-        private readonly HttpClientInterface $httpClient,
-        private readonly string $baseUrl,
-        private readonly string $deployment,
-        private readonly string $apiVersion,
-        private readonly string $key,
+        HttpClientInterface $httpClient,
+        private string $baseUrl,
+        private string $deployment,
+        private string $apiVersion,
+        private string $key,
     ) {
+        $this->httpClient = $httpClient instanceof EventSourceHttpClient ? $httpClient : new EventSourceHttpClient($httpClient);
     }
 
     protected function rawRequest(string $endpoint, array $body): ResponseInterface

diff --git a/src/OpenAI/Platform/OpenAI.php b/src/OpenAI/Platform/OpenAI.php
@@ -5,15 +5,19 @@
 namespace PhpLlm\LlmChain\OpenAI\Platform;
 
 use PhpLlm\LlmChain\OpenAI\Platform;
+use Symfony\Component\HttpClient\EventSourceHttpClient;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
 use Symfony\Contracts\HttpClient\ResponseInterface;
 
-final class OpenAI extends AbstractPlatform implements Platform
+final readonly class OpenAI extends AbstractPlatform implements Platform
 {
+    private EventSourceHttpClient $httpClient;
+
     public function __construct(
-        private readonly HttpClientInterface $httpClient,
-        private readonly string $apiKey,
+        HttpClientInterface $httpClient,
+        private string $apiKey,
     ) {
+        $this->httpClient = $httpClient instanceof EventSourceHttpClient ? $httpClient : new EventSourceHttpClient($httpClient);
     }
 
     protected function rawRequest(string $endpoint, array $body): ResponseInterface