731 lines
24 KiB
TypeScript
731 lines
24 KiB
TypeScript
/**
|
|
* The `OpenAiClient` module defines the low-level Effect service used by the
|
|
* OpenAI integration for Responses API and embedding requests. It builds a
|
|
* configured HTTP client with authentication and OpenAI organization or project
|
|
* headers, exposes helpers for non-streaming responses, SSE response streams,
|
|
* WebSocket response streams, and embeddings, and maps transport or decoding
|
|
* failures into `AiError`.
|
|
*
|
|
* @since 4.0.0
|
|
*/
|
|
import * as Array from "effect/Array"
|
|
import type * as Config from "effect/Config"
|
|
import * as Context from "effect/Context"
|
|
import * as Effect from "effect/Effect"
|
|
import { identity } from "effect/Function"
|
|
import * as Function from "effect/Function"
|
|
import * as Layer from "effect/Layer"
|
|
import * as Predicate from "effect/Predicate"
|
|
import * as Queue from "effect/Queue"
|
|
import * as RcRef from "effect/RcRef"
|
|
import * as Redacted from "effect/Redacted"
|
|
import * as Schema from "effect/Schema"
|
|
import * as Scope from "effect/Scope"
|
|
import * as Semaphore from "effect/Semaphore"
|
|
import * as Stream from "effect/Stream"
|
|
import * as AiError from "effect/unstable/ai/AiError"
|
|
import * as ResponseIdTracker from "effect/unstable/ai/ResponseIdTracker"
|
|
import * as Sse from "effect/unstable/encoding/Sse"
|
|
import * as Headers from "effect/unstable/http/Headers"
|
|
import * as HttpBody from "effect/unstable/http/HttpBody"
|
|
import * as HttpClient from "effect/unstable/http/HttpClient"
|
|
import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest"
|
|
import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse"
|
|
import * as Socket from "effect/unstable/socket/Socket"
|
|
import * as Errors from "./internal/errors.ts"
|
|
import { OpenAiConfig } from "./OpenAiConfig.ts"
|
|
import * as OpenAiSchema from "./OpenAiSchema.ts"
|
|
|
|
// =============================================================================
|
|
// Service Interface
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Effect service interface for the handwritten OpenAI client.
|
|
*
|
|
* **Details**
|
|
*
|
|
* Provides the configured HTTP client plus helpers for Responses API calls, streaming Responses events, and embeddings. Transport and schema decoding failures are mapped to `AiError`.
|
|
*
|
|
* @category models
|
|
* @since 4.0.0
|
|
*/
|
|
export interface Service {
|
|
/**
|
|
* The transformed HTTP client used by this service.
|
|
*/
|
|
readonly client: HttpClient.HttpClient
|
|
|
|
/**
|
|
* Create a response using the OpenAI responses endpoint.
|
|
*/
|
|
readonly createResponse: (
|
|
options: typeof OpenAiSchema.CreateResponse.Encoded
|
|
) => Effect.Effect<
|
|
readonly [body: typeof OpenAiSchema.Response.Type, response: HttpClientResponse.HttpClientResponse],
|
|
AiError.AiError
|
|
>
|
|
|
|
/**
|
|
* Create a streaming response using the OpenAI responses endpoint.
|
|
*/
|
|
readonly createResponseStream: (
|
|
options: Omit<typeof OpenAiSchema.CreateResponse.Encoded, "stream">
|
|
) => Effect.Effect<
|
|
readonly [
|
|
response: HttpClientResponse.HttpClientResponse,
|
|
stream: Stream.Stream<typeof OpenAiSchema.ResponseStreamEvent.Type, AiError.AiError>
|
|
],
|
|
AiError.AiError
|
|
>
|
|
|
|
/**
|
|
* Create embeddings using the OpenAI embeddings endpoint.
|
|
*/
|
|
readonly createEmbedding: (
|
|
options: typeof OpenAiSchema.CreateEmbeddingRequest.Encoded
|
|
) => Effect.Effect<typeof OpenAiSchema.CreateEmbeddingResponse.Type, AiError.AiError>
|
|
}
|
|
|
|
// =============================================================================
|
|
// Service Identifier
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Service tag for the OpenAI client.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use when accessing or providing the OpenAI client service through Effect's
|
|
* context.
|
|
*
|
|
* @see {@link make} for constructing an OpenAI client effectfully
|
|
* @see {@link layer} for providing a client from explicit options
|
|
* @see {@link layerConfig} for providing a client from `Config`
|
|
*
|
|
* @category services
|
|
* @since 4.0.0
|
|
*/
|
|
export class OpenAiClient extends Context.Service<OpenAiClient, Service>()(
|
|
"@effect/ai-openai/OpenAiClient"
|
|
) {}
|
|
|
|
// =============================================================================
|
|
// Options
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Options for configuring the OpenAI client.
|
|
*
|
|
* @category options
|
|
* @since 4.0.0
|
|
*/
|
|
export type Options = {
|
|
/**
|
|
* The OpenAI API key.
|
|
*/
|
|
readonly apiKey?: Redacted.Redacted<string> | undefined
|
|
|
|
/**
|
|
* The base URL for the OpenAI API.
|
|
*
|
|
* @default "https://api.openai.com/v1"
|
|
*/
|
|
readonly apiUrl?: string | undefined
|
|
|
|
/**
|
|
* Optional organization ID for multi-org accounts.
|
|
*/
|
|
readonly organizationId?: Redacted.Redacted<string> | undefined
|
|
|
|
/**
|
|
* Optional project ID for project-scoped requests.
|
|
*/
|
|
readonly projectId?: Redacted.Redacted<string> | undefined
|
|
|
|
/**
|
|
* Optional transformer for the HTTP client.
|
|
*/
|
|
readonly transformClient?: ((client: HttpClient.HttpClient) => HttpClient.HttpClient) | undefined
|
|
}
|
|
|
|
// =============================================================================
|
|
// Constructor
|
|
// =============================================================================
|
|
|
|
const RedactedOpenAiHeaders = {
|
|
OpenAiOrganization: "OpenAI-Organization",
|
|
OpenAiProject: "OpenAI-Project"
|
|
}
|
|
|
|
/**
|
|
* Creates an OpenAI client service with the given options.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use when you need the OpenAI client service value inside an effect.
|
|
*
|
|
* **Details**
|
|
*
|
|
* The returned service uses the current `HttpClient`, prepends `apiUrl` or
|
|
* `https://api.openai.com/v1`, adds the bearer token and optional OpenAI
|
|
* organization/project headers, accepts JSON responses, filters for successful
|
|
* HTTP statuses, and applies `transformClient` when provided.
|
|
*
|
|
* **Gotchas**
|
|
*
|
|
* A scoped `OpenAiConfig.withClientTransform` is applied when request helpers
|
|
* run, after the `transformClient` option supplied to `make`.
|
|
*
|
|
* @see {@link layer} for providing this client from explicit options
|
|
* @see {@link layerConfig} for loading client settings from `Config`
|
|
*
|
|
* @category constructors
|
|
* @since 4.0.0
|
|
*/
|
|
export const make = Effect.fnUntraced(
|
|
function*(
|
|
options: Options
|
|
): Effect.fn.Return<Service, never, HttpClient.HttpClient> {
|
|
const baseClient = yield* HttpClient.HttpClient
|
|
const apiUrl = options.apiUrl ?? "https://api.openai.com/v1"
|
|
|
|
const httpClient = baseClient.pipe(
|
|
HttpClient.mapRequest(Function.flow(
|
|
HttpClientRequest.prependUrl(apiUrl),
|
|
options.apiKey
|
|
? HttpClientRequest.bearerToken(Redacted.value(options.apiKey))
|
|
: identity,
|
|
options.organizationId
|
|
? HttpClientRequest.setHeader(
|
|
RedactedOpenAiHeaders.OpenAiOrganization,
|
|
Redacted.value(options.organizationId)
|
|
)
|
|
: identity,
|
|
options.projectId
|
|
? HttpClientRequest.setHeader(
|
|
RedactedOpenAiHeaders.OpenAiProject,
|
|
Redacted.value(options.projectId)
|
|
)
|
|
: identity,
|
|
HttpClientRequest.acceptJson
|
|
)),
|
|
HttpClient.filterStatusOk,
|
|
options.transformClient
|
|
? options.transformClient
|
|
: identity
|
|
)
|
|
|
|
const resolveHttpClient = Effect.map(
|
|
OpenAiConfig.getOrUndefined,
|
|
(config) =>
|
|
Predicate.isNotUndefined(config?.transformClient)
|
|
? config.transformClient(httpClient)
|
|
: httpClient
|
|
)
|
|
|
|
const decodeResponse = HttpClientResponse.schemaBodyJson(OpenAiSchema.Response)
|
|
|
|
const createResponse = (
|
|
payload: typeof OpenAiSchema.CreateResponse.Encoded
|
|
): Effect.Effect<
|
|
[body: typeof OpenAiSchema.Response.Type, response: HttpClientResponse.HttpClientResponse],
|
|
AiError.AiError
|
|
> =>
|
|
Effect.flatMap(resolveHttpClient, (client) =>
|
|
client.execute(
|
|
HttpClientRequest.post("/responses", {
|
|
body: HttpBody.jsonUnsafe(payload)
|
|
})
|
|
).pipe(
|
|
Effect.flatMap((response) =>
|
|
decodeResponse(response).pipe(
|
|
Effect.map((body): [typeof OpenAiSchema.Response.Type, HttpClientResponse.HttpClientResponse] => [
|
|
body,
|
|
response
|
|
])
|
|
)
|
|
),
|
|
Effect.catchTags({
|
|
HttpClientError: (error) => Errors.mapHttpClientError(error, "createResponse"),
|
|
SchemaError: (error) => Effect.fail(Errors.mapSchemaError(error, "createResponse"))
|
|
})
|
|
))
|
|
|
|
const buildResponseStream = (
|
|
response: HttpClientResponse.HttpClientResponse
|
|
): [
|
|
HttpClientResponse.HttpClientResponse,
|
|
Stream.Stream<typeof OpenAiSchema.ResponseStreamEvent.Type, AiError.AiError>
|
|
] => {
|
|
const stream = response.stream.pipe(
|
|
Stream.decodeText(),
|
|
Stream.pipeThroughChannel(Sse.decodeDataSchema(OpenAiSchema.ResponseStreamEvent)),
|
|
Stream.takeUntil((event) =>
|
|
event.data.type === "response.completed" ||
|
|
event.data.type === "response.incomplete"
|
|
),
|
|
Stream.map((event) => event.data),
|
|
Stream.catchTags({
|
|
// TODO: handle SSE retries
|
|
Retry: (error) => Stream.die(error),
|
|
HttpClientError: (error) => Stream.fromEffect(Errors.mapHttpClientError(error, "createResponseStream")),
|
|
SchemaError: (error) => Stream.fail(Errors.mapSchemaError(error, "createResponseStream"))
|
|
})
|
|
)
|
|
return [response, stream]
|
|
}
|
|
|
|
const createResponseStream: Service["createResponseStream"] = (payload) =>
|
|
Effect.contextWith((services) => {
|
|
const socket = Context.getOrUndefined(services, OpenAiSocket)
|
|
if (socket) return socket.createResponseStream(payload)
|
|
return Effect.flatMap(resolveHttpClient, (client) =>
|
|
client.execute(
|
|
HttpClientRequest.post("/responses", {
|
|
body: HttpBody.jsonUnsafe({ ...payload, stream: true })
|
|
})
|
|
).pipe(
|
|
Effect.map(buildResponseStream),
|
|
Effect.catchTag(
|
|
"HttpClientError",
|
|
(error) => Errors.mapHttpClientError(error, "createResponseStream")
|
|
)
|
|
))
|
|
})
|
|
|
|
const decodeEmbedding = HttpClientResponse.schemaBodyJson(OpenAiSchema.CreateEmbeddingResponse)
|
|
|
|
const createEmbedding = (
|
|
payload: typeof OpenAiSchema.CreateEmbeddingRequest.Encoded
|
|
): Effect.Effect<typeof OpenAiSchema.CreateEmbeddingResponse.Type, AiError.AiError> =>
|
|
Effect.flatMap(resolveHttpClient, (client) =>
|
|
client.execute(
|
|
HttpClientRequest.post("/embeddings", {
|
|
body: HttpBody.jsonUnsafe(payload)
|
|
})
|
|
).pipe(
|
|
Effect.flatMap(decodeEmbedding),
|
|
Effect.catchTags({
|
|
HttpClientError: (error) => Errors.mapHttpClientError(error, "createEmbedding"),
|
|
SchemaError: (error) => Effect.fail(Errors.mapSchemaError(error, "createEmbedding"))
|
|
})
|
|
))
|
|
|
|
return OpenAiClient.of({
|
|
client: httpClient,
|
|
createResponse,
|
|
createResponseStream,
|
|
createEmbedding
|
|
})
|
|
},
|
|
Effect.updateService(
|
|
Headers.CurrentRedactedNames,
|
|
Array.appendAll(Object.values(RedactedOpenAiHeaders))
|
|
)
|
|
)
|
|
|
|
// =============================================================================
|
|
// Layers
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Creates a layer for the OpenAI client with the given options.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use when you already have explicit `Options` values, such as an API key or
|
|
* custom API URL, and want to provide `OpenAiClient` as a `Layer`.
|
|
*
|
|
* @see {@link make} for constructing the client service effectfully
|
|
* @see {@link layerConfig} for loading client settings from `Config`
|
|
*
|
|
* @category layers
|
|
* @since 4.0.0
|
|
*/
|
|
export const layer = (options: Options): Layer.Layer<OpenAiClient, never, HttpClient.HttpClient> =>
|
|
Layer.effect(OpenAiClient, make(options))
|
|
|
|
/**
|
|
* Creates a layer for the OpenAI client from provided `Config` values.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use when you need client settings for OpenAI-compatible APIs to be read from
|
|
* Effect `Config` values while providing `OpenAiClient` as a `Layer`.
|
|
*
|
|
* **Details**
|
|
*
|
|
* Only config values supplied in `options` are loaded. Omitted fields are
|
|
* passed to `make` as `undefined`, and `transformClient` is forwarded as a
|
|
* plain option.
|
|
*
|
|
* @see {@link make} for constructing the client service effectfully
|
|
* @see {@link layer} for providing the client from already-resolved options
|
|
*
|
|
* @category layers
|
|
* @since 4.0.0
|
|
*/
|
|
export const layerConfig = (options?: {
|
|
/**
|
|
* The config value to load for the API key.
|
|
*/
|
|
readonly apiKey?: Config.Config<Redacted.Redacted<string> | undefined> | undefined
|
|
|
|
/**
|
|
* The config value to load for the API URL.
|
|
*/
|
|
readonly apiUrl?: Config.Config<string> | undefined
|
|
|
|
/**
|
|
* The config value to load for the organization ID.
|
|
*/
|
|
readonly organizationId?: Config.Config<Redacted.Redacted<string> | undefined> | undefined
|
|
|
|
/**
|
|
* The config value to load for the project ID.
|
|
*/
|
|
readonly projectId?: Config.Config<Redacted.Redacted<string> | undefined> | undefined
|
|
|
|
/**
|
|
* Optional transformer for the HTTP client.
|
|
*/
|
|
readonly transformClient?: ((client: HttpClient.HttpClient) => HttpClient.HttpClient) | undefined
|
|
}): Layer.Layer<OpenAiClient, Config.ConfigError, HttpClient.HttpClient> =>
|
|
Layer.effect(
|
|
OpenAiClient,
|
|
Effect.gen(function*() {
|
|
const apiKey = Predicate.isNotUndefined(options?.apiKey)
|
|
? yield* options.apiKey :
|
|
undefined
|
|
const apiUrl = Predicate.isNotUndefined(options?.apiUrl)
|
|
? yield* options.apiUrl :
|
|
undefined
|
|
const organizationId = Predicate.isNotUndefined(options?.organizationId)
|
|
? yield* options.organizationId
|
|
: undefined
|
|
const projectId = Predicate.isNotUndefined(options?.projectId)
|
|
? yield* options.projectId :
|
|
undefined
|
|
return yield* make({
|
|
apiKey,
|
|
apiUrl,
|
|
organizationId,
|
|
projectId,
|
|
transformClient: options?.transformClient
|
|
})
|
|
})
|
|
)
|
|
|
|
// =============================================================================
|
|
// Websocket mode
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Response stream event emitted by the OpenAI Responses API.
|
|
*
|
|
* @category Events
|
|
* @since 4.0.0
|
|
*/
|
|
export type ResponseStreamEvent = typeof OpenAiSchema.ResponseStreamEvent.Type
|
|
|
|
/**
|
|
* Service for creating OpenAI response streams over a WebSocket connection.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use when you need direct access to the WebSocket-backed response streaming
|
|
* service rather than wrapping an effect with WebSocket mode.
|
|
*
|
|
* **Details**
|
|
*
|
|
* `createResponseStream` sends a `response.create` message over the WebSocket
|
|
* connection and returns an HTTP response together with a stream of
|
|
* `ResponseStreamEvent` values.
|
|
*
|
|
* **Gotchas**
|
|
*
|
|
* WebSocket response streams are serialized to one request at a time by the
|
|
* shared socket service.
|
|
*
|
|
* @see {@link withWebSocketMode} for enabling WebSocket mode for one effect
|
|
* @see {@link layerWebSocketMode} for providing WebSocket mode through a layer
|
|
*
|
|
* @category Websocket mode
|
|
* @since 4.0.0
|
|
*/
|
|
export class OpenAiSocket extends Context.Service<OpenAiSocket, {
|
|
/**
|
|
* Create a streaming response using the OpenAI responses endpoint.
|
|
*/
|
|
readonly createResponseStream: (
|
|
options: Omit<typeof OpenAiSchema.CreateResponse.Encoded, "stream">
|
|
) => Effect.Effect<
|
|
readonly [
|
|
response: HttpClientResponse.HttpClientResponse,
|
|
stream: Stream.Stream<ResponseStreamEvent, AiError.AiError>
|
|
],
|
|
AiError.AiError
|
|
>
|
|
}>()("@effect/ai-openai/OpenAiClient/OpenAiSocket") {}
|
|
|
|
const makeSocket = Effect.gen(function*() {
|
|
const client = yield* OpenAiClient
|
|
const tracker = yield* ResponseIdTracker.make
|
|
const socketScope = yield* Effect.scope
|
|
const makeRequest = Effect.flatMap(
|
|
OpenAiConfig.getOrUndefined,
|
|
(config) => {
|
|
const httpClient = Predicate.isNotUndefined(config?.transformClient)
|
|
? config.transformClient(client.client)
|
|
: client.client
|
|
return Effect.orDie(httpClient.preprocess(HttpClientRequest.post("/responses")))
|
|
}
|
|
)
|
|
const makeWebSocket = yield* Socket.WebSocketConstructor
|
|
|
|
const decoder = new TextDecoder()
|
|
|
|
const queueRef: RcRef.RcRef<
|
|
{
|
|
readonly send: (message: typeof OpenAiSchema.CreateResponse.Encoded) => Effect.Effect<void, AiError.AiError>
|
|
readonly incoming: Queue.Dequeue<ResponseStreamEvent, AiError.AiError>
|
|
}
|
|
> = yield* RcRef.make({
|
|
idleTimeToLive: 60_000,
|
|
acquire: Effect.gen(function*() {
|
|
const scope = yield* Effect.scope
|
|
const request = yield* makeRequest
|
|
const socket = yield* Socket.makeWebSocket(request.url.replace(/^http/, "ws")).pipe(
|
|
Effect.provideService(Socket.WebSocketConstructor, (url) =>
|
|
makeWebSocket(url, {
|
|
headers: request.headers
|
|
} as any))
|
|
)
|
|
const write = yield* socket.writer
|
|
|
|
yield* Scope.addFinalizerExit(scope, () => {
|
|
tracker.clearUnsafe()
|
|
return Effect.void
|
|
})
|
|
|
|
const incoming = yield* Queue.unbounded<ResponseStreamEvent, AiError.AiError>()
|
|
const send = (message: typeof OpenAiSchema.CreateResponse.Encoded) =>
|
|
write(JSON.stringify({
|
|
type: "response.create",
|
|
...message
|
|
})).pipe(
|
|
Effect.mapError((_error) =>
|
|
AiError.make({
|
|
module: "OpenAiClient",
|
|
method: "createResponseStream",
|
|
reason: new AiError.NetworkError({
|
|
reason: "TransportError",
|
|
request: {
|
|
method: "POST",
|
|
url: request.url,
|
|
urlParams: [],
|
|
hash: undefined,
|
|
headers: request.headers
|
|
},
|
|
description: "Failed to send message over WebSocket"
|
|
})
|
|
})
|
|
)
|
|
)
|
|
|
|
yield* socket.runRaw((msg) => {
|
|
const text = typeof msg === "string" ? msg : decoder.decode(msg)
|
|
try {
|
|
const event = decodeEvent(text)
|
|
if (event.type === "error" && "status" in event) {
|
|
const status = Number(event.status)
|
|
const error = "error" in event ? event.error as typeof ErrorEvent.Type.error : event
|
|
const json = JSON.stringify(error)
|
|
return Effect.fail(
|
|
AiError.make({
|
|
module: "OpenAiClient",
|
|
method: "createResponseStream",
|
|
reason: AiError.reasonFromHttpStatus({
|
|
description: json,
|
|
status: isNaN(status) ? errorTypeToStatus[error.type] ?? 500 : status,
|
|
metadata: error as any,
|
|
http: {
|
|
body: json,
|
|
request: {
|
|
method: "POST",
|
|
url: request.url,
|
|
urlParams: [],
|
|
hash: undefined,
|
|
headers: request.headers
|
|
}
|
|
}
|
|
})
|
|
})
|
|
)
|
|
}
|
|
Queue.offerUnsafe(incoming, event)
|
|
} catch {}
|
|
}).pipe(
|
|
Effect.catchTag("SocketError", (error) =>
|
|
AiError.make({
|
|
module: "OpenAiClient",
|
|
method: "createResponseStream",
|
|
reason: new AiError.NetworkError({
|
|
reason: "TransportError",
|
|
request: {
|
|
method: "POST",
|
|
url: request.url,
|
|
urlParams: [],
|
|
hash: undefined,
|
|
headers: request.headers
|
|
},
|
|
description: error.message
|
|
})
|
|
})),
|
|
Effect.catchCause((cause) => Queue.failCause(incoming, cause)),
|
|
Effect.ensuring(Effect.forkIn(RcRef.invalidate(queueRef), socketScope, {
|
|
startImmediately: true
|
|
})),
|
|
Effect.forkScoped({ startImmediately: true })
|
|
)
|
|
|
|
return { send, incoming } as const
|
|
})
|
|
})
|
|
|
|
// Prime the websocket
|
|
yield* Effect.scoped(RcRef.get(queueRef))
|
|
|
|
// Websocket mode only allows one request at a time
|
|
const semaphore = Semaphore.makeUnsafe(1)
|
|
const request = yield* makeRequest
|
|
|
|
return OpenAiSocket.context({
|
|
createResponseStream(options) {
|
|
const stream = Stream.unwrap(Effect.gen(function*() {
|
|
const scope = yield* Effect.scope
|
|
yield* Effect.acquireRelease(
|
|
semaphore.take(1),
|
|
() => semaphore.release(1),
|
|
{ interruptible: true }
|
|
)
|
|
const { send, incoming } = yield* RcRef.get(queueRef)
|
|
let done = false
|
|
|
|
yield* Scope.addFinalizerExit(
|
|
scope,
|
|
() => done ? Effect.void : RcRef.invalidate(queueRef)
|
|
)
|
|
|
|
yield* send(options).pipe(
|
|
Effect.forkScoped({ startImmediately: true })
|
|
)
|
|
|
|
return Stream.fromQueue(incoming).pipe(
|
|
Stream.takeUntil((e) => {
|
|
done = e.type === "response.completed" || e.type === "response.incomplete"
|
|
return done
|
|
})
|
|
)
|
|
}))
|
|
|
|
return Effect.succeed([
|
|
HttpClientResponse.fromWeb(request, new Response()),
|
|
stream
|
|
])
|
|
}
|
|
}).pipe(
|
|
Context.add(ResponseIdTracker.ResponseIdTracker, tracker)
|
|
)
|
|
})
|
|
|
|
const ErrorEvent = Schema.Struct({
|
|
type: Schema.Literal("error"),
|
|
status: Schema.Number.pipe(
|
|
Schema.withDecodingDefault(Effect.succeed(500))
|
|
),
|
|
error: Schema.Struct({
|
|
type: Schema.String,
|
|
message: Schema.String
|
|
})
|
|
})
|
|
|
|
const errorTypeToStatus: Record<string, number> = {
|
|
invalid_request_error: 400,
|
|
invalid_api_key_error: 401,
|
|
insufficient_quota_error: 429,
|
|
rate_limit_error: 429,
|
|
service_unavailable_error: 503
|
|
}
|
|
|
|
const AllEvents = Schema.Union([ErrorEvent, OpenAiSchema.ResponseStreamEvent])
|
|
const decodeEvent = Schema.decodeUnknownSync(Schema.fromJsonString(AllEvents))
|
|
|
|
/**
|
|
* Uses OpenAI's WebSocket mode for response streams within the provided effect.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use to enable WebSocket mode around one effect that creates OpenAI response
|
|
* streams.
|
|
*
|
|
* **Gotchas**
|
|
*
|
|
* This only works with the following WebSocket constructor layers:
|
|
*
|
|
* - `NodeSocket.layerWebSocketConstructorWS`
|
|
* - `BunSocket.layerWebSocketConstructor`
|
|
*
|
|
* These constructor layers support the non-standard options needed to set the
|
|
* Authorization header.
|
|
*
|
|
* @see {@link layerWebSocketMode} for providing WebSocket mode through a layer
|
|
* @see {@link OpenAiSocket} for direct access to the WebSocket-backed streaming service
|
|
*
|
|
* @category Websocket mode
|
|
* @since 4.0.0
|
|
*/
|
|
export const withWebSocketMode = <A, E, R>(
|
|
effect: Effect.Effect<A, E, R>
|
|
): Effect.Effect<
|
|
A,
|
|
E,
|
|
Exclude<R, OpenAiSocket | ResponseIdTracker.ResponseIdTracker> | OpenAiClient | Socket.WebSocketConstructor
|
|
> =>
|
|
Effect.scopedWith((scope) =>
|
|
Effect.flatMap(
|
|
Scope.provide(makeSocket, scope),
|
|
(services) => Effect.provideContext(effect, services)
|
|
)
|
|
)
|
|
|
|
/**
|
|
* Uses OpenAI's websocket mode for all responses that use the Layer.
|
|
*
|
|
* **When to use**
|
|
*
|
|
* Use to provide WebSocket mode through layer composition for effects that use
|
|
* OpenAI response streaming.
|
|
*
|
|
* **Gotchas**
|
|
*
|
|
* This only works with the following WebSocket constructor layers:
|
|
*
|
|
* - `NodeSocket.layerWebSocketConstructorWS`
|
|
* - `BunSocket.layerWebSocketConstructor`
|
|
*
|
|
* These constructor layers support the non-standard options needed to set the
|
|
* Authorization header.
|
|
*
|
|
* @see {@link withWebSocketMode} for enabling WebSocket mode around a single effect
|
|
*
|
|
* @category Websocket mode
|
|
* @since 4.0.0
|
|
*/
|
|
export const layerWebSocketMode: Layer.Layer<
|
|
OpenAiSocket | ResponseIdTracker.ResponseIdTracker,
|
|
never,
|
|
OpenAiClient | Socket.WebSocketConstructor
|
|
> = Layer.effectContext(makeSocket)
|