diff --git a/packages/llm/src/provider/anthropic-messages.ts b/packages/llm/src/provider/anthropic-messages.ts index 7e9c2c174d..d39068d258 100644 --- a/packages/llm/src/provider/anthropic-messages.ts +++ b/packages/llm/src/provider/anthropic-messages.ts @@ -1,11 +1,9 @@ -import { Cause, Effect, Schema, Stream } from "effect" -import * as Sse from "effect/unstable/encoding/Sse" +import { Effect, Schema, Stream } from "effect" import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http" import { Adapter } from "../adapter" import { capabilities, model as llmModel, type ModelInput } from "../llm" import { InvalidRequestError, - ProviderChunkError, Usage, type CacheHint, type FinishReason, @@ -16,6 +14,9 @@ import { type ToolDefinition, type ToolResultPart, } from "../schema" +import { ProviderShared } from "./shared" + +const ADAPTER = "anthropic-messages" export type AnthropicMessagesModelInput = Omit & { readonly apiKey?: string @@ -151,9 +152,6 @@ interface ParserState { readonly usage?: Usage } -const Json = Schema.fromJsonString(Schema.Unknown) -const decodeJson = Schema.decodeUnknownSync(Json) -const encodeJson = Schema.encodeSync(Json) const AnthropicChunkJson = Schema.fromJsonString(AnthropicChunk) const AnthropicTargetJson = Schema.fromJsonString(AnthropicMessagesTarget) const decodeChunk = Schema.decodeUnknownSync(AnthropicChunkJson) @@ -170,7 +168,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p const resultText = (part: ToolResultPart) => { if (part.result.type === "text" || part.result.type === "error") return String(part.result.value) - return encodeJson(part.result.value) + return ProviderShared.encodeJson(part.result.value) } const lowerTool = (tool: ToolDefinition): AnthropicTool => ({ @@ -327,37 +325,13 @@ const mergeUsage = (left: Usage | undefined, right: Usage | undefined) => { }) } -const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "anthropic-messages", message, raw }) - -const streamError = (cause: Cause.Cause) => { - const failed = cause.reasons.find(Cause.isFailReason)?.error - if (failed instanceof ProviderChunkError) return failed - return chunkError("Failed to read Anthropic Messages stream", Cause.pretty(cause)) -} - -const parseJson = (input: string, message: string) => { - try { - return decodeJson(input) - } catch { - throw chunkError(message, input) - } -} - -const parseChunk = (data: string) => { - try { - return decodeChunk(data) - } catch { - throw chunkError("Invalid Anthropic Messages stream chunk", data) - } -} - const finishToolCall = (tool: ToolAccumulator | undefined) => { if (!tool) return [] return [{ type: "tool-call" as const, id: tool.id, name: tool.name, - input: parseJson(tool.input || "{}", `Invalid JSON input for Anthropic Messages tool call ${tool.name}`), + input: ProviderShared.parseJson(ADAPTER, tool.input || "{}", `Invalid JSON input for Anthropic Messages tool call ${tool.name}`), }] } @@ -400,7 +374,7 @@ const processChunk = (state: ParserState, chunk: AnthropicChunk): readonly [Pars if (chunk.type === "content_block_delta" && chunk.delta?.type === "input_json_delta" && chunk.index !== undefined) { if (!chunk.delta.partial_json) return [state, []] const current = state.tools[chunk.index] - if (!current) throw chunkError("Anthropic Messages tool argument delta is missing its tool call") + if (!current) throw ProviderShared.chunkError(ADAPTER, "Anthropic Messages tool argument delta is missing its tool call") const next = { ...current, input: `${current.input}${chunk.delta.partial_json ?? ""}` } return [{ ...state, tools: { ...state.tools, [chunk.index]: next } }, [ { type: "tool-input-delta", id: next.id, name: next.name, text: chunk.delta.partial_json ?? "" }, @@ -426,24 +400,18 @@ const processChunk = (state: ParserState, chunk: AnthropicChunk): readonly [Pars } const events = (response: HttpClientResponse.HttpClientResponse) => - response.stream.pipe( - Stream.mapError((error) => chunkError("Failed to read Anthropic Messages stream", String(error))), - Stream.decodeText(), - Stream.pipeThroughChannel(Sse.decode()), - Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"), - Stream.mapEffect((event) => - Effect.try({ - try: () => parseChunk(event.data), - catch: (error) => - error instanceof ProviderChunkError ? error : chunkError("Invalid Anthropic Messages stream chunk", event.data), - }), - ), - Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk), - Stream.catchCause((cause) => Stream.fail(streamError(cause))), - ) + ProviderShared.sse({ + adapter: ADAPTER, + response, + readError: "Failed to read Anthropic Messages stream", + invalidChunk: "Invalid Anthropic Messages stream chunk", + decodeChunk, + initial: (): ParserState => ({ tools: {} }), + process: processChunk, + }) export const adapter = Adapter.define({ - id: "anthropic-messages", + id: ADAPTER, protocol: "anthropic-messages", redact: (target) => target, prepare, diff --git a/packages/llm/src/provider/openai-chat.ts b/packages/llm/src/provider/openai-chat.ts index aff2a98914..e1bc5d5e0d 100644 --- a/packages/llm/src/provider/openai-chat.ts +++ b/packages/llm/src/provider/openai-chat.ts @@ -1,11 +1,9 @@ -import { Cause, Effect, Schema, Stream } from "effect" -import * as Sse from "effect/unstable/encoding/Sse" +import { Effect, Schema, Stream } from "effect" import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http" import { Adapter } from "../adapter" import { capabilities, model as llmModel, type ModelInput } from "../llm" import { InvalidRequestError, - ProviderChunkError, Usage, type FinishReason, type ContentPart, @@ -16,6 +14,9 @@ import { type ToolDefinition, type ToolResultPart, } from "../schema" +import { ProviderShared } from "./shared" + +const ADAPTER = "openai-chat" export type OpenAIChatModelInput = Omit & { readonly apiKey?: string @@ -131,9 +132,6 @@ const OpenAIChatChunk = Schema.Struct({ }) type OpenAIChatChunk = Schema.Schema.Type -const Json = Schema.fromJsonString(Schema.Unknown) -const decodeJson = Schema.decodeUnknownSync(Json) -const encodeJson = Schema.encodeSync(Json) const OpenAIChatChunkJson = Schema.fromJsonString(OpenAIChatChunk) const OpenAIChatTargetJson = Schema.fromJsonString(OpenAIChatTarget) const decodeChunk = Schema.decodeUnknownSync(OpenAIChatChunkJson) @@ -161,7 +159,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p const resultText = (part: ToolResultPart) => { if (part.result.type === "text" || part.result.type === "error") return String(part.result.value) - return encodeJson(part.result.value) + return ProviderShared.encodeJson(part.result.value) } const lowerTool = (tool: ToolDefinition): OpenAIChatTool => ({ @@ -188,7 +186,7 @@ const lowerToolCall = (part: ToolCallPart): OpenAIChatAssistantToolCall => ({ type: "function", function: { name: part.name, - arguments: encodeJson(part.input), + arguments: ProviderShared.encodeJson(part.input), }, }) @@ -286,35 +284,11 @@ const mapUsage = (usage: OpenAIChatChunk["usage"]): Usage | undefined => { }) } -const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "openai-chat", message, raw }) - -const streamError = (cause: Cause.Cause) => { - const failed = cause.reasons.find(Cause.isFailReason)?.error - if (failed instanceof ProviderChunkError) return failed - return chunkError("Failed to read OpenAI Chat stream", Cause.pretty(cause)) -} - -const parseJson = (input: string, message: string) => { - try { - return decodeJson(input) - } catch { - throw chunkError(message, input) - } -} - -const parseChunk = (data: string) => { - try { - return decodeChunk(data) - } catch { - throw chunkError("Invalid OpenAI Chat stream chunk", data) - } -} - const pushToolDelta = (tools: Record, delta: OpenAIChatToolCallDelta) => { const current = tools[delta.index] const id = delta.id ?? current?.id const name = delta.function?.name ?? current?.name - if (!id || !name) throw chunkError("OpenAI Chat tool call delta is missing id or name") + if (!id || !name) throw ProviderShared.chunkError(ADAPTER, "OpenAI Chat tool call delta is missing id or name") return { id, @@ -328,7 +302,7 @@ const finishToolCalls = (state: ParserState) => type: "tool-call" as const, id: tool.id, name: tool.name, - input: parseJson(tool.input || "{}", `Invalid JSON input for OpenAI Chat tool call ${tool.name}`), + input: ProviderShared.parseJson(ADAPTER, tool.input || "{}", `Invalid JSON input for OpenAI Chat tool call ${tool.name}`), })) const processChunk = (state: ParserState, chunk: OpenAIChatChunk): readonly [ParserState, ReadonlyArray] => { @@ -363,24 +337,19 @@ const finishEvents = (state: ParserState): ReadonlyArray => { } const events = (response: HttpClientResponse.HttpClientResponse) => - response.stream.pipe( - Stream.mapError((error) => chunkError("Failed to read OpenAI Chat stream", String(error))), - Stream.decodeText(), - Stream.pipeThroughChannel(Sse.decode()), - Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"), - Stream.mapEffect((event) => - Effect.try({ - try: () => parseChunk(event.data), - catch: (error) => - error instanceof ProviderChunkError ? error : chunkError("Invalid OpenAI Chat stream chunk", event.data), - }), - ), - Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk, { onHalt: finishEvents }), - Stream.catchCause((cause) => Stream.fail(streamError(cause))), - ) + ProviderShared.sse({ + adapter: ADAPTER, + response, + readError: "Failed to read OpenAI Chat stream", + invalidChunk: "Invalid OpenAI Chat stream chunk", + decodeChunk, + initial: (): ParserState => ({ tools: {} }), + process: processChunk, + onHalt: finishEvents, + }) export const adapter = Adapter.define({ - id: "openai-chat", + id: ADAPTER, protocol: "openai-chat", redact: (target) => target, prepare, diff --git a/packages/llm/src/provider/openai-responses.ts b/packages/llm/src/provider/openai-responses.ts index ae91dc6dd6..ef84085bdf 100644 --- a/packages/llm/src/provider/openai-responses.ts +++ b/packages/llm/src/provider/openai-responses.ts @@ -1,11 +1,9 @@ -import { Cause, Effect, Schema, Stream } from "effect" -import * as Sse from "effect/unstable/encoding/Sse" +import { Effect, Schema, Stream } from "effect" import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http" import { Adapter } from "../adapter" import { capabilities, model as llmModel, type ModelInput } from "../llm" import { InvalidRequestError, - ProviderChunkError, Usage, type FinishReason, type LLMEvent, @@ -15,6 +13,9 @@ import { type ToolDefinition, type ToolResultPart, } from "../schema" +import { ProviderShared } from "./shared" + +const ADAPTER = "openai-responses" export type OpenAIResponsesModelInput = Omit & { readonly apiKey?: string @@ -111,9 +112,6 @@ const OpenAIResponsesChunk = Schema.Struct({ }) type OpenAIResponsesChunk = Schema.Schema.Type -const Json = Schema.fromJsonString(Schema.Unknown) -const decodeJson = Schema.decodeUnknownSync(Json) -const encodeJson = Schema.encodeSync(Json) const OpenAIResponsesChunkJson = Schema.fromJsonString(OpenAIResponsesChunk) const OpenAIResponsesTargetJson = Schema.fromJsonString(OpenAIResponsesTarget) const decodeChunk = Schema.decodeUnknownSync(OpenAIResponsesChunkJson) @@ -138,7 +136,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p const resultText = (part: ToolResultPart) => { if (part.result.type === "text" || part.result.type === "error") return String(part.result.value) - return encodeJson(part.result.value) + return ProviderShared.encodeJson(part.result.value) } const lowerTool = (tool: ToolDefinition): OpenAIResponsesTool => ({ @@ -162,7 +160,7 @@ const lowerToolCall = (part: ToolCallPart): OpenAIResponsesInputItem => ({ type: "function_call", call_id: part.id, name: part.name, - arguments: encodeJson(part.input), + arguments: ProviderShared.encodeJson(part.input), }) const lowerMessages = Effect.fn("OpenAIResponses.lowerMessages")(function* (request: LLMRequest) { @@ -252,33 +250,9 @@ const mapFinishReason = (chunk: OpenAIResponsesChunk): FinishReason => { return "unknown" } -const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "openai-responses", message, raw }) - -const streamError = (cause: Cause.Cause) => { - const failed = cause.reasons.find(Cause.isFailReason)?.error - if (failed instanceof ProviderChunkError) return failed - return chunkError("Failed to read OpenAI Responses stream", Cause.pretty(cause)) -} - -const parseJson = (input: string, message: string) => { - try { - return decodeJson(input) - } catch { - throw chunkError(message, input) - } -} - -const parseChunk = (data: string) => { - try { - return decodeChunk(data) - } catch { - throw chunkError("Invalid OpenAI Responses stream chunk", data) - } -} - const pushToolDelta = (tools: Record, itemId: string, delta: string) => { const current = tools[itemId] - if (!current) throw chunkError("OpenAI Responses tool argument delta is missing its tool call") + if (!current) throw ProviderShared.chunkError(ADAPTER, "OpenAI Responses tool argument delta is missing its tool call") return { ...current, input: `${current.input}${delta}`, @@ -292,7 +266,7 @@ const finishToolCall = (tools: Record, item: NonNullabl type: "tool-call" as const, id: item.call_id, name: item.name, - input: parseJson(input || "{}", `Invalid JSON input for OpenAI Responses tool call ${item.name}`), + input: ProviderShared.parseJson(ADAPTER, input || "{}", `Invalid JSON input for OpenAI Responses tool call ${item.name}`), }] } @@ -337,24 +311,18 @@ const processChunk = (state: ParserState, chunk: OpenAIResponsesChunk): readonly } const events = (response: HttpClientResponse.HttpClientResponse) => - response.stream.pipe( - Stream.mapError((error) => chunkError("Failed to read OpenAI Responses stream", String(error))), - Stream.decodeText(), - Stream.pipeThroughChannel(Sse.decode()), - Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"), - Stream.mapEffect((event) => - Effect.try({ - try: () => parseChunk(event.data), - catch: (error) => - error instanceof ProviderChunkError ? error : chunkError("Invalid OpenAI Responses stream chunk", event.data), - }), - ), - Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk), - Stream.catchCause((cause) => Stream.fail(streamError(cause))), - ) + ProviderShared.sse({ + adapter: ADAPTER, + response, + readError: "Failed to read OpenAI Responses stream", + invalidChunk: "Invalid OpenAI Responses stream chunk", + decodeChunk, + initial: (): ParserState => ({ tools: {} }), + process: processChunk, + }) export const adapter = Adapter.define({ - id: "openai-responses", + id: ADAPTER, protocol: "openai-responses", redact: (target) => target, prepare, diff --git a/packages/llm/src/provider/shared.ts b/packages/llm/src/provider/shared.ts new file mode 100644 index 0000000000..5f3c149c60 --- /dev/null +++ b/packages/llm/src/provider/shared.ts @@ -0,0 +1,53 @@ +import { Cause, Effect, Schema, Stream } from "effect" +import * as Sse from "effect/unstable/encoding/Sse" +import type { HttpClientResponse } from "effect/unstable/http" +import { ProviderChunkError } from "../schema" + +export const Json = Schema.fromJsonString(Schema.Unknown) +export const decodeJson = Schema.decodeUnknownSync(Json) +export const encodeJson = Schema.encodeSync(Json) + +export const chunkError = (adapter: string, message: string, raw?: string) => + new ProviderChunkError({ adapter, message, raw }) + +export const parseJson = (adapter: string, input: string, message: string) => { + try { + return decodeJson(input) + } catch { + throw chunkError(adapter, message, input) + } +} + +const streamError = (adapter: string, message: string, cause: Cause.Cause) => { + const failed = cause.reasons.find(Cause.isFailReason)?.error + if (failed instanceof ProviderChunkError) return failed + return chunkError(adapter, message, Cause.pretty(cause)) +} + +export const sse = (input: { + readonly adapter: string + readonly response: HttpClientResponse.HttpClientResponse + readonly readError: string + readonly invalidChunk: string + readonly decodeChunk: (data: string) => Chunk + readonly initial: () => State + readonly process: (state: State, chunk: Chunk) => readonly [State, ReadonlyArray] + readonly onHalt?: (state: State) => ReadonlyArray +}): Stream.Stream => + input.response.stream.pipe( + Stream.mapError((error) => chunkError(input.adapter, input.readError, String(error))), + Stream.decodeText(), + Stream.pipeThroughChannel(Sse.decode()), + Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"), + Stream.mapEffect((event) => + Effect.try({ + try: () => input.decodeChunk(event.data), + catch: (error) => + error instanceof ProviderChunkError ? error : chunkError(input.adapter, input.invalidChunk, event.data), + }), + ), + Stream.mapAccum(input.initial, input.process, input.onHalt ? { onHalt: input.onHalt } : undefined), + Stream.catchCause((cause) => Stream.fail(streamError(input.adapter, input.readError, cause))), + ) + +export * as ProviderShared from "./shared"