Merge pull request #44 from outl1ne/wip-streaming-support

wip-streaming-support

README.md

@@ -54,7 +54,7 @@ $assistant = OpenAI::assistants()->create(
 );
 $assistantModified = OpenAI::assistants()->modify($assistant->id, null, 'Allan\'s assistant!');
 $deletedAssistant = OpenAI::assistants()->delete($assistant->id);
-// dd($assistant->response()->json(), $assistantModified->response()->json(), $deletedAssistant->response()->json());
+// dd($assistant->response->json(), $assistantModified->response->json(), $deletedAssistant->response->json());
 ```
 
 Attaching, listing and deleting files.
@@ -84,9 +84,9 @@ $deletedAssistantFile = OpenAI::assistants()->files()->delete($assistant->id, $f
 $deletedAssistant = OpenAI::assistants()->delete($assistant->id);
 $deletedFile = OpenAI::files()->delete($file->id);
 // dd(
-//     $assistantFile->response()->json(),
-//     $assistantFiles->response()->json(),
-//     $deletedAssistantFile->response()->json(),
+//     $assistantFile->response->json(),
+//     $assistantFiles->response->json(),
+//     $deletedAssistantFile->response->json(),
 // );
 ```
 
@@ -97,7 +97,6 @@ $response = OpenAI::chat()->create(
     model: 'gpt-3.5-turbo',
     messages: (new Messages)->system('You are a helpful assistant.')->user('Hello!'),
 );
-$data = $response->response()->json();
 ```
 
 Enable JSON response formatting:
@@ -110,6 +109,17 @@ $response = OpenAI::chat()->create(
 );
 ```
 
+#### Streaming
+
+```php
+$response = OpenAI::chat()->stream(function (string $newChunk, string $message) {
+    echo $newChunk;
+})->create(
+    model: 'gpt-3.5-turbo',
+    messages: (new Messages)->system('You are a helpful assistant.')->user('Hello!'),
+);
+```
+
 ### Embeddings
 
 ```php
@@ -145,7 +155,7 @@ $file = OpenAI::files()->upload(
 $files = OpenAI::files()->list();
 $file2 = OpenAI::files()->retrieve($file->id);
 $deletedFile = OpenAI::files()->delete($file->id);
-// dd($file->response()->json(), $file2->response()->json(), $deletedFile->response()->json());
+// dd($file->response->json(), $file2->response->json(), $deletedFile->response->json());
 ```
 
 Retrieving a file content.
@@ -185,13 +195,13 @@ $messages = OpenAI::threads()->messages()->list($thread->id);
 $deletedThread = OpenAI::threads()->delete($thread->id);
 $deletedAssistant = OpenAI::assistants()->delete($assistant->id);
 // dd(
