Implement Assistants Streaming

README.md

@@ -1426,6 +1426,82 @@ $response->metadata; // []
 $response->toArray(); // ['id' => 'run_4RCYyYzX9m41WQicoJtUQAb8', ...]
 ```
 
+#### `create streamed`
+
+Creates a streamed run.
+
+[OpenAI Assistant Events](https://platform.openai.com/docs/api-reference/assistants-streaming/events)
+
+```php
+$stream = $client->threads()->runs()->createStreamed(
+    threadId: 'thread_tKFLqzRN9n7MnyKKvc1Q7868',
+    parameters: [
+        'assistant_id' => 'asst_gxzBkD1wkKEloYqZ410pT5pd',
+    ],
+);
+
+foreach($stream as $response){
+    $response->event // 'thread.run.created' | 'thread.run.in_progress' | .....
+    $response->data // ThreadResponse | ThreadRunResponse | ThreadRunStepResponse | ThreadRunStepDeltaResponse | ThreadMessageResponse | ThreadMessageDeltaResponse
+}
+
+// ...
+```
+
+#### `create streamed with function calls`
+
+Creates a streamed run with function calls
+
+[OpenAI Assistant Events](https://platform.openai.com/docs/api-reference/assistants-streaming/events)
+
+```php
+$stream = $client->threads()->runs()->createStreamed(
+    threadId: 'thread_tKFLqzRN9n7MnyKKvc1Q7868',
+    parameters: [
+        'assistant_id' => 'asst_gxzBkD1wkKEloYqZ410pT5pd',
+    ],
+);
+
+
+do{
+    foreach($stream as $response){
+        $response->event // 'thread.run.created' | 'thread.run.in_progress' | .....
+        $response->data // ThreadResponse | ThreadRunResponse | ThreadRunStepResponse | ThreadRunStepDeltaResponse | ThreadMessageResponse | ThreadMessageDeltaResponse
+
+        switch($response->event){
+            case 'thread.run.created':
+            case 'thread.run.queued':
+            case 'thread.run.completed':
+            case 'thread.run.cancelling':
+                $run = $response->data;
+                break;
+            case 'thread.run.expired':
+            case 'thread.run.cancelled':
+            case 'thread.run.failed':
+                $run = $response->data;
+                break 3;
+            case 'thread.run.requires_action':
+                // Overwrite the stream with the new stream started by submitting the tool outputs
+                $stream = $client->threads()->runs()->submitToolOutputsStreamed(
+                    threadId: $run->threadId,
+                    runId: $run->id,
+                    parameters: [
+                        'tool_outputs' => [
+                            [
+                                'tool_call_id' => 'call_KSg14X7kZF2WDzlPhpQ168Mj',
+                                'output' => '12',
+                            ]
+                        ],
+                    ]
+                );
+                break;
+        }
+    }
+} while ($run->status != "completed")
+
+// ...
+```
+
 #### `retrieve`
 
 Retrieves a run.

src/Contracts/Resources/ThreadsContract.php

@@ -2,7 +2,9 @@
 
 namespace OpenAI\Contracts\Resources;
 
+use OpenAI\Responses\StreamResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunResponse;
+use OpenAI\Responses\Threads\Runs\ThreadRunStreamResponse;
 use OpenAI\Responses\Threads\ThreadDeleteResponse;
 use OpenAI\Responses\Threads\ThreadResponse;
 
@@ -26,6 +28,16 @@ public function create(array $parameters): ThreadResponse;
      */
     public function createAndRun(array $parameters): ThreadRunResponse;
 
+    /**
+     * Create a thread and run it in one request, returning a stream.
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/createThreadAndRun
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function createAndRunStreamed(array $parameters): StreamResponse;
+
     /**
      * Retrieves a thread.
      *

src/Contracts/Resources/ThreadsRunsContract.php

@@ -2,8 +2,10 @@
 
 namespace OpenAI\Contracts\Resources;
 
+use OpenAI\Responses\StreamResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunListResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunResponse;
+use OpenAI\Responses\Threads\Runs\ThreadRunStreamResponse;
 
 interface ThreadsRunsContract
 {
@@ -16,6 +18,16 @@ interface ThreadsRunsContract
      */
     public function create(string $threadId, array $parameters): ThreadRunResponse;
 
+    /**
+     * Create a streamed run.
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/createRun
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function createStreamed(string $threadId, array $parameters): StreamResponse;
+
     /**
      * Retrieves a run.
      *
@@ -41,6 +53,17 @@ public function modify(string $threadId, string $runId, array $parameters): Thre
      */
     public function submitToolOutputs(string $threadId, string $runId, array $parameters): ThreadRunResponse;
 
+    /**
+     * This endpoint can be used to submit the outputs from the tool calls once they're all completed.
+     * And stream back the response
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/submitToolOutputs
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function submitToolOutputsStreamed(string $threadId, string $runId, array $parameters): StreamResponse;
+
     /**
      * Cancels a run that is `in_progress`.
      *

src/Exceptions/UnknownEventException.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenAI\Exceptions;
+
+use Exception;
+
+final class UnknownEventException extends Exception
+{
+}

src/Resources/Threads.php

@@ -7,14 +7,17 @@
 use OpenAI\Contracts\Resources\ThreadsContract;
 use OpenAI\Contracts\Resources\ThreadsMessagesContract;
 use OpenAI\Contracts\Resources\ThreadsRunsContract;
+use OpenAI\Responses\StreamResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunResponse;
+use OpenAI\Responses\Threads\Runs\ThreadRunStreamResponse;
 use OpenAI\Responses\Threads\ThreadDeleteResponse;
 use OpenAI\Responses\Threads\ThreadResponse;
 use OpenAI\ValueObjects\Transporter\Payload;
 use OpenAI\ValueObjects\Transporter\Response;
 
 final class Threads implements ThreadsContract
 {
+    use Concerns\Streamable;
     use Concerns\Transportable;
 
     /**
@@ -51,6 +54,25 @@ public function createAndRun(array $parameters): ThreadRunResponse
         return ThreadRunResponse::from($response->data(), $response->meta());
     }
 
+    /**
+     * Create a thread and run it in one request, returning a stream.
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/createThreadAndRun
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function createAndRunStreamed(array $parameters): StreamResponse
+    {
+        $parameters = $this->setStreamParameter($parameters);
+
+        $payload = Payload::create('threads/runs', $parameters);
+
+        $response = $this->transporter->requestStream($payload);
+
+        return new StreamResponse(ThreadRunStreamResponse::class, $response);
+    }
+
     /**
      * Retrieves a thread.
      *

src/Resources/ThreadsRuns.php

@@ -6,13 +6,16 @@
 
 use OpenAI\Contracts\Resources\ThreadsRunsContract;
 use OpenAI\Contracts\Resources\ThreadsRunsStepsContract;
+use OpenAI\Responses\StreamResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunListResponse;
 use OpenAI\Responses\Threads\Runs\ThreadRunResponse;
+use OpenAI\Responses\Threads\Runs\ThreadRunStreamResponse;
 use OpenAI\ValueObjects\Transporter\Payload;
 use OpenAI\ValueObjects\Transporter\Response;
 
 final class ThreadsRuns implements ThreadsRunsContract
 {
+    use Concerns\Streamable;
     use Concerns\Transportable;
 
     /**
@@ -32,6 +35,25 @@ public function create(string $threadId, array $parameters): ThreadRunResponse
         return ThreadRunResponse::from($response->data(), $response->meta());
     }
 
+    /**
+     * Creates a streamed run
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/createRun
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function createStreamed(string $threadId, array $parameters): StreamResponse
+    {
+        $parameters = $this->setStreamParameter($parameters);
+
+        $payload = Payload::create('threads/'.$threadId.'/runs', $parameters);
+
+        $response = $this->transporter->requestStream($payload);
+
+        return new StreamResponse(ThreadRunStreamResponse::class, $response);
+    }
+
     /**
      * Retrieves a run.
      *
@@ -81,6 +103,26 @@ public function submitToolOutputs(string $threadId, string $runId, array $parame
         return ThreadRunResponse::from($response->data(), $response->meta());
     }
 
+    /**
+     * This endpoint can be used to submit the outputs from the tool calls once they're all completed.
+     * And stream back the response
+     *
+     * @see https://platform.openai.com/docs/api-reference/runs/submitToolOutputs
+     *
+     * @param  array<string, mixed>  $parameters
+     * @return StreamResponse<ThreadRunStreamResponse>
+     */
+    public function submitToolOutputsStreamed(string $threadId, string $runId, array $parameters): StreamResponse
+    {
+        $parameters = $this->setStreamParameter($parameters);
+
+        $payload = Payload::create('threads/'.$threadId.'/runs/'.$runId.'/submit_tool_outputs', $parameters);
+
+        $response = $this->transporter->requestStream($payload);
+
+        return new StreamResponse(ThreadRunStreamResponse::class, $response);
+    }
+
     /**
      * Cancels a run that is `in_progress`.
      *

src/Resources/ThreadsRunsSteps.php

@@ -23,7 +23,7 @@ public function retrieve(string $threadId, string $runId, string $stepId): Threa
     {
         $payload = Payload::retrieve('threads/'.$threadId.'/runs/'.$runId.'/steps', $stepId);
 
-        /** @var Response<array{id: string, object: string, created_at: int, thread_id: string, assistant_id: string, run_id: string, type: string, status: string, step_details: array{type: 'tool_calls', tool_calls: array<int, array{id: string, type: 'code_interpreter', code_interpreter: array{input: string, outputs: array<int, array{type: 'image', image: array{file_id: string}}|array{type: 'logs', logs: string}>}}|array{id: string, type: 'retrieval', retrieval: array<string, string>}|array{id: string, type: 'function', function: array{name: string, arguments: string, output?: ?string}}>}|array{type: 'message_creation', message_creation: array{message_id: string}}, last_error: ?array{code: string, message: string}, expires_at: ?int, cancelled_at: ?int, failed_at: ?int, completed_at: ?int, metadata?: array<string, string>}> $response */
+        /** @var Response<array{id: string, object: string, created_at: int, thread_id: string, assistant_id: string, run_id: string, type: string, status: string, step_details: array{type: 'tool_calls', tool_calls: array<int, array{id?: string, type: 'code_interpreter', code_interpreter: array{input: string, outputs: array<int, array{type: 'image', image: array{file_id: string}}|array{type: 'logs', logs: string}>}}|array{id: string, type: 'retrieval', retrieval: array<string, string>}|array{id: string, type: 'function', function: array{name: string, arguments: string, output?: ?string}}>}|array{type: 'message_creation', message_creation: array{message_id: string}}, last_error: ?array{code: string, message: string}, expires_at: ?int, cancelled_at: ?int, failed_at: ?int, completed_at: ?int, metadata?: array<string, string>}> $response */
         $response = $this->transporter->requestObject($payload);
 
         return ThreadRunStepResponse::from($response->data(), $response->meta());
@@ -40,7 +40,7 @@ public function list(string $threadId, string $runId, array $parameters = []): T
     {
         $payload = Payload::list('threads/'.$threadId.'/runs/'.$runId.'/steps', $parameters);
 
-        /** @var Response<array{object: string, data: array<int, array{id: string, object: string, created_at: int, thread_id: string, assistant_id: string, run_id: string, type: string, status: string, step_details: array{type: 'tool_calls', tool_calls: array<int, array{id: string, type: 'code_interpreter', code_interpreter: array{input: string, outputs: array<int, array{type: 'image', image: array{file_id: string}}|array{type: 'logs', logs: string}>}}|array{id: string, type: 'retrieval', retrieval: array<string, string>}|array{id: string, type: 'function', function: array{name: string, arguments: string, output: ?string}}>}|array{type: 'message_creation', message_creation: array{message_id: string}}, last_error: ?array{code: string, message: string}, expires_at: ?int, cancelled_at: ?int, failed_at: ?int, completed_at: ?int, metadata?: array<string, string>}>, first_id: ?string, last_id: ?string, has_more: bool}> $response */
+        /** @var Response<array{object: string, data: array<int, array{id: string, object: string, created_at: int, thread_id: string, assistant_id: string, run_id: string, type: string, status: string, step_details: array{type: 'tool_calls', tool_calls: array<int, array{id?: string, type: 'code_interpreter', code_interpreter: array{input: string, outputs: array<int, array{type: 'image', image: array{file_id: string}}|array{type: 'logs', logs: string}>}}|array{id: string, type: 'retrieval', retrieval: array<string, string>}|array{id: string, type: 'function', function: array{name: string, arguments: string, output: ?string}}>}|array{type: 'message_creation', message_creation: array{message_id: string}}, last_error: ?array{code: string, message: string}, expires_at: ?int, cancelled_at: ?int, failed_at: ?int, completed_at: ?int, metadata?: array<string, string>}>, first_id: ?string, last_id: ?string, has_more: bool}> $response */
         $response = $this->transporter->requestObject($payload);
 
         return ThreadRunStepListResponse::from($response->data(), $response->meta());

src/Responses/StreamResponse.php

@@ -37,6 +37,12 @@ public function getIterator(): Generator
         while (! $this->response->getBody()->eof()) {
             $line = $this->readLine($this->response->getBody());
 
+            $event = null;
+            if (str_starts_with($line, 'event:')) {
+                $event = trim(substr($line, strlen('event:')));
+                $line = $this->readLine($this->response->getBody());
+            }
+
             if (! str_starts_with($line, 'data:')) {
                 continue;
             }
@@ -54,6 +60,11 @@ public function getIterator(): Generator
                 throw new ErrorException($response['error']);
             }
 
+            if ($event !== null) {
+                $response['__event'] = $event;
+                $response['__meta'] = $this->meta();
+            }
+
             yield $this->responseClass::from($response);
         }
     }

src/Responses/Threads/Messages/Delta/ThreadMessageDeltaObject.php

@@ -0,0 +1,67 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenAI\Responses\Threads\Messages\Delta;
+
+use OpenAI\Contracts\ResponseContract;
+use OpenAI\Responses\Concerns\ArrayAccessible;
+use OpenAI\Testing\Responses\Concerns\Fakeable;
+
+/**
+ * @implements ResponseContract<array{role: ?string, content: array<int, array{index: int, type: 'image_file', image_file: array{file_id: string}}|array{index: int, type: 'text', text: array{value: string, annotations: array<int, array{type: 'file_citation', text: string, file_citation: array{file_id: string, quote: string}, start_index: int, end_index: int}|array{type: 'file_path', text: string, file_path: array{file_id: string}, start_index: int, end_index: int}>}}>, file_ids: ?array<int, string>}>
+ */
+final class ThreadMessageDeltaObject implements ResponseContract
+{
+    /**
+     * @use ArrayAccessible<array{role: ?string, content: array<int, array{index: int, type: 'image_file', image_file: array{file_id: string}}|array{index: int, type: 'text', text: array{value: string, annotations: array<int, array{type: 'file_citation', text: string, file_citation: array{file_id: string, quote: string}, start_index: int, end_index: int}|array{type: 'file_path', text: string, file_path: array{file_id: string}, start_index: int, end_index: int}>}}>, file_ids: ?array<int, string>}>
+     */
+    use ArrayAccessible;
+
+    use Fakeable;
+
+    /**
+     * @param  array<int, ThreadMessageDeltaResponseContentImageFileObject|ThreadMessageDeltaResponseContentTextObject>  $content
+     * @param  array<int, string>|null  $fileIds
+     */
+    private function __construct(
+        public ?string $role,
+        public array $content,
+        public ?array $fileIds,
+    ) {
+    }
+
+    /**
+     * Acts as static factory, and returns a new Response instance.
+     *
+     * @param  array{role?: string, content: array<int, array{index: int, type: 'image_file', image_file: array{file_id: string}}|array{index: int, type: 'text', text: array{value: string, annotations: array<int, array{type: 'file_citation', text: string, file_citation: array{file_id: string, quote: string}, start_index: int, end_index: int}|array{type: 'file_path', text: string, file_path: array{file_id: string}, start_index: int, end_index: int}>}}>, file_ids?: array<int, string>}  $attributes
+     */
+    public static function from(array $attributes): self
+    {
+        $content = array_map(
+            fn (array $content): ThreadMessageDeltaResponseContentTextObject|ThreadMessageDeltaResponseContentImageFileObject => match ($content['type']) {
+                'text' => ThreadMessageDeltaResponseContentTextObject::from($content),
+                'image_file' => ThreadMessageDeltaResponseContentImageFileObject::from($content),
+            },
+            $attributes['content'],
+        );
+
+        return new self(
+            $attributes['role'] ?? null,
+            $content,
+            $attributes['file_ids'] ?? null,
+        );
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function toArray(): array
+    {
+        return [
+            'role' => $this->role,
+            'content' => array_map(fn (ThreadMessageDeltaResponseContentImageFileObject|ThreadMessageDeltaResponseContentTextObject $content): array => $content->toArray(), $this->content),
+            'file_ids' => $this->fileIds,
+        ];
+    }
+}