DRAFT: Streaming support

composer.json

@@ -22,7 +22,7 @@
     "php": "^8.2",
     "laravel/framework": "^11.0|^12.0",
     "aws/aws-sdk-php": "^3.339",
-    "prism-php/prism": ">=0.77.1"
+    "prism-php/prism": ">=0.80.0"
   },
   "config": {
     "allow-plugins": {
@@ -97,4 +97,4 @@
       ]
     }
   }
-}
+}

src/Bedrock.php

@@ -2,8 +2,10 @@
 
 namespace Prism\Bedrock;
 
+use Aws\BedrockRuntime\BedrockRuntimeClient;
 use Aws\Credentials\Credentials;
 use Aws\Signature\SignatureV4;
+use Generator;
 use Illuminate\Http\Client\PendingRequest;
 use Illuminate\Http\Client\Request;
 use Prism\Bedrock\Enums\BedrockSchema;
@@ -15,6 +17,7 @@
 use Prism\Prism\Providers\Provider;
 use Prism\Prism\Structured\Request as StructuredRequest;
 use Prism\Prism\Structured\Response as StructuredResponse;
+use Prism\Prism\Text\Chunk;
 use Prism\Prism\Text\Request as TextRequest;
 use Prism\Prism\Text\Response as TextResponse;
 
@@ -95,6 +98,23 @@ public function embeddings(EmbeddingRequest $request): EmbeddingsResponse
         return $handler->handle($request);
     }
 
+    #[\Override]
+    /**
+     * @return Generator<Chunk>
+     */
+    public function stream(TextRequest $request): Generator
+    {
+        $schema = BedrockSchema::Converse;
+
+        $handler = $schema->streamHandler();
+
+        $client = $this->bedrockClient();
+
+        $handler = new $handler($this, $client);
+
+        return $handler->handle($request);
+    }
+
     public function schema(PrismRequest $request): BedrockSchema
     {
         $override = $request->providerOptions();
@@ -109,6 +129,11 @@ public function apiVersion(PrismRequest $request): ?string
         return $this->schema($request)->defaultApiVersion();
     }
 
+    protected function bedrockClient(): BedrockRuntimeClient
+    {
+        return app(BedrockClientFactory::class)->make();
+    }
+
     /**
      * @param  array<string, mixed>  $options
      * @param  array<mixed>  $retry

src/BedrockClientFactory.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Prism\Bedrock;
+
+use Aws\BedrockRuntime\BedrockRuntimeClient;
+use GuzzleHttp\HandlerStack;
+
+class BedrockClientFactory
+{
+    public function make(?HandlerStack $handler = null): BedrockRuntimeClient
+    {
+        $config = [
+            'region' => config('prism.providers.bedrock.region', 'eu-central-1'),
+            'version' => config('prism.providers.bedrock.version', 'latest'),
+            'credentials' => [
+                'key' => config('prism.providers.bedrock.api_key', ''),
+                'secret' => config('prism.providers.bedrock.api_secret', ''),
+            ],
+        ];
+
+        if ($handler instanceof \GuzzleHttp\HandlerStack) {
+            $config['http'] = ['handler' => $handler];
+        }
+
+        return new BedrockRuntimeClient($config);
+    }
+}

src/BedrockServiceProvider.php

@@ -43,4 +43,9 @@ protected function registerWithPrism(): void
             return $prismManager;
         });
     }
+
+    protected function registerBedrockClient(): void
+    {
+        $this->app->singleton(BedrockClientFactory::class, fn (): \Prism\Bedrock\BedrockClientFactory => new BedrockClientFactory);
+    }
 }

src/Contracts/BedrockStreamHandler.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace Prism\Bedrock\Contracts;
+
+use Aws\BedrockRuntime\BedrockRuntimeClient;
+use Generator;
+use Prism\Bedrock\Bedrock;
+use Prism\Prism\Text\Request;
+
+abstract class BedrockStreamHandler
+{
+    public function __construct(
+        protected Bedrock $provider,
+        protected BedrockRuntimeClient $client
+    ) {}
+
+    abstract public function handle(Request $request): Generator;
+}

src/Enums/BedrockSchema.php

@@ -9,8 +9,10 @@
 use Prism\Bedrock\Schemas\Anthropic\AnthropicStructuredHandler;
 use Prism\Bedrock\Schemas\Anthropic\AnthropicTextHandler;
 use Prism\Bedrock\Schemas\Cohere\CohereEmbeddingsHandler;
+use Prism\Bedrock\Schemas\Converse\ConverseStreamHandler;
 use Prism\Bedrock\Schemas\Converse\ConverseStructuredHandler;
 use Prism\Bedrock\Schemas\Converse\ConverseTextHandler;
+use Prism\Prism\Exceptions\PrismException;
 
 enum BedrockSchema: string
 {
@@ -30,6 +32,17 @@ public function textHandler(): ?string
         };
     }
 
+    /**
+     * @return null|class-string<ConverseStreamHandler>
+     */
+    public function streamHandler(): ?string
+    {
+        return match ($this) {
+            self::Converse => ConverseStreamHandler::class,
+            default => throw new PrismException('Prism Bedrock only supports streaming for Converse.'),
+        };
+    }
+
     /**
      * @return null|class-string<BedrockStructuredHandler>
      */