-//     $assistant->response()->json(),
-//     $thread->response()->json(),
-//     $message->response()->json(),
-//     $run->response()->json(),
-//     $messages->response()->json(),
-//     $deletedThread->response()->json(),
-//     $deletedAssistant->response()->json(),
+//     $assistant->response->json(),
+//     $thread->response->json(),
+//     $message->response->json(),
+//     $run->response->json(),
+//     $messages->response->json(),
+//     $deletedThread->response->json(),
+//     $deletedAssistant->response->json(),
 // );
 ```
 

src/Capabilities/Capability.php

@@ -11,6 +11,7 @@ class Capability
     public Closure $shouldStoreCallback;
     public Closure $shouldStoreErrorsCallback;
     public Closure $storingCallback;
+    public ?Closure $streamCallback = null;
 
     public function __construct(
         public readonly OpenAI $openAI,
@@ -21,27 +22,33 @@ public function __construct(
         $this->storing(fn ($model) => $model);
     }
 
-    public function shouldStorePending(Closure $callback): self
+    public function shouldStorePending(Closure $callback): static
     {
         $this->shouldStorePendingCallback = $callback;
         return $this;
     }
 
-    public function shouldStore(Closure $callback): self
+    public function shouldStore(Closure $callback): static
     {
         $this->shouldStoreCallback = $callback;
         return $this;
     }
 
-    public function shouldStoreErrors(Closure $callback): self
+    public function shouldStoreErrors(Closure $callback): static
     {
         $this->shouldStoreErrorsCallback = $callback;
         return $this;
     }
 
-    public function storing(Closure $callback): self
+    public function storing(Closure $callback): static
     {
         $this->storingCallback = $callback;
         return $this;
     }
+
+    public function stream(Closure $callback): static
+    {
+        $this->streamCallback = $callback;
+        return $this;
+    }
 }

src/Capabilities/CapabilityClient.php

@@ -2,12 +2,16 @@
 
 namespace Outl1ne\NovaOpenAI\Capabilities;
 
+use Closure;
 use Exception;
-use Outl1ne\NovaOpenAI\Capabilities\Responses\CachedResponse;
 use Outl1ne\NovaOpenAI\OpenAI;
+use Outl1ne\NovaOpenAI\StreamHandler;
+use Outl1ne\NovaOpenAI\Pricing\Calculator;
 use Outl1ne\NovaOpenAI\Models\OpenAIRequest;
+use Illuminate\Http\Client\Response as HttpResponse;
 use Outl1ne\NovaOpenAI\Capabilities\Responses\Response;
-use Outl1ne\NovaOpenAI\Pricing\Calculator;
+use Outl1ne\NovaOpenAI\Capabilities\Responses\CachedResponse;
+use Outl1ne\NovaOpenAI\Capabilities\Responses\StreamChunk;
 
 abstract class CapabilityClient
 {
@@ -24,6 +28,12 @@ public function __construct(
         $this->request = new OpenAIRequest;
         $this->request->method = $this->method;
         $this->request->arguments = [];
+        if ($capability->streamCallback instanceof Closure) {
+            $this->request->appendArgument('stream', true);
+            $this->request->appendArgument('stream_options', [
+                'include_usage' => true,
+            ]);
+        }
     }
 
     public function pending()
@@ -65,6 +75,41 @@ protected function handleResponse(Response $response, ?callable $handleResponse
         return $response;
     }
 
+    protected function isStreamedResponse(HttpResponse $response)
+    {
+        return strpos($response->getHeaderLine('Content-Type'), 'text/event-stream') !== false;
+    }
+
+    protected function handleStreamedResponse(HttpResponse $httpResponse, ?callable $handleResponse = null)
+    {
+        if (!$this->capability->streamCallback instanceof Closure) {
+            throw new Exception('Response is a stream but stream callback is not defined.');
+        }
+
+        $response = (new StreamHandler($httpResponse, $this->capability->streamCallback, function (StreamChunk $streamChunk) {
+            $this->request->status = 'streaming';
+            $this->request->meta = $streamChunk?->meta;
+            $this->request->model_used = $streamChunk?->model;
+            if (($this->capability->shouldStoreCallback)($streamChunk)) {
+                $this->request->save();
+            }
+        }))->handle();
+
+        $this->request->cost = $this->calculateCost($response);
+        $this->request->time_sec = $this->measure();
+        $this->request->status = 'success';
+        $this->request->meta = $response?->meta;
+        $this->request->usage_prompt_tokens = $response->usage?->promptTokens;
+        $this->request->usage_completion_tokens = $response->usage?->completionTokens;
+        $this->request->usage_total_tokens = $response->usage?->totalTokens;
+
+        if ($handleResponse !== null) $handleResponse($response);
+        $this->store($response);
+        $response->request = $this->request;
+
+        return $response;
+    }
+
     protected function handleCachedResponse(CachedResponse $cachedResponse, ?callable $handleResponse = null)
     {
         $this->request->time_sec = $this->measure();

src/Capabilities/Chat/CreateChat.php

@@ -3,10 +3,12 @@
 namespace Outl1ne\NovaOpenAI\Capabilities\Chat;
 
 use Exception;
+use Illuminate\Http\Client\Response;
 use Outl1ne\NovaOpenAI\Capabilities\CapabilityClient;
 use Outl1ne\NovaOpenAI\Capabilities\Chat\Parameters\Messages;
 use Outl1ne\NovaOpenAI\Capabilities\Chat\Responses\ChatResponse;
 use Outl1ne\NovaOpenAI\Capabilities\Chat\Parameters\ResponseFormat;
+use Outl1ne\NovaOpenAI\Capabilities\Chat\Responses\StreamedChatResponse;
 
 class CreateChat extends CapabilityClient
 {
@@ -30,7 +32,7 @@ public function makeRequest(
         ?float $topP = null,
         ?array $tools = null,
         null|string|array $toolChoice = null,
-    ): ChatResponse {
+    ): ChatResponse|StreamedChatResponse {
         $this->request->model_requested = $model;
         $this->request->input = $messages->messages;
         $this->request->appendArgument('response_format', $responseFormat->responseFormat ?? null);
@@ -59,13 +61,16 @@ public function makeRequest(
             ]);
             $response->throw();
 
+            if ($this->isStreamedResponse($response)) {
+                return $this->handleStreamedResponse($response, [$this, 'response']);
+            }
             return $this->handleResponse(new ChatResponse($response), [$this, 'response']);
         } catch (Exception $e) {
             $this->handleException($e);
         }
     }
 
-    protected function response(ChatResponse $response)
+    protected function response(ChatResponse|StreamedChatResponse $response)
     {
         $this->request->output = $response->choices;
     }

src/Capabilities/Chat/Responses/StreamedChatChunk.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace Outl1ne\NovaOpenAI\Capabilities\Chat\Responses;
+
+use Outl1ne\NovaOpenAI\Capabilities\Responses\StreamChunk;
+
+
+class StreamedChatChunk extends StreamChunk
+{
+    public array $choices;
+
+    public function __construct($data)
+    {
+        parent::__construct($data);
+
+        $this->model = $this->data['model'];
+        $this->choices = $this->data['choices'];
+        $this->appendMeta('id', $this->data['id']);
+        $this->appendMeta('object', $this->data['object']);
+        $this->appendMeta('created', $this->data['created']);
+        $this->appendMeta('system_fingerprint', $this->data['system_fingerprint']);
+    }
+}

src/Capabilities/Chat/Responses/StreamedChatResponse.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace Outl1ne\NovaOpenAI\Capabilities\Chat\Responses;
+
+use Illuminate\Http\Client\Response;
+use Outl1ne\NovaOpenAI\Capabilities\Responses\StreamChunk;
+use Outl1ne\NovaOpenAI\Capabilities\Responses\StreamResponse;
+
+class StreamedChatResponse extends StreamResponse
+{
+    public array $choices;
+
+    public function __construct(Response $response, StreamChunk ...$streamChunks)
+    {
+        parent::__construct($response, ...$streamChunks);
+
+        $lastChunk = end($this->streamChunks);
+        $this->model = $lastChunk->model;
+        $this->appendMeta('id', $lastChunk->meta['id']);
+        $this->appendMeta('object', $lastChunk->meta['object']);
+        $this->appendMeta('system_fingerprint', $lastChunk->meta['system_fingerprint'] ?? null);
+
+        $this->choices = $this->createChoices();
+    }
+
+    protected function createChoices(): array
+    {
+        $choices = [];
+
+        foreach ($this->streamChunks as $chunk) {
+            foreach ($chunk->choices as $choice) {
+                $choices[$choice['index']]['index'] = $choice['index'];
+                $choices[$choice['index']]['message']['role'] = $this->handleDelta($choices[$choice['index']]['message']['role'] ?? null, $choice['delta']['role'] ?? null);
+                $choices[$choice['index']]['message']['content'] = $this->handleDelta($choices[$choice['index']]['message']['content'] ?? null, $choice['delta']['content'] ?? null);
+                $choices[$choice['index']]['logprobs'] = $choice['logprobs'];
+                $choices[$choice['index']]['finish_reason'] = $choice['finish_reason'];
+            }
+        }
+
+        return $choices;
+    }
+
+    protected function handleDelta($carry, $chunk): ?string
+    {
+        if ($carry === null && $chunk === null) {
+            return null;
+        }
+
+        if ($carry === null) {
+            return $chunk;
+        }
+
+        return $carry . $chunk;
+    }
+}

src/Capabilities/Responses/Response.php

@@ -18,7 +18,7 @@ class Response
     public ?string $model = null;
 
     public function __construct(
-        protected readonly HttpResponse $response,
+        public readonly HttpResponse $response,
     ) {
         $this->data = $response->json();
         $this->headers = $response->headers();
@@ -27,11 +27,6 @@ public function __construct(
         $this->rateLimit = $this->createRateLimit();
     }
 
-    public function response()
-    {
-        return $this->response;
-    }
-
     protected function createUsage()
     {
         $promptTokens = $this->data['usage']['prompt_tokens'] ?? null;

src/Capabilities/Responses/StreamChunk.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace Outl1ne\NovaOpenAI\Capabilities\Responses;
+
+class StreamChunk
+{
+    use AppendsMeta;
+
+    public ?Usage $usage;
+
+    public ?string $model = null;
+
+    public function __construct(
+        protected readonly array $data,
+    ) {
+
+        $this->usage = $this->createUsage();
+    }
+
+    protected function createUsage()
+    {
+        $promptTokens = $this->data['usage']['prompt_tokens'] ?? null;
+        $completionTokens = $this->data['usage']['completion_tokens'] ?? null;
+        $totalTokens = $this->data['usage']['total_tokens'] ?? null;
+
+        if ($totalTokens === null) return null;
+
+        return new Usage(
+            $promptTokens,
+            $completionTokens,
+            $totalTokens,
+        );
+    }
+}