1import type * as bedrock from "@distilled.cloud/aws/bedrock-runtime";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Stream from "effect/Stream";
6import {
7 AiError,
8 LanguageModel as AiLanguageModel,
9 IdGenerator,
10 Prompt,
11 Response,
12 Tool,
13} from "effect/ai";
14import * as Binding from "../../Binding.ts";
15import type { ConverseRequest } from "./Converse.ts";
16import type { ConverseStreamRequest } from "./ConverseStream.ts";
17
18/**
19 * Inference parameters applied to every request made through the
20 * {@link LanguageModel}. They translate to the Converse API's
21 * `inferenceConfig` (plus `additionalModelRequestFields` for
22 * model-specific extensions like `top_k` or reasoning budgets).
23 */
24export interface LanguageModelParameters {
25 /**
26 * Maximum number of tokens the model may generate in the response.
27 */
28 maxTokens?: number;
29 /**
30 * Sampling temperature. Lower values make the output more deterministic.
31 */
32 temperature?: number;
33 /**
34 * Nucleus-sampling probability mass.
35 */
36 topP?: number;
37 /**
38 * Sequences that stop generation when the model emits them.
39 */
40 stopSequences?: string[];
41 /**
42 * Model-specific request fields passed through verbatim as the Converse
43 * API's `additionalModelRequestFields` — e.g. `{ top_k: 50 }` for
44 * Anthropic models or `{ inferenceConfig: { topK: 5 } }` for Amazon Nova.
45 */
46 additionalModelRequestFields?: unknown;
47}
48
49/**
50 * Options for constructing a Bedrock-backed Effect AI `LanguageModel`.
51 */
52export interface LanguageModelOptions {
53 /**
54 * Default inference parameters. Every field can be overridden per call
55 * with {@link withModelParameters}.
56 */
57 parameters?: LanguageModelParameters;
58}
59
60/**
61 * Per-call overrides applied with {@link withModelParameters}: any
62 * {@link LanguageModelParameters} field plus the model to run the call on.
63 */
64export interface LanguageModelCallParameters extends LanguageModelParameters {
65 /**
66 * The model to run inference on for this call. Must be one of the model
67 * ids the {@link LanguageModel} binding was created with (IAM is scoped to
68 * exactly those).
69 * @default the first bound model id
70 */
71 modelId?: string;
72}
73
74/**
75 * Fiber-scoped {@link LanguageModelCallParameters} consulted on every
76 * generateText / streamText call made through a Bedrock-backed
77 * `LanguageModel`. Set it for a region of your program with
78 * {@link withModelParameters}.
79 */
80export const CurrentModelParameters =
81 Context.Reference<LanguageModelCallParameters>(
82 "AWS.Bedrock.CurrentModelParameters",
83 {
84 defaultValue: () => ({}),
85 },
86 );
87
88/**
89 * Scope per-call inference parameters (and optionally the target model)
90 * onto an Effect or Stream that talks to a Bedrock-backed `LanguageModel`.
91 * Defined fields override the binding's construction-time `parameters`;
92 * everything else falls through to those defaults.
93 *
94 * ```typescript
95 * const response = yield* LanguageModel.generateText({ prompt }).pipe(
96 * Bedrock.withModelParameters({ temperature: 0, maxTokens: 64 }),
97 * );
98 * ```
99 */
100export const withModelParameters =
101 (parameters: LanguageModelCallParameters) =>
102 <S extends Effect.Effect<any, any, any> | Stream.Stream<any, any, any>>(
103 self: S,
104 ): S =>
105 (Effect.isEffect(self)
106 ? Effect.provideService(CurrentModelParameters, parameters)(self)
107 : Stream.provideService(
108 CurrentModelParameters,
109 parameters,
110 )(self as Stream.Stream<any, any, any>)) as S;
111
112/**
113 * Runtime binding that turns an Amazon Bedrock model into an
114 * `effect/ai` {@link AiLanguageModel.LanguageModel} `Layer`, so any
115 * Effect AI program (`LanguageModel.generateText`, `streamText`, `Chat`,
116 * toolkits, ...) runs against Bedrock without code changes.
117 *
118 * Calls are translated to the Bedrock Converse API — Bedrock's unified
119 * messages API that works across all conversational foundation models
120 * (Amazon Nova, Anthropic Claude, Meta Llama, Mistral, ...) — so one binding
121 * covers every model. Bind one model or a list of models: the function is
122 * granted `bedrock:InvokeModel` and `bedrock:InvokeModelWithResponseStream`
123 * scoped to exactly those models, the first is the default, and runtime code
124 * picks between them (and tunes inference parameters) per call with
125 * {@link withModelParameters}. A model reference may be a foundation-model
126 * id, a cross-region inference profile id (e.g. `us.amazon.nova-micro-v1:0`),
127 * or a full Bedrock ARN.
128 *
129 * Model access is an account entitlement — enable the model in the Bedrock
130 * console (Model access) before invoking, otherwise calls fail with
131 * `AccessDeniedException`. Many newer models are only invocable through a
132 * cross-region inference profile id, not their bare foundation-model id.
133 *
134 * ### Effect AI on Bedrock
135 * **Example:** Generate Text
136 * ```typescript
137 * import { LanguageModel } from "effect/ai";
138 *
139 * // init: bind the model and get a LanguageModel Layer
140 * const model = yield* Bedrock.LanguageModel("us.amazon.nova-micro-v1:0", {
141 * parameters: { maxTokens: 1024, temperature: 0.7 },
142 * });
143 *
144 * // runtime: any Effect AI program works against Bedrock
145 * const response = yield* LanguageModel.generateText({
146 * prompt: "Say hello.",
147 * }).pipe(Effect.provide(model));
148 * ```
149 *
150 * **Example:** Stream Text
151 * ```typescript
152 * const parts = LanguageModel.streamText({ prompt }).pipe(
153 * Stream.provide(model),
154 * );
155 * // parts is a Stream of text-start / text-delta / ... / finish parts
156 * ```
157 *
158 * ### Runtime Configuration
159 * **Example:** Override Parameters Per Call
160 * The binding's `parameters` are only defaults — scope overrides onto any
161 * call with `withModelParameters`.
162 * ```typescript
163 * const response = yield* LanguageModel.generateText({ prompt }).pipe(
164 * Bedrock.withModelParameters({ temperature: 0, maxTokens: 64 }),
165 * );
166 * ```
167 *
168 * **Example:** Bind Multiple Models and Pick Per Call
169 * IAM access is fixed at deploy time (scoped to the bound list); which of
170 * those models serves a given request is a runtime decision.
171 * ```typescript
172 * // init: one Layer, IAM for both models, Nova Micro is the default
173 * const model = yield* Bedrock.LanguageModel([
174 * "us.amazon.nova-micro-v1:0",
175 * "us.anthropic.claude-sonnet-4-20250514-v1:0",
176 * ]);
177 *
178 * // runtime: route this call to Claude
179 * const response = yield* LanguageModel.generateText({ prompt }).pipe(
180 * Bedrock.withModelParameters({
181 * modelId: "us.anthropic.claude-sonnet-4-20250514-v1:0",
182 * }),
183 * );
184 * ```
185 *
186 * ### Tool Calling
187 * **Example:** Call Tools with a Toolkit
188 * ```typescript
189 * import { Tool, Toolkit } from "effect/ai";
190 * import * as Schema from "effect/Schema";
191 *
192 * const GetWeather = Tool.make("get_weather", {
193 * description: "Get the current weather for a city.",
194 * parameters: Schema.Struct({ city: Schema.String }),
195 * success: Schema.Struct({ temperatureF: Schema.Number }),
196 * });
197 * const WeatherToolkit = Toolkit.make(GetWeather);
198 *
199 * const response = yield* LanguageModel.generateText({
200 * prompt: "What's the weather in Seattle?",
201 * toolkit: WeatherToolkit,
202 * }).pipe(
203 * Effect.provide(WeatherToolkit.toLayer({
204 * get_weather: ({ city }) => Effect.succeed({ temperatureF: 72 }),
205 * })),
206 * Effect.provide(model),
207 * );
208 * ```
209 *
210 * @binding
211 */
212export interface LanguageModel extends Binding.Service<
213 LanguageModel,
214 "AWS.Bedrock.LanguageModel",
215 (
216 model: string | readonly [string, ...string[]],
217 options?: LanguageModelOptions,
218 ) => Effect.Effect<Layer.Layer<AiLanguageModel.LanguageModel>>
219> {}
220export const LanguageModel = Binding.Service<LanguageModel>(
221 "AWS.Bedrock.LanguageModel",
222);
223
224/**
225 * The already-bound Converse callables the adapter drives. Produced by
226 * `yield* Bedrock.Converse(model)` / `yield* Bedrock.ConverseStream(model)`
227 * (or any function of the same shape).
228 */
229export interface MakeLanguageModelOptions {
230 /** Non-streaming Converse callable (modelId already bound). */
231 readonly converse: (
232 request: ConverseRequest,
233 ) => Effect.Effect<bedrock.ConverseResponse, bedrock.ConverseError>;
234 /** Streaming Converse callable (modelId already bound). */
235 readonly converseStream: (
236 request: ConverseStreamRequest,
237 ) => Effect.Effect<
238 bedrock.ConverseStreamResponse,
239 bedrock.ConverseStreamError
240 >;
241 /** Default inference parameters for every call. */
242 readonly parameters?: LanguageModelParameters;
243}
244
245/**
246 * Provide an {@link AiLanguageModel.LanguageModel} layer backed by the
247 * supplied Bedrock Converse callables.
248 */
249export const makeLanguageModelLayer = (
250 options: MakeLanguageModelOptions,
251): Layer.Layer<AiLanguageModel.LanguageModel> =>
252 Layer.effect(AiLanguageModel.LanguageModel, makeLanguageModel(options));
253
254/**
255 * Build an {@link AiLanguageModel.LanguageModel} that proxies generateText /
256 * streamText through the Bedrock Converse API.
257 */
258export const makeLanguageModel = ({
259 converse,
260 converseStream,
261 parameters,
262}: MakeLanguageModelOptions): Effect.Effect<AiLanguageModel.LanguageModel> =>
263 AiLanguageModel.make({
264 generateText: (options) =>
265 Effect.gen(function* () {
266 // Read the fiber-scoped per-call overrides (withModelParameters)
267 // and merge them over the binding's construction-time defaults.
268 const call = yield* CurrentModelParameters;
269 const request = toConverseRequest({
270 options,
271 parameters: mergeParameters(parameters, call),
272 modelId: call.modelId,
273 });
274 const response = yield* converse(request).pipe(
275 Effect.mapError((cause) => toAiError(cause, "generateText")),
276 );
277 return toResponseParts(response);
278 }),
279 streamText: (options) =>
280 Stream.unwrap(
281 Effect.gen(function* () {
282 const idGen = yield* IdGenerator.IdGenerator;
283 const call = yield* CurrentModelParameters;
284 const request = toConverseRequest({
285 options,
286 parameters: mergeParameters(parameters, call),
287 modelId: call.modelId,
288 });
289 const response = yield* converseStream(request).pipe(
290 Effect.mapError((cause) => toAiError(cause, "streamText")),
291 );
292 return parseStream(response, idGen);
293 }),
294 ),
295 });
296
297// ---------------------------------------------------------------------------
298// Prompt → Converse messages
299//
300// The Converse API separates system prompts from the message list, requires
301// strict user/assistant alternation, and carries tool results inside *user*
302// messages — so conversion merges consecutive same-role messages after
303// mapping.
304// ---------------------------------------------------------------------------
305
306const IMAGE_FORMATS: Record<string, bedrock.ImageFormat> = {
307 "image/png": "png",
308 "image/jpeg": "jpeg",
309 "image/gif": "gif",
310 "image/webp": "webp",
311};
312
313const base64ToUint8Array = (data: string): Uint8Array => {
314 const binary = atob(data);
315 const bytes = new Uint8Array(binary.length);
316 for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i);
317 return bytes;
318};
319
320const fileToImageBlock = (
321 data: string | Uint8Array | URL,
322 mediaType: string,
323): bedrock.ContentBlock | undefined => {
324 const format = IMAGE_FORMATS[mediaType];
325 // Converse only accepts inline bytes (or S3 locations) — remote URLs and
326 // non-image documents are not representable, so they are dropped.
327 if (format === undefined || data instanceof URL) return undefined;
328 const bytes =
329 data instanceof Uint8Array
330 ? data
331 : base64ToUint8Array(
332 data.startsWith("data:") ? data.slice(data.indexOf(",") + 1) : data,
333 );
334 return { image: { format, source: { bytes } } };
335};
336
337const toolCallInput = (params: unknown): unknown => {
338 if (typeof params !== "string") return params ?? {};
339 try {
340 return JSON.parse(params);
341 } catch {
342 return {};
343 }
344};
345
346interface ConverseMessages {
347 readonly system: bedrock.SystemContentBlock[];
348 readonly messages: bedrock.Message[];
349}
350
351const toUserContent = (
352 parts: Prompt.UserMessage["content"],
353): bedrock.ContentBlock[] =>
354 parts.flatMap((p): bedrock.ContentBlock[] => {
355 if (p.type === "text") return p.text.length > 0 ? [{ text: p.text }] : [];
356 if (p.type === "file") {
357 const block = fileToImageBlock(p.data, p.mediaType);
358 return block === undefined ? [] : [block];
359 }
360 return [];
361 });
362
363const toAssistantContent = (
364 parts: Prompt.AssistantMessage["content"],
365): bedrock.ContentBlock[] =>
366 parts.flatMap((p): bedrock.ContentBlock[] => {
367 // Reasoning parts are not replayed: Bedrock requires the original
368 // cryptographic signature alongside replayed reasoning, which the
369 // framework does not round-trip. Models tolerate its absence.
370 if (p.type === "text") return p.text.length > 0 ? [{ text: p.text }] : [];
371 if (p.type === "tool-call") {
372 return [
373 {
374 toolUse: {
375 toolUseId: p.id,
376 name: p.name,
377 input: toolCallInput(p.params),
378 },
379 },
380 ];
381 }
382 return [];
383 });
384
385const toToolResultContent = (
386 parts: Prompt.ToolMessage["content"],
387): bedrock.ContentBlock[] =>
388 parts.flatMap((p): bedrock.ContentBlock[] =>
389 p.type === "tool-result"
390 ? [
391 {
392 toolResult: {
393 toolUseId: p.id,
394 content:
395 typeof p.result === "string"
396 ? [{ text: p.result }]
397 : [{ json: p.result ?? {} }],
398 ...(p.isFailure ? { status: "error" as const } : {}),
399 },
400 },
401 ]
402 : [],
403 );
404
405const convertPrompt = (prompt: Prompt.Prompt): ConverseMessages => {
406 const system: bedrock.SystemContentBlock[] = [];
407 const messages: bedrock.Message[] = [];
408
409 const append = (
410 role: bedrock.ConversationRole,
411 content: bedrock.ContentBlock[],
412 ) => {
413 if (content.length === 0) return;
414 const last = messages[messages.length - 1];
415 // Converse requires user/assistant alternation; merge consecutive
416 // same-role messages (tool results become user messages, so a tool
417 // message followed by a user message must fuse).
418 if (last !== undefined && last.role === role) {
419 last.content.push(...content);
420 } else {
421 messages.push({ role, content });
422 }
423 };
424
425 for (const message of prompt.content) {
426 switch (message.role) {
427 case "system":
428 if (message.content.length > 0) system.push({ text: message.content });
429 break;
430 case "user":
431 append("user", toUserContent(message.content));
432 break;
433 case "assistant":
434 append("assistant", toAssistantContent(message.content));
435 break;
436 case "tool":
437 append("user", toToolResultContent(message.content));
438 break;
439 }
440 }
441
442 return { system, messages };
443};
444
445// ---------------------------------------------------------------------------
446// Tools / toolChoice
447// ---------------------------------------------------------------------------
448
449const toToolSpec = (tool: Tool.Any): bedrock.Tool => ({
450 toolSpec: {
451 name: tool.name,
452 description: Tool.getDescription(tool),
453 inputSchema: { json: Tool.getJsonSchema(tool) },
454 },
455});
456
457const hasToolBlocks = (messages: ReadonlyArray<bedrock.Message>): boolean =>
458 messages.some((m) =>
459 m.content.some(
460 (block) => block.toolUse !== undefined || block.toolResult !== undefined,
461 ),
462 );
463
464const toToolConfig = (
465 tools: ReadonlyArray<Tool.Any>,
466 toolChoice: AiLanguageModel.ProviderOptions["toolChoice"],
467 messages: ReadonlyArray<bedrock.Message>,
468): bedrock.ToolConfiguration | undefined => {
469 if (tools.length === 0) return undefined;
470 const mapped = tools.map(toToolSpec);
471
472 if (toolChoice === "none") {
473 // Converse has no "none" mode. Omit toolConfig entirely — unless the
474 // conversation already contains toolUse/toolResult blocks, which the API
475 // rejects without a toolConfig; then send the tools with auto choice.
476 return hasToolBlocks(messages)
477 ? { tools: mapped, toolChoice: { auto: {} } }
478 : undefined;
479 }
480 if (toolChoice === "required") {
481 return { tools: mapped, toolChoice: { any: {} } };
482 }
483 if (typeof toolChoice === "object" && "tool" in toolChoice) {
484 return {
485 tools: mapped,
486 toolChoice: { tool: { name: toolChoice.tool } },
487 };
488 }
489 if (typeof toolChoice === "object" && "oneOf" in toolChoice) {
490 const allowed = new Set(toolChoice.oneOf);
491 return {
492 tools: mapped.filter(
493 (t) => t.toolSpec !== undefined && allowed.has(t.toolSpec.name),
494 ),
495 toolChoice: toolChoice.mode === "required" ? { any: {} } : { auto: {} },
496 };
497 }
498 return { tools: mapped, toolChoice: { auto: {} } };
499};
500
501// ---------------------------------------------------------------------------
502// Request assembly
503// ---------------------------------------------------------------------------
504
505/**
506 * Merge per-call overrides over construction-time defaults, field by field —
507 * an undefined override field never clobbers a configured default.
508 */
509const mergeParameters = (
510 defaults: LanguageModelParameters | undefined,
511 overrides: LanguageModelCallParameters,
512): LanguageModelParameters => ({
513 maxTokens: overrides.maxTokens ?? defaults?.maxTokens,
514 temperature: overrides.temperature ?? defaults?.temperature,
515 topP: overrides.topP ?? defaults?.topP,
516 stopSequences: overrides.stopSequences ?? defaults?.stopSequences,
517 additionalModelRequestFields:
518 overrides.additionalModelRequestFields ??
519 defaults?.additionalModelRequestFields,
520});
521
522const toConverseRequest = ({
523 options,
524 parameters,
525 modelId,
526}: {
527 readonly options: AiLanguageModel.ProviderOptions;
528 readonly parameters: LanguageModelParameters | undefined;
529 readonly modelId?: string | undefined;
530}): ConverseRequest & ConverseStreamRequest => {
531 const { system, messages } = convertPrompt(options.prompt);
532 const toolConfig = toToolConfig(options.tools, options.toolChoice, messages);
533 const inferenceConfig: bedrock.InferenceConfiguration = {
534 ...(parameters?.maxTokens !== undefined
535 ? { maxTokens: parameters.maxTokens }
536 : {}),
537 ...(parameters?.temperature !== undefined
538 ? { temperature: parameters.temperature }
539 : {}),
540 ...(parameters?.topP !== undefined ? { topP: parameters.topP } : {}),
541 ...(parameters?.stopSequences !== undefined
542 ? { stopSequences: parameters.stopSequences }
543 : {}),
544 };
545 return {
546 messages,
547 ...(modelId !== undefined ? { modelId } : {}),
548 ...(system.length > 0 ? { system } : {}),
549 ...(Object.keys(inferenceConfig).length > 0 ? { inferenceConfig } : {}),
550 ...(toolConfig !== undefined ? { toolConfig } : {}),
551 ...(parameters?.additionalModelRequestFields !== undefined
552 ? {
553 additionalModelRequestFields: parameters.additionalModelRequestFields,
554 }
555 : {}),
556 };
557};
558
559// ---------------------------------------------------------------------------
560// Finish reason / usage mapping
561// ---------------------------------------------------------------------------
562
563const mapStopReason = (
564 raw: bedrock.StopReason | undefined,
565): Response.FinishReason => {
566 switch (raw) {
567 case "end_turn":
568 case "stop_sequence":
569 return "stop";
570 case "tool_use":
571 return "tool-calls";
572 case "max_tokens":
573 case "model_context_window_exceeded":
574 return "length";
575 case "guardrail_intervened":
576 case "content_filtered":
577 return "content-filter";
578 case "malformed_model_output":
579 case "malformed_tool_use":
580 return "error";
581 case undefined:
582 return "unknown";
583 default:
584 return "other";
585 }
586};
587
588const mapUsage = (usage: bedrock.TokenUsage | undefined): Response.Usage => {
589 // Bedrock's `inputTokens` excludes cache reads/writes, which are reported
590 // separately — so `total` is the sum of all three.
591 const input = usage?.inputTokens ?? 0;
592 const cacheRead = usage?.cacheReadInputTokens ?? 0;
593 const cacheWrite = usage?.cacheWriteInputTokens ?? 0;
594 return new Response.Usage({
595 inputTokens: {
596 uncached: input,
597 total: input + cacheRead + cacheWrite,
598 cacheRead,
599 cacheWrite,
600 },
601 outputTokens: {
602 total: usage?.outputTokens ?? 0,
603 text: 0,
604 reasoning: 0,
605 },
606 });
607};
608
609// ---------------------------------------------------------------------------
610// generateText: ConverseResponse → Response.PartEncoded[]
611// ---------------------------------------------------------------------------
612
613const toResponseParts = (
614 response: bedrock.ConverseResponse,
615): Array<Response.PartEncoded> => {
616 const content = response.output.message?.content ?? [];
617 const parts = content.flatMap((block): Response.PartEncoded[] => {
618 if (block.text !== undefined && block.text.length > 0) {
619 return [{ type: "text", text: block.text }];
620 }
621 if (block.reasoningContent?.reasoningText !== undefined) {
622 const text = block.reasoningContent.reasoningText.text;
623 return text.length > 0 ? [{ type: "reasoning", text }] : [];
624 }
625 if (block.toolUse !== undefined) {
626 return [
627 {
628 type: "tool-call",
629 id: block.toolUse.toolUseId,
630 name: block.toolUse.name,
631 params: block.toolUse.input ?? {},
632 },
633 ];
634 }
635 return [];
636 });
637 return [
638 ...parts,
639 {
640 type: "finish",
641 reason: mapStopReason(response.stopReason),
642 usage: mapUsage(response.usage),
643 response: undefined,
644 },
645 ];
646};
647
648// ---------------------------------------------------------------------------
649// streamText: ConverseStreamOutput events → Stream<Response.StreamPartEncoded>
650//
651// Bedrock's stream is fully structured: content blocks are opened by index
652// (`contentBlockStart` for tool use; text/reasoning open implicitly on the
653// first delta), advanced by `contentBlockDelta`, and closed by
654// `contentBlockStop`. `messageStop` carries the stop reason and `metadata`
655// carries usage. State threads through `Stream.mapAccumEffect`; the
656// per-event `parts` buffer is mutable but scoped to one event.
657// ---------------------------------------------------------------------------
658
659interface OpenBlock {
660 readonly kind: "text" | "reasoning" | "tool";
661 readonly id: string;
662}
663
664interface StreamState {
665 readonly blocks: ReadonlyMap<number, OpenBlock>;
666 readonly usage: bedrock.TokenUsage | undefined;
667 readonly stopReason: bedrock.StopReason | undefined;
668}
669
670const initialStreamState: StreamState = {
671 blocks: new Map(),
672 usage: undefined,
673 stopReason: undefined,
674};
675
676type StreamParts = Array<Response.StreamPartEncoded>;
677
678// Text/reasoning blocks open implicitly on their first delta; tool blocks
679// are opened by handleBlockStart (Bedrock names them via contentBlockStart).
680const openBlock = (
681 state: StreamState,
682 index: number,
683 block: OpenBlock & { readonly kind: "text" | "reasoning" },
684 parts: StreamParts,
685): StreamState => {
686 parts.push(
687 block.kind === "text"
688 ? { type: "text-start", id: block.id }
689 : { type: "reasoning-start", id: block.id },
690 );
691 const blocks = new Map(state.blocks);
692 blocks.set(index, block);
693 return { ...state, blocks };
694};
695
696const closeBlock = (
697 state: StreamState,
698 index: number,
699 parts: StreamParts,
700): StreamState => {
701 const block = state.blocks.get(index);
702 if (block === undefined) return state;
703 parts.push(
704 block.kind === "text"
705 ? { type: "text-end", id: block.id }
706 : block.kind === "reasoning"
707 ? { type: "reasoning-end", id: block.id }
708 : { type: "tool-params-end", id: block.id },
709 );
710 const blocks = new Map(state.blocks);
711 blocks.delete(index);
712 return { ...state, blocks };
713};
714
715const handleBlockStart = (
716 state: StreamState,
717 event: bedrock.ContentBlockStartEvent,
718 parts: StreamParts,
719): StreamState => {
720 const toolUse = event.start.toolUse;
721 if (toolUse === undefined) return state;
722 parts.push({
723 type: "tool-params-start",
724 id: toolUse.toolUseId,
725 name: toolUse.name,
726 });
727 const blocks = new Map(state.blocks);
728 blocks.set(event.contentBlockIndex, { kind: "tool", id: toolUse.toolUseId });
729 return { ...state, blocks };
730};
731
732const handleBlockDelta = (
733 state: StreamState,
734 event: bedrock.ContentBlockDeltaEvent,
735 parts: StreamParts,
736 idGen: IdGenerator.Service,
737): Effect.Effect<StreamState> =>
738 Effect.gen(function* () {
739 const index = event.contentBlockIndex;
740 const delta = event.delta;
741 let s = state;
742
743 if (delta.text !== undefined) {
744 let block = s.blocks.get(index);
745 if (block === undefined) {
746 const opened = {
747 kind: "text",
748 id: yield* idGen.generateId(),
749 } as const;
750 s = openBlock(s, index, opened, parts);
751 block = opened;
752 }
753 if (delta.text.length > 0) {
754 parts.push({ type: "text-delta", id: block.id, delta: delta.text });
755 }
756 return s;
757 }
758
759 if (delta.reasoningContent !== undefined) {
760 const text = delta.reasoningContent.text;
761 // Signature / redacted-content deltas carry no displayable text.
762 if (text === undefined) return s;
763 let block = s.blocks.get(index);
764 if (block === undefined) {
765 const opened = {
766 kind: "reasoning",
767 id: yield* idGen.generateId(),
768 } as const;
769 s = openBlock(s, index, opened, parts);
770 block = opened;
771 }
772 if (text.length > 0) {
773 parts.push({ type: "reasoning-delta", id: block.id, delta: text });
774 }
775 return s;
776 }
777
778 if (delta.toolUse !== undefined) {
779 const block = s.blocks.get(index);
780 if (block !== undefined && delta.toolUse.input.length > 0) {
781 parts.push({
782 type: "tool-params-delta",
783 id: block.id,
784 delta: delta.toolUse.input,
785 });
786 }
787 return s;
788 }
789
790 return s;
791 });
792
793const streamExceptionOf = (event: bedrock.ConverseStreamOutput): unknown =>
794 event.internalServerException ??
795 event.modelStreamErrorException ??
796 event.validationException ??
797 event.throttlingException ??
798 event.serviceUnavailableException;
799
800const handleStreamEvent = (
801 state: StreamState,
802 event: bedrock.ConverseStreamOutput,
803 idGen: IdGenerator.Service,
804): Effect.Effect<
805 readonly [StreamState, ReadonlyArray<Response.StreamPartEncoded>],
806 AiError.AiError
807> =>
808 Effect.gen(function* () {
809 const exception = streamExceptionOf(event);
810 if (exception !== undefined) {
811 return yield* Effect.fail(toAiError(exception, "streamText"));
812 }
813 const parts: StreamParts = [];
814 let s = state;
815 if (event.contentBlockStart !== undefined) {
816 s = handleBlockStart(s, event.contentBlockStart, parts);
817 } else if (event.contentBlockDelta !== undefined) {
818 s = yield* handleBlockDelta(s, event.contentBlockDelta, parts, idGen);
819 } else if (event.contentBlockStop !== undefined) {
820 s = closeBlock(s, event.contentBlockStop.contentBlockIndex, parts);
821 } else if (event.messageStop !== undefined) {
822 s = { ...s, stopReason: event.messageStop.stopReason };
823 } else if (event.metadata !== undefined) {
824 s = { ...s, usage: event.metadata.usage };
825 }
826 return [s, parts] as const;
827 });
828
829const finalizeStream = (
830 state: StreamState,
831): ReadonlyArray<Response.StreamPartEncoded> => {
832 const parts: StreamParts = [];
833 let s = state;
834 for (const index of [...s.blocks.keys()].sort((a, b) => a - b)) {
835 s = closeBlock(s, index, parts);
836 }
837 parts.push({
838 type: "finish",
839 reason: mapStopReason(s.stopReason),
840 usage: mapUsage(s.usage),
841 response: undefined,
842 });
843 return parts;
844};
845
846const parseStream = (
847 response: bedrock.ConverseStreamResponse,
848 idGen: IdGenerator.Service,
849): Stream.Stream<Response.StreamPartEncoded, AiError.AiError> => {
850 const events = response.stream;
851 if (events === undefined) {
852 return Stream.fromIterable(finalizeStream(initialStreamState));
853 }
854 return events.pipe(
855 Stream.mapError((cause) => toAiError(cause, "streamText")),
856 Stream.mapAccumEffect(
857 () => initialStreamState,
858 (state, event) => handleStreamEvent(state, event, idGen),
859 { onHalt: (state) => finalizeStream(state) },
860 ),
861 );
862};
863
864// ---------------------------------------------------------------------------
865// Error mapping
866// ---------------------------------------------------------------------------
867
868const toAiError = (
869 cause: unknown,
870 method: "generateText" | "streamText",
871): AiError.AiError => {
872 const tagged = cause as { _tag?: string; message?: string } | undefined;
873 const description = [
874 tagged?._tag ?? (cause instanceof Error ? cause.name : undefined),
875 tagged?.message ?? "Bedrock Converse request failed",
876 ]
877 .filter((s): s is string => s !== undefined && s.length > 0)
878 .join(": ");
879 return AiError.AiError.make({
880 module: "AWS.Bedrock.LanguageModel",
881 method,
882 reason: new AiError.UnknownError({ description }),
883 });
884};