Implement responses API streaming

README.md

@@ -10,7 +10,7 @@ OpenAI APIs for XP
 
 This library implements OpenAI APIs with a low-level abstraction approach, supporting their REST and realtime APIs, request and response streaming, function calling and TikToken encoding.
 
-Completions
+Quick start
 -----------
 Using the REST API, see https://platform.openai.com/docs/api-reference/making-requests
 
@@ -19,12 +19,11 @@ use com\openai\rest\OpenAIEndpoint;
 use util\cmd\Console;
 
 $ai= new OpenAIEndpoint('https://'.getenv('OPENAI_API_KEY').'@api.openai.com/v1');
-$payload= [
-  'model'    => 'gpt-4o-mini',
-  'messages' => [['role' => 'user', 'content' => $prompt]],
-];
 
-Console::writeLine($ai->api('/chat/completions')->invoke($payload));
+Console::writeLine($ai->api('/responses')->invoke([
+  'model' => 'gpt-4o-mini',
+  'input' => $prompt,
+]));
 ```
 
 Streaming
@@ -36,19 +35,18 @@ use com\openai\rest\OpenAIEndpoint;
 use util\cmd\Console;
 
 $ai= new OpenAIEndpoint('https://'.getenv('OPENAI_API_KEY').'@api.openai.com/v1');
-$payload= [
-  'model'    => 'gpt-4o-mini',
-  'messages' => [['role' => 'user', 'content' => $prompt]],
-];
 
-$stream= $ai->api('/chat/completions')->stream($payload);
-foreach ($stream->deltas('content') as $delta) {
-  Console::write($delta);
+$events= $ai->api('/responses')->stream([
+  'model' => 'gpt-4o-mini',
+  'input' => $prompt,
+]);
+foreach ($events as $type => $value) {
+  Console::write('<', $type, '> ', $value);
 }
 Console::writeLine();
 ```
 
-To access the result object after streaming, use `$stream->result()`. It contains the choices list as well as model, filter results and usage information.
+To access the result object, check for the *response.completed* event type and use its value. It contains the outuputs as well as model, filter results and usage information.
 
 TikToken
 --------
@@ -197,9 +195,9 @@ $functions= (new Functions())->register('weather', new Weather());
 
 $ai= new OpenAIEndpoint('https://'.getenv('OPENAI_API_KEY').'@api.openai.com/v1');
 $payload= [
-  'model'    => 'gpt-4o-mini',
-  'tools'    => new Tools($functions),
-  'messages' => [['role' => 'user', 'content' => $prompt]],
+  'model' => 'gpt-4o-mini',
+  'tools' => new Tools($functions),
+  'input' => [['type' => 'message', 'role' => 'user', 'content' => $content]],
 ];
 ```
 
@@ -213,23 +211,24 @@ use util\cmd\Console;
 // ...setup code from above...
 
 $calls= $functions->calls()->catching(fn($t) => $t->printStackTrace());
-complete: $result= $ai->api('/chat/completions')->invoke($payload));
-
-// If tool calls are requested, invoke them and return to next completion cycle
-if ('tool_calls' === ($result['choices'][0]['finish_reason'] ?? null)) {
-  $payload['messages'][]= $result['choices'][0]['message'];
-
-  foreach ($result['choices'][0]['message']['tool_calls'] as $call) {
-    $return= $calls->call($call['function']['name'], $call['function']['arguments']);
-    $payload['messages'][]= [
-      'role'         => 'tool',
-      'tool_call_id' => $call['id'],
-      'content'      => $return,
-    ];
-  }
-
-  goto complete;
+next: $result= $ai->api('/responses')->invoke($payload));
+
+// If function calls are requested, invoke them and return to next response cycle
+$invokations= false;
+foreach ($result['output'] as $output) {
+  if ('function_call' !== $output['type']) continue;
+
+  $invokations= true;
+  $return= $calls->call($call['name'], $call['arguments']);
+
+  $payload['input'][]= $call;
+  $payload['input'][]= [
+    'type'    => 'function_call_output',
+    'call_id' => $call['call_id'],
+    'output'  => $return,
+  ];
 }
+if ($invokations) goto next;
 
 // Print out final result
 Console::writeLine($result);
@@ -256,7 +255,7 @@ class Memory {
 // ...shortened for brevity...
 
 $context= ['user' => $user];
-$return= $calls->call($call['function']['name'], $call['function']['arguments'], $context);
+$return= $calls->call($call['name'], $call['arguments'], $context);
 ```
 
 Azure OpenAI
