feat: introducing async responses for parallel calling (#141)

Author: chr-hertel

README.md

@@ -430,6 +430,33 @@ dump($vectors[0]->getData()); // Array of float values
 1. **OpenAI's Emebddings**: [embeddings-openai.php](examples/embeddings-openai.php)
 1. **Voyage's Embeddings**: [embeddings-voyage.php](examples/embeddings-voyage.php)
 
+### Parallel Platform Calls
+
+Platform supports multiple model calls in parallel, which can be useful to speed up the processing:
+
+```php
+// Initialize Platform & Model
+
+foreach ($inputs as $input) {
+    $responses[] = $platform->request($model, $input);
+}
+
+foreach ($responses as $response) {
+    echo $response->getContent().PHP_EOL;
+}
+```
+
+> [!NOTE]
+> This requires cURL and the `ext-curl` extension to be installed.
+
+#### Code Examples
+
+1. **Parallel GPT Calls**: [parallel-chat-gpt.php](examples/parallel-chat-gpt.php)
+1. **Parallel Embeddings Calls**: [parallel-embeddings.php](examples/parallel-embeddings.php)
+
+> [!NOTE]
+> Please be aware that some embeddings models also support batch processing out of the box.
+
 ### Input & Output Processing
 
 The behavior of the Chain is extendable with services that implement `InputProcessor` and/or `OutputProcessor`

examples/parallel-chat-gpt.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+use PhpLlm\LlmChain\Bridge\OpenAI\GPT;
+use PhpLlm\LlmChain\Bridge\OpenAI\PlatformFactory;
+use PhpLlm\LlmChain\Model\Message\Message;
+use PhpLlm\LlmChain\Model\Message\MessageBag;
+use Symfony\Component\Dotenv\Dotenv;
+
+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 = PlatformFactory::create($_ENV['OPENAI_API_KEY']);
+$llm = new GPT(GPT::GPT_4O_MINI, [
+    'temperature' => 0.5, // default options for the model
+]);
+
+$messages = new MessageBag(
+    Message::forSystem('You will be given a letter and you answer with only the next letter of the alphabet.'),
+);
+
+echo 'Initiating parallel calls to GPT on platform ...'.PHP_EOL;
+$responses = [];
+foreach (range('A', 'D') as $letter) {
+    echo ' - Request for the letter '.$letter.' initiated.'.PHP_EOL;
+    $responses[] = $platform->request($llm, $messages->with(Message::ofUser($letter)));
+}
+
+echo 'Waiting for the responses ...'.PHP_EOL;
+foreach ($responses as $response) {
+    echo 'Next Letter: '.$response->getContent().PHP_EOL;
+}

examples/parallel-embeddings.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+use PhpLlm\LlmChain\Bridge\OpenAI\Embeddings;
+use PhpLlm\LlmChain\Bridge\OpenAI\PlatformFactory;
+use PhpLlm\LlmChain\Model\Response\VectorResponse;
+use Symfony\Component\Dotenv\Dotenv;
+
+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 = PlatformFactory::create($_ENV['OPENAI_API_KEY']);
+$ada = new Embeddings(Embeddings::TEXT_ADA_002);
+$small = new Embeddings(Embeddings::TEXT_3_SMALL);
+$large = new Embeddings(Embeddings::TEXT_3_LARGE);
+
+echo 'Initiating parallel embeddings calls to platform ...'.PHP_EOL;
+$responses = [];
+foreach (['ADA' => $ada, 'Small' => $small, 'Large' => $large] as $name => $model) {
+    echo ' - Request for model '.$name.' initiated.'.PHP_EOL;
+    $responses[] = $platform->request($model, 'Hello, world!');
+}
+
+echo 'Waiting for the responses ...'.PHP_EOL;
+foreach ($responses as $response) {
+    assert($response instanceof VectorResponse);
+    echo 'Dimensions: '.$response->getContent()[0]->getDimensions().PHP_EOL;
+}

src/Chain.php

@@ -13,6 +13,7 @@
 use PhpLlm\LlmChain\Exception\MissingModelSupport;
 use PhpLlm\LlmChain\Model\LanguageModel;
 use PhpLlm\LlmChain\Model\Message\MessageBag;
+use PhpLlm\LlmChain\Model\Response\AsyncResponse;
 use PhpLlm\LlmChain\Model\Response\ResponseInterface;
 
 final readonly class Chain implements ChainInterface
@@ -55,6 +56,10 @@ public function call(MessageBag $messages, array $options = []): ResponseInterfa
 
         $response = $this->platform->request($this->llm, $messages, $options = $input->getOptions());
 
+        if ($response instanceof AsyncResponse) {
+            $response = $response->unwrap();
+        }
+
         $output = new Output($this->llm, $response, $messages, $options);
         array_map(fn (OutputProcessor $processor) => $processor->processOutput($output), $this->outputProcessor);
 

src/Model/Response/AsyncResponse.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpLlm\LlmChain\Model\Response;
+
+use PhpLlm\LlmChain\Platform\ResponseConverter;
+use Symfony\Contracts\HttpClient\ResponseInterface as HttpResponse;
+
+final class AsyncResponse implements ResponseInterface
+{
+    private bool $isConverted = false;
+    private ResponseInterface $convertedResponse;
+
+    /**
+     * @param array<string, mixed> $options
+     */
+    public function __construct(
+        private readonly ResponseConverter $responseConverter,
+        private readonly HttpResponse $response,
+        private readonly array $options = [],
+    ) {
+    }
+
+    public function getContent(): string|iterable|object|null
+    {
+        return $this->unwrap()->getContent();
+    }
+
+    public function unwrap(): ResponseInterface
+    {
+        if (!$this->isConverted) {
+            $this->convertedResponse = $this->responseConverter->convert($this->response, $this->options);
+            $this->isConverted = true;
+        }
+
+        return $this->convertedResponse;
+    }
+}

src/Platform.php

@@ -7,6 +7,7 @@
 use PhpLlm\LlmChain\Exception\InvalidArgumentException;
 use PhpLlm\LlmChain\Exception\RuntimeException;
 use PhpLlm\LlmChain\Model\Model;
+use PhpLlm\LlmChain\Model\Response\AsyncResponse;
 use PhpLlm\LlmChain\Model\Response\ResponseInterface;
 use PhpLlm\LlmChain\Platform\ModelClient;
 use PhpLlm\LlmChain\Platform\ResponseConverter;
@@ -80,7 +81,7 @@ private function convertResponse(Model $model, object|array|string $input, HttpR
     {
         foreach ($this->responseConverter as $responseConverter) {
             if ($responseConverter->supports($model, $input)) {
-                return $responseConverter->convert($response, $options);
+                return new AsyncResponse($responseConverter, $response, $options);
             }
         }