Open History to see the last commit
1import type { RuntimeContext } from "alchemy/RuntimeContext";
2import { AiError, LanguageModel, Response, Tool } from "effect/ai";
3import { toCodecOpenAI } from "effect/ai/OpenAiStructuredOutput";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Sse from "effect/encoding/Sse";
7import * as FetchHttpClient from "effect/http/FetchHttpClient";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import type * as HttpClientResponse from "effect/http/HttpClientResponse";
11import * as Layer from "effect/Layer";
12import type * as Redacted from "effect/Redacted";
13import * as Schema from "effect/Schema";
14import * as Stream from "effect/Stream";
15
16/** Runtime-only connection obtained from QueryAIGateway, never an account API key. */
17export interface LanguageModelClient {
18 /** OpenAI-compatible base URL ending in /v1. */
19 readonly chatBaseUrl: Effect.Effect<string, never, RuntimeContext>;
20 /** Redacted credential with ai_gateway:invoke scope. */
21 readonly token: Effect.Effect<Redacted.Redacted<string>, never, RuntimeContext>;
22}
23
24/** Configuration for Neon's OpenAI-compatible Chat Completions endpoint. */
25export interface LanguageModelOptions {
26 /** Model ID from Neon's catalog that supports Chat Completions. */
27 readonly model: string;
28 /** Model-dependent generation settings, sent without silently rewriting them. */
29 readonly parameters?: {
30 /** Sampling temperature. */
31 readonly temperature?: number;
32 /** Maximum output tokens; mutually exclusive with maxCompletionTokens. */
33 readonly maxTokens?: number;
34 /** Completion budget for models using max_completion_tokens. */
35 readonly maxCompletionTokens?: number;
36 /** Nucleus sampling probability. */
37 readonly topP?: number;
38 /** Provider-supported deterministic sampling seed. */
39 readonly seed?: number;
40 /** Repetition penalty based on token frequency. */
41 readonly frequencyPenalty?: number;
42 /** Repetition penalty based on token presence. */
43 readonly presencePenalty?: number;
44 };
45}
46
47/**
48 * Effect AI over Neon's documented `/v1/chat/completions` route. Supports text,
49 * image inputs, function tools, JSON-schema output and cancellable SSE streams.
50 * Tool and structured-output support depends on the selected model. Responses-only
51 * models (including Codex), native Anthropic thinking/cache controls, audio,
52 * provider-executed tools and image generation are not supported by this adapter.
53 * Use the bound client's dialect URLs with a native SDK for those APIs.
54 *
55 * ### Generate text
56 * **Example:** A request-scoped Effect model
57 * ```typescript
58 * const ai = yield* Neon.QueryAIGateway(gateway);
59 * const model = ai.model({ model: "gpt-5-mini" });
60 * // Inside the Function or Worker request handler:
61 * const reply = yield* LanguageModel.generateText({ prompt: "Say hello." }).pipe(
62 * Effect.provide(model),
63 * );
64 * ```
65 *
66 * ### Generate structured output
67 * **Example:** Decode the response with an Effect Schema
68 * ```typescript
69 * const reply = yield* LanguageModel.generateObject({
70 * prompt: "Return a short greeting.",
71 * schema: Schema.Struct({ greeting: Schema.String }),
72 * }).pipe(Effect.provide(ai.model({ model: "gpt-5-mini" })));
73 * ```
74 *
75 * @layer
76 * @provides effect/ai/LanguageModel
77 * @product AI Gateway
78 * @category AI Gateway
79 */
80export const makeLanguageModelLayer = (
81 options: LanguageModelOptions & { readonly client: LanguageModelClient },
82): Layer.Layer<LanguageModel.LanguageModel, never, RuntimeContext> =>
83 Layer.effect(LanguageModel.LanguageModel, makeLanguageModel(options)).pipe(
84 Layer.provide(FetchHttpClient.layer),
85 );
86
87/** Construct with an injected Effect HTTP client for custom transports and protocol tests. */
88export const makeLanguageModel = ({
89 client,
90 model,
91 parameters,
92}: LanguageModelOptions & {
93 readonly client: LanguageModelClient;
94}): Effect.Effect<LanguageModel.LanguageModel, never, RuntimeContext | HttpClient.HttpClient> =>
95 Effect.gen(function* () {
96 const http = yield* HttpClient.HttpClient;
97 const baseUrl = yield* client.chatBaseUrl;
98 const token = yield* client.token;
99 const send = (options: LanguageModel.ProviderOptions, stream: boolean) =>
100 Effect.gen(function* () {
101 const body = yield* requestBody(options, model, parameters, stream);
102 const request = yield* HttpClientRequest.bodyJson(body)(
103 HttpClientRequest.post(`${baseUrl.replace(/\/$/, "")}/chat/completions`).pipe(
104 HttpClientRequest.bearerToken(token),
105 HttpClientRequest.accept(stream ? "text/event-stream" : "application/json"),
106 ),
107 ).pipe(Effect.mapError(() => invalidRequest("Request is not JSON serializable")));
108 const response = yield* http.execute(request).pipe(
109 Effect.mapError(() =>
110 error(
111 AiError.NetworkError.make({
112 reason: "TransportError",
113 method: "POST",
114 url: "/v1/chat/completions",
115 description: "Neon AI Gateway transport failed",
116 }),
117 ),
118 ),
119 );
120 if (response.status < 200 || response.status >= 300) {
121 return yield* httpError(response);
122 }
123 return response;
124 });
125 return yield* LanguageModel.make({
126 codecTransformer: toCodecOpenAI,
127 generateText: (options) =>
128 Effect.gen(function* () {
129 const response = yield* send(options, false);
130 const raw = yield* response.json.pipe(
131 Effect.mapError(() => invalidOutput("Expected a JSON completion")),
132 );
133 const decoded = yield* decode(Completion, raw);
134 const choice = decoded.choices[0];
135 if (!choice) return yield* invalidOutput("Completion has no choices");
136 const parts: Array<Response.PartEncoded> = [];
137 const message = choice.message;
138 if (message.refusal)
139 return yield* error(
140 AiError.ContentPolicyError.make({
141 description: "Model refused the request",
142 }),
143 );
144 if (message.reasoning_content)
145 parts.push({ type: "reasoning", text: message.reasoning_content });
146 if (typeof message.content === "string") {
147 if (message.content) parts.push({ type: "text", text: message.content });
148 } else if (message.content) {
149 for (const block of message.content) parts.push({ type: "text", text: block.text });
150 }
151 for (const call of message.tool_calls ?? []) {
152 parts.push({
153 type: "tool-call",
154 id: call.id,
155 name: call.function.name,
156 params: yield* parseArguments(call.function.arguments),
157 });
158 }
159 parts.push({
160 type: "finish",
161 reason: finishReason(choice.finish_reason),
162 usage: usage(decoded.usage),
163 });
164 return parts;
165 }),
166 streamText: (options) =>
167 Stream.unwrap(
168 Effect.gen(function* () {
169 const response = yield* send(options, true);
170 return completionStream(response);
171 }),
172 ),
173 });
174 });
175
176const error = (reason: AiError.AiError["reason"]) =>
177 AiError.make({
178 module: "Neon.LanguageModel",
179 method: "chatCompletions",
180 reason,
181 });
182const invalidOutput = (description: string) =>
183 error(AiError.InvalidOutputError.make({ description }));
184const invalidRequest = (description: string) =>
185 error(AiError.InvalidRequestError.make({ description }));
186const decode = <S extends Schema.ConstraintDecoder<unknown>>(schema: S, value: unknown) =>
187 Schema.decodeUnknownEffect(schema)(value).pipe(
188 Effect.mapError(() => invalidOutput("Invalid Chat Completions response shape")),
189 );
190const parseArguments = (value: string) =>
191 Schema.decodeEffect(Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)))(
192 value,
193 ).pipe(Effect.mapError(() => invalidOutput("Invalid function-call JSON arguments")));
194
195const TokenCount = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0));
196const Usage = Schema.Struct({
197 prompt_tokens: Schema.optional(TokenCount),
198 completion_tokens: Schema.optional(TokenCount),
199 prompt_tokens_details: Schema.Struct({ cached_tokens: Schema.optional(TokenCount) }).pipe(
200 Schema.NullOr,
201 Schema.optional,
202 ),
203 completion_tokens_details: Schema.Struct({ reasoning_tokens: Schema.optional(TokenCount) }).pipe(
204 Schema.NullOr,
205 Schema.optional,
206 ),
207});
208const Content = Schema.NullOr(
209 Schema.Union([
210 Schema.String,
211 Schema.Array(Schema.Struct({ type: Schema.Literal("text"), text: Schema.String })),
212 ]),
213);
214const Completion = Schema.Struct({
215 choices: Schema.Array(
216 Schema.Struct({
217 finish_reason: Schema.NullOr(Schema.String),
218 message: Schema.Struct({
219 content: Schema.optional(Content),
220 reasoning_content: Schema.String.pipe(Schema.NullOr, Schema.optional),
221 refusal: Schema.String.pipe(Schema.NullOr, Schema.optional),
222 tool_calls: Schema.Struct({
223 id: Schema.String,
224 type: Schema.Literal("function"),
225 function: Schema.Struct({
226 name: Schema.String,
227 arguments: Schema.String,
228 }),
229 }).pipe(Schema.Array, Schema.optional),
230 }),
231 }),
232 ),
233 usage: Usage.pipe(Schema.NullOr, Schema.optional),
234});
235const Chunk = Schema.Struct({
236 choices: Schema.Array(
237 Schema.Struct({
238 index: TokenCount,
239 finish_reason: Schema.String.pipe(Schema.NullOr, Schema.optional),
240 delta: Schema.Struct({
241 content: Schema.String.pipe(Schema.NullOr, Schema.optional),
242 reasoning_content: Schema.String.pipe(Schema.NullOr, Schema.optional),
243 refusal: Schema.String.pipe(Schema.NullOr, Schema.optional),
244 tool_calls: Schema.Struct({
245 index: TokenCount,
246 id: Schema.optional(Schema.String),
247 type: Schema.optional(Schema.Literal("function")),
248 function: Schema.optional(
249 Schema.Struct({
250 name: Schema.optional(Schema.String),
251 arguments: Schema.optional(Schema.String),
252 }),
253 ),
254 }).pipe(Schema.Array, Schema.optional),
255 }),
256 }),
257 ),
258 usage: Usage.pipe(Schema.NullOr, Schema.optional),
259});
260const ErrorBody = Schema.Struct({
261 error: Schema.Struct({
262 code: Schema.optional(Schema.Union([Schema.String, Schema.Number, Schema.Null])),
263 message: Schema.optional(Schema.String),
264 }),
265});
266
267const httpError = (
268 response: HttpClientResponse.HttpClientResponse,
269): Effect.Effect<never, AiError.AiError> =>
270 Effect.gen(function* () {
271 // Never retain upstream bodies, headers or request objects in public errors.
272 const body = yield* response.json.pipe(
273 Effect.flatMap(Schema.decodeUnknownEffect(ErrorBody)),
274 Effect.orElseSucceed(() => undefined),
275 );
276 const status = response.status;
277 if (status === 403 && body?.error.message === "ai gateway not enabled for account")
278 return yield* error(AiError.QuotaExhaustedError.make({}));
279 if (status === 401 || status === 403)
280 return yield* error(
281 AiError.AuthenticationError.make({
282 kind: status === 401 ? "InvalidKey" : "InsufficientPermissions",
283 }),
284 );
285 if (
286 status === 402 ||
287 body?.error.code === "REQUEST_LIMIT_EXCEEDED" ||
288 body?.error.code === "insufficient_quota"
289 )
290 return yield* error(AiError.QuotaExhaustedError.make({}));
291 if (status === 429) {
292 const seconds = Number(response.headers["retry-after"]);
293 return yield* error(
294 AiError.RateLimitError.make(
295 Number.isFinite(seconds) && seconds >= 0 ? { retryAfter: Duration.seconds(seconds) } : {},
296 ),
297 );
298 }
299 if (status >= 500)
300 return yield* error(
301 AiError.InternalProviderError.make({
302 description: `Neon AI Gateway returned HTTP ${status}`,
303 }),
304 );
305 return yield* invalidRequest(
306 `Neon AI Gateway rejected the request (HTTP ${status}); check the model, endpoint and parameters`,
307 );
308 });
309
310const finishReason = (reason: string | null | undefined): Response.FinishReason => {
311 switch (reason) {
312 case "stop":
313 return "stop";
314 case "length":
315 return "length";
316 case "tool_calls":
317 return "tool-calls";
318 case "content_filter":
319 return "content-filter";
320 case null:
321 case undefined:
322 return "unknown";
323 default:
324 return "other";
325 }
326};
327const usage = (value: typeof Usage.Type | null | undefined) => {
328 const input = value?.prompt_tokens;
329 const cached = value?.prompt_tokens_details?.cached_tokens;
330 const output = value?.completion_tokens;
331 const reasoning = value?.completion_tokens_details?.reasoning_tokens;
332 return Response.Usage.make({
333 inputTokens: {
334 total: input,
335 uncached: input === undefined ? undefined : Math.max(0, input - (cached ?? 0)),
336 cacheRead: cached,
337 cacheWrite: undefined,
338 },
339 outputTokens: {
340 total: output,
341 text: output === undefined ? undefined : Math.max(0, output - (reasoning ?? 0)),
342 reasoning,
343 },
344 });
345};
346
347const requestBody = (
348 options: LanguageModel.ProviderOptions,
349 model: string,
350 parameters: LanguageModelOptions["parameters"],
351 stream: boolean,
352) =>
353 Effect.gen(function* () {
354 if (parameters?.maxTokens !== undefined && parameters.maxCompletionTokens !== undefined)
355 return yield* invalidRequest("Set maxTokens or maxCompletionTokens, not both");
356 if (options.tools.some(Tool.isProviderDefined))
357 return yield* invalidRequest("Provider-executed tools require a native provider API");
358 const messages: Array<Record<string, unknown>> = [];
359 for (const message of options.prompt.content) {
360 if (message.role === "system") {
361 messages.push({ role: "system", content: message.content });
362 } else if (message.role === "user") {
363 const content: Array<unknown> = [];
364 for (const part of message.content) {
365 if (part.type === "text") content.push({ type: "text", text: part.text });
366 else if (part.type === "file" && part.mediaType.startsWith("image/")) {
367 const url = yield* Effect.sync(() => {
368 if (part.data instanceof URL) return part.data.toString();
369 if (part.data instanceof Uint8Array) {
370 let binary = "";
371 for (const byte of part.data) binary += String.fromCharCode(byte);
372 return `data:${part.mediaType};base64,${btoa(binary)}`;
373 }
374 return /^(data:|https?:)/.test(part.data)
375 ? part.data
376 : `data:${part.mediaType};base64,${part.data}`;
377 });
378 content.push({ type: "image_url", image_url: { url } });
379 } else return yield* invalidRequest("Only text and image user inputs are supported");
380 }
381 messages.push({ role: "user", content });
382 } else if (message.role === "assistant") {
383 const text: Array<string> = [];
384 const toolCalls: Array<unknown> = [];
385 for (const part of message.content) {
386 if (part.type === "text") text.push(part.text);
387 else if (part.type === "tool-call")
388 toolCalls.push({
389 id: part.id,
390 type: "function",
391 function: {
392 name: part.name,
393 arguments: yield* stringify(part.params),
394 },
395 });
396 else if (part.type !== "reasoning")
397 return yield* invalidRequest("Unsupported assistant content for Chat Completions");
398 }
399 messages.push({
400 role: "assistant",
401 content: text.join("") || null,
402 ...(toolCalls.length ? { tool_calls: toolCalls } : {}),
403 });
404 } else {
405 for (const part of message.content) {
406 if (part.type !== "tool-result")
407 return yield* invalidRequest("Unsupported tool message for Chat Completions");
408 messages.push({
409 role: "tool",
410 tool_call_id: part.id,
411 content: typeof part.result === "string" ? part.result : yield* stringify(part.result),
412 });
413 }
414 }
415 }
416 const choice = options.toolChoice;
417 const selected =
418 typeof choice === "object" && "oneOf" in choice
419 ? options.tools.filter((tool) => choice.oneOf.includes(tool.name))
420 : options.tools;
421 const tools = yield* Effect.try({
422 try: () =>
423 selected.map((tool) => ({
424 type: "function",
425 function: {
426 name: tool.name,
427 description: Tool.getDescription(tool),
428 parameters: Tool.getJsonSchema(tool),
429 },
430 })),
431 catch: () => invalidRequest("Tool schema cannot be represented as JSON Schema"),
432 });
433 const responseFormat = options.responseFormat;
434 const jsonSchema =
435 responseFormat.type === "json"
436 ? yield* Effect.try({
437 try: () => toCodecOpenAI(responseFormat.schema).jsonSchema,
438 catch: () =>
439 error(
440 AiError.UnsupportedSchemaError.make({
441 description: "Schema is not supported by OpenAI structured output",
442 }),
443 ),
444 })
445 : undefined;
446 return {
447 model,
448 messages,
449 stream,
450 ...(stream ? { stream_options: { include_usage: true } } : {}),
451 ...(tools.length
452 ? {
453 tools,
454 tool_choice:
455 typeof choice === "object"
456 ? "tool" in choice
457 ? { type: "function", function: { name: choice.tool } }
458 : (choice.mode ?? "auto")
459 : choice,
460 }
461 : {}),
462 ...(jsonSchema
463 ? {
464 response_format: {
465 type: "json_schema",
466 json_schema: {
467 name: responseFormat.type === "json" ? responseFormat.objectName : "response",
468 schema: jsonSchema,
469 strict: true,
470 },
471 },
472 }
473 : {}),
474 temperature: parameters?.temperature,
475 max_tokens: parameters?.maxTokens,
476 max_completion_tokens: parameters?.maxCompletionTokens,
477 top_p: parameters?.topP,
478 seed: parameters?.seed,
479 frequency_penalty: parameters?.frequencyPenalty,
480 presence_penalty: parameters?.presencePenalty,
481 };
482 });
483const stringify = (value: unknown) =>
484 Effect.try({
485 try: () => JSON.stringify(value),
486 catch: () => invalidRequest("Prompt contains a non-JSON value"),
487 });
488
489const completionStream = (
490 response: HttpClientResponse.HttpClientResponse,
491): Stream.Stream<Response.StreamPartEncoded, AiError.AiError> =>
492 Stream.unwrap(
493 Effect.sync(() => {
494 let done = false;
495 let reason: string | undefined;
496 let tokenUsage: typeof Usage.Type | null | undefined;
497 let textStarted = false;
498 let reasoningStarted = false;
499 const calls = new Map<
500 number,
501 { id: string; name: string; arguments: string; started: boolean }
502 >();
503 const chunks = response.stream.pipe(
504 Stream.mapError(() => invalidOutput("Chat Completions stream transport failed")),
505 Stream.decodeText(),
506 Stream.pipeThroughChannel(Sse.decode<AiError.AiError, unknown>()),
507 Stream.mapError(() => invalidOutput("Invalid or interrupted SSE stream")),
508 Stream.takeUntil((event) => event.data === "[DONE]"),
509 Stream.mapEffect((event) =>
510 Effect.gen(function* () {
511 const parts: Array<Response.StreamPartEncoded> = [];
512 if (event.data === "[DONE]") {
513 done = true;
514 return parts;
515 }
516 if (!event.data) return parts;
517 if (event.event === "error")
518 return yield* error(
519 AiError.InternalProviderError.make({
520 description: "Neon AI Gateway stream reported an error",
521 }),
522 );
523 const chunk = yield* Schema.decodeEffect(Schema.fromJsonString(Chunk))(event.data).pipe(
524 Effect.mapError(() => invalidOutput("Invalid Chat Completions stream chunk")),
525 );
526 if (chunk.usage != null) tokenUsage = chunk.usage;
527 for (const choice of chunk.choices) {
528 if (choice.index !== 0) continue;
529 if (choice.finish_reason != null) reason = choice.finish_reason;
530 const delta = choice.delta;
531 if (delta.refusal)
532 return yield* error(
533 AiError.ContentPolicyError.make({
534 description: "Model refused the request",
535 }),
536 );
537 if (delta.reasoning_content) {
538 if (!reasoningStarted) {
539 reasoningStarted = true;
540 parts.push({ type: "reasoning-start", id: "reasoning" });
541 }
542 parts.push({
543 type: "reasoning-delta",
544 id: "reasoning",
545 delta: delta.reasoning_content,
546 });
547 }
548 if (delta.content) {
549 if (!textStarted) {
550 textStarted = true;
551 parts.push({ type: "text-start", id: "text" });
552 }
553 parts.push({
554 type: "text-delta",
555 id: "text",
556 delta: delta.content,
557 });
558 }
559 for (const deltaCall of delta.tool_calls ?? []) {
560 const call = calls.get(deltaCall.index) ?? {
561 id: "",
562 name: "",
563 arguments: "",
564 started: false,
565 };
566 if (deltaCall.id) {
567 if (call.id && call.id !== deltaCall.id)
568 return yield* invalidOutput("Tool call ID changed within a stream");
569 call.id = deltaCall.id;
570 }
571 if (deltaCall.function?.name) {
572 if (call.started && call.name !== deltaCall.function.name)
573 return yield* invalidOutput("Tool name changed within a stream");
574 call.name = deltaCall.function.name;
575 }
576 const args = deltaCall.function?.arguments ?? "";
577 call.arguments += args;
578 if (!call.started && call.id && call.name) {
579 call.started = true;
580 parts.push({
581 type: "tool-params-start",
582 id: call.id,
583 name: call.name,
584 });
585 if (call.arguments)
586 parts.push({
587 type: "tool-params-delta",
588 id: call.id,
589 delta: call.arguments,
590 });
591 } else if (call.started && args)
592 parts.push({
593 type: "tool-params-delta",
594 id: call.id,
595 delta: args,
596 });
597 calls.set(deltaCall.index, call);
598 }
599 }
600 return parts;
601 }),
602 ),
603 Stream.flatMap(Stream.fromIterable),
604 );
605 const finish = Stream.unwrap(
606 Effect.gen(function* () {
607 if (!done || !reason)
608 return yield* invalidOutput(
609 "Chat Completions stream ended before its finish reason and [DONE]",
610 );
611 const parts: Array<Response.StreamPartEncoded> = [];
612 if (reasoningStarted) parts.push({ type: "reasoning-end", id: "reasoning" });
613 if (textStarted) parts.push({ type: "text-end", id: "text" });
614 for (const call of calls.values()) {
615 if (!call.started) return yield* invalidOutput("Incomplete streamed tool identity");
616 parts.push({ type: "tool-params-end", id: call.id });
617 if (reason !== "length" && reason !== "content_filter") {
618 parts.push({
619 type: "tool-call",
620 id: call.id,
621 name: call.name,
622 params: yield* parseArguments(call.arguments),
623 });
624 }
625 }
626 parts.push({
627 type: "finish",
628 reason: finishReason(reason),
629 usage: usage(tokenUsage),
630 });
631 return Stream.fromIterable(parts);
632 }),
633 );
634 return Stream.concat(chunks, finish);
635 }),
636 );