@@ -271,12 +270,11 @@ $ai= new AzureAIEndpoint(
   'https://'.getenv('AZUREAI_API_KEY').'@example.openai.azure.com/openai/deployments/mini',
   '2024-02-01'
 );
-$payload= [
-  'model'    => 'gpt-4o-mini',
-  'messages' => [['role' => 'user', 'content' => $prompt]],
-];
 
-Console::writeLine($ai->api('/chat/completions')->invoke($payload));
+Console::writeLine($ai->api('/responses')->invoke([
+  'model' => 'gpt-4o-mini',
+  'input' => $prompt,
+]));
 ```
 
 Distributing requests
@@ -293,12 +291,11 @@ $endpoints= [
 ];
 
 $ai= new Distributed($endpoints, new ByRemainingRequests());
-$payload= [
-  'model'    => 'gpt-4o-mini',
-  'messages' => [['role' => 'user', 'content' => $prompt]],
-];
 
-Console::writeLine($ai->api('/chat/completions')->invoke($payload));
+Console::writeLine($ai->api('/responses')->invoke([
+  'model' => 'gpt-4o-mini',
+  'input' => $prompt,
+]));
 foreach ($endpoints as $i => $endpoint) {
   Console::writeLine('Endpoint #', $i, ': ', $endpoint->rateLimit());
 }
@@ -354,6 +351,29 @@ $api= new RealtimeApi('wss://example.openai.azure.com/openai/realtime?'.
 $session= $api->connect(['api-key' => getenv('AZUREAI_API_KEY')]);
 ```
 
+Completions API
+---------------
+To use the legacy (but industry standard) chat completions API:
+
+```php
+use com\openai\rest\OpenAIEndpoint;
+use util\cmd\Console;
+
+$ai= new OpenAIEndpoint('https://'.getenv('OPENAI_API_KEY').'@api.openai.com/v1');
+
+$flow= $ai->api('/chat/completions')->flow([
+  'model'    => 'gpt-4o-mini',
+  'messages' => [['role' => 'user', 'content' => $prompt]],
+]);
+$flow= $ai->api('/chat/completions')->flow($payload);
+foreach ($flow->deltas() as $type => $delta) {
+  Console::writeLine('<', $type, '> ', $delta);
+}
+Console::writeLine();
+```
+
+The result object is computed from the streamed deltas and can be retrieved by accessing *$flow->result()*.
+
 See also
 --------
 * https://github.com/openai/tiktoken/

src/main/php/com/openai/rest/Api.class.php

@@ -1,17 +1,45 @@
 <?php namespace com\openai\rest;
 
+use com\openai\tools\Functions;
 use webservices\rest\{RestResource, RestResponse, RestUpload, UnexpectedStatus};
 
