diff --git a/src/Gateway/Bedrock/BedrockTextGateway.php b/src/Gateway/Bedrock/BedrockTextGateway.php index dd0d05616..27ed9089a 100644 --- a/src/Gateway/Bedrock/BedrockTextGateway.php +++ b/src/Gateway/Bedrock/BedrockTextGateway.php @@ -247,6 +247,7 @@ protected function processTextStream( $currentReasoningText = ''; $currentReasoningSignature = ''; $currentReasoningRedacted = ''; + $hasReasoningBlocks = false; $stopReason = 'stop'; $emitTextStart = function () use (&$textId, $invocationId, $timestamp) { @@ -301,21 +302,24 @@ protected function processTextStream( if (isset($delta['text'])) { $currentBlockType = 'text'; - if ($emittedEvent = $emitTextStart()) { - yield $emittedEvent; - } + if ($delta['text'] !== '') { + if ($emittedEvent = $emitTextStart()) { + yield $emittedEvent; + } - $assistantText .= $delta['text']; - $currentText .= $delta['text']; + $assistantText .= $delta['text']; + $currentText .= $delta['text']; - yield (new TextDelta( - (string) Str::uuid(), - $textId, - $delta['text'], - $timestamp, - ))->withInvocationId($invocationId); - } elseif (isset($delta['reasoningContent']['text'])) { + yield (new TextDelta( + (string) Str::uuid(), + $textId, + $delta['text'], + $timestamp, + ))->withInvocationId($invocationId); + } + } elseif (isset($delta['reasoningContent']['text']) && $delta['reasoningContent']['text'] !== '') { $currentBlockType = 'reasoning'; + $hasReasoningBlocks = true; if ($emittedEvent = $emitReasoningStart()) { yield $emittedEvent; @@ -329,16 +333,18 @@ protected function processTextStream( $delta['reasoningContent']['text'], $timestamp, ))->withInvocationId($invocationId); - } elseif (isset($delta['reasoningContent']['signature'])) { + } elseif (isset($delta['reasoningContent']['signature']) && $delta['reasoningContent']['signature'] !== '') { $currentBlockType = 'reasoning'; + $hasReasoningBlocks = true; if ($emittedEvent = $emitReasoningStart()) { yield $emittedEvent; } $currentReasoningSignature .= $delta['reasoningContent']['signature']; - } elseif (isset($delta['reasoningContent']['redactedContent'])) { + } elseif (isset($delta['reasoningContent']['redactedContent']) && $delta['reasoningContent']['redactedContent'] !== '') { $currentBlockType = 'reasoning'; + $hasReasoningBlocks = true; if ($emittedEvent = $emitReasoningStart()) { yield $emittedEvent; @@ -363,12 +369,15 @@ protected function processTextStream( ], ]; } else { + $reasoningText = ['text' => $currentReasoningText]; + + if ($currentReasoningSignature !== '') { + $reasoningText['signature'] = $currentReasoningSignature; + } + $responseContent[$index] = [ 'reasoningContent' => [ - 'reasoningText' => [ - 'text' => $currentReasoningText, - 'signature' => $currentReasoningSignature, - ], + 'reasoningText' => $reasoningText, ], ]; } @@ -383,14 +392,16 @@ protected function processTextStream( $currentReasoningSignature = ''; $currentReasoningRedacted = ''; $reasoningId = ''; - } elseif ($currentBlockType === 'text' && $textId !== '') { + } elseif ($currentBlockType === 'text') { $responseContent[$index] = ['text' => $currentText]; - yield (new TextEnd( - (string) Str::uuid(), - $textId, - $timestamp, - ))->withInvocationId($invocationId); + if ($textId !== '') { + yield (new TextEnd( + (string) Str::uuid(), + $textId, + $timestamp, + ))->withInvocationId($invocationId); + } $currentText = ''; $textId = ''; @@ -457,6 +468,15 @@ protected function processTextStream( $finishReason = FinishReason::Stop; } + $providerContentBlocks = array_values($responseContent); + + if (! $hasReasoningBlocks) { + $providerContentBlocks = array_values(array_filter( + $providerContentBlocks, + fn (array $block) => ! isset($block['text']) || $block['text'] !== '', + )); + } + return new StepResponse( text: $assistantText, toolCalls: $toolCalls, @@ -464,7 +484,7 @@ protected function processTextStream( usage: $totalUsage, meta: new Meta($provider->name(), $model), structured: $structuredOutput !== null ? $this->decodeStructuredOutput($structuredOutput) : null, - providerContentBlocks: array_values($responseContent), + providerContentBlocks: $providerContentBlocks, ); } diff --git a/src/Streaming/Events/TextDelta.php b/src/Streaming/Events/TextDelta.php index 476b03a12..60f0ece97 100644 --- a/src/Streaming/Events/TextDelta.php +++ b/src/Streaming/Events/TextDelta.php @@ -17,14 +17,22 @@ public function __construct( /** * Combine the text deltas in the given collection of events into a single string. + * + * Deltas from a multi-step generation carry a distinct message ID per step, + * and each step's text is a self-contained utterance (typically narration + * around a tool call). Steps are therefore joined with a blank line instead + * of being run together mid-sentence. */ public static function combine(Collection|array $events): string { $events = is_array($events) ? new Collection($events) : $events; return $events->whereInstanceOf(TextDelta::class) - ->map(fn (TextDelta $event) => $event->delta) - ->join(''); + ->groupBy(fn (TextDelta $event) => $event->messageId) + ->map(fn (Collection $deltas) => $deltas->map(fn (TextDelta $event) => $event->delta)->join('')) + ->filter(fn (string $text) => trim($text) !== '') + ->values() + ->join("\n\n"); } /** diff --git a/tests/Feature/Providers/Bedrock/StreamingTest.php b/tests/Feature/Providers/Bedrock/StreamingTest.php index dd5fa6ac0..4246bb2f4 100644 --- a/tests/Feature/Providers/Bedrock/StreamingTest.php +++ b/tests/Feature/Providers/Bedrock/StreamingTest.php @@ -120,6 +120,181 @@ ->and($assistantTurn['content'][1]['toolUse']['name'])->toBe('FixedNumberGenerator'); }); + test('streaming does not round-trip an empty text block on follow-up tool step', function () { + $mock = new MockHandler([ + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['text' => '']), + $this->contentBlockStop(0), + $this->contentBlockStart(1, ['toolUse' => ['toolUseId' => 't1', 'name' => 'FixedNumberGenerator']]), + $this->contentBlockDelta(1, ['toolUse' => ['input' => '{}']]), + $this->contentBlockStop(1), + $this->messageStop('tool_use'), + ]]), + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['text' => 'Done']), + $this->contentBlockStop(0), + $this->messageStop('end_turn'), + ]]), + ]); + + $gateway = $this->gatewayWithClient($this->bedrockClient($mock)); + + iterator_to_array( + (new TextGenerationLoop($gateway))->stream( + 'inv-1', + $this->bedrockProvider(), + 'anthropic.claude-sonnet-5', + null, + tools: [new FixedNumberGenerator], + ), + preserve_keys: false, + ); + + $assistantTurn = $mock->getLastCommand()->toArray()['messages'][0]; + + expect($assistantTurn['content'])->toHaveCount(1) + ->and($assistantTurn['content'][0])->toHaveKey('toolUse'); + }); + + test('streaming preserves empty text blocks between reasoning blocks on follow-up tool step', function () { + $mock = new MockHandler([ + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['reasoningContent' => ['text' => 'I should call the tool']]), + $this->contentBlockDelta(0, ['reasoningContent' => ['signature' => 'sig-1']]), + $this->contentBlockStop(0), + $this->contentBlockStart(1), + $this->contentBlockDelta(1, ['text' => '']), + $this->contentBlockStop(1), + $this->contentBlockStart(2), + $this->contentBlockDelta(2, ['reasoningContent' => ['text' => 'I need the result']]), + $this->contentBlockDelta(2, ['reasoningContent' => ['signature' => 'sig-2']]), + $this->contentBlockStop(2), + $this->contentBlockStart(3, ['toolUse' => ['toolUseId' => 't1', 'name' => 'FixedNumberGenerator']]), + $this->contentBlockDelta(3, ['toolUse' => ['input' => '{}']]), + $this->contentBlockStop(3), + $this->messageStop('tool_use'), + ]]), + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['text' => 'Done']), + $this->contentBlockStop(0), + $this->messageStop('end_turn'), + ]]), + ]); + + $gateway = $this->gatewayWithClient($this->bedrockClient($mock)); + + iterator_to_array( + (new TextGenerationLoop($gateway))->stream( + 'inv-1', + $this->bedrockProvider(), + 'anthropic.claude-sonnet-5', + null, + tools: [new FixedNumberGenerator], + ), + preserve_keys: false, + ); + + $assistantTurn = $mock->getLastCommand()->toArray()['messages'][0]; + + expect($assistantTurn['content'])->toHaveCount(4) + ->and($assistantTurn['content'][0])->toBe([ + 'reasoningContent' => ['reasoningText' => ['text' => 'I should call the tool', 'signature' => 'sig-1']], + ]) + ->and($assistantTurn['content'][1])->toBe(['text' => '']) + ->and($assistantTurn['content'][2])->toBe([ + 'reasoningContent' => ['reasoningText' => ['text' => 'I need the result', 'signature' => 'sig-2']], + ]) + ->and($assistantTurn['content'][3]['toolUse']['toolUseId'])->toBe('t1') + ->and($assistantTurn['content'][3]['toolUse']['name'])->toBe('FixedNumberGenerator'); + }); + + test('streaming does not round-trip an empty reasoning block on follow-up tool step', function (array $delta) { + $mock = new MockHandler([ + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, $delta), + $this->contentBlockStop(0), + $this->contentBlockStart(1, ['toolUse' => ['toolUseId' => 't1', 'name' => 'FixedNumberGenerator']]), + $this->contentBlockDelta(1, ['toolUse' => ['input' => '{}']]), + $this->contentBlockStop(1), + $this->messageStop('tool_use'), + ]]), + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['text' => 'Done']), + $this->contentBlockStop(0), + $this->messageStop('end_turn'), + ]]), + ]); + + $gateway = $this->gatewayWithClient($this->bedrockClient($mock)); + + iterator_to_array( + (new TextGenerationLoop($gateway))->stream( + 'inv-1', + $this->bedrockProvider(), + 'anthropic.claude-sonnet-5', + null, + tools: [new FixedNumberGenerator], + ), + preserve_keys: false, + ); + + $assistantTurn = $mock->getLastCommand()->toArray()['messages'][0]; + + expect($assistantTurn['content'])->toHaveCount(1) + ->and($assistantTurn['content'][0])->toHaveKey('toolUse'); + })->with([ + 'text' => [['reasoningContent' => ['text' => '']]], + 'signature' => [['reasoningContent' => ['signature' => '']]], + 'redacted content' => [['reasoningContent' => ['redactedContent' => '']]], + ]); + + test('streaming omits an empty reasoning signature on follow-up tool step', function () { + $mock = new MockHandler([ + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['reasoningContent' => ['text' => 'I should call the tool']]), + $this->contentBlockStop(0), + $this->contentBlockStart(1, ['toolUse' => ['toolUseId' => 't1', 'name' => 'FixedNumberGenerator']]), + $this->contentBlockDelta(1, ['toolUse' => ['input' => '{}']]), + $this->contentBlockStop(1), + $this->messageStop('tool_use'), + ]]), + new Result(['stream' => [ + $this->contentBlockStart(0), + $this->contentBlockDelta(0, ['text' => 'Done']), + $this->contentBlockStop(0), + $this->messageStop('end_turn'), + ]]), + ]); + + $gateway = $this->gatewayWithClient($this->bedrockClient($mock)); + + iterator_to_array( + (new TextGenerationLoop($gateway))->stream( + 'inv-1', + $this->bedrockProvider(), + 'openai.gpt-oss-120b-1:0', + null, + tools: [new FixedNumberGenerator], + ), + preserve_keys: false, + ); + + $assistantTurn = $mock->getLastCommand()->toArray()['messages'][0]; + + expect($assistantTurn['content'][0])->toBe([ + 'reasoningContent' => [ + 'reasoningText' => ['text' => 'I should call the tool'], + ], + ]); + }); + test('streaming round-trips redacted reasoning block', function () { $mock = new MockHandler([ new Result(['stream' => [ diff --git a/tests/Unit/Streaming/TextDeltaTest.php b/tests/Unit/Streaming/TextDeltaTest.php new file mode 100644 index 000000000..44679e101 --- /dev/null +++ b/tests/Unit/Streaming/TextDeltaTest.php @@ -0,0 +1,57 @@ +toBe('Hello there!'); +}); + +test('combine separates text from different messages with a blank line', function () { + $events = [ + textDelta('message-1', 'Let me look that up.'), + new ToolCall(uniqid(), new Data\ToolCall('call-1', 'get_weather', ['city' => 'Copenhagen']), time()), + new ToolResult(uniqid(), new Data\ToolResult('call-1', 'get_weather', ['city' => 'Copenhagen'], '12°C'), true, null, time()), + textDelta('message-2', 'It is '), + textDelta('message-2', '12°C in Copenhagen.'), + ]; + + expect(TextDelta::combine($events))->toBe("Let me look that up.\n\nIt is 12°C in Copenhagen."); +}); + +test('combine drops whitespace-only messages', function () { + $events = [ + textDelta('message-1', 'First.'), + textDelta('message-2', "\n"), + textDelta('message-3', 'Second.'), + ]; + + expect(TextDelta::combine($events))->toBe("First.\n\nSecond."); +}); + +test('combine ignores events that are not text deltas', function () { + $events = [ + new StreamStart(uniqid(), 'fake', 'fake-model', time()), + textDelta('message-1', 'Only text.'), + ]; + + expect(TextDelta::combine($events))->toBe('Only text.'); +}); + +test('combine returns an empty string when there are no text deltas', function () { + expect(TextDelta::combine([]))->toBe(''); +});