feat(ai): add speech generation with streaming media routes (#50883)

This commit is contained in:
Shoubhit Dash
2026-09-23 17:35:23 +05:30
committed by GitHub
parent 9c8a63e852
commit f526727178
49 changed files with 2659 additions and 54 deletions
+2
View File
@@ -100,6 +100,8 @@ Media does not fit the SSE-frames-to-event-state-machine LLM route. `MediaRoute.
`MediaProtocol.queued` is the submit-then-poll kind every video route uses: `start` (body + decode into `{ token, snapshot }`), `status`, `result`, and optional `cancel`, each addressed by a route-owned `token` whose `Schema.Codec` makes it serializable. `MediaRoute.inline` and `MediaRoute.queued` compose the two kinds with `Endpoint` and `Auth`; the queued route decodes the token once at the boundary (`start` output or `resume` input) and closes over it in a token-free `GenerationRoute` (`status`/`result`/`cancel` are plain Effects), so `Generation` never sees the token's shape and only carries the encoded JSON for persistence. Polls reuse the route's auth and deployment headers plus the request's `http` overlay after `start`, and resolve relative paths against the route base URL (provider-issued absolute URLs such as fal's `status_url` pass through). `result` is always its own GET even when the provider returns output inside the status document, so `Generation.await` behaves the same after `start` and after `resume`. `PollContext.auth` carries only what `Auth` added so protocols can hand download credentials to output assets as transient `Media.Asset.headers` (Veo) — never part of `source` or JSON. Status strings map through a per-protocol `STATUS` table via `MediaProtocol.status`; terminal generations without output fail through `output.ended` / `output.contentPolicy` with the provider document on `reason.body`. `Generation.AwaitOptions` (`{ poll?: Poll }`) is the one options type for `await`, `events`, `Video.generate`, and `Video.stream`.
`MediaProtocol.stream` is the incremental kind every speech route uses, with the same discipline as LLM protocols. `MediaRoute.stream` submits the caller's request as `MediaProtocol.Addressed<Request>` (`{ ...request, mode }`, `mode: "generate" | "stream"`), so one provider stays one protocol: `body.from`, the endpoint path, and `frames` read `request.mode` to pick the body, path, and framing. `frames(bytes, context)` returns frames — `Framing.sse`, `Framing.lines`, `Framing.document` (a single-document response shaped like a streamed record), or the raw `bytes` for chunked audio. `initial()` is fresh per-response parser state; `step` folds each frame into it and emits modality events; `finish(state, context)` runs once after the last frame with the request, body, and observed `http` (header-only usage lives there) and emits exactly one terminal event or fails with `MediaProtocol.incomplete`. Keep parser state to real accumulators and derive anything the request or body determines in `finish`. `generate` runs the same stream and folds it with the modality's `collect`. Request-derived URL parameters go on the JSON body's `query`, applied before route and caller `http.query`. Decode frames with `MediaProtocol.decodeFrame` and raise stream-time failures with `MediaProtocol.frameError` (the frame stays on `reason.body`); protocols never thread HTTP context, because the route fills `reason.http` on stream errors that lack it. Speech protocols share `protocols/utils/speech-stream.ts` for deltas, timestamps, voice ids, PCM and container descriptions, and the terminal asset.
### URL Construction
`Endpoint` owns `{ baseURL, path, query }`. Each protocol route includes a canonical endpoint when the provider has one (e.g. `https://api.openai.com/v1`); provider helpers override endpoint fields by configuring the route before selecting a model. Generic OpenAI-compatible routes have no canonical URL and require configuration before execution.
+86 -1
View File
@@ -682,6 +682,90 @@ const generation = await ai.video.start({ model, prompt })
const video = await generation.await({ poll: { interval: 10_000 }, signal })
```
## Speech generation
Speech (text-to-speech) is one request whose response is parsed incrementally, so every route supports both
`Speech.generate` (the whole file) and `Speech.stream` (audio chunks as they arrive). Models come from `.speech(...)`
selectors on the `OpenAI`, `Google` (Gemini TTS), `ElevenLabs`, `Cartesia`, and `Deepgram` facades. Common fields
(`voice`, `format`, `speed`, `language`, `instructions`, `timestamps`) lower natively or fail with a typed `AIError`
before any network call; provider-native controls live under `providerOptions`, inferred from the selected model.
```ts
import { Media, Speech, SpeechClient, SpeechEvent } from "@opencode/ai"
import { ElevenLabs, OpenAI } from "@opencode/ai/providers"
const openai = OpenAI.configure({ apiKey: process.env.OPENAI_API_KEY })
// The whole file, written to disk.
const program = Effect.gen(function* () {
const response = yield* Speech.generate({
model: openai.speech("gpt-4o-mini-tts"),
text: "Hello from OpenCode.",
voice: "coral",
format: "mp3",
instructions: "Warm and unhurried.",
})
response.audio // Media.Asset with bytes; headerless PCM carries info.encoding / sampleRate / channels
response.usage // undefined: OpenAI reports tokens only on SSE streams (Gemini: tokens; ElevenLabs: credits; Deepgram: characters)
yield* Media.write(response.audio, "hello.mp3")
})
// Chunks as they arrive: audio-delta* (interleaved with timestamps) then one finish carrying the assembled asset.
const events = Speech.stream({
model: ElevenLabs.configure({ apiKey }).speech("eleven_flash_v2_5"),
text: "Hello from OpenCode.",
voice: "JBFqnCBsd6RMkjVDRZzb",
format: "pcm",
timestamps: true,
}).pipe(
Stream.tap((event) => {
if (SpeechEvent.is.audioDelta(event)) return play(event.chunk)
if (SpeechEvent.is.timestamps(event)) return highlight(event.items) // { text, startSeconds, endSeconds }[]
return Effect.void
}),
)
```
`voice` is the provider's own identifier — a name on OpenAI and Gemini (`"coral"`, `"Kore"`), a voice id on
ElevenLabs and Cartesia. `{ id }` selects an OpenAI custom voice (`{ id: "voice_1234" }`) and means the same as the
plain string elsewhere. There is no cross-provider voice catalog. `format` is the container-level word (`mp3`, `wav`,
`pcm`, `opus`, `aac`, `flac`); sample rates and bitrates live under `providerOptions`, and a value the route cannot
produce fails as `UnsupportedOperation`. Streams buffer every chunk so `finish` can carry the whole clip.
`SpeechClient.layer` needs `RequestExecutor.Service`.
Provider notes:
- **OpenAI** streams over SSE (`stream_format: "sse"`), which is also the only place it reports token usage; `tts-1`
and `tts-1-hd` do not support SSE and stream the raw audio body instead. `pcm` is 24 kHz 16-bit mono. `language`
and `timestamps` are not supported.
- **Gemini TTS** returns raw 16-bit PCM only (`audio/L16;codec=pcm;rate=24000`), so any `format` other than `pcm`
fails typed; wrap the samples yourself. Style is directed in the text, so `instructions` and `speed` fail typed.
Only `gemini-3.1-flash-tts-preview` and later support streaming. Two-speaker audio goes through
`providerOptions.speechConfig.multiSpeakerVoiceConfig`.
- **ElevenLabs** requires `voice` (the path voice id) and authenticates with `xi-api-key`. `format` maps to the
`output_format` query parameter (`mp3_44100_128`, `pcm_24000`, `wav_24000`, `opus_48000_64`);
`providerOptions.outputFormat` sets the exact string. WAV is only available from `generate`. `timestamps: true`
selects the `with-timestamps` endpoints and yields character-level alignment. `instructions` is not supported.
- **Cartesia** requires `voice` and pins `Cartesia-Version`. `generate` defaults to MP3 from `/tts/bytes`; streams
and `timestamps: true` (word-level) use `/tts/sse`, which only serves raw PCM. `providerOptions.sampleRate`,
`bitRate`, and `encoding` complete `output_format`. No usage is reported.
- **Deepgram** Aura's voice is the model id (`aura-2-thalia-en`), so `voice` and `language` fail typed. `format`
and `providerOptions` lower to query parameters (`encoding`, `container`, `sample_rate`, `bit_rate`); `pcm` is
`linear16` without a container. Auth is `Authorization: Token <DEEPGRAM_API_KEY>`.
The promise client mirrors the Effect API; `ai.speech.stream` is an `AsyncIterable`.
```ts
import { ai } from "@opencode/ai/promise"
const response = await ai.speech.generate({ model, text: "Hello from OpenCode.", voice: "coral" })
await Bun.write("hello.mp3", await ai.run(response.audio.bytes()))
for await (const event of ai.speech.stream({ model, text: "Hello from OpenCode.", voice: "coral" })) {
if (event.type === "audio-delta") player.write(event.chunk)
}
```
## Public API
- **`LLM.request({...})`** — build a provider-neutral `LLMRequest`. Accepts ergonomic inputs (`system: string`, `prompt: string`) that normalize into the canonical Schema classes.
@@ -693,7 +777,8 @@ const video = await generation.await({ poll: { interval: 10_000 }, signal })
- **`ImageClient`** — Effect service and layer for image execution, parallel to `LLMClient`.
- **`Media`** — the shared asset type (`Media.Asset`, `Media.Source`) and constructors used by messages, tool results, and media requests.
- **`Generation`** — provider-neutral handle for an in-flight media generation (`await`, `refresh`, `cancel`, `events`) used by queued media routes.
- **`@opencode/ai/promise`** — `AI.make({ layer? })` and a default `ai` client exposing `llm` and `image` as Promise / `AsyncIterable` APIs.
- **`Speech.request` / `Speech.generate` / `Speech.stream`** — text-to-speech through a provider-neutral request; `SpeechClient` is its Effect service and layer.
- **`@opencode/ai/promise`** — `AI.make({ layer? })` and a default `ai` client exposing `llm`, `image`, `video`, and `speech` as Promise / `AsyncIterable` APIs.
## Testing
+62 -12
View File
@@ -1,6 +1,6 @@
# Media generation in `@opencode/ai` — public API direction
Status: phases 1–2 implemented; phases 3–5 proposal.
Status: phases 1–2 implemented; phase 3 Speech implemented, Transcription pending; phases 4–5 proposal.
## Goal
@@ -181,23 +181,71 @@ hints (none of the four providers emit one). Later providers: Luma, Kling, MiniM
#### Speech (TTS)
Shipped in phase 3 (`src/speech.ts`, `src/speech-client.ts`, protocols `openai-speech`, `google-speech`,
`elevenlabs-speech`, `cartesia-speech`, `deepgram-speech`; new `ElevenLabs`, `Cartesia`, and `Deepgram` facades).
```ts
const request = Speech.request({
model: elevenlabs.speech("eleven_v3"),
model: elevenlabs.speech("eleven_flash_v2_5"),
text: "Hello from OpenCode.",
voice: "JBFqnCBsd6RMkjVDRZzb", // name, uuid, or { id } — provider-normalized
voice: "JBFqnCBsd6RMkjVDRZzb", // provider-native identifier, or { id }
format: "mp3", // mp3 | wav | pcm | opus | aac | flac | (string & {})
speed: 1.0,
language: "en",
instructions: "Warm, unhurried.",
providerOptions: { stability: 0.5 },
instructions: "Warm, unhurried.", // only OpenAI; elsewhere fails typed
timestamps: true, // request alignment; routes without it fail typed
providerOptions: { voice_settings: { stability: 0.5 } },
})
const response = yield* Speech.generate(request) // SpeechResponse: audio: Media.Asset, timestamps?, usage
yield* Speech.stream(request) // Stream<SpeechEvent>: audio-delta { chunk } | timestamps { words } | finish
const response = yield* Speech.generate(request) // SpeechResponse: audio: Media.Asset, timestamps?, usage?, providerMetadata?
yield* Speech.stream(request) // Stream<SpeechEvent>: audio-delta { chunk } | timestamps { items } | finish { audio, usage? }
```
Streaming TTS is first-class on day one: OpenAI `stream_format: sse`, ElevenLabs `/stream`, Cartesia SSE, Deepgram chunked. Input-streaming TTS (WS, text arrives incrementally) is a later `Speech.session(...)` scoped resource, not part of `generate`.
Execution is `MediaProtocol.stream` for every provider: one request whose body is framed and folded by a `step`
state machine, with `generate` running the same stream and collecting it. The route submits the request with its
`mode` (`"generate" | "stream"`), which lets one provider stay one protocol — OpenAI adds `stream_format: "sse"` (except `tts-1`/`tts-1-hd`, which stream raw bytes), ElevenLabs appends
`/stream`, Cartesia switches `/tts/bytes` to `/tts/sse`, Gemini switches `generateContent` to
`streamGenerateContent`. The terminal `finish` event carries the assembled asset (every provider's stream is
concatenable chunks), so stream consumers also get the whole file and `generate` is just "take `finish`, gather
`timestamps`". The cost is memory: a stream holds every chunk until `finish`, so even a consumer that only plays deltas
keeps the whole clip in memory. That is bounded by the providers' input text limits (a few minutes of audio); a
long-form or session API would need an opt-out.
**Voice.** `voice?: string | { id: string }`. A string is passed through as the provider's native identifier — a
name on OpenAI and Gemini, a voice id on ElevenLabs (path segment) and Cartesia. `{ id }` selects an OpenAI custom
voice and is treated as the plain string on routes that do not distinguish custom from built-in. Deepgram's voice is
the model id (`aura-2-thalia-en`), so `voice` is `unsupported` there. There is no cross-provider voice catalog or
name→id resolution. Multi-speaker (Gemini `speechConfig.multiSpeakerVoiceConfig`) and per-voice settings
(ElevenLabs `voice_settings`) go through `providerOptions`.
**Format and PCM.** `format` is container-level; provider sample rates and bitrates live under `providerOptions`
(ElevenLabs `outputFormat`, Cartesia `sampleRate`/`bitRate`/`encoding`, Deepgram `encoding`/`container`/`sampleRate`/
`bitRate`). Each protocol maps `format` to its native value (ElevenLabs `mp3_44100_128`/`pcm_24000`/`wav_24000`/
`opus_48000_64`, Cartesia `{ container, encoding, sample_rate }`, Deepgram `encoding`+`container`) and declares the
asset's media type rather than sniffing, because headerless PCM can look like an MPEG frame sync. Headerless PCM
always carries `info.encoding`, `info.sampleRate`, and `info.channels`; its media type is the provider's declaration
(Gemini `audio/L16;codec=pcm;rate=24000`, Deepgram's `content-type`) or `audio/pcm`. Gemini returns PCM only, so any
other `format` is rejected rather than wrapped as WAV by the route. Every `format` value a route cannot produce (unknown
to it, a container on Cartesia SSE, WAV on an ElevenLabs stream, anything but PCM on Gemini) fails the same way as an
unsupported field: `UnsupportedOperation` with `operation: "media.format"`.
**Timestamps.** `timestamps: true` on the request asks for alignment. ElevenLabs selects the `with-timestamps`
endpoints (character-level, NDJSON when streaming); Cartesia sets `add_timestamps` on `/tts/sse` (word-level; a
`generate` with timestamps collects the SSE stream). OpenAI, Gemini, and Deepgram reject it.
Common-field lowering per provider:
| Provider | `voice` | `speed` | `language` | `instructions` | `timestamps` | Usage |
|---|---|---|---|---|---|---|
| OpenAI | `voice` (name or `{ id }`) | `speed` | unsupported | `instructions` | unsupported | `tokens` from SSE `speech.audio.done` only |
| Gemini | `prebuiltVoiceConfig.voiceName` | unsupported | `speechConfig.languageCode` | unsupported (direct in text) | unsupported | `tokens` from `usageMetadata` |
| ElevenLabs | path voice id (required) | `voice_settings.speed` | `language_code` | unsupported | `with-timestamps` | `credits` from `character-cost` header |
| Cartesia | `voice` (required) | `generation_config.speed` | `language` | unsupported | `add_timestamps` | none |
| Deepgram | unsupported (voice is the model) | `speed` query | unsupported | unsupported | unsupported | `characters` from `dg-char-count` header |
Deferred: `Speech.session(...)` — input-streaming TTS where text arrives incrementally over a WebSocket (ElevenLabs
`stream-input`, Cartesia WebSocket contexts, Deepgram WebSocket speak) — is a separate scoped resource, not part of
`generate`/`stream`, and ships with the realtime work in phase 5.
#### Transcription (STT)
@@ -293,8 +341,10 @@ Existing facades gain per-modality selectors; the modality routes each facade pr
| `Google` | Gemini | Gemini-native (default), `imagen` | Veo | Gemini TTS | Gemini transcribe | |
| `XAI` | ✓ | ✓ | ✓ | | | |
| `ElevenLabs` | | | | ✓ | Scribe | soundEffect, music |
| `Cartesia` | | | | ✓ | | |
| `Deepgram` | | | | Aura | ✓ | |
| `Fal` | | ✓ | ✓ | | | |
| `Replicate`, `Runway`, `Luma`, `Kling`, `MiniMax`, `Deepgram`, `Cartesia`, `AssemblyAI`, `BlackForestLabs`, `Stability` | | per provider | | | | |
| `Replicate`, `Runway`, `Luma`, `Kling`, `MiniMax`, `AssemblyAI`, `BlackForestLabs`, `Stability` | | per provider | | | | |
New facades follow the existing one-file-per-provider rule. Package entrypoints are modality-specific, such as `@opencode/ai/providers/openai/images`, and return the concrete model.
@@ -306,9 +356,9 @@ Media does not fit the LLM four-axis route (SSE frames → event state machine)
- `MediaProtocol.inline` — `body.from(request)` (JSON, multipart, or query), `response.decode(response)` (JSON, or binary body → `Media.Asset`).
- `MediaProtocol.queued` — `start` (body + decode to `{ token, snapshot }`), `status`, `result`, optional `cancel`, `pollHint`, and a `token` codec. `result` is always a separate GET (against the status document for Veo/xAI/Runway, fal's `response_url` otherwise) so `await` after `start` and after `resume` share one path. `PollContext.auth` hands the auth headers the route sent to the protocol for output URLs that need them (Veo downloads); they become transient `Media.Asset.headers`, never part of `source`. There is no separate `download` step: `Media.Asset.bytes()` downloads through the executor with those headers. `MediaRoute.inline(...)` / `MediaRoute.queued(...)` compose each kind with endpoint and auth; the queued route decodes the token once and hands `Generation` a token-free `{ status, result, cancel? }`.
- `MediaProtocol.stream` — framing + `step` state machine emitting modality events, same discipline as LLM protocols.
- `MediaProtocol.stream` — `body.from(request)` over the request plus its `mode`, `frames` (a function that picks the framing for the call: `Framing.sse`, `lines`, `document`, or the raw bytes), fresh per-response `initial()` state, `step` emitting modality events, and `finish(state, context)` — with the observed response for header-only usage — emitting exactly one terminal event or failing as an incomplete stream. The route fills `reason.http` on stream errors. `MediaRoute.stream(...)` exposes `stream` and `generate` (the same stream folded by the modality's `collect`).
`MediaRoute.inline` / `MediaRoute.queued` compose one protocol kind with endpoint/auth; `ImageModel`/`VideoModel` share the `MediaModel` base (`src/media-model.ts`).
`MediaRoute.inline` / `MediaRoute.queued` / `MediaRoute.stream` compose one protocol kind with endpoint/auth; `ImageModel`/`VideoModel`/`SpeechModel` share the `MediaModel` base (`src/media-model.ts`).
### LLM integration
@@ -338,7 +388,7 @@ Foundation + Image ship together as the reference implementation, serially. Vide
1. **Foundation** — per-modality selectors, `Media`, `Generation`, `Poll`, `Usage` union, `MediaProtocol` kinds, `@opencode/ai/promise` with `llm` + `image`. Port the five existing image protocols onto it. Unify `MediaPart` and add the `media` LLM event (fixes Gemini image output being dropped).
2. **Video** — ✅ Veo, xAI, fal, Runway shipped (`MediaProtocol.queued`, `Video.start/generate/resume/stream`, promise `ai.video`). Deferred: `Video.complete` (webhooks), Luma, Kling, MiniMax, Replicate.
3. **Speech + Transcription** — OpenAI, ElevenLabs, Gemini TTS, Deepgram, Cartesia, AssemblyAI. Streaming TTS from the start.
3. **Speech + Transcription** — ✅ Speech: OpenAI, Gemini TTS, ElevenLabs, Cartesia, Deepgram shipped (`MediaProtocol.stream`, `Speech.generate/stream`, promise `ai.speech`). Pending: Transcription (OpenAI, ElevenLabs Scribe, Gemini, Deepgram, AssemblyAI). Deferred: `Speech.session` (WebSocket input streaming).
4. **Image queued routes and partials** — BFL, fal, Replicate, Stability; OpenAI `partial_images` streaming.
5. **Later** — ElevenLabs music/SFX, Lyria, `Speech.session` / `Transcription.session`, realtime.
+35
View File
@@ -134,6 +134,41 @@ const PROVIDERS: ReadonlyArray<Provider> = [
"X-Runway-Version": "2024-11-06",
}),
},
{
id: "elevenlabs",
label: "ElevenLabs",
tier: "canary",
note: "ElevenLabs text-to-speech recorded tests",
vars: [{ name: "ELEVENLABS_API_KEY" }],
validate: (env) =>
HttpClientRequest.get("https://api.elevenlabs.io/v1/models").pipe(
HttpClientRequest.setHeader("xi-api-key", Redacted.value(Redacted.make(env.ELEVENLABS_API_KEY))),
executeRequest,
),
},
{
id: "cartesia",
label: "Cartesia",
tier: "canary",
note: "Cartesia text-to-speech recorded tests",
vars: [{ name: "CARTESIA_API_KEY" }],
validate: (env) =>
validateBearer("https://api.cartesia.ai/voices?limit=1", Redacted.make(env.CARTESIA_API_KEY), {
"Cartesia-Version": "2026-08-14",
}),
},
{
id: "deepgram",
label: "Deepgram",
tier: "canary",
note: "Deepgram Aura text-to-speech recorded tests",
vars: [{ name: "DEEPGRAM_API_KEY" }],
validate: (env) =>
HttpClientRequest.get("https://api.deepgram.com/v1/projects").pipe(
HttpClientRequest.setHeader("authorization", `Token ${Redacted.value(Redacted.make(env.DEEPGRAM_API_KEY))}`),
executeRequest,
),
},
{
id: "cloudflare-ai-gateway",
label: "Cloudflare AI Gateway",
+19
View File
@@ -48,6 +48,25 @@ export type {
VideoRoute,
} from "./video.js"
export { Video } from "./video.js"
export { SpeechClient } from "./speech-client.js"
export {
SpeechEvent,
SpeechModel,
SpeechModelSchema,
SpeechRequest,
SpeechResponse,
SpeechTimestamp,
SpeechVoice,
} from "./speech.js"
export type {
SpeechFormat,
SpeechModelOptions,
SpeechOptions,
SpeechRequestFor,
SpeechRequestInput,
SpeechRoute,
} from "./speech.js"
export { Speech } from "./speech.js"
export { Media } from "./media.js"
export { Generation } from "./generation.js"
export type {
+3 -3
View File
@@ -6,8 +6,8 @@ import { AIError, HttpOptions, InvalidRequestError, ModelID, ProviderID } from "
/**
* What every media model carries: ids, the configured route, and deployment `http` overlays. Modality classes
* (`ImageModel`, `VideoModel`) extend it with their route type and a nominal marker so one cannot stand in for the
* other in requests.
* (`ImageModel`, `VideoModel`, `SpeechModel`) extend it with their route type and a nominal marker so one cannot stand
* in for the other in requests.
*/
export class MediaModel<Route, Options> {
declare protected readonly _Options: (options: Options) => Options
@@ -44,7 +44,7 @@ export namespace MediaModel {
}
}
/** Compose a protocol route input with one deployment through `MediaRoute.inline` or `MediaRoute.queued`. */
/** Compose a protocol route input with one deployment through `MediaRoute.inline`, `queued`, or `stream`. */
export const composeRoute = <Request extends MediaRoute.MediaRequest, Protocol, Route>(
compose: (input: MediaRoute.Composition<Request> & { readonly protocol: Protocol }) => Route,
route: MediaModel.RouteInput<Request, Protocol>,
+19 -3
View File
@@ -7,12 +7,14 @@ import { LLMClient } from "./route/client.js"
import { RequestExecutor } from "./route/executor.js"
import { LanguageModel, LLMRequest } from "./schema/index.js"
import type { RequestInput } from "./llm.js"
import { Speech, SpeechModel, SpeechRequest, type SpeechRequestInput } from "./speech.js"
import { SpeechClient } from "./speech-client.js"
import { Video, VideoModel, VideoRequest, type VideoOptions, type VideoRequestInput } from "./video.js"
import { VideoClient } from "./video-client.js"
/**
* Promise-first entrypoint for scripts and non-Effect callers. One `ManagedRuntime` hosts the LLM, image, and video
* clients over a request executor; every method runs the corresponding Effect API and rethrows `AIError` unchanged.
* Promise-first entrypoint for scripts and non-Effect callers. One `ManagedRuntime` hosts the LLM, image, video, and
* speech clients over a request executor; every method runs the corresponding Effect API and rethrows `AIError` unchanged.
*/
export interface Options {
/** Executor layer; defaults to `RequestExecutor.fetchLayer`. Inject a recorder or middleware here. */
@@ -27,6 +29,7 @@ export type Services =
| Layer.Success<typeof LLMClient.layer>
| Layer.Success<typeof ImageClient.layer>
| Layer.Success<typeof VideoClient.layer>
| Layer.Success<typeof SpeechClient.layer>
| RequestExecutor.Service
/** Promise view of a `Generation`: its snapshot plus `await`, `refresh`, and `cancel` returning promises. */
@@ -53,7 +56,7 @@ const abortEffect = (signal: AbortSignal | undefined) =>
export const make = (options: Options = {}) => {
const runtime = ManagedRuntime.make(
Layer.mergeAll(LLMClient.layer, ImageClient.layer, VideoClient.layer).pipe(
Layer.mergeAll(LLMClient.layer, ImageClient.layer, VideoClient.layer, SpeechClient.layer).pipe(
Layer.provideMerge(options.layer ?? RequestExecutor.fetchLayer),
),
)
@@ -88,6 +91,8 @@ export const make = (options: Options = {}) => {
input instanceof ImageRequest ? input : Image.request(input)
const videoRequest = (input: VideoRequestInput | VideoRequest) =>
input instanceof VideoRequest ? input : Video.request(input)
const speechRequest = (input: SpeechRequestInput | SpeechRequest) =>
input instanceof SpeechRequest ? input : Speech.request(input)
return {
run,
@@ -122,6 +127,17 @@ export const make = (options: Options = {}) => {
options?: AwaitOptions & RunOptions,
) => iterate(Video.stream(videoRequest(input), { poll: options?.poll }), options),
},
speech: {
request: Speech.request,
generate: <const Model extends SpeechModel>(
input: SpeechRequestInput<Model> | SpeechRequest,
options?: RunOptions,
) => run(Speech.generate(speechRequest(input)), options),
stream: <const Model extends SpeechModel>(
input: SpeechRequestInput<Model> | SpeechRequest,
options?: RunOptions,
) => iterate(Speech.stream(speechRequest(input)), options),
},
dispose: () => runtime.dispose(),
}
}
@@ -28,6 +28,7 @@ import { Lifecycle } from "./utils/lifecycle.js"
import { MistralToolID } from "./utils/mistral-tool-id.js"
import { ToolSchemaProjection } from "./utils/tool-schema.js"
import { ToolStream } from "./utils/tool-stream.js"
import { concatBytes } from "../utils/bytes.js"
const ADAPTER = "bedrock-converse"
@@ -524,14 +525,7 @@ interface ParserState {
readonly reasoningRedactedContent: Readonly<Record<number, ReadonlyArray<Uint8Array>>>
}
const encodeRedactedContent = (chunks: ReadonlyArray<Uint8Array>) => {
const bytes = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0))
chunks.reduce((offset, chunk) => {
bytes.set(chunk, offset)
return offset + chunk.length
}, 0)
return Encoding.encodeBase64(bytes)
}
const encodeRedactedContent = (chunks: ReadonlyArray<Uint8Array>) => Encoding.encodeBase64(concatBytes(chunks))
const step = (state: ParserState, event: BedrockEvent) =>
Effect.gen(function* () {
@@ -0,0 +1,195 @@
import { Effect, Schema } from "effect"
import { classifyProviderFailure } from "../provider-error.js"
import { Framing } from "../route/framing.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { AIError, ProviderID, mergeJsonRecords } from "../schema/index.js"
import { SpeechModel, type SpeechEvent, type SpeechRequestFor } from "../speech.js"
import { ProviderShared, optionalNull } from "./shared.js"
import { SpeechStream } from "./utils/speech-stream.js"
const ADAPTER = "cartesia-speech"
const NAME = "Cartesia"
const PROVIDER = ProviderID.make("cartesia")
export const DEFAULT_BASE_URL = "https://api.cartesia.ai"
export const API_VERSION = "2026-08-14"
export const BYTES_PATH = "/tts/bytes"
export const SSE_PATH = "/tts/sse"
const DEFAULT_SAMPLE_RATE = 44100
const DEFAULT_BIT_RATE = 128000
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type CartesiaSpeechString<Known extends string> = Known | (string & {})
export type CartesiaEncoding = SpeechStream.PcmEncoding
export type CartesiaSpeechOptions = {
readonly sampleRate?: 8000 | 16000 | 22050 | 24000 | 44100 | 48000
readonly bitRate?: 32000 | 64000 | 96000 | 128000 | 192000
readonly encoding?: CartesiaEncoding
readonly generation_config?: {
readonly volume?: number
readonly emotion?: CartesiaSpeechString<"neutral" | "calm" | "angry" | "content" | "sad" | "scared">
}
readonly pronunciation_dict_id?: string
} & Record<string, unknown>
export type Request = SpeechRequestFor<CartesiaSpeechOptions>
// ---------------------------------------------------------------------------
// 3. Streaming event schema
// ---------------------------------------------------------------------------
/** `phoneme_timestamps` and future record types are ignored. */
const SseEvent = Schema.Struct({
type: Schema.String,
data: Schema.optional(Schema.Uint8ArrayFromBase64),
word_timestamps: Schema.optional(
Schema.Struct({
words: Schema.Array(Schema.String),
start: Schema.Array(Schema.Number),
end: Schema.Array(Schema.Number),
}),
),
status_code: Schema.optional(Schema.Number),
title: Schema.optional(Schema.String),
message: Schema.optional(Schema.String),
error_code: optionalNull(Schema.String),
})
const decodeEvent = MediaProtocol.decodeFrame(ADAPTER, NAME, SseEvent)
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
interface State extends SpeechStream.Audio {
readonly done: boolean
}
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
/** Timestamps exist only on the SSE endpoint, so a `generate` that asks for them collects an SSE stream. */
const usesSse = (request: MediaProtocol.Addressed<Request>) => request.mode === "stream" || request.timestamps === true
const CONTAINERS: Readonly<Record<string, "raw" | "wav" | "mp3">> = { pcm: "raw", wav: "wav", mp3: "mp3" }
const outputFormat = Effect.fn("CartesiaSpeech.outputFormat")(function* (request: MediaProtocol.Addressed<Request>) {
const sse = usesSse(request)
const format = request.format ?? (sse ? "pcm" : "mp3")
const container = CONTAINERS[format]
if (container === undefined)
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} supports the pcm, wav, and mp3 formats, not "${format}"`,
)
if (sse && container !== "raw")
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} streams and timestamps only raw PCM; request format "pcm" instead of "${format}"`,
)
const sampleRate = request.providerOptions?.sampleRate ?? DEFAULT_SAMPLE_RATE
if (container === "mp3")
return { container, sample_rate: sampleRate, bit_rate: request.providerOptions?.bitRate ?? DEFAULT_BIT_RATE }
return { container, encoding: request.providerOptions?.encoding ?? "pcm_s16le", sample_rate: sampleRate }
})
const fromRequest = Effect.fn("CartesiaSpeech.fromRequest")(function* (request: MediaProtocol.Addressed<Request>) {
const voice = SpeechStream.voiceID(request.voice)
if (voice === undefined)
return yield* ProviderShared.invalidRequest(`${NAME} requires a voice id; pass it as \`voice\``)
const { sampleRate: _sampleRate, bitRate: _bitRate, encoding: _encoding, ...native } = request.providerOptions ?? {}
return MediaProtocol.json(
mergeJsonRecords(
{
model_id: request.model.id,
transcript: request.text,
voice,
output_format: yield* outputFormat(request),
language: request.language,
generation_config: request.speed === undefined ? undefined : { speed: request.speed },
add_timestamps: request.timestamps === true ? true : undefined,
},
native,
request.http?.body,
) ?? {},
)
})
// ---------------------------------------------------------------------------
// 6. Stream parsing
// ---------------------------------------------------------------------------
const onEvent = Effect.fn("CartesiaSpeech.onEvent")(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
if (event.type === "chunk" && event.data !== undefined) return SpeechStream.delta(state, event.data)
if (event.type === "timestamps" && event.word_timestamps !== undefined) {
const words = event.word_timestamps
return [state, SpeechStream.timestamps(words.words, words.start, words.end)] as const
}
if (event.type === "done") return [{ ...state, done: true }, []] as const
if (event.type === "error")
return yield* new AIError({
reason: classifyProviderFailure({
message: `${NAME} stream failed${event.title === undefined ? "" : ` (${event.title})`}: ${event.message ?? "unknown error"}`,
status: event.status_code,
rawBody: frame,
}),
})
return [state, []] as const
})
const finish = Effect.fn("CartesiaSpeech.finish")(function* (
state: State,
context: MediaProtocol.ResponseContext<Request>,
) {
if (usesSse(context.request) && !state.done) return yield* MediaProtocol.incomplete(ADAPTER)
const format = yield* outputFormat(context.request)
return yield* SpeechStream.finish(
ADAPTER,
state,
format.container === "raw"
? SpeechStream.pcm(format.encoding, format.sample_rate)
: SpeechStream.container(format.container, format.sample_rate),
)
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.stream<Request, SpeechEvent, string | Uint8Array, State>({
id: ADAPTER,
name: NAME,
unsupported: ["instructions"],
body: { from: fromRequest },
frames: (bytes, context) => (usesSse(context.request) ? Framing.sse.frame(bytes) : bytes),
initial: () => ({ chunks: [], done: false }),
step: SpeechStream.step(onEvent),
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
SpeechModel.fromRoute<CartesiaSpeechOptions, string | Uint8Array, State>(
{
id: ADAPTER,
provider: PROVIDER,
protocol,
baseURL: DEFAULT_BASE_URL,
headers: { "Cartesia-Version": API_VERSION },
path: ({ request }) => (usesSse(request) ? SSE_PATH : BYTES_PATH),
},
input,
)
export const CartesiaSpeech = {
protocol,
model,
} as const
@@ -0,0 +1,145 @@
import { Effect } from "effect"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { SpeechModel, type SpeechEvent, type SpeechRequestFor } from "../speech.js"
import { SpeechStream } from "./utils/speech-stream.js"
const ADAPTER = "deepgram-speech"
const NAME = "Deepgram"
const PROVIDER = ProviderID.make("deepgram")
export const DEFAULT_BASE_URL = "https://api.deepgram.com"
export const PATH = "/v1/speak"
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type DeepgramSpeechString<Known extends string> = Known | (string & {})
export type DeepgramEncoding = DeepgramSpeechString<"linear16" | "mulaw" | "alaw" | "mp3" | "opus" | "flac" | "aac">
export type DeepgramSpeechOptions = {
readonly encoding?: DeepgramEncoding
readonly container?: DeepgramSpeechString<"wav" | "ogg" | "none">
readonly sampleRate?: number
readonly bitRate?: number
readonly mip_opt_out?: boolean
readonly tag?: string
} & Record<string, unknown>
export type Request = SpeechRequestFor<DeepgramSpeechOptions>
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
type State = SpeechStream.Audio
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
const FORMATS: Readonly<Record<string, { readonly encoding: string; readonly container?: string }>> = {
mp3: { encoding: "mp3" },
wav: { encoding: "linear16", container: "wav" },
pcm: { encoding: "linear16", container: "none" },
opus: { encoding: "opus" },
flac: { encoding: "flac" },
aac: { encoding: "aac" },
}
const queryValue = (value: unknown) =>
typeof value === "string" || typeof value === "number" || typeof value === "boolean" ? String(value) : undefined
const queryParameters = (request: Request) => {
const format = request.format === undefined ? undefined : FORMATS[request.format]
const { encoding, container, sampleRate, bitRate, ...native } = request.providerOptions ?? {}
return Object.fromEntries(
Object.entries({
...native,
model: request.model.id,
encoding: encoding ?? format?.encoding,
container: container ?? format?.container,
sample_rate: sampleRate,
bit_rate: bitRate,
speed: request.speed,
}).flatMap(([key, value]) => {
const text = queryValue(value)
return text === undefined ? [] : [[key, text]]
}),
)
}
const fromRequest = Effect.fn("DeepgramSpeech.fromRequest")(function* (request: Request) {
if (
request.format !== undefined &&
FORMATS[request.format] === undefined &&
request.providerOptions?.encoding === undefined
)
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} has no encoding for format "${request.format}"; pass providerOptions.encoding`,
)
return MediaProtocol.json(
mergeJsonRecords({ text: request.text }, request.http?.body) ?? {},
queryParameters(request),
)
})
// ---------------------------------------------------------------------------
// 6. Stream parsing
// ---------------------------------------------------------------------------
const HEADERLESS_ENCODINGS: Readonly<Record<string, SpeechStream.PcmEncoding>> = {
linear16: "pcm_s16le",
mulaw: "pcm_mulaw",
alaw: "pcm_alaw",
}
const finish = (state: State, context: MediaProtocol.ResponseContext<Request>) => {
const headers = context.http.headers
const mediaType = headers["content-type"]
const query = queryParameters(context.request)
const encoding = HEADERLESS_ENCODINGS[query.encoding ?? ""]
const requestID = headers["dg-request-id"]
const modelName = headers["dg-model-name"]
return SpeechStream.finish(ADAPTER, state, {
...(query.container === "none" && encoding !== undefined
? SpeechStream.pcm(encoding, SpeechStream.sampleRate(mediaType), mediaType)
: // Deepgram's default encoding is MP3; WAV is a container around any encoding.
{ mediaType, info: { format: query.container === "wav" ? "wav" : (query.encoding ?? "mp3") } }),
usage: SpeechStream.headerUsage("characters", headers["dg-char-count"]),
providerMetadata:
requestID === undefined && modelName === undefined
? undefined
: { deepgram: { requestId: requestID, modelName } },
})
}
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.stream<Request, SpeechEvent, Uint8Array, State>({
id: ADAPTER,
name: NAME,
unsupported: ["voice", "language", "instructions", "timestamps"],
body: { from: fromRequest },
frames: (bytes) => bytes,
initial: () => ({ chunks: [] }),
step: (state, frame) => Effect.succeed(SpeechStream.delta(state, frame)),
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
SpeechModel.fromRoute<DeepgramSpeechOptions, Uint8Array, State>(
{ id: ADAPTER, provider: PROVIDER, protocol, baseURL: DEFAULT_BASE_URL, path: PATH },
input,
)
export const DeepgramSpeech = {
protocol,
model,
} as const
@@ -0,0 +1,218 @@
import { Effect, Schema } from "effect"
import { Framing } from "../route/framing.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords } from "../schema/index.js"
import { SpeechModel, type SpeechEvent, type SpeechRequestFor } from "../speech.js"
import { ProviderShared, optionalNull } from "./shared.js"
import { SpeechStream } from "./utils/speech-stream.js"
const ADAPTER = "elevenlabs-speech"
const NAME = "ElevenLabs"
const PROVIDER = ProviderID.make("elevenlabs")
export const DEFAULT_BASE_URL = "https://api.elevenlabs.io"
export const PATH = "/v1/text-to-speech"
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type ElevenLabsSpeechString<Known extends string> = Known | (string & {})
export type ElevenLabsOutputFormat = ElevenLabsSpeechString<
| "mp3_22050_32"
| "mp3_24000_48"
| "mp3_44100_32"
| "mp3_44100_64"
| "mp3_44100_96"
| "mp3_44100_128"
| "mp3_44100_192"
| "pcm_8000"
| "pcm_16000"
| "pcm_22050"
| "pcm_24000"
| "pcm_32000"
| "pcm_44100"
| "pcm_48000"
| "wav_8000"
| "wav_16000"
| "wav_22050"
| "wav_24000"
| "wav_32000"
| "wav_44100"
| "wav_48000"
| "ulaw_8000"
| "alaw_8000"
| "opus_48000_32"
| "opus_48000_64"
| "opus_48000_96"
| "opus_48000_128"
| "opus_48000_192"
>
export type ElevenLabsSpeechOptions = {
readonly outputFormat?: ElevenLabsOutputFormat
readonly voice_settings?: {
readonly stability?: number
readonly similarity_boost?: number
readonly style?: number
readonly use_speaker_boost?: boolean
}
readonly seed?: number
readonly apply_text_normalization?: ElevenLabsSpeechString<"auto" | "on" | "off">
} & Record<string, unknown>
export type Request = SpeechRequestFor<ElevenLabsSpeechOptions>
// ---------------------------------------------------------------------------
// 3. Streaming event schema
// ---------------------------------------------------------------------------
const Alignment = Schema.Struct({
characters: Schema.Array(Schema.String),
character_start_times_seconds: Schema.Array(Schema.Number),
character_end_times_seconds: Schema.Array(Schema.Number),
})
const TimestampedAudio = Schema.Struct({
audio_base64: Schema.Uint8ArrayFromBase64,
alignment: optionalNull(Alignment),
})
const decodeRecord = MediaProtocol.decodeFrame(ADAPTER, NAME, TimestampedAudio)
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
type State = SpeechStream.Audio
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
const OUTPUT_FORMATS: Readonly<Record<string, string>> = {
mp3: "mp3_44100_128",
pcm: "pcm_24000",
wav: "wav_24000",
opus: "opus_48000_64",
}
/** WAV is served only by the non-streaming endpoints. */
const outputFormat = Effect.fn("ElevenLabsSpeech.outputFormat")(function* (request: MediaProtocol.Addressed<Request>) {
const format = request.providerOptions?.outputFormat ?? OUTPUT_FORMATS[request.format ?? "mp3"]
if (format === undefined)
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} has no default output format for "${request.format}"; pass providerOptions.outputFormat`,
)
if (request.mode === "stream" && format.startsWith("wav_"))
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} streams mp3, pcm, opus, ulaw, and alaw but not "${format}"; use generate for WAV`,
)
return format
})
const fromRequest = Effect.fn("ElevenLabsSpeech.fromRequest")(function* (request: MediaProtocol.Addressed<Request>) {
if (request.voice === undefined)
return yield* ProviderShared.invalidRequest(`${NAME} requires a voice id; pass it as \`voice\``)
const { outputFormat: _outputFormat, ...native } = request.providerOptions ?? {}
return MediaProtocol.json(
mergeJsonRecords(
{
text: request.text,
model_id: request.model.id,
language_code: request.language,
voice_settings: request.speed === undefined ? undefined : { speed: request.speed },
},
native,
request.http?.body,
) ?? {},
{ output_format: yield* outputFormat(request) },
)
})
const path = (request: MediaProtocol.Addressed<Request>) =>
`${PATH}/${encodeURIComponent(SpeechStream.voiceID(request.voice) ?? "")}${request.mode === "stream" ? "/stream" : ""}${
request.timestamps === true ? "/with-timestamps" : ""
}`
// ---------------------------------------------------------------------------
// 6. Stream parsing
// ---------------------------------------------------------------------------
const onRecord = Effect.fn("ElevenLabsSpeech.onRecord")(function* (state: State, frame: string) {
const record = yield* decodeRecord(frame)
const [next, events] = SpeechStream.delta(state, record.audio_base64)
const alignment = record.alignment
if (!alignment) return [next, events] as const
return [
next,
[
...events,
...SpeechStream.timestamps(
alignment.characters,
alignment.character_start_times_seconds,
alignment.character_end_times_seconds,
),
],
] as const
})
const PCM_CODECS: Readonly<Record<string, SpeechStream.PcmEncoding>> = {
pcm: "pcm_s16le",
ulaw: "pcm_mulaw",
alaw: "pcm_alaw",
}
const describeOutput = (format: string) => {
const [codec = format, rate] = format.split("_")
const sampleRate = rate === undefined ? undefined : Number(rate)
const encoding = PCM_CODECS[codec]
return encoding === undefined ? SpeechStream.container(codec, sampleRate) : SpeechStream.pcm(encoding, sampleRate)
}
const finish = Effect.fn("ElevenLabsSpeech.finish")(function* (
state: State,
context: MediaProtocol.ResponseContext<Request>,
) {
const requestID = context.http.headers["request-id"]
return yield* SpeechStream.finish(ADAPTER, state, {
...describeOutput(yield* outputFormat(context.request)),
// `character-cost` is billed credits, not a character count (3 for 20 characters on `eleven_flash_v2_5`).
usage: SpeechStream.headerUsage("credits", context.http.headers["character-cost"]),
providerMetadata: requestID === undefined ? undefined : { elevenlabs: { requestId: requestID } },
})
})
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.stream<Request, SpeechEvent, string | Uint8Array, State>({
id: ADAPTER,
name: NAME,
unsupported: ["instructions"],
body: { from: fromRequest },
frames: (bytes, context) => {
if (context.request.timestamps !== true) return bytes
return context.request.mode === "stream" ? Framing.lines.frame(bytes) : Framing.document.frame(bytes)
},
initial: () => ({ chunks: [] }),
step: SpeechStream.step(onRecord),
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
SpeechModel.fromRoute<ElevenLabsSpeechOptions, string | Uint8Array, State>(
{ id: ADAPTER, provider: PROVIDER, protocol, baseURL: DEFAULT_BASE_URL, path: ({ request }) => path(request) },
input,
)
export const ElevenLabsSpeech = {
protocol,
model,
} as const
+212
View File
@@ -0,0 +1,212 @@
import { Effect, Schema } from "effect"
import { Framing } from "../route/framing.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { AIError, ContentPolicyError, ProviderID, mergeJsonRecords } from "../schema/index.js"
import { SpeechModel, type SpeechEvent, type SpeechRequestFor } from "../speech.js"
import { ProviderShared } from "./shared.js"
import { SpeechStream } from "./utils/speech-stream.js"
const ADAPTER = "google-speech"
const NAME = "Google Speech"
const PROVIDER = ProviderID.make("google")
export const DEFAULT_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
const DEFAULT_SAMPLE_RATE = 24000
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
/** Style is directed in the text itself, and `speechConfig.multiSpeakerVoiceConfig` excludes `voice`. */
export type GoogleSpeechOptions = {
readonly temperature?: number
readonly seed?: number
readonly speechConfig?: {
readonly multiSpeakerVoiceConfig?: {
readonly speakerVoiceConfigs: ReadonlyArray<{
readonly speaker: string
readonly voiceConfig: { readonly prebuiltVoiceConfig: { readonly voiceName: string } }
}>
}
}
} & Record<string, unknown>
export type Request = SpeechRequestFor<GoogleSpeechOptions>
// ---------------------------------------------------------------------------
// 3. Streaming event schema
// ---------------------------------------------------------------------------
const GenerateContentChunk = Schema.Struct({
candidates: Schema.optional(
Schema.Array(
Schema.Struct({
content: Schema.optional(
Schema.Struct({
parts: Schema.optional(
Schema.Array(
Schema.Struct({
text: Schema.optional(Schema.String),
inlineData: Schema.optional(
Schema.Struct({ mimeType: Schema.String, data: Schema.Uint8ArrayFromBase64 }),
),
}),
),
),
}),
),
finishReason: Schema.optional(Schema.String),
}),
),
),
promptFeedback: Schema.optional(
Schema.Struct({ blockReason: Schema.optional(Schema.String), blockReasonMessage: Schema.optional(Schema.String) }),
),
usageMetadata: Schema.optional(
Schema.Struct({
promptTokenCount: Schema.optional(Schema.Number),
candidatesTokenCount: Schema.optional(Schema.Number),
totalTokenCount: Schema.optional(Schema.Number),
}),
),
modelVersion: Schema.optional(Schema.String),
responseId: Schema.optional(Schema.String),
})
type GenerateContentChunk = Schema.Schema.Type<typeof GenerateContentChunk>
const decodeChunk = MediaProtocol.decodeFrame(ADAPTER, NAME, GenerateContentChunk)
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
interface State extends SpeechStream.Audio {
readonly mimeType?: string
readonly usage?: GenerateContentChunk["usageMetadata"]
readonly finishReason?: string
readonly modelVersion?: string
readonly responseId?: string
}
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
const fromRequest = Effect.fn("GoogleSpeech.fromRequest")(function* (request: MediaProtocol.Addressed<Request>) {
if (request.format !== undefined && request.format !== "pcm")
return yield* SpeechStream.unsupportedFormat(
PROVIDER,
ADAPTER,
`${NAME} only returns raw PCM; request format "pcm" or omit it, then wrap the samples yourself`,
)
const voiceName = SpeechStream.voiceID(request.voice)
return MediaProtocol.json(
mergeJsonRecords(
{
contents: [{ role: "user", parts: [{ text: request.text }] }],
generationConfig: mergeJsonRecords(
{
responseModalities: ["AUDIO"],
speechConfig: {
voiceConfig: voiceName === undefined ? undefined : { prebuiltVoiceConfig: { voiceName } },
languageCode: request.language,
},
},
request.providerOptions,
),
},
request.http?.body,
) ?? {},
)
})
// Only `gemini-3.1-flash-tts-preview` and later stream; earlier TTS models reject `streamGenerateContent`.
const path = (request: MediaProtocol.Addressed<Request>) =>
request.mode === "stream"
? `/models/${request.model.id}:streamGenerateContent?alt=sse`
: `/models/${request.model.id}:generateContent`
// ---------------------------------------------------------------------------
// 6. Stream parsing
// ---------------------------------------------------------------------------
const step = Effect.fn("GoogleSpeech.step")(function* (state: State, frame: string) {
const chunk = yield* decodeChunk(frame)
const blockReason = chunk.promptFeedback?.blockReason
if (blockReason !== undefined)
return yield* new AIError({
reason: new ContentPolicyError({
message: `${NAME} blocked the prompt (${blockReason})${
chunk.promptFeedback?.blockReasonMessage === undefined ? "" : `: ${chunk.promptFeedback.blockReasonMessage}`
}`,
body: frame,
}),
})
const candidate = chunk.candidates?.[0]
const audio = (candidate?.content?.parts ?? []).flatMap((part) =>
part.inlineData === undefined ? [] : [part.inlineData],
)
const next: State = {
...state,
mimeType: state.mimeType ?? audio[0]?.mimeType,
usage: chunk.usageMetadata ?? state.usage,
finishReason: candidate?.finishReason ?? state.finishReason,
modelVersion: chunk.modelVersion ?? state.modelVersion,
responseId: chunk.responseId ?? state.responseId,
}
return [next, audio.flatMap((part) => SpeechStream.delta(next, part.data)[1])] as const
})
const finish = (state: State) => {
const usage = state.usage
const sampleRate = SpeechStream.sampleRate(state.mimeType) ?? DEFAULT_SAMPLE_RATE
return SpeechStream.finish(ADAPTER, state, {
...SpeechStream.pcm("pcm_s16le", sampleRate, state.mimeType ?? `audio/L16;codec=pcm;rate=${sampleRate}`),
usage:
usage === undefined
? undefined
: {
type: "tokens",
input: usage.promptTokenCount,
output: usage.candidatesTokenCount,
total: ProviderShared.totalTokens(
usage.promptTokenCount,
usage.candidatesTokenCount,
usage.totalTokenCount,
),
details: { google: usage },
},
providerMetadata: {
google: { finishReason: state.finishReason, modelVersion: state.modelVersion, responseId: state.responseId },
},
detail: state.finishReason === undefined ? undefined : `finish reason: ${state.finishReason}`,
})
}
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.stream<Request, SpeechEvent, string, State>({
id: ADAPTER,
name: NAME,
unsupported: ["instructions", "speed", "timestamps"],
body: { from: fromRequest },
// `generateContent` answers with one document shaped exactly like a streamed chunk, so it is a single frame.
frames: (bytes, context) =>
context.request.mode === "stream" ? Framing.sse.frame(bytes) : Framing.document.frame(bytes),
initial: () => ({ chunks: [] }),
step,
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
SpeechModel.fromRoute<GoogleSpeechOptions, string, State>(
{ id: ADAPTER, provider: PROVIDER, protocol, baseURL: DEFAULT_BASE_URL, path: ({ request }) => path(request) },
input,
)
export const GoogleSpeech = {
protocol,
model,
} as const
+141
View File
@@ -0,0 +1,141 @@
import { Effect, Schema } from "effect"
import { Framing } from "../route/framing.js"
import { MediaProtocol } from "../route/media-protocol.js"
import { MediaRoute } from "../route/media.js"
import { ProviderID, mergeJsonRecords, type MediaUsage } from "../schema/index.js"
import { SpeechModel, type SpeechEvent, type SpeechRequestFor } from "../speech.js"
import { SpeechStream } from "./utils/speech-stream.js"
const ADAPTER = "openai-speech"
const NAME = "OpenAI Speech"
const PROVIDER = ProviderID.make("openai")
export const DEFAULT_BASE_URL = "https://api.openai.com/v1"
export const PATH = "/audio/speech"
/** `pcm` is raw 24 kHz, 16-bit signed little-endian mono samples without a header. */
const PCM_SAMPLE_RATE = 24000
// ---------------------------------------------------------------------------
// 1. Public model input
// ---------------------------------------------------------------------------
export type OpenAISpeechOptions = Record<string, unknown>
export type Request = SpeechRequestFor<OpenAISpeechOptions>
// ---------------------------------------------------------------------------
// 3. Streaming event schema
// ---------------------------------------------------------------------------
const SpeechStreamEvent = Schema.Union([
Schema.Struct({ type: Schema.Literal("speech.audio.delta"), audio: Schema.Uint8ArrayFromBase64 }),
Schema.Struct({
type: Schema.Literal("speech.audio.done"),
usage: Schema.optional(
Schema.Struct({
input_tokens: Schema.optional(Schema.Number),
output_tokens: Schema.optional(Schema.Number),
total_tokens: Schema.optional(Schema.Number),
}),
),
}),
])
const decodeEvent = MediaProtocol.decodeFrame(ADAPTER, NAME, SpeechStreamEvent)
// ---------------------------------------------------------------------------
// 4. Parser state
// ---------------------------------------------------------------------------
interface State extends SpeechStream.Audio {
readonly done: boolean
readonly usage?: MediaUsage
}
// ---------------------------------------------------------------------------
// 5. Request body construction
// ---------------------------------------------------------------------------
// `sse` is not supported for `tts-1` or `tts-1-hd`; those models stream the raw audio body instead.
const supportsSse = (model: string) => !/^tts-1(-hd)?(-|$)/.test(model)
const fromRequest = Effect.fn("OpenAISpeech.fromRequest")(function* (request: MediaProtocol.Addressed<Request>) {
return MediaProtocol.json(
mergeJsonRecords(
{
model: request.model.id,
input: request.text,
voice: request.voice,
instructions: request.instructions,
response_format: request.format,
speed: request.speed,
stream_format: request.mode === "stream" && supportsSse(request.model.id) ? "sse" : undefined,
},
request.providerOptions,
request.http?.body,
) ?? {},
)
})
// ---------------------------------------------------------------------------
// 6. Stream parsing
// ---------------------------------------------------------------------------
const isSse = (body: MediaProtocol.Body) => body.type === "json" && body.value.stream_format === "sse"
const onEvent = Effect.fn("OpenAISpeech.onEvent")(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
if (event.type === "speech.audio.delta") return SpeechStream.delta(state, event.audio)
const usage = event.usage
return [
{
...state,
done: true,
usage:
usage === undefined
? undefined
: {
type: "tokens" as const,
input: usage.input_tokens,
output: usage.output_tokens,
total: usage.total_tokens,
details: { openai: usage },
},
},
[],
] as const
})
const finish = (state: State, context: MediaProtocol.ResponseContext<Request>) => {
if (isSse(context.body) && !state.done) return Effect.fail(MediaProtocol.incomplete(ADAPTER))
const format = context.request.format ?? "mp3"
return SpeechStream.finish(ADAPTER, state, {
...(format === "pcm" ? SpeechStream.pcm("pcm_s16le", PCM_SAMPLE_RATE) : SpeechStream.container(format)),
usage: state.usage,
})
}
// ---------------------------------------------------------------------------
// 7. Protocol and route
// ---------------------------------------------------------------------------
export const protocol = MediaProtocol.stream<Request, SpeechEvent, string | Uint8Array, State>({
id: ADAPTER,
name: NAME,
unsupported: ["language", "timestamps"],
body: { from: fromRequest },
frames: (bytes, context) => (isSse(context.body) ? Framing.sse.frame(bytes) : bytes),
initial: () => ({ chunks: [], done: false }),
step: SpeechStream.step(onEvent),
finish,
})
export const model = (input: MediaRoute.ModelInput) =>
SpeechModel.fromRoute<OpenAISpeechOptions, string | Uint8Array, State>(
{ id: ADAPTER, provider: PROVIDER, protocol, baseURL: DEFAULT_BASE_URL, path: PATH },
input,
)
export const OpenAISpeech = {
protocol,
model,
} as const
@@ -0,0 +1,118 @@
import { Effect } from "effect"
import { Media } from "../../media.js"
import { MediaProtocol } from "../../route/media-protocol.js"
import type { AIError, MediaUsage, ProviderID, ProviderMetadata } from "../../schema/index.js"
import {
SpeechAudioDeltaEvent,
SpeechFinishEvent,
SpeechTimestampsEvent,
type SpeechEvent,
type SpeechVoice,
} from "../../speech.js"
import { concatBytes } from "../../utils/bytes.js"
import { ProviderShared } from "../shared.js"
export interface Audio {
/** Appended in place: the route creates fresh state for each response through `initial`. */
readonly chunks: Array<Uint8Array>
}
export type StepResult<State> = readonly [State, ReadonlyArray<SpeechEvent>]
/** Empty chunks (keep-alive records) emit nothing. */
export const delta = <State extends Audio>(state: State, chunk: Uint8Array): StepResult<State> => {
if (chunk.length === 0) return [state, []]
state.chunks.push(chunk)
return [state, [SpeechAudioDeltaEvent.make({ chunk })]]
}
export const step =
<State extends Audio>(onRecord: (state: State, frame: string) => Effect.Effect<StepResult<State>, AIError>) =>
(state: State, frame: string | Uint8Array) =>
typeof frame === "string" ? onRecord(state, frame) : Effect.succeed(delta(state, frame))
export const timestamps = (
texts: ReadonlyArray<string>,
starts: ReadonlyArray<number>,
ends: ReadonlyArray<number>,
): ReadonlyArray<SpeechEvent> =>
texts.length === 0
? []
: [
SpeechTimestampsEvent.make({
items: texts.map((text, index) => ({ text, startSeconds: starts[index] ?? 0, endSeconds: ends[index] ?? 0 })),
}),
]
export const voiceID = (voice: SpeechVoice | undefined) => (typeof voice === "object" ? voice.id : voice)
const CONTAINER_MEDIA_TYPES: Readonly<Record<string, string>> = {
mp3: "audio/mpeg",
wav: "audio/wav",
opus: "audio/ogg",
aac: "audio/aac",
flac: "audio/flac",
}
export const container = (format: string, sampleRate?: number) => ({
mediaType: CONTAINER_MEDIA_TYPES[format],
info: { format, sampleRate },
})
const PCM_MEDIA_TYPES = {
pcm_s16le: "audio/pcm",
pcm_f32le: "audio/pcm",
pcm_mulaw: "audio/mulaw",
pcm_alaw: "audio/alaw",
} as const
export type PcmEncoding = keyof typeof PCM_MEDIA_TYPES
export const pcm = (encoding: PcmEncoding, sampleRate: number | undefined, mediaType?: string) => ({
mediaType: mediaType ?? PCM_MEDIA_TYPES[encoding],
info: { format: "pcm", encoding, sampleRate, channels: 1 },
})
export const sampleRate = (mediaType: string | undefined) => {
const rate = /rate=(\d+)/i.exec(mediaType ?? "")?.[1]
return rate === undefined ? undefined : Number(rate)
}
export const unsupportedFormat = (provider: ProviderID, route: string, message: string) =>
ProviderShared.unsupportedOperation({ operation: "media.format", provider, route, message })
/** A declared `mediaType` wins over sniffing: headerless PCM can start with bytes that look like an MPEG frame sync. */
export const finish = (
route: string,
state: Audio,
output: {
readonly mediaType: string | undefined
readonly info?: Media.Info
readonly usage?: MediaUsage
readonly providerMetadata?: ProviderMetadata
readonly detail?: string
},
): Effect.Effect<ReadonlyArray<SpeechEvent>, AIError> => {
if (state.chunks.length === 0)
return Effect.fail(
MediaProtocol.frameError(
route,
`The provider returned no audio${output.detail === undefined ? "" : ` (${output.detail})`}`,
),
)
return Effect.succeed([
SpeechFinishEvent.make({
audio: Media.bytes(concatBytes(state.chunks), output.mediaType, { info: output.info }),
usage: output.usage,
providerMetadata: output.providerMetadata,
}),
])
}
export const headerUsage = (type: "characters" | "credits", value: string | undefined): MediaUsage | undefined => {
const amount = Number(value)
if (!Number.isFinite(amount)) return undefined
return type === "credits" ? { type, credits: amount } : { type, characters: amount }
}
export * as SpeechStream from "./speech-stream.js"
+35
View File
@@ -0,0 +1,35 @@
import { AuthOptions, type ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { CartesiaSpeech, DEFAULT_BASE_URL } from "../protocols/cartesia-speech.js"
export type { CartesiaEncoding, CartesiaSpeechOptions } from "../protocols/cartesia-speech.js"
export const id = ProviderID.make("cartesia")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => AuthOptions.bearer(options, "CARTESIA_API_KEY")
export const configure = (input: Config = {}) => {
const speech = (modelID: string | ModelID) =>
CartesiaSpeech.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
speech,
configure,
}
}
export const provider = configure()
export const speech = provider.speech
+41
View File
@@ -0,0 +1,41 @@
import { Auth } from "../route/auth.js"
import type { ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { DEFAULT_BASE_URL, DeepgramSpeech } from "../protocols/deepgram-speech.js"
export type { DeepgramEncoding, DeepgramSpeechOptions } from "../protocols/deepgram-speech.js"
export const id = ProviderID.make("deepgram")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => {
if ("auth" in options && options.auth) return options.auth
return Auth.optional("apiKey" in options ? options.apiKey : undefined, "apiKey")
.orElse(Auth.config("DEEPGRAM_API_KEY"))
.pipe(Auth.scheme("Token"))
}
export const configure = (input: Config = {}) => {
const speech = (modelID: string | ModelID) =>
DeepgramSpeech.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
speech,
configure,
}
}
export const provider = configure()
export const speech = provider.speech
+41
View File
@@ -0,0 +1,41 @@
import { Auth } from "../route/auth.js"
import type { ProviderAuthOption } from "../route/auth-options.js"
import { HttpOptions, ProviderID, type ModelID } from "../schema/index.js"
import { DEFAULT_BASE_URL, ElevenLabsSpeech } from "../protocols/elevenlabs-speech.js"
export type { ElevenLabsOutputFormat, ElevenLabsSpeechOptions } from "../protocols/elevenlabs-speech.js"
export const id = ProviderID.make("elevenlabs")
const baseURL = DEFAULT_BASE_URL
export type Config = ProviderAuthOption<"optional"> & {
readonly baseURL?: string
readonly headers?: Record<string, string>
readonly http?: HttpOptions.Input
}
const auth = (options: ProviderAuthOption<"optional">) => {
if ("auth" in options && options.auth) return options.auth
return Auth.optional("apiKey" in options ? options.apiKey : undefined, "apiKey")
.orElse(Auth.config("ELEVENLABS_API_KEY"))
.pipe(Auth.header("xi-api-key"))
}
export const configure = (input: Config = {}) => {
const speech = (modelID: string | ModelID) =>
ElevenLabsSpeech.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL ?? baseURL,
headers: input.headers,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
return {
id,
speech,
configure,
}
}
export const provider = configure()
export const speech = provider.speech
+4
View File
@@ -5,9 +5,11 @@ import type { ProviderPackage } from "../provider-package.js"
import { HttpOptions, ProviderID, mergeHttpOptions, type ModelID } from "../schema/index.js"
import { Gemini } from "../protocols/gemini.js"
import { GoogleImages } from "../protocols/google-images.js"
import { GoogleSpeech } from "../protocols/google-speech.js"
import { GoogleVideo } from "../protocols/google-video.js"
export type { GoogleImageOptions } from "../protocols/google-images.js"
export type { GoogleSpeechOptions } from "../protocols/google-speech.js"
export type { GoogleVideoOptions } from "../protocols/google-video.js"
export type GeminiOptionsInput = Gemini.OptionsInput
export type GeminiProviderOptionsInput = Gemini.ProviderOptionsInput
@@ -54,6 +56,7 @@ export const configure = (input: Config = {}) => {
model: (modelID: string | ModelID) => route.model<Gemini.ProviderOptionsInput>({ id: modelID }),
image: (modelID: string | ModelID) => GoogleImages.model(media(modelID)),
video: (modelID: string | ModelID) => GoogleVideo.model(media(modelID)),
speech: (modelID: string | ModelID) => GoogleSpeech.model(media(modelID)),
configure,
}
}
@@ -73,3 +76,4 @@ export const model: ProviderPackage.Definition<Settings, Gemini.ProviderOptionsI
export const image = provider.image
export const video = provider.video
export const speech = provider.speech
+3
View File
@@ -5,11 +5,14 @@ export * as AmazonBedrock from "./amazon-bedrock.js"
export * as AmazonBedrockMantle from "./amazon-bedrock-mantle.js"
export * as Azure from "./azure.js"
export * as Baseten from "./baseten.js"
export * as Cartesia from "./cartesia.js"
export * as Cerebras from "./cerebras.js"
export * as CloudflareAIGateway from "./cloudflare-ai-gateway.js"
export * as CloudflareWorkersAI from "./cloudflare-workers-ai.js"
export * as DeepInfra from "./deepinfra.js"
export * as Deepgram from "./deepgram.js"
export * as DeepSeek from "./deepseek.js"
export * as ElevenLabs from "./elevenlabs.js"
export * as Fal from "./fal.js"
export * as Fireworks from "./fireworks.js"
export * as Google from "./google.js"
+16 -11
View File
@@ -6,9 +6,11 @@ import * as OpenAIChat from "../protocols/openai-chat.js"
import * as OpenAIResponses from "../protocols/openai-responses.js"
import { withOpenAIOptions, type OpenAIProviderOptionsInput } from "./openai-options.js"
import { OpenAIImages, type OpenAIImageString } from "../protocols/openai-images.js"
import { OpenAISpeech } from "../protocols/openai-speech.js"
export type { OpenAIOptionsInput, OpenAIResponseIncludable } from "./openai-options.js"
export type { OpenAIImageOptions } from "../protocols/openai-images.js"
export type { OpenAISpeechOptions } from "../protocols/openai-speech.js"
export const id = ProviderID.make("openai")
@@ -95,17 +97,18 @@ export const configure = (input: Config = {}) => {
id,
compatibility: { supportsPromptCacheKey: true },
})
const image = (modelID: string | ModelID) =>
OpenAIImages.model({
id: modelID,
auth: auth(input),
baseURL: input.baseURL,
headers: input.headers,
http: mergeHttpOptions(
input.http === undefined ? undefined : HttpOptions.make(input.http),
input.queryParams === undefined ? undefined : new HttpOptions({ query: input.queryParams }),
),
})
const media = (modelID: string | ModelID) => ({
id: modelID,
auth: auth(input),
baseURL: input.baseURL,
headers: input.headers,
http: mergeHttpOptions(
input.http === undefined ? undefined : HttpOptions.make(input.http),
input.queryParams === undefined ? undefined : new HttpOptions({ query: input.queryParams }),
),
})
const image = (modelID: string | ModelID) => OpenAIImages.model(media(modelID))
const speech = (modelID: string | ModelID) => OpenAISpeech.model(media(modelID))
return {
id,
@@ -113,6 +116,7 @@ export const configure = (input: Config = {}) => {
responses,
chat,
image,
speech,
configure,
}
}
@@ -159,3 +163,4 @@ export const chatModel: ProviderPackage.Definition<Settings, OpenAIProviderOptio
export const responses = provider.responses
export const chat = provider.chat
export const image = provider.image
export const speech = provider.speech
+18 -1
View File
@@ -1,4 +1,4 @@
import type { Stream } from "effect"
import { Stream } from "effect"
import * as ProviderShared from "../protocols/shared.js"
import type { AIError } from "../schema/index.js"
@@ -12,6 +12,8 @@ import type { AIError } from "../schema/index.js"
* `[DONE]`; protocols that use it as a terminal select `sseWithDone`.
* - AWS event stream — length-prefixed binary frames with CRC checksums.
* Each emitted frame is one parsed binary event record.
* - Media streams — newline-delimited JSON (`lines`) or the whole body as one
* frame (`document`); chunked binary bodies need no framing.
*
* The frame type is opaque to this layer; the protocol's event schema decodes
* each frame before its state machine handles it.
@@ -38,4 +40,19 @@ export const sseEvents = (events: ReadonlySet<string>): Definition<string> => ({
frame: (bytes) => ProviderShared.sseFraming(bytes, events),
})
export const lines: Definition<string> = {
id: "lines",
frame: (bytes) =>
bytes.pipe(
Stream.decodeText(),
Stream.splitLines,
Stream.filter((line) => line.trim().length > 0),
),
}
export const document: Definition<string> = {
id: "document",
frame: (bytes) => Stream.fromEffect(Stream.mkString(bytes.pipe(Stream.decodeText()))),
}
export * as Framing from "./framing.js"
+65 -3
View File
@@ -1,4 +1,4 @@
import { Clock, Duration, Effect, Schema } from "effect"
import { Clock, Duration, Effect, Schema, type Stream } from "effect"
import { HttpClientResponse } from "effect/unstable/http"
import type { Snapshot, Status } from "../generation.js"
import { Media } from "../media.js"
@@ -16,11 +16,16 @@ import {
// Bodies
// ---------------------------------------------------------------------------
/** JSON `query` is appended to the endpoint URL before the route and caller `http.query` overlays. */
export type Body =
| { readonly type: "json"; readonly value: Record<string, unknown> }
| { readonly type: "json"; readonly value: Record<string, unknown>; readonly query?: Record<string, string> }
| { readonly type: "multipart"; readonly value: FormData }
export const json = (value: Record<string, unknown>): Body => ({ type: "json", value })
export const json = (value: Record<string, unknown>, query?: Record<string, string>): Body => ({
type: "json",
value,
query,
})
export const multipart = (value: FormData): Body => ({ type: "multipart", value })
// ---------------------------------------------------------------------------
@@ -119,6 +124,42 @@ export const queued = <Request, Response, Token>(
...input,
})
export type Mode = "generate" | "stream"
export type Addressed<Request> = Request & { readonly mode: Mode }
export interface ResponseContext<Request> extends DecodeContext<Addressed<Request>> {
readonly http: HttpContext
}
/**
* One request whose body is parsed incrementally, like LLM protocols: `frames` → `step`* → `finish`. `generate` and
* `stream` share this state machine; `request.mode` lets a protocol pick a different body, path, or framing.
*/
export interface Streamed<Request, Event, Frame, State> {
readonly kind: "stream"
readonly id: string
readonly name: string
/** Common request fields this protocol cannot lower; the route rejects them before `body.from` runs. */
readonly unsupported?: ReadonlyArray<keyof Request & string>
readonly body: { readonly from: (request: Addressed<Request>) => Effect.Effect<Body, AIError> }
readonly frames: (
bytes: Stream.Stream<Uint8Array, AIError>,
context: DecodeContext<Addressed<Request>>,
) => Stream.Stream<Frame, AIError>
readonly initial: () => State
readonly step: (state: State, frame: Frame) => Effect.Effect<readonly [State, ReadonlyArray<Event>], AIError>
/** Emit exactly one terminal event, or fail when the provider stopped before completing. */
readonly finish: (state: State, context: ResponseContext<Request>) => Effect.Effect<ReadonlyArray<Event>, AIError>
}
export const stream = <Request, Event, Frame, State>(
input: Omit<Streamed<Request, Event, Frame, State>, "kind">,
): Streamed<Request, Event, Frame, State> => ({
kind: "stream",
...input,
})
// ---------------------------------------------------------------------------
// Response helpers
// ---------------------------------------------------------------------------
@@ -203,6 +244,27 @@ export const status = <Table extends Record<string, Status>>(
return Effect.succeed(normalized)
}
export const frameError = (route: string, message: string, body?: string, cause?: unknown) =>
new AIError({ reason: new InvalidProviderOutputError({ route, message, body, cause }) })
export const incomplete = (route: string) =>
new AIError({
reason: new InvalidProviderOutputError({
route,
message: "The provider response ended unexpectedly.",
classification: "incomplete-stream",
}),
})
/** Schema-decode one JSON stream frame. Decode failures keep the frame as `reason.body`. */
export const decodeFrame = <A>(route: string, name: string, schema: Schema.Codec<A, unknown>) => {
const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(schema))
return (frame: string) =>
decode(frame).pipe(
Effect.mapError((cause) => frameError(route, `${name} sent an invalid stream event`, frame, cause)),
)
}
/** A `url` asset whose provider-declared retention window starts now. */
export const expiringUrl = (url: string, retention: Duration.Duration, options?: Parameters<typeof Media.url>[1]) =>
Clock.currentTimeMillis.pipe(
+93 -7
View File
@@ -1,12 +1,20 @@
import { Effect, Schema } from "effect"
import { Effect, Schema, Stream } from "effect"
import { Headers, HttpClientRequest, type HttpClientResponse } from "effect/unstable/http"
import { Auth, type AuthInput } from "./auth.js"
import { Endpoint } from "./endpoint.js"
import type { Interface } from "./executor-service.js"
import { RequestExecutor } from "./executor.js"
import { MediaProtocol } from "./media-protocol.js"
import { Generation, type Route as GenerationRoute } from "../generation.js"
import { ProviderShared } from "../protocols/shared.js"
import { AIError, HttpOptions, InvalidRequestError, ProviderID, mergeHttpOptions } from "../schema/index.js"
import {
AIError,
AIErrorReason,
HttpOptions,
InvalidRequestError,
ProviderID,
mergeHttpOptions,
} from "../schema/index.js"
import { sanitizeSurrogates } from "../utils/sanitize.js"
export type Execute = Interface["execute"]
@@ -52,6 +60,15 @@ export interface QueuedRoute<Request extends MediaRequest, Response> {
) => Effect.Effect<Generation<Response>, AIError>
}
/** One request whose response parses into events; `generate` runs the same stream and collects it. */
export interface StreamRoute<Request extends MediaRequest, Event, Response> {
readonly id: string
readonly provider: ProviderID
readonly protocol: string
readonly stream: (request: Request, execute: Execute) => Stream.Stream<Event, AIError>
readonly generate: (request: Request, execute: Execute) => Effect.Effect<Response, AIError>
}
export interface Composition<Request extends MediaRequest> {
readonly id: string
readonly provider: string | ProviderID
@@ -69,6 +86,12 @@ export interface QueuedInput<Request extends MediaRequest, Response, Token> exte
readonly protocol: MediaProtocol.Queued<Request, Response, Token>
}
export interface StreamInput<Request extends MediaRequest, Event, Response, Frame, State>
extends Composition<MediaProtocol.Addressed<Request>> {
readonly protocol: MediaProtocol.Streamed<Request, Event, Frame, State>
readonly collect: (events: ReadonlyArray<Event>) => Effect.Effect<Response, AIError>
}
/**
* Compose an inline media protocol with an endpoint and auth into a runnable route. The route owns the transport
* plumbing every media protocol would otherwise duplicate: option merging, surrogate sanitizing, unsupported-field
@@ -165,16 +188,73 @@ export const queued = <Request extends MediaRequest, Response, Token>(
return { id: input.id, provider: transport.provider, protocol: protocol.id, start, resume }
}
/** Compose a streaming media protocol; `generate` runs the same stream in `generate` mode and folds it with `collect`. */
export const stream = <Request extends MediaRequest, Event, Response, Frame, State>(
input: StreamInput<Request, Event, Response, Frame, State>,
): StreamRoute<Request, Event, Response> => {
const transport = makeTransport(input)
const protocol = input.protocol
const events = (request: Request, execute: Execute, mode: MediaProtocol.Mode) =>
Stream.unwrap(
Effect.gen(function* () {
const submitted = yield* transport.submit(
{ ...request, mode },
{ unsupported: protocol.unsupported, from: protocol.body.from },
execute,
)
const http = RequestExecutor.responseHttp(submitted.response)
return Stream.suspend(() => {
// Parser state is local to one response, exactly like `Route.make`'s LLM stream loop.
let state = protocol.initial()
return protocol.frames(RequestExecutor.responseStream(submitted.response), submitted.context).pipe(
Stream.mapEffect((frame) =>
protocol.step(state, frame).pipe(
Effect.map(([next, output]) => {
state = next
return output
}),
),
),
Stream.flattenIterable,
Stream.concat(
Stream.suspend(() => Stream.fromIterableEffect(protocol.finish(state, { ...submitted.context, http }))),
),
Stream.mapError((error) =>
error.reason.http !== undefined
? error
: new AIError({
reason: AIErrorReason.make({
...error.reason,
message: error.reason.message,
cause: error.reason.cause,
http,
}),
}),
),
)
})
}),
)
return {
id: input.id,
provider: transport.provider,
protocol: protocol.id,
stream: (request, execute) => events(request, execute, "stream"),
generate: (request, execute) =>
events(request, execute, "generate").pipe(Stream.runCollect, Effect.flatMap(input.collect)),
}
}
// ---------------------------------------------------------------------------
// Transport plumbing shared by both kinds
// Transport plumbing shared by every kind
// ---------------------------------------------------------------------------
const makeTransport = <Request extends MediaRequest>(input: Composition<Request>) => {
const provider = ProviderID.make(input.provider)
const routeHttp = input.headers === undefined ? undefined : new HttpOptions({ headers: input.headers })
const authorize = Auth.toEffect(input.auth)
const withQuery = (url: URL, http: HttpOptions | undefined) => {
for (const [key, value] of Object.entries(http?.query ?? {})) url.searchParams.set(key, value)
const withQuery = (url: URL, query: Record<string, string> | undefined) => {
for (const [key, value] of Object.entries(query ?? {})) url.searchParams.set(key, value)
return url
}
return {
@@ -195,7 +275,13 @@ const makeTransport = <Request extends MediaRequest>(input: Composition<Request>
// Sanitize after merging so model-level overlays are covered; the model value is restored, not sanitized.
const resolved: Request = { ...sanitizeSurrogates({ ...request, http }), model: request.model }
const body = yield* protocol.from(resolved)
const url = withQuery(Endpoint.render(input.endpoint, { request: resolved, body }), http)
const url = withQuery(
withQuery(
Endpoint.render(input.endpoint, { request: resolved, body }),
body.type === "json" ? body.query : undefined,
),
http?.query,
)
const encoded = body.type === "json" ? ProviderShared.encodeJson(body.value) : "[multipart/form-data]"
const baseHeaders = Headers.fromInput(http?.headers)
const headers = yield* authorize({
@@ -230,7 +316,7 @@ const makeTransport = <Request extends MediaRequest>(input: Composition<Request>
/^https?:\/\//.test(path)
? new URL(path)
: new URL(`${ProviderShared.trimBaseUrl(input.endpoint.baseURL ?? "")}${path}`),
http,
http?.query,
)
for (const [key, value] of Object.entries(input.endpoint.query ?? {})) url.searchParams.set(key, value)
const base = Headers.fromInput(http?.headers)
+51
View File
@@ -0,0 +1,51 @@
import { Context, Effect, Layer, Stream } from "effect"
import { RequestExecutor } from "./route/executor.js"
import type { AIError } from "./schema/index.js"
import type { SpeechEvent, SpeechOptions, SpeechRequestFor, SpeechResponse } from "./speech.js"
export interface Interface {
readonly generate: <Options extends SpeechOptions>(
request: SpeechRequestFor<Options>,
) => Effect.Effect<SpeechResponse, AIError>
readonly stream: <Options extends SpeechOptions>(
request: SpeechRequestFor<Options>,
) => Stream.Stream<SpeechEvent, AIError>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SpeechClient") {}
export const generate = <Options extends SpeechOptions>(
request: SpeechRequestFor<Options>,
): Effect.Effect<SpeechResponse, AIError, Service> =>
Effect.gen(function* () {
const client = yield* Service
return yield* client.generate(request)
})
export const stream = <Options extends SpeechOptions>(
request: SpeechRequestFor<Options>,
): Stream.Stream<SpeechEvent, AIError, Service> =>
Stream.unwrap(
Effect.gen(function* () {
const client = yield* Service
return client.stream(request)
}),
)
export const layer: Layer.Layer<Service, never, RequestExecutor.Service> = Layer.effect(
Service,
Effect.gen(function* () {
const executor = yield* RequestExecutor.Service
return Service.of({
generate: (request) => request.model.route.generate(request, executor.execute),
stream: (request) => request.model.route.stream(request, executor.execute),
})
}),
)
export const SpeechClient = {
Service,
layer,
generate,
stream,
} as const
+217
View File
@@ -0,0 +1,217 @@
import { Effect, Schema, Stream } from "effect"
import { Media } from "./media.js"
import { MediaModel, composeRoute, tryRequest } from "./media-model.js"
import { MediaRoute } from "./route/media.js"
import type { MediaProtocol } from "./route/media-protocol.js"
import { AIError, HttpOptions, MediaUsage, ProviderMetadata } from "./schema/index.js"
import { SpeechClient, Service } from "./speech-client.js"
// ---------------------------------------------------------------------------
// Model
// ---------------------------------------------------------------------------
export type SpeechOptions = Record<string, unknown>
export type SpeechRoute<Options extends SpeechOptions = SpeechOptions> = MediaRoute.StreamRoute<
SpeechRequestFor<Options>,
SpeechEvent,
SpeechResponse
>
export class SpeechModel<Options extends SpeechOptions = SpeechOptions> extends MediaModel<
SpeechRoute<Options>,
Options
> {
declare protected readonly _SpeechModel: void
static make<Options extends SpeechOptions = SpeechOptions>(input: MediaModel.Input<SpeechRoute<Options>>) {
return new SpeechModel<Options>(input)
}
/** Compose a streaming speech protocol with its canonical path into a model for one deployment. */
static fromRoute<Options extends SpeechOptions = SpeechOptions, Frame = unknown, State = unknown>(
route: SpeechModel.RouteInput<Options, Frame, State>,
input: MediaRoute.ModelInput,
) {
return new SpeechModel<Options>({
id: input.id,
provider: route.provider,
http: input.http,
route: composeRoute(
(composition) => MediaRoute.stream({ ...composition, collect: collectResponse }),
route,
input,
),
})
}
}
export namespace SpeechModel {
export type RouteInput<
Options extends SpeechOptions = SpeechOptions,
Frame = unknown,
State = unknown,
> = MediaModel.RouteInput<
MediaProtocol.Addressed<SpeechRequestFor<Options>>,
MediaProtocol.Streamed<SpeechRequestFor<Options>, SpeechEvent, Frame, State>
>
}
export const SpeechModelSchema = Schema.declare((value): value is SpeechModel => value instanceof SpeechModel, {
expected: "Speech.Model",
})
// ---------------------------------------------------------------------------
// Request
// ---------------------------------------------------------------------------
/** Provider-native: a name on OpenAI and Gemini, a voice id on ElevenLabs and Cartesia; `{ id }` is an OpenAI custom voice. */
export const SpeechVoice = Schema.Union([Schema.String, Schema.Struct({ id: Schema.String })]).annotate({
identifier: "Speech.Voice",
})
export type SpeechVoice = Schema.Schema.Type<typeof SpeechVoice>
export type SpeechFormat = "mp3" | "wav" | "pcm" | "opus" | "aac" | "flac" | (string & {})
/** Granularity is provider-native: characters on ElevenLabs, words on Cartesia. */
export const SpeechTimestamp = Schema.Struct({
text: Schema.String,
startSeconds: Schema.Number,
endSeconds: Schema.Number,
}).annotate({ identifier: "Speech.Timestamp" })
export type SpeechTimestamp = Schema.Schema.Type<typeof SpeechTimestamp>
export class SpeechRequest extends Schema.Class<SpeechRequest>("Speech.Request")({
model: SpeechModelSchema,
text: Schema.String,
voice: Schema.optional(SpeechVoice),
format: Schema.optional(Schema.String),
speed: Schema.optional(Schema.Number),
language: Schema.optional(Schema.String),
instructions: Schema.optional(Schema.String),
timestamps: Schema.optional(Schema.Boolean),
providerOptions: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
http: Schema.optional(HttpOptions),
}) {
declare protected readonly _SpeechRequest: void
}
export type SpeechRequestFor<Options extends SpeechOptions = SpeechOptions> = Omit<
SpeechRequest,
"model" | "providerOptions"
> & {
readonly model: SpeechModel<Options>
readonly providerOptions?: Options
}
export type SpeechModelOptions<Model> = Model extends SpeechModel<infer Options> ? Options : never
export type SpeechRequestInput<Model extends SpeechModel = SpeechModel> = Omit<
ConstructorParameters<typeof SpeechRequest>[0],
"model" | "providerOptions" | "http" | "format"
> & {
readonly model: Model
readonly format?: SpeechFormat
readonly providerOptions?: NoInfer<SpeechModelOptions<Model>>
readonly http?: HttpOptions.Input
}
// ---------------------------------------------------------------------------
// Response and events
// ---------------------------------------------------------------------------
export class SpeechResponse extends Schema.Class<SpeechResponse>("Speech.Response")({
/** The complete audio. Headerless PCM carries `info.encoding`, `info.sampleRate`, and `info.channels`. */
audio: Media.AssetSchema,
timestamps: Schema.optional(Schema.Array(SpeechTimestamp)),
usage: Schema.optional(MediaUsage),
notices: Schema.optional(Schema.Array(Media.Notice)),
providerMetadata: Schema.optional(ProviderMetadata),
}) {}
export const SpeechAudioDeltaEvent = Schema.Struct({
type: Schema.tag("audio-delta"),
chunk: Schema.Uint8Array,
}).annotate({ identifier: "Speech.Event.AudioDelta" })
export const SpeechTimestampsEvent = Schema.Struct({
type: Schema.tag("timestamps"),
items: Schema.Array(SpeechTimestamp),
}).annotate({ identifier: "Speech.Event.Timestamps" })
/** `audio` is every `audio-delta` chunk concatenated, so the route holds the whole clip in memory until `finish`. */
export const SpeechFinishEvent = Schema.Struct({
type: Schema.tag("finish"),
audio: Media.AssetSchema,
usage: Schema.optional(MediaUsage),
notices: Schema.optional(Schema.Array(Media.Notice)),
providerMetadata: Schema.optional(ProviderMetadata),
}).annotate({ identifier: "Speech.Event.Finish" })
const speechEventTagged = Schema.Union([SpeechAudioDeltaEvent, SpeechTimestampsEvent, SpeechFinishEvent]).pipe(
Schema.toTaggedUnion("type"),
)
export const SpeechEvent = Object.assign(speechEventTagged, {
is: {
audioDelta: speechEventTagged.guards["audio-delta"],
timestamps: speechEventTagged.guards.timestamps,
finish: speechEventTagged.guards.finish,
},
})
export type SpeechEvent = Schema.Schema.Type<typeof speechEventTagged>
const collectResponse = (events: ReadonlyArray<SpeechEvent>): Effect.Effect<SpeechResponse> => {
const finish = events.find(SpeechEvent.is.finish)
// Every speech protocol's `finish` emits the terminal event or fails, so a completed stream always has one.
if (finish === undefined) return Effect.die(new Error("The speech stream completed without a finish event"))
const timestamps = events.filter(SpeechEvent.is.timestamps).flatMap((event) => event.items)
return Effect.succeed(
new SpeechResponse({
audio: finish.audio,
timestamps: timestamps.length === 0 ? undefined : timestamps,
usage: finish.usage,
notices: finish.notices,
providerMetadata: finish.providerMetadata,
}),
)
}
// ---------------------------------------------------------------------------
// Request-shaped call API
// ---------------------------------------------------------------------------
export function request<const Model extends SpeechModel>(
input: SpeechRequestInput<Model>,
): SpeechRequestFor<SpeechModelOptions<Model>>
export function request(input: SpeechRequest): SpeechRequest
export function request(input: SpeechRequest | SpeechRequestInput) {
if (input instanceof SpeechRequest) return input
return new SpeechRequest({
...input,
http: input.http === undefined ? undefined : HttpOptions.make(input.http),
})
}
const requestEffect = (input: SpeechRequest | SpeechRequestInput) => tryRequest(() => request(input))
export function generate<const Model extends SpeechModel>(
input: SpeechRequestInput<Model>,
): Effect.Effect<SpeechResponse, AIError, Service>
export function generate(input: SpeechRequest): Effect.Effect<SpeechResponse, AIError, Service>
export function generate(input: SpeechRequest | SpeechRequestInput) {
return requestEffect(input).pipe(Effect.flatMap((request) => SpeechClient.generate(request)))
}
export function stream<const Model extends SpeechModel>(
input: SpeechRequestInput<Model>,
): Stream.Stream<SpeechEvent, AIError, Service>
export function stream(input: SpeechRequest): Stream.Stream<SpeechEvent, AIError, Service>
export function stream(input: SpeechRequest | SpeechRequestInput) {
return Stream.unwrap(requestEffect(input).pipe(Effect.map((request) => SpeechClient.stream(request))))
}
export const Speech = {
request,
generate,
stream,
} as const
+9
View File
@@ -0,0 +1,9 @@
export const concatBytes = (chunks: ReadonlyArray<Uint8Array>) => {
if (chunks.length === 1) return chunks[0]
const bytes = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0))
chunks.reduce((offset, chunk) => {
bytes.set(chunk, offset)
return offset + chunk.length
}, 0)
return bytes
}
+7 -2
View File
@@ -19,9 +19,12 @@ export const detectMediaType = (bytes: Uint8Array): string | undefined => {
if (startsWith(bytes, [0x25, 0x50, 0x44, 0x46])) return "application/pdf"
if (bytes.length >= 12 && ascii(bytes, 4, 8) === "ftyp") return "video/mp4"
if (startsWith(bytes, [0x1a, 0x45, 0xdf, 0xa3])) return "video/webm"
if (startsWith(bytes, [0x49, 0x44, 0x33]) || startsWith(bytes, [0xff, 0xfb]) || startsWith(bytes, [0xff, 0xf3]))
return "audio/mpeg"
if (startsWith(bytes, [0x49, 0x44, 0x33])) return "audio/mpeg"
// An 11-bit frame sync; layer bits `00` mark AAC ADTS, any other layer is MPEG audio.
if (bytes.length >= 2 && bytes[0] === 0xff && (bytes[1] & 0xe0) === 0xe0)
return (bytes[1] & 0x06) === 0 ? "audio/aac" : "audio/mpeg"
if (startsWith(bytes, [0x4f, 0x67, 0x67, 0x53])) return "audio/ogg"
if (startsWith(bytes, [0x66, 0x4c, 0x61, 0x43])) return "audio/flac"
return undefined
}
@@ -37,6 +40,8 @@ const EXTENSIONS: Readonly<Record<string, string>> = {
mp3: "audio/mpeg",
wav: "audio/wav",
ogg: "audio/ogg",
flac: "audio/flac",
aac: "audio/aac",
txt: "text/plain",
md: "text/markdown",
csv: "text/csv",
+18
View File
@@ -8,6 +8,9 @@ import {
LLMClient,
Media,
Provider,
Speech,
SpeechClient,
SpeechEvent,
Video,
VideoClient,
} from "@opencode/ai"
@@ -15,11 +18,15 @@ import { Route, Protocol, WebSocketTransport } from "@opencode/ai/route"
import { Provider as ProviderSubpath } from "@opencode/ai/provider"
import {
Baseten,
Cartesia,
CloudflareAIGateway,
CloudflareWorkersAI,
Deepgram,
DeepSeek,
ElevenLabs,
Fal,
Fireworks,
Google,
OpenCodeZen,
OpenAI,
OpenAICompatible,
@@ -53,6 +60,10 @@ describe("public exports", () => {
expect(Video.start).toBeFunction()
expect(Video.resume).toBeFunction()
expect(VideoClient.layer).toBeDefined()
expect(Speech.generate).toBeFunction()
expect(Speech.stream).toBeFunction()
expect(SpeechClient.layer).toBeDefined()
expect(SpeechEvent.is.audioDelta).toBeFunction()
expect(Generation).toBeFunction()
expect(Provider.make).toBeFunction()
expect(ProviderSubpath.make).toBe(Provider.make)
@@ -104,6 +115,13 @@ describe("public exports", () => {
expect(Fal.configure({ apiKey: "fixture" }).video("fal-ai/veo3.1").route.id).toBe("fal-video")
expect(Runway.configure({ apiKey: "fixture" }).video("gen4.5").route.id).toBe("runway-video")
expect(Runway.provider.video).toBe(Runway.video)
expect(OpenAI.configure({ apiKey: "fixture" }).speech("gpt-4o-mini-tts").route.id).toBe("openai-speech")
expect(Google.configure({ apiKey: "fixture" }).speech("gemini-2.5-flash-preview-tts").route.id).toBe(
"google-speech",
)
expect(ElevenLabs.configure({ apiKey: "fixture" }).speech("eleven_flash_v2_5").route.id).toBe("elevenlabs-speech")
expect(Cartesia.configure({ apiKey: "fixture" }).speech("sonic-3").route.id).toBe("cartesia-speech")
expect(Deepgram.configure({ apiKey: "fixture" }).speech("aura-2-thalia-en").route.id).toBe("deepgram-speech")
})
test("protocol barrels expose supported low-level routes", () => {
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+7
View File
@@ -63,6 +63,13 @@ describe("Media", () => {
expect(Media.detectMediaType(WEBP)).toBe("image/webp")
expect(Media.detectMediaType(PDF)).toBe("application/pdf")
expect(Media.detectMediaType(Uint8Array.from([1, 2, 3]))).toBeUndefined()
expect(Media.detectMediaType(new TextEncoder().encode("ID3\x04"))).toBe("audio/mpeg")
expect(Media.detectMediaType(Uint8Array.from([0xff, 0xfb, 0x90, 0x64]))).toBe("audio/mpeg")
expect(Media.detectMediaType(Uint8Array.from([0xff, 0xf3, 0x90, 0x64]))).toBe("audio/mpeg")
expect(Media.detectMediaType(Uint8Array.from([0xff, 0xf1, 0x50, 0x80]))).toBe("audio/aac")
expect(Media.detectMediaType(new TextEncoder().encode("RIFF\0\0\0\0WAVEfmt "))).toBe("audio/wav")
expect(Media.detectMediaType(new TextEncoder().encode("OggS\0\x02"))).toBe("audio/ogg")
expect(Media.detectMediaType(new TextEncoder().encode("fLaC\0\0\0\x22"))).toBe("audio/flac")
expect(Media.bytes(PNG).mediaType).toBe("image/png")
expect(Media.bytes(Uint8Array.from([1, 2, 3])).mediaType).toBe("application/octet-stream")
expect(Media.bytes(Uint8Array.from([1, 2, 3]), "image/x-custom").mediaType).toBe("image/x-custom")
+30 -3
View File
@@ -1,7 +1,7 @@
import { describe, expect, test } from "bun:test"
import { Effect, Layer } from "effect"
import { HttpClientRequest } from "effect/unstable/http"
import { AIError, LLMEvent, Media } from "../src/index.js"
import { AIError, LLMEvent, Media, SpeechEvent } from "../src/index.js"
import { RequestExecutor } from "../src/route.js"
import { AI } from "../src/promise.js"
import { OpenAI, Runway } from "../src/providers.js"
@@ -17,8 +17,8 @@ const chatBody = sseEvents(
)
/**
* Executor layer that answers chat completions with SSE text, image generations with one base64 PNG, and Runway video
* tasks with a queued submission that succeeds on the second poll.
* Executor layer that answers chat completions with SSE text, image generations with one base64 PNG, Runway video
* tasks with a queued submission that succeeds on the second poll, and speech with raw audio or SSE audio deltas.
*/
const executor = (seen: Array<string>) =>
RequestExecutor.layer.pipe(
@@ -33,6 +33,17 @@ const executor = (seen: Array<string>) =>
})
if (web.url.endsWith("/chat/completions"))
return input.respond(chatBody, { headers: { "content-type": "text/event-stream" } })
if (web.url.endsWith("/audio/speech"))
return JSON.parse(input.text).stream_format === "sse"
? input.respond(
sseEvents(
{ type: "speech.audio.delta", audio: "AQI=" },
{ type: "speech.audio.delta", audio: "Aw==" },
{ type: "speech.audio.done", usage: { input_tokens: 4, output_tokens: 8, total_tokens: 12 } },
),
{ headers: { "content-type": "text/event-stream" } },
)
: input.respond(Uint8Array.from([1, 2, 3]), { headers: { "content-type": "audio/pcm" } })
if (web.url.endsWith("/text_to_video"))
return input.respond(JSON.stringify({ id: "task_1" }), { headers: { "content-type": "application/json" } })
if (web.url.endsWith("/tasks/task_1")) {
@@ -123,6 +134,21 @@ describe("AI promise client", () => {
await ai.dispose()
})
test("generates and streams speech over the same runtime", async () => {
const ai = AI.make({ layer: executor([]) })
const model = openai.speech("gpt-4o-mini-tts")
const response = await ai.speech.generate({ model, text: "Hello", voice: "coral", format: "pcm" })
expect(await ai.run(response.audio.bytes())).toEqual(Uint8Array.from([1, 2, 3]))
const events: Array<SpeechEvent> = []
for await (const event of ai.speech.stream({ model, text: "Hello", voice: "coral" })) events.push(event)
expect(events.map((event) => event.type)).toEqual(["audio-delta", "audio-delta", "finish"])
const finish = events.find(SpeechEvent.is.finish)
expect(await ai.run(finish!.audio.bytes())).toEqual(Uint8Array.from([1, 2, 3]))
await ai.dispose()
})
test("rethrows AIError unchanged and honors abort signals", async () => {
const ai = AI.make({ layer: executor([]) })
@@ -148,6 +174,7 @@ describe("AI promise client", () => {
expect(typeof AI.ai.llm.generate).toBe("function")
expect(typeof AI.ai.image.generate).toBe("function")
expect(typeof AI.ai.video.start).toBe("function")
expect(typeof AI.ai.speech.generate).toBe("function")
await AI.ai.dispose()
})
})
@@ -0,0 +1,47 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Speech, SpeechEvent } from "../../src/index.js"
import { Cartesia } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { TEXT, collectSpeech } from "./speech-recording.js"
const model = Cartesia.configure({
apiKey: process.env.CARTESIA_API_KEY ?? "fixture",
}).speech("sonic-3")
// "Katie", the voice from Cartesia's quickstart.
const voice = "f786b574-daa5-4673-aa0c-cbe3e8534c02"
const recorded = recordedTests({
prefix: "cartesia-speech",
provider: "cartesia",
protocol: "cartesia-speech",
requires: ["CARTESIA_API_KEY"],
options: { redact: { allowRequestHeaders: ["cartesia-version"] } },
})
describe("Cartesia Speech recorded", () => {
recorded.effect("generates speech", () =>
Effect.gen(function* () {
const response = yield* Speech.generate({ model, text: TEXT, voice, format: "mp3" })
expect(response.audio.mediaType).toBe("audio/mpeg")
expect(response.audio.info).toMatchObject({ format: "mp3", sampleRate: 44100 })
expect((yield* response.audio.bytes()).length).toBeGreaterThan(0)
}),
)
recorded.effect("streams speech with timestamps", () =>
Effect.gen(function* () {
const { events, finish } = yield* collectSpeech(
Speech.stream({ model, text: TEXT, voice, timestamps: true, providerOptions: { sampleRate: 24000 } }),
)
const words = events.filter(SpeechEvent.is.timestamps).flatMap((event) => event.items)
expect(finish.audio.mediaType).toBe("audio/pcm")
expect(finish.audio.info).toMatchObject({ encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
expect(words.length).toBeGreaterThan(0)
expect(words.every((item) => item.endSeconds >= item.startSeconds)).toBe(true)
}),
)
})
@@ -0,0 +1,41 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Speech } from "../../src/index.js"
import { Deepgram } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { TEXT, collectSpeech } from "./speech-recording.js"
const model = Deepgram.configure({
apiKey: process.env.DEEPGRAM_API_KEY ?? "fixture",
}).speech("aura-2-thalia-en")
const recorded = recordedTests({
prefix: "deepgram-speech",
provider: "deepgram",
protocol: "deepgram-speech",
requires: ["DEEPGRAM_API_KEY"],
// Usage and the request id arrive only as response headers.
options: { redact: { allowResponseHeaders: ["dg-char-count", "dg-request-id", "dg-model-name"] } },
})
describe("Deepgram Speech recorded", () => {
recorded.effect("generates speech", () =>
Effect.gen(function* () {
const response = yield* Speech.generate({ model, text: TEXT })
expect(response.audio.mediaType).toBe("audio/mpeg")
expect(response.audio.info).toEqual({ format: "mp3" })
expect((yield* response.audio.bytes()).length).toBeGreaterThan(0)
expect(response.usage).toEqual({ type: "characters", characters: expect.any(Number) })
}),
)
recorded.effect("streams speech", () =>
Effect.gen(function* () {
const { finish } = yield* collectSpeech(Speech.stream({ model, text: TEXT, format: "pcm" }))
expect(finish.audio.info).toMatchObject({ format: "pcm", encoding: "pcm_s16le", channels: 1 })
expect(finish.usage).toEqual({ type: "characters", characters: expect.any(Number) })
}),
)
})
@@ -0,0 +1,54 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Speech, SpeechEvent } from "../../src/index.js"
import { ElevenLabs } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { TEXT, collectSpeech } from "./speech-recording.js"
const model = ElevenLabs.configure({
apiKey: process.env.ELEVENLABS_API_KEY ?? "fixture",
}).speech("eleven_flash_v2_5")
// "George", an ElevenLabs premade voice.
const voice = "JBFqnCBsd6RMkjVDRZzb"
const recorded = recordedTests({
prefix: "elevenlabs-speech",
provider: "elevenlabs",
protocol: "elevenlabs-speech",
requires: ["ELEVENLABS_API_KEY"],
// Usage and the request id arrive only as response headers.
options: { redact: { allowResponseHeaders: ["character-cost", "request-id"] } },
})
describe("ElevenLabs Speech recorded", () => {
recorded.effect("generates speech", () =>
Effect.gen(function* () {
const response = yield* Speech.generate({ model, text: TEXT, voice })
expect(response.audio.mediaType).toBe("audio/mpeg")
expect(response.audio.info).toMatchObject({ format: "mp3", sampleRate: 44100 })
expect((yield* response.audio.bytes()).length).toBeGreaterThan(0)
expect(response.usage).toEqual({ type: "credits", credits: 3 })
}),
)
recorded.effect("streams speech with timestamps", () =>
Effect.gen(function* () {
const { events, finish } = yield* collectSpeech(
Speech.stream({ model, text: TEXT, voice, format: "pcm", timestamps: true }),
)
const timestamps = events.filter(SpeechEvent.is.timestamps).flatMap((event) => event.items)
expect(finish.audio.mediaType).toBe("audio/pcm")
expect(finish.audio.info).toMatchObject({ encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
expect(
timestamps
.map((item) => item.text)
.join("")
.trim(),
).toBe(TEXT)
expect(timestamps.every((item) => item.endSeconds >= item.startSeconds)).toBe(true)
}),
)
})
@@ -0,0 +1,45 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Speech } from "../../src/index.js"
import { Google } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { TEXT, collectSpeech } from "./speech-recording.js"
const google = Google.configure({ apiKey: process.env.GOOGLE_GENERATIVE_AI_API_KEY ?? "fixture" })
const recorded = recordedTests({
prefix: "google-speech",
provider: "google",
protocol: "google-speech",
requires: ["GOOGLE_GENERATIVE_AI_API_KEY"],
})
describe("Google Speech recorded", () => {
recorded.effect("generates speech", () =>
Effect.gen(function* () {
const response = yield* Speech.generate({
model: google.speech("gemini-2.5-flash-preview-tts"),
text: `Say: ${TEXT}`,
voice: "Kore",
})
expect(response.audio.mediaType).toMatch(/^audio\/l16/i)
expect(response.audio.info).toMatchObject({ encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
expect((yield* response.audio.bytes()).length).toBeGreaterThan(0)
expect(response.usage?.type).toBe("tokens")
}),
)
// Only Gemini 3.1 TTS and later support `streamGenerateContent`.
recorded.effect("streams speech", () =>
Effect.gen(function* () {
const { finish } = yield* collectSpeech(
Speech.stream({ model: google.speech("gemini-3.1-flash-tts-preview"), text: `Say: ${TEXT}`, voice: "Kore" }),
)
expect(finish.audio.mediaType).toMatch(/^audio\/l16/i)
expect(finish.audio.info).toMatchObject({ encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
expect(finish.usage?.type).toBe("tokens")
}),
)
})
@@ -0,0 +1,38 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Speech } from "../../src/index.js"
import { OpenAI } from "../../src/providers.js"
import { recordedTests } from "../recorded-test.js"
import { TEXT, collectSpeech } from "./speech-recording.js"
const model = OpenAI.configure({
apiKey: process.env.OPENAI_API_KEY ?? "fixture",
}).speech("gpt-4o-mini-tts")
const recorded = recordedTests({
prefix: "openai-speech",
provider: "openai",
protocol: "openai-speech",
requires: ["OPENAI_API_KEY"],
})
describe("OpenAI Speech recorded", () => {
recorded.effect("generates speech", () =>
Effect.gen(function* () {
const response = yield* Speech.generate({ model, text: TEXT, voice: "coral" })
expect(response.audio.mediaType).toBe("audio/mpeg")
expect((yield* response.audio.bytes()).length).toBeGreaterThan(0)
}),
)
recorded.effect("streams speech", () =>
Effect.gen(function* () {
const { finish } = yield* collectSpeech(Speech.stream({ model, text: TEXT, voice: "coral", format: "pcm" }))
expect(finish.audio.mediaType).toBe("audio/pcm")
expect(finish.audio.info).toMatchObject({ encoding: "pcm_s16le", sampleRate: 24000, channels: 1 })
expect(finish.usage?.type).toBe("tokens")
}),
)
})
@@ -0,0 +1,19 @@
import { expect } from "bun:test"
import { Effect, Stream } from "effect"
import { SpeechEvent } from "../../src/index.js"
export const TEXT = "Hello from OpenCode."
export const collectSpeech = <E, R>(stream: Stream.Stream<SpeechEvent, E, R>) =>
Effect.gen(function* () {
const events = Array.from(yield* Stream.runCollect(stream))
const finish = events.at(-1)
const deltas = events.filter(SpeechEvent.is.audioDelta)
expect(deltas.length).toBeGreaterThan(0)
expect(events.filter(SpeechEvent.is.finish)).toHaveLength(1)
if (!finish || !SpeechEvent.is.finish(finish)) throw new Error("Expected the speech stream to end with finish")
const audio = yield* finish.audio.bytes()
expect(audio.length).toBeGreaterThan(0)
expect(audio.length).toBe(deltas.reduce((total, event) => total + event.chunk.length, 0))
return { events, finish }
})
+4
View File
@@ -7,10 +7,12 @@ import { fileURLToPath } from "node:url"
import { LLMClient, RequestExecutor } from "../src/route.js"
import { ImageClient } from "../src/image-client.js"
import { VideoClient } from "../src/video-client.js"
import { SpeechClient } from "../src/speech-client.js"
import { EvaluationClient } from "../src/experimental/evaluation-client.js"
import type { Service as EvaluationClientService } from "../src/experimental/evaluation-client.js"
import type { Service as ImageClientService } from "../src/image-client.js"
import type { Service as VideoClientService } from "../src/video-client.js"
import type { Service as SpeechClientService } from "../src/speech-client.js"
import type { Service as LLMClientService } from "../src/route/client.js"
import type { Service as RequestExecutorService } from "../src/route/executor.js"
import {
@@ -27,6 +29,7 @@ type RecordedEnv =
| LLMClientService
| ImageClientService
| VideoClientService
| SpeechClientService
| EvaluationClientService
| Socket.WebSocketConstructor
@@ -103,6 +106,7 @@ export const recordedTests = (options: RecordedTestsOptions) =>
LLMClient.layer.pipe(Layer.provide(requestExecutor)),
ImageClient.layer.pipe(Layer.provide(requestExecutor)),
VideoClient.layer.pipe(Layer.provide(requestExecutor)),
SpeechClient.layer.pipe(Layer.provide(requestExecutor)),
EvaluationClient.layer.pipe(Layer.provide(requestExecutor)),
webSocket,
)
+131
View File
@@ -0,0 +1,131 @@
import { describe, expect } from "bun:test"
import { Effect, Encoding, Layer, Stream } from "effect"
import { Speech, SpeechClient, SpeechEvent } from "../src/index.js"
import { Cartesia, Deepgram, ElevenLabs, Google, OpenAI } from "../src/providers.js"
import { it } from "./lib/effect.js"
import { dynamicResponse } from "./lib/http.js"
import { sseEvents } from "./lib/sse.js"
// Recorded tests in test/provider cover lowering and decoding; these cover what recordings cannot.
const layer = (handler: Parameters<typeof dynamicResponse>[0]) =>
SpeechClient.layer.pipe(Layer.provideMerge(dynamicResponse(handler)))
const respond = (body: string | ReadableStream<Uint8Array>, contentType: string) =>
layer((input) => Effect.succeed(input.respond(body, { headers: { "content-type": contentType } })))
const collect = <E, R>(stream: Stream.Stream<SpeechEvent, E, R>) =>
stream.pipe(
Stream.runCollect,
Effect.map((events) => Array.from(events)),
)
const openai = OpenAI.configure({ apiKey: "test", baseURL: "https://openai.test/v1" }).speech("gpt-4o-mini-tts")
const elevenlabs = ElevenLabs.configure({ apiKey: "test", baseURL: "https://elevenlabs.test" }).speech(
"eleven_flash_v2_5",
)
const cartesia = Cartesia.configure({ apiKey: "test", baseURL: "https://cartesia.test" }).speech("sonic-3")
const google = Google.configure({ apiKey: "test", baseURL: "https://google.test/v1beta" }).speech(
"gemini-2.5-flash-preview-tts",
)
const deepgram = Deepgram.configure({ apiKey: "test", baseURL: "https://deepgram.test" }).speech("aura-2-thalia-en")
const voice = "JBFqnCBsd6RMkjVDRZzb"
describe("Speech", () => {
it.effect("rejects what a provider cannot produce before sending anything", () =>
Effect.gen(function* () {
const errors = yield* Effect.all(
[
Speech.generate({ model: openai, text: "Hi", timestamps: true }),
Speech.generate({ model: google, text: "Hi", format: "mp3" }),
Speech.generate({ model: google, text: "Hi", instructions: "Warm." }),
collect(Speech.stream({ model: elevenlabs, text: "Hi", voice, format: "wav" })),
collect(Speech.stream({ model: cartesia, text: "Hi", voice, format: "mp3" })),
Speech.generate({ model: deepgram, text: "Hi", voice: "thalia" }),
].map((effect) => Effect.flip(effect)),
)
expect(errors.map((error) => [error.reason._tag, "operation" in error.reason && error.reason.operation])).toEqual(
[
["UnsupportedOperation", "media.timestamps"],
["UnsupportedOperation", "media.format"],
["UnsupportedOperation", "media.instructions"],
["UnsupportedOperation", "media.format"],
["UnsupportedOperation", "media.format"],
["UnsupportedOperation", "media.voice"],
],
)
}).pipe(Effect.provide(layer(() => Effect.die("an unsupported request reached the network")))),
)
it.effect("classifies stream failures and keeps the provider payload and HTTP context", () =>
Effect.gen(function* () {
const badFrame = JSON.stringify({ type: "speech.audio.delta", audio: "not base64!" })
const cartesiaError = { type: "error", done: true, status_code: 400, title: "Invalid model", message: "Nope" }
const blocked = JSON.stringify({ promptFeedback: { blockReason: "PROHIBITED_CONTENT" } })
const [truncated, invalid, provider, policy] = yield* Effect.all(
[
collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
Effect.provide(respond(sseEvents({ type: "speech.audio.delta", audio: "AQI=" }), "text/event-stream")),
),
collect(Speech.stream({ model: openai, text: "Hi" })).pipe(
Effect.provide(respond(sseEvents(badFrame), "text/event-stream")),
),
collect(Speech.stream({ model: cartesia, text: "Hi", voice })).pipe(
Effect.provide(respond(sseEvents(cartesiaError), "text/event-stream")),
),
Speech.generate({ model: google, text: "Hi" }).pipe(Effect.provide(respond(blocked, "application/json"))),
].map((effect) => Effect.flip(effect)),
)
expect(truncated.reason).toMatchObject({ _tag: "InvalidProviderOutput", classification: "incomplete-stream" })
expect(truncated.reason.http?.status).toBe(200)
expect(invalid.reason).toMatchObject({ _tag: "InvalidProviderOutput", body: badFrame })
expect(invalid.reason.http?.status).toBe(200)
expect(provider.reason).toMatchObject({ _tag: "InvalidRequest", body: JSON.stringify(cartesiaError) })
expect(provider.message).toBe("Cartesia stream failed (Invalid model): Nope")
expect(policy.reason).toMatchObject({ _tag: "ContentPolicy", body: blocked })
}),
)
it.effect("parses ElevenLabs timestamped records split across network chunks", () =>
Effect.gen(function* () {
const record = (bytes: ReadonlyArray<number>, character: string, start: number) =>
JSON.stringify({
audio_base64: Encoding.encodeBase64(Uint8Array.from(bytes)),
alignment: {
characters: [character],
character_start_times_seconds: [start],
character_end_times_seconds: [start + 0.1],
},
})
const first = record([1, 2], "H", 0)
const second = record([3], "i", 0.1)
const encoder = new TextEncoder()
const body = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode(`${first}\n${second.slice(0, 10)}`))
controller.enqueue(encoder.encode(`${second.slice(10)}\n`))
controller.close()
},
})
const events = yield* collect(Speech.stream({ model: elevenlabs, text: "Hi", voice, timestamps: true })).pipe(
Effect.provide(respond(body, "application/json")),
)
expect(events.map((event) => event.type)).toEqual([
"audio-delta",
"timestamps",
"audio-delta",
"timestamps",
"finish",
])
expect(events.filter(SpeechEvent.is.timestamps).flatMap((event) => event.items)).toEqual([
{ text: "H", startSeconds: 0, endSeconds: 0.1 },
{ text: "i", startSeconds: 0.1, endSeconds: 0.2 },
])
const finish = events.find(SpeechEvent.is.finish)
expect(yield* finish!.audio.bytes()).toEqual(Uint8Array.from([1, 2, 3]))
}),
)
})
+31
View File
@@ -0,0 +1,31 @@
import type { Stream } from "effect"
import { Speech, type SpeechEvent } from "../src/index.js"
import { ElevenLabs, OpenAI, Runway } from "../src/providers.js"
type StreamItem<T> = T extends Stream.Stream<infer A, infer _E, infer _R> ? A : never
type Equal<A, B> = [A, B] extends [B, A] ? true : false
type Assert<T extends true> = T
const elevenlabs = ElevenLabs.configure({ apiKey: "test" }).speech("eleven_flash_v2_5")
Speech.generate({
model: elevenlabs,
text: "Hello",
voice: "JBFqnCBsd6RMkjVDRZzb",
format: "future-format",
providerOptions: { outputFormat: "future_format", voice_settings: { stability: 0.4 }, futureOption: true },
})
Speech.stream({ model: OpenAI.configure({ apiKey: "test" }).speech("gpt-4o-mini-tts"), text: "Hi", voice: { id: "v" } })
// @ts-expect-error Known provider options are inferred from the selected model.
Speech.generate({ model: elevenlabs, text: "Hello", providerOptions: { voice_settings: { stability: "high" } } })
// @ts-expect-error Voices are provider-native strings or `{ id }` objects.
Speech.generate({ model: elevenlabs, text: "Hello", voice: { name: "George" } })
// @ts-expect-error Image models cannot be used for speech requests.
Speech.request({ model: OpenAI.configure({ apiKey: "test" }).image("gpt-image-2"), text: "Hello" })
// @ts-expect-error Video models cannot be used for speech requests.
Speech.request({ model: Runway.configure({ apiKey: "test" }).video("gen4.5"), text: "Hello" })
const streamed = Speech.stream({ model: elevenlabs, text: "Hello" })
type StreamEvent = Assert<Equal<StreamItem<typeof streamed>, SpeechEvent>>
void (true satisfies StreamEvent)