+/** @see https://platform.openai.com/docs/guides/responses-vs-chat-completions */
 class Api {
   const JSON= 'application/json';
-  const STREAMING= ['stream' => true, 'stream_options' => ['include_usage' => true]];
+  const EVENTS= 'text/event-stream';
 
   private $resource, $rateLimit;
+  private $adapt= null;
 
   /** Creates a new API instance from a given REST resource */
   public function __construct(RestResource $resource, RateLimit $rateLimit) {
     $this->resource= $resource;
     $this->rateLimit= $rateLimit;
+
+    // In the legacy completions API, streaming requires an option to include usage and tools
+    // are formatted in a substructure, see https://github.com/xp-forge/openai/issues/20
+    if (0 === substr_compare($resource->uri()->path(), '/completions', -12)) {
+      $structure= function($tools) {
+        foreach ($tools->selection as $select) {
+          if ($select instanceof Functions) {
+            foreach ($select->schema() as $name => $function) {
+              yield ['type' => 'function', 'function' => [
+                'name'        => $name,
+                'description' => $function['description'],
+                'parameters'  => $function['input'],
+              ]];
+            }
+          } else {
+            yield $select;
+          }
+        }
+      };
+      $this->adapt= function($payload) use($structure) {
+        if ($payload['stream'] ?? null) $payload['stream_options']= ['include_usage' => true];
+        if ($payload['tools'] ?? null) $payload['tools']= [...$structure($payload['tools'])];
+        return $payload;
+      };
+    }
   }
 
   /**
@@ -23,7 +51,7 @@ public function __construct(RestResource $resource, RateLimit $rateLimit) {
    * @throws webservices.rest.UnexpectedStatus
    */
   public function transmit($payload, $mime= self::JSON): RestResponse {
-    $r= $this->resource->post($payload, $mime);
+    $r= $this->resource->post($this->adapt ? ($this->adapt)($payload) : $payload, $mime);
     $this->rateLimit->update($r->header('x-ratelimit-remaining-requests'));
     if (200 === $r->status()) return $r;
 
@@ -50,9 +78,15 @@ public function invoke(array $payload) {
     return $this->transmit($payload)->value();
   }
 
-  /** Streams API response */
-  public function stream(array $payload): EventStream {
-    $this->resource->accepting('text/event-stream');
-    return new EventStream($this->transmit(self::STREAMING + $payload)->stream());
+  /** Yields events from a streamed response */
+  public function stream(array $payload): Events {
+    $this->resource->accepting(self::EVENTS);
+    return new Events($this->transmit(['stream' => true] + $payload)->stream());
+  }
+
+  /** Completions API flow: Yield deltas and compute result */
+  public function flow(array $payload): Flow {
+    $this->resource->accepting(self::EVENTS);
+    return new Flow($this->transmit(['stream' => true] + $payload)->stream());
   }
 }

src/main/php/com/openai/rest/Events.class.php

@@ -0,0 +1,41 @@
+<?php namespace com\openai\rest;
+
+use IteratorAggregate, Traversable;
+use io\streams\{InputStream, StringReader};
+use lang\IllegalStateException;
+use util\Objects;
+
+/**
+ * Streams events using SSE with JSON data
+ *
+ * @see   https://platform.openai.com/docs/guides/structured-outputs?api-mode=responses
+ * @see   https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events
+ * @test  com.openai.unittest.EventsTest
+ */
+class Events implements IteratorAggregate {
+  const DONE= '[DONE]';
+  private $stream;
+
+  /** Creates a new event stream */
+  public function __construct(InputStream $stream) {
+    $this->stream= $stream;
+  }
+
+  /** Returns events while reading */
+  public function getIterator(): Traversable {
+    $r= new StringReader($this->stream);
+    $event= null;
+
+    // Read all lines starting with `event` or `data`, ignore others
+    while (null !== ($line= $r->readLine())) {
+      // echo "\n<<< $line\n";
+      if (0 === strncmp($line, 'event: ', 6)) {
+        $event= substr($line, 7);
+      } else if (0 === strncmp($line, 'data: ', 5)) {
+        $data= substr($line, 6);
+        if (self::DONE !== $data) yield $event => json_decode($data, true);
+        $event= null;
+      }
+    }
+  }
+}

src/main/php/com/openai/rest/EventStream.class.php → src/main/php/com/openai/rest/Flow.class.php

@@ -5,17 +5,17 @@
 use util\Objects;
 
 /**
- * OpenAI API event stream
+ * OpenAI API completion flow
  *
  * Note: While these event streams are based on server-sent events, they do not
  * utilize their full extent - there are no event types, IDs or multiline data.
  * This implementation can be a bit simpler because of that.
  *
  * @see   https://platform.openai.com/docs/guides/production-best-practices/streaming
  * @see   https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events
- * @test  com.openai.unittest.EventStreamTest
+ * @test  com.openai.unittest.FlowTest
  */
-class EventStream {
+class Flow {
   private $stream;
   private $result= null;
 

src/main/php/com/openai/rest/RestEndpoint.class.php

@@ -13,11 +13,12 @@ public function __construct($endpoint) {
       foreach ($tools->selection as $select) {
         if ($select instanceof Functions) {
           foreach ($select->schema() as $name => $function) {
-            yield ['type' => 'function', 'function' => [
+            yield [
+              'type'        => 'function',
               'name'        => $name,
               'description' => $function['description'],
               'parameters'  => $function['input'],
-            ]];
+            ];
           }
         } else {
           yield $select;