Skip to content

Commit bfb0ed9

Browse files
e4see4se
authored
fix(OpenAI): optimize streamed response processing (#795)
* Support streaming compaction output * Optimize streamed response parsing
1 parent 9b84990 commit bfb0ed9

2 files changed

Lines changed: 217 additions & 22 deletions

File tree

‎src/Responses/StreamResponse.php‎

Lines changed: 95 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,10 @@
1717
*/
1818
final class StreamResponse implements ResponseHasMetaInformationContract, ResponseStreamContract
1919
{
20+
private const STREAM_READ_SIZE = 64 * 1024;
21+
22+
private string $lineBuffer = '';
23+
2024
/**
2125
* Creates a new Stream Response instance.
2226
*
@@ -34,64 +38,133 @@ public function __construct(
3438
*/
3539
public function getIterator(): Generator
3640
{
37-
while (! $this->response->getBody()->eof()) {
38-
$line = $this->readLine($this->response->getBody());
41+
$body = $this->response->getBody();
42+
$event = null;
3943

40-
$event = null;
44+
while (($line = $this->readLine($body)) !== null) {
4145
if (str_starts_with($line, 'event:')) {
4246
$event = trim(substr($line, strlen('event:')));
43-
$line = $this->readLine($this->response->getBody());
47+
48+
unset($line);
49+
50+
continue;
4451
}
4552

4653
if (! str_starts_with($line, 'data:')) {
54+
$event = null;
55+
56+
unset($line);
57+
4758
continue;
4859
}
4960

50-
$data = trim(substr($line, strlen('data:')));
61+
$data = substr($line, strlen('data:'));
62+
63+
unset($line);
64+
65+
if (strlen($data) <= 16 && trim($data) === '[DONE]') {
66+
unset($data);
5167

52-
if ($data === '[DONE]') {
5368
break;
5469
}
5570

56-
/** @var array{error?: array{message: string|array<int, string>, type: string, code: string}, type?: string} $response */
57-
$response = json_decode($data, true, flags: JSON_THROW_ON_ERROR);
71+
/** @var array{error?: array{message: string|array<int, string>, type: string, code: string}, type?: string} $attributes */
72+
$attributes = json_decode($data, true, flags: JSON_THROW_ON_ERROR);
73+
74+
unset($data);
5875

59-
if (isset($response['error'])) {
60-
throw new ErrorException($response['error'], $this->response);
76+
if (isset($attributes['error'])) {
77+
throw new ErrorException($attributes['error'], $this->response);
6178
}
6279

6380
$skippableTypes = ['ping', 'keepalive', 'response.keep_alive'];
64-
if (isset($response['type']) && in_array($response['type'], $skippableTypes, true)) {
81+
82+
if (isset($attributes['type']) && in_array($attributes['type'], $skippableTypes, true)) {
83+
$event = null;
84+
85+
unset($attributes);
86+
6587
continue;
6688
}
6789

6890
if ($event !== null) {
69-
$response['__event'] = $event;
91+
$attributes['__event'] = $event;
7092
}
71-
$response['__meta'] = $this->meta();
7293

73-
yield $this->responseClass::from($response);
94+
$attributes['__meta'] = $this->meta();
95+
96+
$streamEvent = $this->responseClass::from($attributes);
97+
98+
$event = null;
99+
100+
unset($attributes);
101+
102+
yield $streamEvent;
103+
104+
unset($streamEvent);
74105
}
75106
}
76107

77108
/**
78109
* Read a line from the stream.
79110
*/
80-
private function readLine(StreamInterface $stream): string
111+
private function readLine(StreamInterface $stream): ?string
81112
{
82-
$buffer = '';
113+
$newLinePosition = strpos($this->lineBuffer, "\n");
114+
115+
if ($newLinePosition !== false) {
116+
$lineLength = $newLinePosition + 1;
117+
118+
if ($lineLength === strlen($this->lineBuffer)) {
119+
$line = $this->lineBuffer;
120+
$this->lineBuffer = '';
121+
122+
return $line;
123+
}
124+
125+
$line = substr($this->lineBuffer, 0, $lineLength);
126+
$this->lineBuffer = substr($this->lineBuffer, $lineLength);
127+
128+
return $line;
129+
}
83130

84131
while (! $stream->eof()) {
85-
if ('' === ($byte = $stream->read(1))) {
86-
return $buffer;
132+
$chunk = $stream->read(self::STREAM_READ_SIZE);
133+
134+
if ($chunk === '') {
135+
continue;
87136
}
88-
$buffer .= $byte;
89-
if ($byte === "\n") {
90-
break;
137+
138+
// Split the fresh chunk before appending it to avoid copying a large buffered line.
139+
$newLinePosition = strpos($chunk, "\n");
140+
141+
if ($newLinePosition === false) {
142+
$this->lineBuffer .= $chunk;
143+
144+
continue;
145+
}
146+
147+
$lineLength = $newLinePosition + 1;
148+
$line = $this->lineBuffer;
149+
$this->lineBuffer = substr($chunk, $lineLength);
150+
151+
if ($lineLength === strlen($chunk)) {
152+
$line .= $chunk;
153+
} else {
154+
$line .= substr($chunk, 0, $lineLength);
91155
}
156+
157+
return $line;
92158
}
93159

94-
return $buffer;
160+
if ($this->lineBuffer === '') {
161+
return null;
162+
}
163+
164+
$line = $this->lineBuffer;
165+
$this->lineBuffer = '';
166+
167+
return $line;
95168
}
96169

97170
public function meta(): MetaInformation

‎tests/Responses/StreamResponse.php‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
<?php
2+
3+
use GuzzleHttp\Psr7\Response;
4+
use GuzzleHttp\Psr7\StreamDecoratorTrait;
5+
use GuzzleHttp\Psr7\Utils;
6+
use OpenAI\Responses\Responses\CreateStreamedResponse;
7+
use OpenAI\Responses\Responses\Output\OutputCompaction;
8+
use OpenAI\Responses\StreamResponse;
9+
use Psr\Http\Message\StreamInterface;
10+
11+
final class StreamResponseReadTrackingStream implements StreamInterface
12+
{
13+
use StreamDecoratorTrait;
14+
15+
private StreamInterface $stream;
16+
17+
public int $readCalls = 0;
18+
19+
public int $largestRead = 0;
20+
21+
public ?int $emptyReadCall = null;
22+
23+
public ?int $maximumBytesPerRead = null;
24+
25+
public function read($length): string
26+
{
27+
$this->readCalls++;
28+
$this->largestRead = max($this->largestRead, $length);
29+
30+
if ($this->readCalls === $this->emptyReadCall) {
31+
return '';
32+
}
33+
34+
if ($this->maximumBytesPerRead !== null) {
35+
$length = min($length, $this->maximumBytesPerRead);
36+
}
37+
38+
return $this->stream->read($length);
39+
}
40+
}
41+
42+
test('reads large SSE events in chunks without losing buffered events', function () {
43+
$largeEncryptedContent = str_repeat('a', 128 * 1024);
44+
45+
$events = [
46+
[
47+
'type' => 'response.output_item.done',
48+
'output_index' => 0,
49+
'sequence_number' => 1,
50+
'item' => [
51+
'id' => 'cmp_large',
52+
'encrypted_content' => $largeEncryptedContent,
53+
'type' => 'compaction',
54+
'created_by' => 'user',
55+
],
56+
],
57+
[
58+
'type' => 'response.output_item.done',
59+
'output_index' => 1,
60+
'sequence_number' => 2,
61+
'item' => [
62+
'id' => 'cmp_buffered',
63+
'encrypted_content' => 'buffered content',
64+
'type' => 'compaction',
65+
'created_by' => 'user',
66+
],
67+
],
68+
];
69+
70+
$body = implode('', array_map(
71+
fn (array $event): string => "event: response.output_item.done\n".
72+
'data: '.json_encode($event, flags: JSON_THROW_ON_ERROR)."\n\n",
73+
$events,
74+
)).'data: [DONE]';
75+
76+
$stream = new StreamResponseReadTrackingStream(Utils::streamFor($body));
77+
$response = new Response(body: $stream);
78+
$streamResponse = new StreamResponse(CreateStreamedResponse::class, $response);
79+
80+
$result = iterator_to_array($streamResponse);
81+
82+
expect($result)
83+
->toHaveCount(2)
84+
->and($result[0]->response->item)
85+
->toBeInstanceOf(OutputCompaction::class)
86+
->encryptedContent->toBe($largeEncryptedContent)
87+
->and($result[1]->response->item)
88+
->toBeInstanceOf(OutputCompaction::class)
89+
->encryptedContent->toBe('buffered content')
90+
->and($stream->largestRead)->toBe(64 * 1024)
91+
->and($stream->readCalls)->toBeLessThan(10);
92+
});
93+
94+
test('retries an empty read before EOF without losing the buffered line', function () {
95+
$attributes = [
96+
'type' => 'response.output_item.done',
97+
'output_index' => 0,
98+
'sequence_number' => 1,
99+
'item' => [
100+
'id' => 'cmp_after_empty_read',
101+
'encrypted_content' => 'complete content',
102+
'type' => 'compaction',
103+
'created_by' => 'user',
104+
],
105+
];
106+
$body = 'data: '.json_encode($attributes, flags: JSON_THROW_ON_ERROR)."\n";
107+
108+
$stream = new StreamResponseReadTrackingStream(Utils::streamFor($body));
109+
$stream->maximumBytesPerRead = 10;
110+
$stream->emptyReadCall = 2;
111+
112+
$response = new Response(body: $stream);
113+
$streamResponse = new StreamResponse(CreateStreamedResponse::class, $response);
114+
115+
$result = iterator_to_array($streamResponse);
116+
117+
expect($result)
118+
->toHaveCount(1)
119+
->and($result[0]->response->item)
120+
->toBeInstanceOf(OutputCompaction::class)
121+
->encryptedContent->toBe('complete content');
122+
});

0 commit comments

Comments
 (0)