Merge pull request #2 from xp-forge/feature/rest-api

Implement REST API

README.md

@@ -28,14 +28,44 @@ $tokens= Encoding::for('omni')->load($source)->encode('Hello World!');
 
 Completions
 -----------
-*Coming soon*
+Using the REST API, see https://platform.openai.com/docs/api-reference/making-requests
 
-Embeddings
-----------
-*Coming soon*
+```php
+use util\cmd\Console;
+use com\openai\rest\OpenAIEndpoint;
+
+$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));
+```
 
 Streaming
 ---------
+The REST API can use server-sent events to stream responses, see https://platform.openai.com/docs/api-reference/streaming
+
+```php
+use util\cmd\Console;
+use com\openai\rest\OpenAIEndpoint;
+
+$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);
+}
+Console::writeLine();
+```
+
+Embeddings
+----------
 *Coming soon*
 
 Functions

composer.json

@@ -7,6 +7,7 @@
   "keywords": ["module", "xp"],
   "require" : {
     "xp-framework/core": "^12.0 | ^11.0 | ^10.0",
+    "xp-forge/rest-client": "^5.6",
     "php" : ">=7.4.0"
   },
   "require-dev" : {

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

@@ -0,0 +1,34 @@
+<?php namespace com\openai\rest;
+
+use webservices\rest\{RestResource, UnexpectedStatus};
+
+class Api {
+  private $resource;
+
+  /** Creates a new API instance from a given REST resource */
+  public function __construct(RestResource $resource) {
+    $this->resource= $resource;
+  }
+
+  /** Invokes API and returns result */
+  public function invoke(array $payload) {
+    $r= $this->resource
+      ->accepting('application/json')
+      ->post(['stream' => false] + $payload, 'application/json')
+    ;
+    if (200 === $r->status()) return $r->value();
+
+    throw new UnexpectedStatus($r);
+  }
+
+  /** Streams API response */
+  public function stream(array $payload): EventStream {
+    $r= $this->resource
+      ->accepting('text/event-stream')
+      ->post(['stream' => true] + $payload, 'application/json')
+    ;
+    if (200 === $r->status()) return new EventStream($r->stream());
+
+    throw new UnexpectedStatus($r);
+  }
+}

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

@@ -0,0 +1,120 @@
+<?php namespace com\openai\rest;
+
+use io\streams\{InputStream, StringReader};
+use lang\IllegalStateException;
+use util\Objects;
+
+/**
+ * OpenAI API event stream
+ *
+ * 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
+ */
+class EventStream {
+  private $stream;
+  private $result= null;
+
+  /** Creates a new event stream */
+  public function __construct(InputStream $stream) {
+    $this->stream= $stream;
+  }
+
+  /**
+   * Apply a given value with a delta
+   *
+   * @param  var $result
+   * @param  string|int $field
+   * @param  var $delta
+   * @return void
+   * @throws lang.IllegalStateException
+   */
+  private function apply(&$result, $field, $delta) {
+    if (null === $delta) {
+      // NOOP
+    } else if (is_string($delta)) {
+      $result[$field]??= '';
+      $result[$field].= $delta;
+    } else if (is_int($delta) || is_float($delta)) {
+      $result[$field]??= 0;
+      $result[$field]+= $delta;
+    } else if (is_array($delta)) {
+      if (isset($delta['index'])) {
+        $ptr= &$result[$delta['index']];
+        unset($delta['index']);
+      } else {
+        $ptr= &$result[$field];
+      }
+      $ptr??= [];
+      foreach ($delta as $key => $val) {
+        $this->apply($ptr, $key, $val);
+      }
+    } else {
+      throw new IllegalStateException('Cannot apply delta '.Objects::stringOf($delta));
+    }
+  }
+
+  /**
+   * Merge a given value with the result, yielding any deltas
+   *
+   * @param  var $result
+   * @param  var $value
+   * @return iterable
+   * @throws lang.IllegalStateException
+   */
+  private function merge(&$result, $value) {
+    if (is_array($value)) {
+      $result??= [];
+      foreach ($value as $key => $val) {
+        if ('delta' === $key) {
+          foreach ($val as $field => $delta) {
+            yield $field => $delta;
+            $this->apply($result['message'], $field, $delta);
+          }
+        } else {
+          yield from $this->merge($result[$key], $val);
+        }
+      }
+    } else {
+      $result= $value;
+    }
+  }
+
+  /**
+   * Returns delta pairs while reading
+   *
+   * @throws lang.IllegalStateException
+   */
+  public function deltas(?string $filter= null): iterable {
+    if (null !== $this->result) {
+      throw new IllegalStateException('Event stream already consumed');
+    }
+
+    $r= new StringReader($this->stream);
+    while (null !== ($line= $r->readLine())) {
+      if (0 !== strncmp($line, 'data: ', 5)) continue;
+      // echo "\n<<< $line\n";
+
+      // Last chunk is "data: [DONE]"
+      $data= substr($line, 6);
+      if ('[DONE]' === $data) break;
+
+      // Process deltas, applying them to our result while simultaneously
+      // yielding them back to our caller.
+      foreach ($this->merge($this->result, json_decode($data, true)) as $field => $delta) {
+        if (null === $filter || $filter === $field) yield $field => $delta;
+      }
+    }
+    $this->stream->close();
+  }
+
+  /** Returns the result, fetching deltas if necessary */
+  public function result(): array {
+    if (null === $this->result) iterator_count($this->deltas());
+    return $this->result;
+  }
+}

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

@@ -0,0 +1,21 @@
+<?php namespace com\openai\rest;
+
+use webservices\rest\Endpoint;
+
+class OpenAIEndpoint {
+  private $endpoint;
+
+  /**
+   * Creates a new OpenAI endpoint
+   *
+   * @param  string|util.URI|webservices.rest.Endpoint
+   */
+  public function __construct($arg) {
+    $this->endpoint= $arg instanceof Endpoint ? $arg : new Endpoint($arg);
+  }
+
+  /** Returns an API */
+  public function api(string $path, array $segments= []): Api {
+    return new Api($this->endpoint->resource(ltrim($path, '/'), $segments));
+  }
+}

src/test/php/com/openai/unittest/EventStreamTest.class.php

@@ -0,0 +1,134 @@
+<?php namespace com\openai\unittest;
+
+use com\openai\rest\EventStream;
+use io\streams\{InputStream, MemoryInputStream};
+use test\{Assert, Test, Values};
+use lang\IllegalStateException;
+
+class EventStreamTest {
+
+  /** Streams contents */
+  private function contentStream(): array {
+    return [
+      'data: {"choices":[{"delta":{"role":"assistant"}}]}',
+      'data: {"choices":[{"delta":{"content":"Test"}}]}',
+      'data: {"choices":[{"delta":{"content":"ed"}}]}',
+      'data: [DONE]'
+    ];
+  }
+
+  /** Streams tool calls */
+  private function toolCallStream(): array {
+    return [
+      'data: {"choices":[{"delta":{"role":"assistant"}}]}',
+      'data: {"choices":[{"delta":{"tool_calls":[{"type":"function","function":{"name":"search","arguments":""}}]}}]}',
+      'data: {"choices":[{"delta":{"tool_calls":[{"function":{"arguments":"{"}}]}}]}',
+      'data: {"choices":[{"delta":{"tool_calls":[{"function":{"arguments":"}"}}]}}]}',
+      'data: {"choices":[{"delta":{},"finish_reason":"function_call"}]}',
+      'data: [DONE]'
+    ];
+  }
+
+  /** Returns input */
+  private function input(array $lines): InputStream {
+    return new MemoryInputStream(implode("\n\n", $lines));
+  }
+
+  /** Maps deltas to a list of pairs */
+  private function pairsOf(iterable $deltas): array {
+    $r= [];
+    foreach ($deltas as $field => $delta) {
+      $r[]= [$field => $delta];
+    }
+    return $r;
+  }
+
+  /** Filtered deltas */
+  private function filtered(): iterable {
+    yield [null, [['role' => 'assistant'], ['content' => 'Test'], ['content' => 'ed']]];
+    yield ['role', [['role' => 'assistant']]];
+    yield ['content', [['content' => 'Test'], ['content' => 'ed']]];
+  }
+
+  #[Test]
+  public function can_create() {
+    new EventStream($this->input([]));
+  }
+
+  #[Test]
+  public function receive_done_as_first_token() {
+    $events= ['data: [DONE]'];
+    Assert::equals([], $this->pairsOf((new EventStream($this->input($events)))->deltas()));
+  }
+
+  #[Test]
+  public function does_not_continue_reading_after_done() {
+    $events= ['data: [DONE]', '', 'data: "Test"'];
+    Assert::equals([], $this->pairsOf((new EventStream($this->input($events)))->deltas()));
+  }
+
+  #[Test]
+  public function deltas() {
+    Assert::equals(
+      [['role' => 'assistant'], ['content' => 'Test'], ['content' => 'ed']],
+      $this->pairsOf((new EventStream($this->input($this->contentStream())))->deltas())
+    );
+  }
+
+  #[Test]
+  public function deltas_throws_if_already_consumed() {
+    $events= new EventStream($this->input($this->contentStream()));
+    iterator_count($events->deltas());
+
+    Assert::throws(IllegalStateException::class, fn() => iterator_count($events->deltas()));
+  }
+
+  #[Test]
+  public function ignores_newlines() {
+    Assert::equals(
+      [['role' => 'assistant'], ['content' => 'Test'], ['content' => 'ed']],
+      $this->pairsOf((new EventStream($this->input(['', ...$this->contentStream()])))->deltas())
+    );
+  }
+
+  #[Test, Values(from: 'filtered')]
+  public function filtered_deltas($filter, $expected) {
+    Assert::equals(
+      $expected,
+      $this->pairsOf((new EventStream($this->input($this->contentStream())))->deltas($filter))
+    );
+  }
+
+  #[Test]
+  public function result() {
+    Assert::equals(
+      ['choices' => [['message' => ['role' => 'assistant', 'content' => 'Tested']]]],
+      (new EventStream($this->input($this->contentStream())))->result()
+    );
+  }
+
+  #[Test]
+  public function tool_call_deltas() {
+    Assert::equals(
+      [
+        ['role' => 'assistant'],
+        ['tool_calls' => [['type' => 'function', 'function' => ['name' => 'search', 'arguments' => '']]]],
+        ['tool_calls' => [['function' => ['arguments' => '{']]]],
+        ['tool_calls' => [['function' => ['arguments' => '}']]]],
+      ],
+      $this->pairsOf((new EventStream($this->input($this->toolCallStream())))->deltas())
+    );
+  }
+
+  #[Test]
+  public function tool_call_result() {
+    $calls= [['type' => 'function', 'function' => ['name' => 'search', 'arguments' => '{}']]];
+    Assert::equals(
+      ['choices' => [[
+        'message'       => ['role' => 'assistant', 'tool_calls' => $calls],
+        'finish_reason' => 'function_call',
+      ]]],
+      (new EventStream($this->input($this->toolCallStream())))->result()
+    );
+  }
+}