mirror of
https://github.com/RooVetGit/Roo-Code.git
synced 2026-09-11 22:51:26 +00:00
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.
This commit is contained in:
Roo Code
parent
88f9d0fb06
commit
d068136a1c
3 changed files with 350 additions and 584 deletions
|
|
@ -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<typeof createVertexAnthropic>
|
||||
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<string, unknown> = {}
|
||||
|
||||
// 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<number>()
|
||||
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<typeof streamText>[0] = {
|
||||
model: this.provider(modelConfig.id),
|
||||
system: systemPrompt,
|
||||
...({
|
||||
systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
|
||||
} as Record<string, unknown>),
|
||||
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<string, Record<string, unknown>>,
|
||||
): 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<string, Record<string, unknown>> }[],
|
||||
targetOriginalIndices: Set<number>,
|
||||
cacheProviderOption: Record<string, Record<string, unknown>>,
|
||||
): 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<typeof streamText>[0]["model"] {
|
||||
return this.provider(id)
|
||||
}
|
||||
|
||||
getModel() {
|
||||
|
|
@ -364,50 +122,4 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple
|
|||
...params,
|
||||
}
|
||||
}
|
||||
|
||||
async completePrompt(prompt: string): Promise<string> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<typeof createAnthropic>
|
||||
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<string, unknown> = {}
|
||||
|
||||
// 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<number>()
|
||||
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<typeof streamText>[0] = {
|
||||
model: this.provider(modelConfig.id),
|
||||
system: systemPrompt,
|
||||
...({
|
||||
systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
|
||||
} as Record<string, unknown>),
|
||||
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<string, Record<string, unknown>>,
|
||||
): 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<string, Record<string, unknown>> }[],
|
||||
targetOriginalIndices: Set<number>,
|
||||
cacheProviderOption: Record<string, Record<string, unknown>>,
|
||||
): 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<typeof streamText>[0]["model"] {
|
||||
return this.provider(id)
|
||||
}
|
||||
|
||||
getModel() {
|
||||
|
|
@ -329,50 +93,4 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa
|
|||
...params,
|
||||
}
|
||||
}
|
||||
|
||||
async completePrompt(prompt: string): Promise<string> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
336
src/api/providers/base-anthropic-ai-sdk-handler.ts
Normal file
336
src/api/providers/base-anthropic-ai-sdk-handler.ts
Normal file
|
|
@ -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<typeof streamText>[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<string, unknown> = {}
|
||||
|
||||
// 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<number>()
|
||||
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<typeof streamText>[0] = {
|
||||
model: this.getProviderModel(modelConfig.id),
|
||||
system: systemPrompt,
|
||||
...({
|
||||
systemProviderOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
|
||||
} as Record<string, unknown>),
|
||||
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<string, Record<string, unknown>>,
|
||||
): 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<string, Record<string, unknown>> }[],
|
||||
targetOriginalIndices: Set<number>,
|
||||
cacheProviderOption: Record<string, Record<string, unknown>>,
|
||||
): 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<string> {
|
||||
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
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue