From d068136a1cf796f8739340518824c1b7d72ba164 Mon Sep 17 00:00:00 2001 From: Roo Code Date: Mon, 9 Feb 2026 21:11:34 +0000 Subject: [PATCH] refactor: extract shared Anthropic AI SDK logic into BaseAnthropicAiSdkHandler Extract ~200 lines of duplicated logic between AnthropicHandler and AnthropicVertexHandler into a shared base class. The following methods now live in BaseAnthropicAiSdkHandler: - createMessage (stream processing, cache control, thinking capture) - processUsageMetrics (Anthropic cache token extraction) - applyCacheControlToAiSdkMessages (message-level cache placement) - completePrompt (generateText wrapper with error handling) - getThoughtSignature / getRedactedThinkingBlocks (thinking state) - isAiSdkProvider (returns true) Subclasses only provide constructor setup, getProviderModel(), and getModel() with provider-specific model resolution. --- src/api/providers/anthropic-vertex.ts | 300 +--------------- src/api/providers/anthropic.ts | 298 +--------------- .../base-anthropic-ai-sdk-handler.ts | 336 ++++++++++++++++++ 3 files changed, 350 insertions(+), 584 deletions(-) create mode 100644 src/api/providers/base-anthropic-ai-sdk-handler.ts diff --git a/src/api/providers/anthropic-vertex.ts b/src/api/providers/anthropic-vertex.ts index 685c8628b0..2606622402 100644 --- a/src/api/providers/anthropic-vertex.ts +++ b/src/api/providers/anthropic-vertex.ts @@ -1,43 +1,26 @@ -import type { Anthropic } from "@anthropic-ai/sdk" import { createVertexAnthropic } from "@ai-sdk/google-vertex/anthropic" -import { streamText, generateText, ToolSet } from "ai" +import { streamText } from "ai" import { type ModelInfo, type VertexModelId, vertexDefaultModelId, vertexModels, - ANTHROPIC_DEFAULT_MAX_TOKENS, VERTEX_1M_CONTEXT_MODEL_IDS, - ApiProviderError, } from "@roo-code/types" -import { TelemetryService } from "@roo-code/telemetry" import type { ApiHandlerOptions } from "../../shared/api" -import { shouldUseReasoningBudget } from "../../shared/api" -import type { ApiStream, ApiStreamUsageChunk } from "../transform/stream" import { getModelParams } from "../transform/model-params" -import { - convertToAiSdkMessages, - convertToolsForAiSdk, - processAiSdkStreamPart, - mapToolChoice, - handleAiSdkError, -} from "../transform/ai-sdk" -import { calculateApiCostAnthropic } from "../../shared/cost" import { DEFAULT_HEADERS } from "./constants" -import { BaseProvider } from "./base-provider" -import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata } from "../index" +import { BaseAnthropicAiSdkHandler } from "./base-anthropic-ai-sdk-handler" // https://docs.anthropic.com/en/api/claude-on-vertex-ai -export class AnthropicVertexHandler extends BaseProvider implements SingleCompletionHandler { +export class AnthropicVertexHandler extends BaseAnthropicAiSdkHandler { protected options: ApiHandlerOptions + protected readonly providerName = "Vertex (Anthropic)" private provider: ReturnType - private readonly providerName = "Vertex (Anthropic)" - private lastThoughtSignature: string | undefined - private lastRedactedThinkingBlocks: Array<{ type: "redacted_thinking"; data: string }> = [] constructor(options: ApiHandlerOptions) { super() @@ -83,233 +66,8 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple }) } - override async *createMessage( - systemPrompt: string, - messages: Anthropic.Messages.MessageParam[], - metadata?: ApiHandlerCreateMessageMetadata, - ): ApiStream { - const modelConfig = this.getModel() - - // Reset thinking state for this request - this.lastThoughtSignature = undefined - this.lastRedactedThinkingBlocks = [] - - // Convert messages to AI SDK format - const aiSdkMessages = convertToAiSdkMessages(messages) - - // Convert tools to AI SDK format - const openAiTools = this.convertToolsForOpenAI(metadata?.tools) - const aiSdkTools = convertToolsForAiSdk(openAiTools) as ToolSet | undefined - - // Build Anthropic provider options - const anthropicProviderOptions: Record = {} - - // Configure thinking/reasoning if the model supports it - const isThinkingEnabled = - shouldUseReasoningBudget({ model: modelConfig.info, settings: this.options }) && - modelConfig.reasoning && - modelConfig.reasoningBudget - - if (isThinkingEnabled) { - anthropicProviderOptions.thinking = { - type: "enabled", - budgetTokens: modelConfig.reasoningBudget, - } - } - - // Forward parallelToolCalls setting - // When parallelToolCalls is explicitly false, disable parallel tool use - if (metadata?.parallelToolCalls === false) { - anthropicProviderOptions.disableParallelToolUse = true - } - - /** - * Vertex API has specific limitations for prompt caching: - * 1. Maximum of 4 blocks can have cache_control - * 2. Only text blocks can be cached (images and other content types cannot) - * 3. Cache control can only be applied to user messages, not assistant messages - * - * Our caching strategy: - * - Cache the system prompt (1 block) - * - Cache the last text block of the second-to-last user message (1 block) - * - Cache the last text block of the last user message (1 block) - * This ensures we stay under the 4-block limit while maintaining effective caching - * for the most relevant context. - */ - const cacheProviderOption = { anthropic: { cacheControl: { type: "ephemeral" as const } } } - - const userMsgIndices = messages.reduce( - (acc, msg, index) => (msg.role === "user" ? [...acc, index] : acc), - [] as number[], - ) - - const targetIndices = new Set() - const lastUserMsgIndex = userMsgIndices[userMsgIndices.length - 1] ?? -1 - const secondLastUserMsgIndex = userMsgIndices[userMsgIndices.length - 2] ?? -1 - - if (lastUserMsgIndex >= 0) targetIndices.add(lastUserMsgIndex) - if (secondLastUserMsgIndex >= 0) targetIndices.add(secondLastUserMsgIndex) - - if (targetIndices.size > 0) { - this.applyCacheControlToAiSdkMessages(messages, aiSdkMessages, targetIndices, cacheProviderOption) - } - - // Build streamText request - // Cast providerOptions to any to bypass strict JSONObject typing — the AI SDK accepts the correct runtime values - const requestOptions: Parameters[0] = { - model: this.provider(modelConfig.id), - system: systemPrompt, - ...({ - systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, - } as Record), - messages: aiSdkMessages, - temperature: modelConfig.temperature, - maxOutputTokens: modelConfig.maxTokens ?? ANTHROPIC_DEFAULT_MAX_TOKENS, - tools: aiSdkTools, - toolChoice: mapToolChoice(metadata?.tool_choice), - ...(Object.keys(anthropicProviderOptions).length > 0 && { - providerOptions: { anthropic: anthropicProviderOptions } as any, - }), - } - - try { - const result = streamText(requestOptions) - - for await (const part of result.fullStream) { - // Capture thinking signature from stream events - // The AI SDK's @ai-sdk/anthropic emits the signature as a reasoning-delta - // event with providerMetadata.anthropic.signature - const partAny = part as any - if (partAny.providerMetadata?.anthropic?.signature) { - this.lastThoughtSignature = partAny.providerMetadata.anthropic.signature - } - - // Capture redacted thinking blocks from stream events - if (partAny.providerMetadata?.anthropic?.redactedData) { - this.lastRedactedThinkingBlocks.push({ - type: "redacted_thinking", - data: partAny.providerMetadata.anthropic.redactedData, - }) - } - - for (const chunk of processAiSdkStreamPart(part)) { - yield chunk - } - } - - // Yield usage metrics at the end, including cache metrics from providerMetadata - const usage = await result.usage - const providerMetadata = await result.providerMetadata - if (usage) { - yield this.processUsageMetrics(usage, modelConfig.info, providerMetadata) - } - } catch (error) { - const errorMessage = error instanceof Error ? error.message : String(error) - TelemetryService.instance.captureException( - new ApiProviderError(errorMessage, this.providerName, modelConfig.id, "createMessage"), - ) - throw handleAiSdkError(error, this.providerName) - } - } - - /** - * Process usage metrics from the AI SDK response, including Anthropic's cache metrics. - */ - private processUsageMetrics( - usage: { inputTokens?: number; outputTokens?: number }, - info: ModelInfo, - providerMetadata?: Record>, - ): ApiStreamUsageChunk { - const inputTokens = usage.inputTokens ?? 0 - const outputTokens = usage.outputTokens ?? 0 - - // Extract cache metrics from Anthropic's providerMetadata - const anthropicMeta = providerMetadata?.anthropic as - | { cacheCreationInputTokens?: number; cacheReadInputTokens?: number } - | undefined - const cacheWriteTokens = anthropicMeta?.cacheCreationInputTokens ?? 0 - const cacheReadTokens = anthropicMeta?.cacheReadInputTokens ?? 0 - - const { totalCost } = calculateApiCostAnthropic( - info, - inputTokens, - outputTokens, - cacheWriteTokens, - cacheReadTokens, - ) - - return { - type: "usage", - inputTokens, - outputTokens, - cacheWriteTokens: cacheWriteTokens > 0 ? cacheWriteTokens : undefined, - cacheReadTokens: cacheReadTokens > 0 ? cacheReadTokens : undefined, - totalCost, - } - } - - /** - * Apply cacheControl providerOptions to the correct AI SDK messages by walking - * the original Anthropic messages and converted AI SDK messages in parallel. - * - * convertToAiSdkMessages() can split a single Anthropic user message (containing - * tool_results + text) into 2 AI SDK messages (tool role + user role). This method - * accounts for that split so cache control lands on the right message. - */ - private applyCacheControlToAiSdkMessages( - originalMessages: Anthropic.Messages.MessageParam[], - aiSdkMessages: { role: string; providerOptions?: Record> }[], - targetOriginalIndices: Set, - cacheProviderOption: Record>, - ): void { - let aiSdkIdx = 0 - for (let origIdx = 0; origIdx < originalMessages.length; origIdx++) { - const origMsg = originalMessages[origIdx] - - if (typeof origMsg.content === "string") { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } else if (origMsg.role === "user") { - const hasToolResults = origMsg.content.some((part) => (part as { type: string }).type === "tool_result") - const hasNonToolContent = origMsg.content.some( - (part) => (part as { type: string }).type === "text" || (part as { type: string }).type === "image", - ) - - if (hasToolResults && hasNonToolContent) { - const userMsgIdx = aiSdkIdx + 1 - if (targetOriginalIndices.has(origIdx) && userMsgIdx < aiSdkMessages.length) { - aiSdkMessages[userMsgIdx].providerOptions = { - ...aiSdkMessages[userMsgIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx += 2 - } else if (hasToolResults) { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } else { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } - } else { - aiSdkIdx++ - } - } + protected getProviderModel(id: string): Parameters[0]["model"] { + return this.provider(id) } getModel() { @@ -364,50 +122,4 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple ...params, } } - - async completePrompt(prompt: string): Promise { - const { id, temperature } = this.getModel() - - try { - const { text } = await generateText({ - model: this.provider(id), - prompt, - maxOutputTokens: ANTHROPIC_DEFAULT_MAX_TOKENS, - temperature, - }) - - return text - } catch (error) { - TelemetryService.instance.captureException( - new ApiProviderError( - error instanceof Error ? error.message : String(error), - this.providerName, - id, - "completePrompt", - ), - ) - throw handleAiSdkError(error, this.providerName) - } - } - - /** - * Returns the thinking signature captured from the last Anthropic response. - * Claude models with extended thinking return a cryptographic signature - * which must be round-tripped back for multi-turn conversations with tool use. - */ - getThoughtSignature(): string | undefined { - return this.lastThoughtSignature - } - - /** - * Returns any redacted thinking blocks captured from the last Anthropic response. - * Anthropic returns these when safety filters trigger on reasoning content. - */ - getRedactedThinkingBlocks(): Array<{ type: "redacted_thinking"; data: string }> | undefined { - return this.lastRedactedThinkingBlocks.length > 0 ? this.lastRedactedThinkingBlocks : undefined - } - - override isAiSdkProvider(): boolean { - return true - } } diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index f6ee47e130..5a38afe07f 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -1,41 +1,19 @@ -import type { Anthropic } from "@anthropic-ai/sdk" import { createAnthropic } from "@ai-sdk/anthropic" -import { streamText, generateText, ToolSet } from "ai" +import { streamText } from "ai" -import { - type ModelInfo, - type AnthropicModelId, - anthropicDefaultModelId, - anthropicModels, - ANTHROPIC_DEFAULT_MAX_TOKENS, - ApiProviderError, -} from "@roo-code/types" -import { TelemetryService } from "@roo-code/telemetry" +import { type ModelInfo, type AnthropicModelId, anthropicDefaultModelId, anthropicModels } from "@roo-code/types" import type { ApiHandlerOptions } from "../../shared/api" -import { shouldUseReasoningBudget } from "../../shared/api" -import type { ApiStream, ApiStreamUsageChunk } from "../transform/stream" import { getModelParams } from "../transform/model-params" -import { - convertToAiSdkMessages, - convertToolsForAiSdk, - processAiSdkStreamPart, - mapToolChoice, - handleAiSdkError, -} from "../transform/ai-sdk" -import { calculateApiCostAnthropic } from "../../shared/cost" import { DEFAULT_HEADERS } from "./constants" -import { BaseProvider } from "./base-provider" -import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata } from "../index" +import { BaseAnthropicAiSdkHandler } from "./base-anthropic-ai-sdk-handler" -export class AnthropicHandler extends BaseProvider implements SingleCompletionHandler { - private options: ApiHandlerOptions +export class AnthropicHandler extends BaseAnthropicAiSdkHandler { + protected options: ApiHandlerOptions + protected readonly providerName = "Anthropic" private provider: ReturnType - private readonly providerName = "Anthropic" - private lastThoughtSignature: string | undefined - private lastRedactedThinkingBlocks: Array<{ type: "redacted_thinking"; data: string }> = [] constructor(options: ApiHandlerOptions) { super() @@ -70,222 +48,8 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa }) } - override async *createMessage( - systemPrompt: string, - messages: Anthropic.Messages.MessageParam[], - metadata?: ApiHandlerCreateMessageMetadata, - ): ApiStream { - const modelConfig = this.getModel() - - // Reset thinking state for this request - this.lastThoughtSignature = undefined - this.lastRedactedThinkingBlocks = [] - - // Convert messages to AI SDK format - const aiSdkMessages = convertToAiSdkMessages(messages) - - // Convert tools to AI SDK format - const openAiTools = this.convertToolsForOpenAI(metadata?.tools) - const aiSdkTools = convertToolsForAiSdk(openAiTools) as ToolSet | undefined - - // Build Anthropic provider options - const anthropicProviderOptions: Record = {} - - // Configure thinking/reasoning if the model supports it - const isThinkingEnabled = - shouldUseReasoningBudget({ model: modelConfig.info, settings: this.options }) && - modelConfig.reasoning && - modelConfig.reasoningBudget - - if (isThinkingEnabled) { - anthropicProviderOptions.thinking = { - type: "enabled", - budgetTokens: modelConfig.reasoningBudget, - } - } - - // Forward parallelToolCalls setting - // When parallelToolCalls is explicitly false, disable parallel tool use - if (metadata?.parallelToolCalls === false) { - anthropicProviderOptions.disableParallelToolUse = true - } - - // Apply cache control to user messages - // Strategy: cache the last 2 user messages (write-to-cache + read-from-cache) - const cacheProviderOption = { anthropic: { cacheControl: { type: "ephemeral" as const } } } - - const userMsgIndices = messages.reduce( - (acc, msg, index) => (msg.role === "user" ? [...acc, index] : acc), - [] as number[], - ) - - const targetIndices = new Set() - const lastUserMsgIndex = userMsgIndices[userMsgIndices.length - 1] ?? -1 - const secondLastUserMsgIndex = userMsgIndices[userMsgIndices.length - 2] ?? -1 - - if (lastUserMsgIndex >= 0) targetIndices.add(lastUserMsgIndex) - if (secondLastUserMsgIndex >= 0) targetIndices.add(secondLastUserMsgIndex) - - if (targetIndices.size > 0) { - this.applyCacheControlToAiSdkMessages(messages, aiSdkMessages, targetIndices, cacheProviderOption) - } - - // Build streamText request - // Cast providerOptions to any to bypass strict JSONObject typing — the AI SDK accepts the correct runtime values - const requestOptions: Parameters[0] = { - model: this.provider(modelConfig.id), - system: systemPrompt, - ...({ - systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, - } as Record), - messages: aiSdkMessages, - temperature: modelConfig.temperature, - maxOutputTokens: modelConfig.maxTokens ?? ANTHROPIC_DEFAULT_MAX_TOKENS, - tools: aiSdkTools, - toolChoice: mapToolChoice(metadata?.tool_choice), - ...(Object.keys(anthropicProviderOptions).length > 0 && { - providerOptions: { anthropic: anthropicProviderOptions } as any, - }), - } - - try { - const result = streamText(requestOptions) - - for await (const part of result.fullStream) { - // Capture thinking signature from stream events - // The AI SDK's @ai-sdk/anthropic emits the signature as a reasoning-delta - // event with providerMetadata.anthropic.signature - const partAny = part as any - if (partAny.providerMetadata?.anthropic?.signature) { - this.lastThoughtSignature = partAny.providerMetadata.anthropic.signature - } - - // Capture redacted thinking blocks from stream events - if (partAny.providerMetadata?.anthropic?.redactedData) { - this.lastRedactedThinkingBlocks.push({ - type: "redacted_thinking", - data: partAny.providerMetadata.anthropic.redactedData, - }) - } - - for (const chunk of processAiSdkStreamPart(part)) { - yield chunk - } - } - - // Yield usage metrics at the end, including cache metrics from providerMetadata - const usage = await result.usage - const providerMetadata = await result.providerMetadata - if (usage) { - yield this.processUsageMetrics(usage, modelConfig.info, providerMetadata) - } - } catch (error) { - const errorMessage = error instanceof Error ? error.message : String(error) - TelemetryService.instance.captureException( - new ApiProviderError(errorMessage, this.providerName, modelConfig.id, "createMessage"), - ) - throw handleAiSdkError(error, this.providerName) - } - } - - /** - * Process usage metrics from the AI SDK response, including Anthropic's cache metrics. - */ - private processUsageMetrics( - usage: { inputTokens?: number; outputTokens?: number }, - info: ModelInfo, - providerMetadata?: Record>, - ): ApiStreamUsageChunk { - const inputTokens = usage.inputTokens ?? 0 - const outputTokens = usage.outputTokens ?? 0 - - // Extract cache metrics from Anthropic's providerMetadata - const anthropicMeta = providerMetadata?.anthropic as - | { cacheCreationInputTokens?: number; cacheReadInputTokens?: number } - | undefined - const cacheWriteTokens = anthropicMeta?.cacheCreationInputTokens ?? 0 - const cacheReadTokens = anthropicMeta?.cacheReadInputTokens ?? 0 - - const { totalCost } = calculateApiCostAnthropic( - info, - inputTokens, - outputTokens, - cacheWriteTokens, - cacheReadTokens, - ) - - return { - type: "usage", - inputTokens, - outputTokens, - cacheWriteTokens: cacheWriteTokens > 0 ? cacheWriteTokens : undefined, - cacheReadTokens: cacheReadTokens > 0 ? cacheReadTokens : undefined, - totalCost, - } - } - - /** - * Apply cacheControl providerOptions to the correct AI SDK messages by walking - * the original Anthropic messages and converted AI SDK messages in parallel. - * - * convertToAiSdkMessages() can split a single Anthropic user message (containing - * tool_results + text) into 2 AI SDK messages (tool role + user role). This method - * accounts for that split so cache control lands on the right message. - */ - private applyCacheControlToAiSdkMessages( - originalMessages: Anthropic.Messages.MessageParam[], - aiSdkMessages: { role: string; providerOptions?: Record> }[], - targetOriginalIndices: Set, - cacheProviderOption: Record>, - ): void { - let aiSdkIdx = 0 - for (let origIdx = 0; origIdx < originalMessages.length; origIdx++) { - const origMsg = originalMessages[origIdx] - - if (typeof origMsg.content === "string") { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } else if (origMsg.role === "user") { - const hasToolResults = origMsg.content.some((part) => (part as { type: string }).type === "tool_result") - const hasNonToolContent = origMsg.content.some( - (part) => (part as { type: string }).type === "text" || (part as { type: string }).type === "image", - ) - - if (hasToolResults && hasNonToolContent) { - const userMsgIdx = aiSdkIdx + 1 - if (targetOriginalIndices.has(origIdx) && userMsgIdx < aiSdkMessages.length) { - aiSdkMessages[userMsgIdx].providerOptions = { - ...aiSdkMessages[userMsgIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx += 2 - } else if (hasToolResults) { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } else { - if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { - aiSdkMessages[aiSdkIdx].providerOptions = { - ...aiSdkMessages[aiSdkIdx].providerOptions, - ...cacheProviderOption, - } - } - aiSdkIdx++ - } - } else { - aiSdkIdx++ - } - } + protected getProviderModel(id: string): Parameters[0]["model"] { + return this.provider(id) } getModel() { @@ -329,50 +93,4 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa ...params, } } - - async completePrompt(prompt: string): Promise { - const { id, temperature } = this.getModel() - - try { - const { text } = await generateText({ - model: this.provider(id), - prompt, - maxOutputTokens: ANTHROPIC_DEFAULT_MAX_TOKENS, - temperature, - }) - - return text - } catch (error) { - TelemetryService.instance.captureException( - new ApiProviderError( - error instanceof Error ? error.message : String(error), - this.providerName, - id, - "completePrompt", - ), - ) - throw handleAiSdkError(error, this.providerName) - } - } - - /** - * Returns the thinking signature captured from the last Anthropic response. - * Claude models with extended thinking return a cryptographic signature - * which must be round-tripped back for multi-turn conversations with tool use. - */ - getThoughtSignature(): string | undefined { - return this.lastThoughtSignature - } - - /** - * Returns any redacted thinking blocks captured from the last Anthropic response. - * Anthropic returns these when safety filters trigger on reasoning content. - */ - getRedactedThinkingBlocks(): Array<{ type: "redacted_thinking"; data: string }> | undefined { - return this.lastRedactedThinkingBlocks.length > 0 ? this.lastRedactedThinkingBlocks : undefined - } - - override isAiSdkProvider(): boolean { - return true - } } diff --git a/src/api/providers/base-anthropic-ai-sdk-handler.ts b/src/api/providers/base-anthropic-ai-sdk-handler.ts new file mode 100644 index 0000000000..30047f28a1 --- /dev/null +++ b/src/api/providers/base-anthropic-ai-sdk-handler.ts @@ -0,0 +1,336 @@ +import type { Anthropic } from "@anthropic-ai/sdk" +import { streamText, generateText, ToolSet } from "ai" + +import { type ModelInfo, ANTHROPIC_DEFAULT_MAX_TOKENS, ApiProviderError } from "@roo-code/types" +import { TelemetryService } from "@roo-code/telemetry" + +import type { ApiHandlerOptions } from "../../shared/api" +import { shouldUseReasoningBudget } from "../../shared/api" + +import type { ApiStream, ApiStreamUsageChunk } from "../transform/stream" +import { + convertToAiSdkMessages, + convertToolsForAiSdk, + processAiSdkStreamPart, + mapToolChoice, + handleAiSdkError, +} from "../transform/ai-sdk" +import { calculateApiCostAnthropic } from "../../shared/cost" + +import { BaseProvider } from "./base-provider" +import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata } from "../index" + +/** + * Return type shared by all Anthropic-based getModel() implementations. + * Subclasses may return additional fields (e.g. `betas`) via intersection. + */ +export interface AnthropicModelConfig { + id: string + info: ModelInfo + temperature: number | undefined + maxTokens: number | undefined + reasoning: unknown + reasoningBudget: number | undefined +} + +/** + * Shared base class for Anthropic-based AI SDK providers. + * + * Both `AnthropicHandler` (direct API) and `AnthropicVertexHandler` (Vertex AI) + * use the Vercel AI SDK with Anthropic models. This base class extracts the + * common stream processing, cache control, usage metrics, thinking/reasoning + * capture, and error handling logic so it lives in one place. + * + * Subclasses only need to: + * - Set up their SDK provider in the constructor + * - Implement `getProviderModel()` to return the AI SDK language model + * - Implement `getModel()` with provider-specific model resolution + */ +export abstract class BaseAnthropicAiSdkHandler extends BaseProvider implements SingleCompletionHandler { + protected abstract options: ApiHandlerOptions + protected abstract readonly providerName: string + + private lastThoughtSignature: string | undefined + private lastRedactedThinkingBlocks: Array<{ type: "redacted_thinking"; data: string }> = [] + + /** + * Returns the AI SDK language model instance for the given model ID. + * Each subclass wraps its own SDK provider (e.g. `createAnthropic`, `createVertexAnthropic`). + */ + protected abstract getProviderModel(id: string): Parameters[0]["model"] + + abstract override getModel(): AnthropicModelConfig + + override async *createMessage( + systemPrompt: string, + messages: Anthropic.Messages.MessageParam[], + metadata?: ApiHandlerCreateMessageMetadata, + ): ApiStream { + const modelConfig = this.getModel() + + // Reset thinking state for this request + this.lastThoughtSignature = undefined + this.lastRedactedThinkingBlocks = [] + + // Convert messages to AI SDK format + const aiSdkMessages = convertToAiSdkMessages(messages) + + // Convert tools to AI SDK format + const openAiTools = this.convertToolsForOpenAI(metadata?.tools) + const aiSdkTools = convertToolsForAiSdk(openAiTools) as ToolSet | undefined + + // Build Anthropic provider options + const anthropicProviderOptions: Record = {} + + // Configure thinking/reasoning if the model supports it + const isThinkingEnabled = + shouldUseReasoningBudget({ model: modelConfig.info, settings: this.options }) && + modelConfig.reasoning && + modelConfig.reasoningBudget + + if (isThinkingEnabled) { + anthropicProviderOptions.thinking = { + type: "enabled", + budgetTokens: modelConfig.reasoningBudget, + } + } + + // Forward parallelToolCalls setting + // When parallelToolCalls is explicitly false, disable parallel tool use + if (metadata?.parallelToolCalls === false) { + anthropicProviderOptions.disableParallelToolUse = true + } + + /** + * Anthropic prompt caching strategy: + * - Cache the system prompt (1 block) + * - Cache the last text block of the second-to-last user message (1 block) + * - Cache the last text block of the last user message (1 block) + * This ensures we stay under the 4-block limit while maintaining effective caching + * for the most relevant context. + */ + const cacheProviderOption = { anthropic: { cacheControl: { type: "ephemeral" as const } } } + + const userMsgIndices = messages.reduce( + (acc, msg, index) => (msg.role === "user" ? [...acc, index] : acc), + [] as number[], + ) + + const targetIndices = new Set() + const lastUserMsgIndex = userMsgIndices[userMsgIndices.length - 1] ?? -1 + const secondLastUserMsgIndex = userMsgIndices[userMsgIndices.length - 2] ?? -1 + + if (lastUserMsgIndex >= 0) targetIndices.add(lastUserMsgIndex) + if (secondLastUserMsgIndex >= 0) targetIndices.add(secondLastUserMsgIndex) + + if (targetIndices.size > 0) { + this.applyCacheControlToAiSdkMessages(messages, aiSdkMessages, targetIndices, cacheProviderOption) + } + + // Build streamText request + // Cast providerOptions to any to bypass strict JSONObject typing — the AI SDK accepts the correct runtime values + const requestOptions: Parameters[0] = { + model: this.getProviderModel(modelConfig.id), + system: systemPrompt, + ...({ + systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } }, + } as Record), + messages: aiSdkMessages, + temperature: modelConfig.temperature, + maxOutputTokens: modelConfig.maxTokens ?? ANTHROPIC_DEFAULT_MAX_TOKENS, + tools: aiSdkTools, + toolChoice: mapToolChoice(metadata?.tool_choice), + ...(Object.keys(anthropicProviderOptions).length > 0 && { + providerOptions: { anthropic: anthropicProviderOptions } as any, + }), + } + + try { + const result = streamText(requestOptions) + + for await (const part of result.fullStream) { + // Capture thinking signature from stream events + // The AI SDK's @ai-sdk/anthropic emits the signature as a reasoning-delta + // event with providerMetadata.anthropic.signature + const partAny = part as any + if (partAny.providerMetadata?.anthropic?.signature) { + this.lastThoughtSignature = partAny.providerMetadata.anthropic.signature + } + + // Capture redacted thinking blocks from stream events + if (partAny.providerMetadata?.anthropic?.redactedData) { + this.lastRedactedThinkingBlocks.push({ + type: "redacted_thinking", + data: partAny.providerMetadata.anthropic.redactedData, + }) + } + + for (const chunk of processAiSdkStreamPart(part)) { + yield chunk + } + } + + // Yield usage metrics at the end, including cache metrics from providerMetadata + const usage = await result.usage + const providerMetadata = await result.providerMetadata + if (usage) { + yield this.processUsageMetrics(usage, modelConfig.info, providerMetadata) + } + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error) + TelemetryService.instance.captureException( + new ApiProviderError(errorMessage, this.providerName, modelConfig.id, "createMessage"), + ) + throw handleAiSdkError(error, this.providerName) + } + } + + /** + * Process usage metrics from the AI SDK response, including Anthropic's cache metrics. + */ + private processUsageMetrics( + usage: { inputTokens?: number; outputTokens?: number }, + info: ModelInfo, + providerMetadata?: Record>, + ): ApiStreamUsageChunk { + const inputTokens = usage.inputTokens ?? 0 + const outputTokens = usage.outputTokens ?? 0 + + // Extract cache metrics from Anthropic's providerMetadata + const anthropicMeta = providerMetadata?.anthropic as + | { cacheCreationInputTokens?: number; cacheReadInputTokens?: number } + | undefined + const cacheWriteTokens = anthropicMeta?.cacheCreationInputTokens ?? 0 + const cacheReadTokens = anthropicMeta?.cacheReadInputTokens ?? 0 + + const { totalCost } = calculateApiCostAnthropic( + info, + inputTokens, + outputTokens, + cacheWriteTokens, + cacheReadTokens, + ) + + return { + type: "usage", + inputTokens, + outputTokens, + cacheWriteTokens: cacheWriteTokens > 0 ? cacheWriteTokens : undefined, + cacheReadTokens: cacheReadTokens > 0 ? cacheReadTokens : undefined, + totalCost, + } + } + + /** + * Apply cacheControl providerOptions to the correct AI SDK messages by walking + * the original Anthropic messages and converted AI SDK messages in parallel. + * + * convertToAiSdkMessages() can split a single Anthropic user message (containing + * tool_results + text) into 2 AI SDK messages (tool role + user role). This method + * accounts for that split so cache control lands on the right message. + */ + private applyCacheControlToAiSdkMessages( + originalMessages: Anthropic.Messages.MessageParam[], + aiSdkMessages: { role: string; providerOptions?: Record> }[], + targetOriginalIndices: Set, + cacheProviderOption: Record>, + ): void { + let aiSdkIdx = 0 + for (let origIdx = 0; origIdx < originalMessages.length; origIdx++) { + const origMsg = originalMessages[origIdx] + + if (typeof origMsg.content === "string") { + if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { + aiSdkMessages[aiSdkIdx].providerOptions = { + ...aiSdkMessages[aiSdkIdx].providerOptions, + ...cacheProviderOption, + } + } + aiSdkIdx++ + } else if (origMsg.role === "user") { + const hasToolResults = origMsg.content.some( + (part: { type: string }) => (part as { type: string }).type === "tool_result", + ) + const hasNonToolContent = origMsg.content.some( + (part: { type: string }) => + (part as { type: string }).type === "text" || (part as { type: string }).type === "image", + ) + + if (hasToolResults && hasNonToolContent) { + const userMsgIdx = aiSdkIdx + 1 + if (targetOriginalIndices.has(origIdx) && userMsgIdx < aiSdkMessages.length) { + aiSdkMessages[userMsgIdx].providerOptions = { + ...aiSdkMessages[userMsgIdx].providerOptions, + ...cacheProviderOption, + } + } + aiSdkIdx += 2 + } else if (hasToolResults) { + if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { + aiSdkMessages[aiSdkIdx].providerOptions = { + ...aiSdkMessages[aiSdkIdx].providerOptions, + ...cacheProviderOption, + } + } + aiSdkIdx++ + } else { + if (targetOriginalIndices.has(origIdx) && aiSdkIdx < aiSdkMessages.length) { + aiSdkMessages[aiSdkIdx].providerOptions = { + ...aiSdkMessages[aiSdkIdx].providerOptions, + ...cacheProviderOption, + } + } + aiSdkIdx++ + } + } else { + aiSdkIdx++ + } + } + } + + async completePrompt(prompt: string): Promise { + const { id, temperature } = this.getModel() + + try { + const { text } = await generateText({ + model: this.getProviderModel(id), + prompt, + maxOutputTokens: ANTHROPIC_DEFAULT_MAX_TOKENS, + temperature, + }) + + return text + } catch (error) { + TelemetryService.instance.captureException( + new ApiProviderError( + error instanceof Error ? error.message : String(error), + this.providerName, + id, + "completePrompt", + ), + ) + throw handleAiSdkError(error, this.providerName) + } + } + + /** + * Returns the thinking signature captured from the last Anthropic response. + * Claude models with extended thinking return a cryptographic signature + * which must be round-tripped back for multi-turn conversations with tool use. + */ + getThoughtSignature(): string | undefined { + return this.lastThoughtSignature + } + + /** + * Returns any redacted thinking blocks captured from the last Anthropic response. + * Anthropic returns these when safety filters trigger on reasoning content. + */ + getRedactedThinkingBlocks(): Array<{ type: "redacted_thinking"; data: string }> | undefined { + return this.lastRedactedThinkingBlocks.length > 0 ? this.lastRedactedThinkingBlocks : undefined + } + + override isAiSdkProvider(): boolean { + return true + } +}