src/Exceptions/BedrockException.php

@@ -0,0 +1,5 @@
+<?php
+
+namespace Prism\Bedrock\Exceptions;
+
+class BedrockException extends \Exception {}

src/HandlesStream.php

@@ -0,0 +1,164 @@
+<?php
+
+namespace Prism\Bedrock;
+
+use Aws\Result;
+use Generator;
+use Prism\Bedrock\Exceptions\BedrockException;
+use Prism\Prism\Enums\ChunkType;
+use Prism\Prism\Exceptions\PrismException;
+use Prism\Prism\Providers\Anthropic\Maps\FinishReasonMap;
+use Prism\Prism\Text\Chunk;
+use Prism\Prism\Text\Request;
+
+trait HandlesStream
+{
+    public function processStream(Result $result, Request $request, int $depth = 0): Generator
+    {
+        $this->state->reset();
+
+        $this->validateToolCallDepth($request, $depth);
+
+        yield from $this->processStreamChunks($result, $request, $depth);
+
+        if ($this->state->hasToolCalls()) {
+            yield from $this->handleToolUseFinish($request, $depth);
+        }
+    }
+
+    protected function validateToolCallDepth(Request $request, int $depth): void
+    {
+        if ($depth >= $request->maxSteps()) {
+            throw new PrismException('Maximum tool call chain depth exceeded');
+        }
+    }
+
+    protected function processStreamChunks(Result $result, Request $request, int $depth): Generator
+    {
+
+        foreach ($result->get('stream') as $event) {
+
+            $outcome = $this->processChunk($event, $request, $depth);
+
+            if ($outcome instanceof Generator) {
+                yield from $outcome;
+            }
+
+            if ($outcome instanceof Chunk) {
+                yield $outcome;
+            }
+        }
+    }
+
+    protected function processChunk(array $chunk, Request $request, int $depth): Generator|Chunk|null
+    {
+        return match (array_key_first($chunk) ?? null) {
+            'messageStart' => $this->handleMessageStart(data_get($chunk, 'messageStart')),
+            'contentBlockDelta' => $this->handleContentBlockDelta(data_get($chunk, 'contentBlockDelta')),
+            'contentBlockStop' => $this->handleContentBlockStop(),
+            'messageStop' => $this->handleMessageStop(data_get($chunk, 'messageStop'), $depth),
+            'metadata' => $this->handleMetadata(data_get($chunk, 'metadata')),
+            'modelStreamErrorException' => $this->handleException(data_get($chunk, 'modelStreamErrorException')),
+            'serviceUnavailableException' => $this->handleException(data_get($chunk, 'serviceUnavailableException')),
+            'throttlingException' => $this->handleException(data_get($chunk, 'throttlingException')),
+            'validationException' => $this->handleException(data_get($chunk, 'validationException')),
+            default => null,
+        };
+    }
+
+    protected function handleMessageStart(array $chunk): null
+    {
+        // {
+        // messageStart: {
+        //      role: assistant
+        // }
+
+        return null;
+    }
+
+    protected function handleContentBlockDelta(array $chunk): ?Chunk
+    {
+        if ($text = data_get($chunk, 'delta.text')) {
+            return $this->handleTextBlockDelta($text, (int) data_get($chunk, 'contentBlockIndex'));
+        }
+
+        if ($reasoningContent = data_get($chunk, 'delta.reasoningContent')) {
+            return $this->handleReasoningContentBlockDelta($reasoningContent);
+        }
+
+        if ($toolUse = data_get($chunk, 'delta.toolUse')) {
+            return $this->handleToolUseBlockDelta($toolUse);
+        }
+
+        return null;
+    }
+
+    protected function handleTextBlockDelta(string $text, int $contentBlockIndex): Chunk
+    {
+        $this->state->appendText($text);
+
+        return new Chunk(
+            text: $text,
+            additionalContent: [
+                'contentBlockIndex' => $contentBlockIndex,
+            ],
+            chunkType: ChunkType::Text
+        );
+    }
+
+    protected function handleReasoningContentBlockDelta(array $reasoningContent): Chunk
+    {
+        $text = data_get($reasoningContent, 'reasoningText.text', '');
+        $signature = data_get($reasoningContent, 'reasoningText.signature', '');
+
+        $this->state->appendThinking($text);
+        $this->state->appendThinkingSignature($signature);
+
+        return new Chunk(
+            text: $text,
+            chunkType: ChunkType::Thinking
+        );
+    }
+
+    protected function handleContentBlockStop(): void
+    {
+        $this->state->resetContentBlock();
+    }
+
+    protected function handleMessageStop(array $chunk, int $depth): Generator|Chunk
+    {
+        $this->state->setStopReason(data_get($chunk, 'stopReason'));
+
+        if ($this->state->isToolUseFinish()) {
+            return $this->handleToolUseFinish($chunk, $depth);
+        }
+
+        return new Chunk(
+            text: $this->state->text(),
+            finishReason: FinishReasonMap::map($this->state->stopReason()),
+            additionalContent: $this->state->buildAdditionalContent(),
+            chunkType: ChunkType::Meta
+        );
+    }
+
+    protected function handleMetadata(array $chunk): void
+    {
+        // {"metadata":{"usage":{"inputTokens":11,"outputTokens":48,"totalTokens":59},"metrics":{"latencyMs":1269}}}
+        // not sure yet where to store this information.
+    }
+
+    protected function handleException(array $chunk): void
+    {
+        throw new BedrockException(data_get($chunk, 'message'));
+    }
+
+    protected function handleToolUseBlockDelta(array $toolUse): void
+    {
+        throw new \Exception('Tool use not yet supported');
+    }
+
+    public function handleToolUseFinish(Request $request, int $depth): void
+    {
+        throw new \Exception('Tool use not yet supported');
+    }
+}

src/Schemas/Anthropic/Maps/MessageMap.php

@@ -8,8 +8,8 @@
 use Exception;
 use Prism\Prism\Contracts\Message;
 use Prism\Prism\Exceptions\PrismException;
+use Prism\Prism\ValueObjects\Media\Image;
 use Prism\Prism\ValueObjects\Messages\AssistantMessage;
-use Prism\Prism\ValueObjects\Messages\Support\Image;
 use Prism\Prism\ValueObjects\Messages\SystemMessage;
 use Prism\Prism\ValueObjects\Messages\ToolResultMessage;
 use Prism\Prism\ValueObjects\Messages\UserMessage;