diff --git a/.changeset/swift-windows-appear.md b/.changeset/swift-windows-appear.md new file mode 100644 index 000000000..0b9f31af8 --- /dev/null +++ b/.changeset/swift-windows-appear.md @@ -0,0 +1,102 @@ +--- +"@voltagent/core": patch +--- + +feat: add retry/fallback hooks and middleware retry feedback + +### Model Retry and Fallback + +Configure ordered model candidates with per-model retry limits. + +```ts +import { Agent } from "@voltagent/core"; +import { anthropic } from "@ai-sdk/anthropic"; + +const agent = new Agent({ + name: "Support", + instructions: "Answer support questions with short, direct replies.", + model: [ + { id: "primary", model: "openai/gpt-4o-mini", maxRetries: 2 }, + { id: "fallback", model: anthropic("claude-3-5-sonnet"), maxRetries: 1 }, + ], +}); +``` + +- `maxRetries` is per model (total attempts = `maxRetries + 1`). +- If retries are exhausted, VoltAgent tries the next enabled model. + +### Middleware Retry Feedback + +Middleware can request a retry. The retry reason and metadata are added as a system +message for the next attempt. + +```ts +import { Agent, createOutputMiddleware } from "@voltagent/core"; + +const requireSignature = createOutputMiddleware({ + name: "RequireSignature", + handler: ({ output, abort }) => { + if (!output.includes("-- Support")) { + abort("Missing signature", { retry: true, metadata: { signature: "-- Support" } }); + } + return output; + }, +}); + +const agent = new Agent({ + name: "Support", + instructions: "Answer support questions with short, direct replies.", + model: "openai/gpt-4o-mini", + maxMiddlewareRetries: 1, + outputMiddlewares: [requireSignature], +}); +``` + +### Input Middleware Example + +Input middleware can rewrite user input before guardrails and hooks. + +```ts +import { Agent, createInputMiddleware } from "@voltagent/core"; + +const normalizeInput = createInputMiddleware({ + name: "NormalizeInput", + handler: ({ input }) => { + if (typeof input !== "string") return input; + return input.trim(); + }, +}); + +const agent = new Agent({ + name: "Support", + instructions: "Answer support questions with short, direct replies.", + model: "openai/gpt-4o-mini", + inputMiddlewares: [normalizeInput], +}); +``` + +### Retry and Fallback Hooks + +Track retries and fallbacks in hooks. `onRetry` runs for LLM and middleware retries. + +```ts +const agent = new Agent({ + name: "RetryHooks", + instructions: "Answer support questions with short, direct replies.", + model: "openai/gpt-4o-mini", + hooks: { + onRetry: async (args) => { + if (args.source === "llm") { + console.log(`LLM retry ${args.nextAttempt}/${args.maxRetries + 1} for ${args.modelName}`); + return; + } + console.log( + `Middleware retry ${args.retryCount + 1}/${args.maxRetries + 1} for ${args.middlewareId ?? "unknown"}` + ); + }, + onFallback: async ({ stage, fromModel, nextModel }) => { + console.log(`Fallback (${stage}) from ${fromModel} to ${nextModel ?? "next"}`); + }, + }, +}); +``` diff --git a/examples/README.md b/examples/README.md index 2e98161e4..fa48fd1fb 100644 --- a/examples/README.md +++ b/examples/README.md @@ -83,6 +83,8 @@ Create a multi-agent research workflow where different AI agents collaborate to ## All Examples - [Base Starter](./base) — Minimal VoltAgent starter with a single agent, memory, and dev server. +- [Retries and Fallbacks](./with-retries-fallback) — Model fallback list with per-model retries and agent-level defaults. +- [Middleware](./with-middleware) — Input/output middleware with retry feedback. - [PlanAgents](./with-planagents) — Quickstart for PlanAgents with planning, filesystem tools, and subagent tasks. - [Slack](./with-slack) — Slack app mention bot that replies in the same channel/thread via VoltOps Slack actions. - [Airtable](./with-airtable) — React to new Airtable records and write updates back using VoltOps Airtable actions. diff --git a/examples/with-middleware/.env.example b/examples/with-middleware/.env.example new file mode 100644 index 000000000..3cac181df --- /dev/null +++ b/examples/with-middleware/.env.example @@ -0,0 +1 @@ +OPENAI_API_KEY=your_openai_api_key_here diff --git a/examples/with-middleware/README.md b/examples/with-middleware/README.md new file mode 100644 index 000000000..4cd3a3c5f --- /dev/null +++ b/examples/with-middleware/README.md @@ -0,0 +1,28 @@ +# Middleware Example + +This example shows input and output middleware with retry feedback. + +## Scaffold + +```bash +npm create voltagent-app@latest -- --example with-middleware +``` + +## Run + +1. Set `OPENAI_API_KEY` in `.env`. +2. Install dependencies: + +```bash +pnpm install +``` + +3. Start the dev server: + +```bash +pnpm dev +``` + +## Files + +- `src/index.ts` registers input and output middlewares and enables middleware retries. diff --git a/examples/with-middleware/package.json b/examples/with-middleware/package.json new file mode 100644 index 000000000..4a9d0808a --- /dev/null +++ b/examples/with-middleware/package.json @@ -0,0 +1,38 @@ +{ + "name": "voltagent-example-with-middleware", + "author": "", + "dependencies": { + "@ai-sdk/openai": "^3.0.0", + "@voltagent/cli": "^0.1.21", + "@voltagent/core": "^2.1.2", + "@voltagent/libsql": "^2.0.2", + "@voltagent/logger": "^2.0.2", + "@voltagent/server-hono": "^2.0.3", + "ai": "^6.0.0", + "zod": "^3.25.76" + }, + "devDependencies": { + "@types/node": "^24.2.1", + "tsx": "^4.19.3", + "typescript": "^5.8.2" + }, + "keywords": [ + "agent", + "ai", + "voltagent" + ], + "license": "MIT", + "private": true, + "repository": { + "type": "git", + "url": "https://github.com/VoltAgent/voltagent.git", + "directory": "examples/with-middleware" + }, + "scripts": { + "build": "tsc", + "dev": "tsx watch --env-file=.env ./src", + "start": "node dist/index.js", + "volt": "volt" + }, + "type": "module" +} diff --git a/examples/with-middleware/src/index.ts b/examples/with-middleware/src/index.ts new file mode 100644 index 000000000..18fc9f851 --- /dev/null +++ b/examples/with-middleware/src/index.ts @@ -0,0 +1,61 @@ +import { + Agent, + Memory, + VoltAgent, + createInputMiddleware, + createOutputMiddleware, +} from "@voltagent/core"; +import { LibSQLMemoryAdapter } from "@voltagent/libsql"; +import { createPinoLogger } from "@voltagent/logger"; +import { honoServer } from "@voltagent/server-hono"; + +const logger = createPinoLogger({ + name: "with-middleware", + level: "info", +}); + +const memory = new Memory({ + storage: new LibSQLMemoryAdapter(), +}); + +const normalizeInput = createInputMiddleware({ + name: "NormalizeInput", + handler: ({ input }) => { + if (typeof input !== "string") { + return input; + } + return input.trim(); + }, +}); + +const requireSignature = createOutputMiddleware({ + name: "RequireSignature", + handler: ({ output, abort }) => { + if (!output.includes("-- Support")) { + abort( + 'Retry required. Respond again and end the response with "-- Support". Do not omit it.', + { + retry: true, + metadata: { signature: "-- Support" }, + }, + ); + } + return output; + }, +}); + +const agent = new Agent({ + name: "MiddlewareAgent", + instructions: "Answer support questions with short, direct replies. ", + model: "openai/gpt-4o-mini", + memory, + inputMiddlewares: [normalizeInput], + outputMiddlewares: [requireSignature], + maxMiddlewareRetries: 1, +}); + +new VoltAgent({ + agents: { agent }, + server: honoServer(), + logger, +}); diff --git a/examples/with-middleware/tsconfig.json b/examples/with-middleware/tsconfig.json new file mode 100644 index 000000000..cee90c6f3 --- /dev/null +++ b/examples/with-middleware/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "strict": true, + "outDir": "dist", + "skipLibCheck": true + }, + "include": ["src"], + "exclude": ["node_modules", "dist"] +} diff --git a/examples/with-retries-fallback/.env.example b/examples/with-retries-fallback/.env.example new file mode 100644 index 000000000..121e76c45 --- /dev/null +++ b/examples/with-retries-fallback/.env.example @@ -0,0 +1 @@ +OPENAI_API_KEY=your_openai_api_key_here \ No newline at end of file diff --git a/examples/with-retries-fallback/.gitignore b/examples/with-retries-fallback/.gitignore new file mode 100644 index 000000000..0ca39c007 --- /dev/null +++ b/examples/with-retries-fallback/.gitignore @@ -0,0 +1,3 @@ +node_modules +dist +.DS_Store diff --git a/examples/with-retries-fallback/README.md b/examples/with-retries-fallback/README.md new file mode 100644 index 000000000..21d68b841 --- /dev/null +++ b/examples/with-retries-fallback/README.md @@ -0,0 +1,28 @@ +# Retries and Fallbacks Example + +This example configures a fallback model list with per-model and agent-level retry settings. + +## Scaffold + +```bash +npm create voltagent-app@latest -- --example with-retries-fallback +``` + +## Run + +1. Set `OPENAI_API_KEY` in `.env`. +2. Install dependencies: + +```bash +pnpm install +``` + +3. Start the dev server: + +```bash +pnpm dev +``` + +## Files + +- `src/index.ts` sets `model` to a list and configures `maxRetries`. diff --git a/examples/with-retries-fallback/package.json b/examples/with-retries-fallback/package.json new file mode 100644 index 000000000..f67988610 --- /dev/null +++ b/examples/with-retries-fallback/package.json @@ -0,0 +1,38 @@ +{ + "name": "voltagent-example-with-retries-fallback", + "author": "", + "dependencies": { + "@ai-sdk/openai": "^3.0.0", + "@voltagent/cli": "^0.1.21", + "@voltagent/core": "^2.1.2", + "@voltagent/libsql": "^2.0.2", + "@voltagent/logger": "^2.0.2", + "@voltagent/server-hono": "^2.0.3", + "ai": "^6.0.0", + "zod": "^3.25.76" + }, + "devDependencies": { + "@types/node": "^24.2.1", + "tsx": "^4.19.3", + "typescript": "^5.8.2" + }, + "keywords": [ + "agent", + "ai", + "voltagent" + ], + "license": "MIT", + "private": true, + "repository": { + "type": "git", + "url": "https://github.com/VoltAgent/voltagent.git", + "directory": "examples/with-retries-fallback" + }, + "scripts": { + "build": "tsc", + "dev": "tsx watch --env-file=.env ./src", + "start": "node dist/index.js", + "volt": "volt" + }, + "type": "module" +} diff --git a/examples/with-retries-fallback/src/index.ts b/examples/with-retries-fallback/src/index.ts new file mode 100644 index 000000000..fbfcbc750 --- /dev/null +++ b/examples/with-retries-fallback/src/index.ts @@ -0,0 +1,27 @@ +import { Agent, Memory, VoltAgent } from "@voltagent/core"; +import { LibSQLMemoryAdapter } from "@voltagent/libsql"; +import { createPinoLogger } from "@voltagent/logger"; +import { honoServer } from "@voltagent/server-hono"; + +const logger = createPinoLogger({ + name: "with-retries-fallback", + level: "info", +}); + +const memory = new Memory({ + storage: new LibSQLMemoryAdapter(), +}); + +const agent = new Agent({ + name: "RetriesFallbackAgent", + instructions: "Answer support questions with short, direct replies.", + model: "openai/gpt-4o-mini", + maxRetries: 1, + memory, +}); + +new VoltAgent({ + agents: { agent }, + server: honoServer(), + logger, +}); diff --git a/examples/with-retries-fallback/tsconfig.json b/examples/with-retries-fallback/tsconfig.json new file mode 100644 index 000000000..cee90c6f3 --- /dev/null +++ b/examples/with-retries-fallback/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "esModuleInterop": true, + "forceConsistentCasingInFileNames": true, + "strict": true, + "outDir": "dist", + "skipLibCheck": true + }, + "include": ["src"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/core/src/agent/agent.spec-d.ts b/packages/core/src/agent/agent.spec-d.ts index 86ce71cd3..c40a8f909 100644 --- a/packages/core/src/agent/agent.spec-d.ts +++ b/packages/core/src/agent/agent.spec-d.ts @@ -287,6 +287,12 @@ describe("Agent Type System", () => { const dynamicModel: AgentModelValue = async () => mockModel; expectTypeOf(dynamicModel).toMatchTypeOf(); + + const fallbackModels: AgentModelValue = [ + { model: "openai/gpt-4o-mini", maxRetries: 2 }, + { model: async () => mockModel, enabled: false }, + ]; + expectTypeOf(fallbackModels).toMatchTypeOf(); }); it("should handle ToolsDynamicValue", () => { diff --git a/packages/core/src/agent/agent.spec.ts b/packages/core/src/agent/agent.spec.ts index e18a62c26..d4ef6c357 100644 --- a/packages/core/src/agent/agent.spec.ts +++ b/packages/core/src/agent/agent.spec.ts @@ -1524,7 +1524,291 @@ describe("Agent", () => { }); }); + describe("Middleware", () => { + it("runs input middleware before input guardrails", async () => { + const inputMiddleware = ({ input }: { input: string | UIMessage[] }) => { + if (typeof input === "string") { + return `${input}-middleware`; + } + return input; + }; + + const inputGuardrail = ({ input }: { input: string | UIMessage[] }) => { + if (input !== "hello-middleware") { + throw new Error("Guardrail saw unexpected input"); + } + return { pass: true }; + }; + + const agent = new Agent({ + name: "MiddlewareAgent", + instructions: "Test", + model: mockModel as any, + inputMiddlewares: [inputMiddleware], + inputGuardrails: [inputGuardrail as any], + }); + + vi.mocked(ai.generateText).mockResolvedValue({ + text: "ok", + content: [{ type: "text", text: "ok" }], + reasoning: [], + files: [], + sources: [], + toolCalls: [], + toolResults: [], + finishReason: "stop", + usage: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + warnings: [], + request: {}, + response: { + id: "test-response", + modelId: "test-model", + timestamp: new Date(), + messages: [], + }, + steps: [], + } as any); + + const result = await agent.generateText("hello"); + expect(result.text).toBe("ok"); + }); + + it("runs output middleware before output guardrails", async () => { + const outputMiddleware = ({ output }: { output: string }) => `${output}-middleware`; + const outputGuardrail = ({ output }: { output: string }) => { + if (output !== "base-middleware") { + throw new Error("Guardrail saw unexpected output"); + } + return { pass: true }; + }; + + const agent = new Agent({ + name: "MiddlewareAgent", + instructions: "Test", + model: mockModel as any, + outputMiddlewares: [outputMiddleware], + outputGuardrails: [outputGuardrail as any], + }); + + vi.mocked(ai.generateText).mockResolvedValue({ + text: "base", + content: [{ type: "text", text: "base" }], + reasoning: [], + files: [], + sources: [], + toolCalls: [], + toolResults: [], + finishReason: "stop", + usage: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + warnings: [], + request: {}, + response: { + id: "test-response", + modelId: "test-model", + timestamp: new Date(), + messages: [], + }, + steps: [], + } as any); + + const result = await agent.generateText("hello"); + expect(result.text).toBe("base-middleware"); + }); + + it("retries when middleware requests retry", async () => { + const outputMiddleware = ({ + output, + retryCount, + abort, + }: { + output: string; + retryCount: number; + abort: (reason?: string, options?: { retry?: boolean }) => never; + }) => { + if (retryCount === 0) { + abort("retry", { retry: true }); + } + return `${output}-ok`; + }; + + const agent = new Agent({ + name: "MiddlewareRetryAgent", + instructions: "Test", + model: mockModel as any, + outputMiddlewares: [outputMiddleware], + maxMiddlewareRetries: 1, + }); + + vi.mocked(ai.generateText).mockResolvedValue({ + text: "base", + content: [{ type: "text", text: "base" }], + reasoning: [], + files: [], + sources: [], + toolCalls: [], + toolResults: [], + finishReason: "stop", + usage: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + warnings: [], + request: {}, + response: { + id: "test-response", + modelId: "test-model", + timestamp: new Date(), + messages: [], + }, + steps: [], + } as any); + + const result = await agent.generateText("hello"); + expect(result.text).toBe("base-ok"); + expect(vi.mocked(ai.generateText)).toHaveBeenCalledTimes(2); + }); + }); + describe("Error Handling", () => { + it("should fall back to the next model when the primary fails", async () => { + const fallbackModel = new MockLanguageModelV3({ + modelId: "fallback-model", + doGenerate: { + content: [{ type: "text", text: "Fallback response" }], + finishReason: "stop", + usage: { + inputTokens: 10, + outputTokens: 5, + totalTokens: 15, + inputTokenDetails: { + noCacheTokens: 10, + cacheReadTokens: 0, + cacheWriteTokens: 0, + }, + outputTokenDetails: { + textTokens: 5, + reasoningTokens: 0, + }, + }, + warnings: [], + }, + }); + + const agent = new Agent({ + name: "TestAgent", + instructions: "Test", + model: [ + { model: mockModel as any, maxRetries: 0 }, + { model: fallbackModel as any, maxRetries: 1 }, + ], + }); + + const mockResponse = { + text: "Fallback response", + content: [{ type: "text", text: "Fallback response" }], + reasoning: [], + files: [], + sources: [], + toolCalls: [], + toolResults: [], + finishReason: "stop", + usage: { + inputTokens: 10, + outputTokens: 5, + totalTokens: 15, + }, + warnings: [], + request: {}, + response: { + id: "test-response", + modelId: "fallback-model", + timestamp: new Date(), + messages: [], + }, + steps: [], + }; + + vi.mocked(ai.generateText).mockImplementation(async (args: any) => { + if (args.model === mockModel) { + throw new Error("Primary model failed"); + } + if (args.model === fallbackModel) { + return mockResponse as any; + } + throw new Error("Unexpected model"); + }); + + const result = await agent.generateText("Test"); + + expect(result.text).toBe("Fallback response"); + expect(vi.mocked(ai.generateText)).toHaveBeenCalledTimes(2); + expect(vi.mocked(ai.generateText).mock.calls[0][0].model).toBe(mockModel); + expect(vi.mocked(ai.generateText).mock.calls[1][0].model).toBe(fallbackModel); + expect(vi.mocked(ai.generateText).mock.calls[1][0].maxRetries).toBe(0); + }); + + it("should retry the same model before returning a response", async () => { + const agent = new Agent({ + name: "RetryAgent", + instructions: "Test", + model: mockModel as any, + maxRetries: 2, + }); + + const mockResponse = { + text: "Retry response", + content: [{ type: "text", text: "Retry response" }], + reasoning: [], + files: [], + sources: [], + toolCalls: [], + toolResults: [], + finishReason: "stop", + usage: { + inputTokens: 10, + outputTokens: 5, + totalTokens: 15, + }, + warnings: [], + request: {}, + response: { + id: "retry-response", + modelId: "test-model", + timestamp: new Date(), + messages: [], + }, + steps: [], + }; + + let callCount = 0; + vi.mocked(ai.generateText).mockImplementation(async () => { + callCount += 1; + if (callCount < 3) { + const error = new Error("Transient error"); + (error as any).isRetryable = true; + throw error; + } + return mockResponse as any; + }); + + const result = await agent.generateText("Test"); + + expect(result.text).toBe("Retry response"); + expect(vi.mocked(ai.generateText)).toHaveBeenCalledTimes(3); + for (const call of vi.mocked(ai.generateText).mock.calls) { + expect(call[0].maxRetries).toBe(0); + } + }); + it("should handle model errors gracefully", async () => { const agent = new Agent({ name: "TestAgent", @@ -1939,6 +2223,16 @@ describe("Agent", () => { expect(agent.getModelName()).toBe("test-model"); }); + it("should get model name from fallback list", () => { + const agent = new Agent({ + name: "TestAgent", + instructions: "Test", + model: [{ model: mockModel as any }, { model: "mock/secondary" }], + }); + + expect(agent.getModelName()).toBe("test-model"); + }); + it("should unregister agent", () => { const agent = new Agent({ name: "TestAgent", diff --git a/packages/core/src/agent/agent.ts b/packages/core/src/agent/agent.ts index e8fb7a875..fa16fa080 100644 --- a/packages/core/src/agent/agent.ts +++ b/packages/core/src/agent/agent.ts @@ -28,6 +28,7 @@ import { type LanguageModelUsage, type Output, type Warning, + consumeStream, convertToModelMessages, createTextStreamResponse, createUIMessageStream, @@ -69,7 +70,9 @@ import { createVoltAgentError, isBailError, isClientHTTPError, + isMiddlewareAbortError, isToolDeniedError, + isVoltAgentError, } from "./errors"; import { type AgentEvalHost, @@ -107,6 +110,14 @@ import { MemoryPersistQueue, } from "./memory-persist-queue"; import { sanitizeMessagesForModel } from "./message-normalizer"; +import { + type NormalizedInputMiddleware, + type NormalizedOutputMiddleware, + normalizeInputMiddlewareList, + normalizeOutputMiddlewareList, + runInputMiddlewares, + runOutputMiddlewares, +} from "./middleware"; import { type GuardrailPipeline, createAsyncIterableReadable, @@ -122,21 +133,26 @@ import type { AgentFeedbackOptions, AgentFullState, AgentGuardrailState, + AgentModelConfig, AgentModelValue, AgentOptions, AgentSummarizationOptions, DynamicValue, DynamicValueOptions, InputGuardrail, + InputMiddleware, InstructionsDynamicValue, OperationContext, OutputGuardrail, + OutputMiddleware, SupervisorConfig, } from "./types"; const BUFFER_CONTEXT_KEY = Symbol("conversationBuffer"); const QUEUE_CONTEXT_KEY = Symbol("memoryPersistQueue"); const STEP_PERSIST_COUNT_KEY = Symbol("persistedStepCount"); +const ABORT_LISTENER_ATTACHED_KEY = Symbol("abortListenerAttached"); +const MIDDLEWARE_RETRY_FEEDBACK_KEY = Symbol("middlewareRetryFeedback"); const DEFAULT_FEEDBACK_KEY = "satisfaction"; // ============================================================================ @@ -323,6 +339,7 @@ function createDeferred(): Deferred { } const asyncGeneratorFunction = Object.getPrototypeOf(async function* () {}).constructor; +const DEFAULT_LLM_MAX_RETRIES = 3; function isAsyncGeneratorFunction( value: unknown, @@ -387,6 +404,11 @@ export interface BaseGenerationOptions extends Partial { inputGuardrails?: InputGuardrail[]; outputGuardrails?: OutputGuardrail[]; + // Middleware (can override agent-level middlewares) + inputMiddlewares?: InputMiddleware[]; + outputMiddlewares?: OutputMiddleware[]; + maxMiddlewareRetries?: number; + // Provider-specific options providerOptions?: ProviderOptions; @@ -443,6 +465,7 @@ export class Agent { readonly temperature?: number; readonly maxOutputTokens?: number; readonly maxSteps: number; + readonly maxRetries: number; readonly stopWhen?: StopWhen; readonly markdown: boolean; readonly voice?: Voice; @@ -463,6 +486,9 @@ export class Agent { private readonly feedbackOptions?: AgentFeedbackOptions | boolean; private readonly inputGuardrails: NormalizedInputGuardrail[]; private readonly outputGuardrails: NormalizedOutputGuardrail[]; + private readonly inputMiddlewares: NormalizedInputMiddleware[]; + private readonly outputMiddlewares: NormalizedOutputMiddleware[]; + private readonly maxMiddlewareRetries: number; private readonly observabilityAuthWarningState: ObservabilityFlushState = { authWarningLogged: false, }; @@ -478,6 +504,7 @@ export class Agent { this.temperature = options.temperature; this.maxOutputTokens = options.maxOutputTokens; this.maxSteps = options.maxSteps || 5; + this.maxRetries = options.maxRetries ?? DEFAULT_LLM_MAX_RETRIES; this.stopWhen = options.stopWhen; this.markdown = options.markdown ?? false; this.voice = options.voice; @@ -489,6 +516,9 @@ export class Agent { this.feedbackOptions = options.feedback; this.inputGuardrails = normalizeInputGuardrailList(options.inputGuardrails || []); this.outputGuardrails = normalizeOutputGuardrailList(options.outputGuardrails || []); + this.inputMiddlewares = normalizeInputMiddlewareList(options.inputMiddlewares || []); + this.outputMiddlewares = normalizeOutputMiddlewareList(options.outputMiddlewares || []); + this.maxMiddlewareRetries = options.maxMiddlewareRetries ?? 0; // Initialize logger - always use LoggerProxy for consistency // If external logger is provided, it will be used by LoggerProxy @@ -559,265 +589,360 @@ export class Agent { const rootSpan = oc.traceContext.getRootSpan(); return await oc.traceContext.withSpan(rootSpan, async () => { const guardrailSet = this.resolveGuardrailSets(options); - const buffer = this.getConversationBuffer(oc); - const persistQueue = this.getMemoryPersistQueue(oc); + const middlewareSet = this.resolveMiddlewareSets(options); + const maxMiddlewareRetries = this.resolveMiddlewareRetries(options); + let middlewareRetryCount = 0; const feedbackPromise = feedbackOptions && feedbackClient ? this.createFeedbackMetadata(oc, options) : null; let effectiveInput: typeof input = input; try { - effectiveInput = await executeInputGuardrails( - input, - oc, - guardrailSet.input, - "generateText", - this, - ); - - const { messages, uiMessages, model, tools, maxSteps } = await this.prepareExecution( - effectiveInput, - oc, - options, - ); + while (true) { + try { + if (middlewareRetryCount > 0) { + this.resetOperationAttemptState(oc); + } - const modelName = this.getModelName(model); - const contextLimit = options?.contextLimit; + const buffer = this.getConversationBuffer(oc); + const persistQueue = this.getMemoryPersistQueue(oc); - // Add model attributes and all options - addModelAttributesToSpan( - rootSpan, - modelName, - options, - this.maxOutputTokens, - this.temperature, - ); + effectiveInput = await runInputMiddlewares( + input, + oc, + middlewareSet.input, + "generateText", + this, + middlewareRetryCount, + ); - // Add context to span - const contextMap = Object.fromEntries(oc.context.entries()); - if (Object.keys(contextMap).length > 0) { - rootSpan.setAttribute("agent.context", safeStringify(contextMap)); - } + effectiveInput = await executeInputGuardrails( + effectiveInput, + oc, + guardrailSet.input, + "generateText", + this, + ); - // Add messages (serialize to JSON string) - rootSpan.setAttribute("agent.messages", safeStringify(messages)); - rootSpan.setAttribute("agent.messages.ui", safeStringify(uiMessages)); + const { messages, uiMessages, modelName, tools, maxSteps } = + await this.prepareExecution(effectiveInput, oc, options); + const contextLimit = options?.contextLimit; - // Add agent state snapshot for remote observability - const agentState = this.getFullState(); - rootSpan.setAttribute("agent.stateSnapshot", safeStringify(agentState)); + // Add model attributes and all options + addModelAttributesToSpan( + rootSpan, + modelName, + options, + this.maxOutputTokens, + this.temperature, + ); - // Log generation start with only event-specific context - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_START, - `Starting text generation with ${modelName}`, - ), - { - event: LogEvents.AGENT_GENERATION_STARTED, - operationType: "text", - contextLimit, - memoryEnabled: !!this.memoryManager.getMemory(), - model: modelName, - messageCount: messages?.length || 0, - input: effectiveInput, - }, - ); + // Add context to span + const contextMap = Object.fromEntries(oc.context.entries()); + if (Object.keys(contextMap).length > 0) { + rootSpan.setAttribute("agent.context", safeStringify(contextMap)); + } - // Call hooks - await this.getMergedHooks(options).onStart?.({ agent: this, context: oc }); + // Add messages (serialize to JSON string) + rootSpan.setAttribute("agent.messages", safeStringify(messages)); + rootSpan.setAttribute("agent.messages.ui", safeStringify(uiMessages)); - // Event tracking now handled by OpenTelemetry spans + // Add agent state snapshot for remote observability + const agentState = this.getFullState(); + rootSpan.setAttribute("agent.stateSnapshot", safeStringify(agentState)); - // Setup abort signal listener - this.setupAbortSignalListener(oc); + // Log generation start with only event-specific context + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_START, + `Starting text generation with ${modelName}`, + ), + { + event: LogEvents.AGENT_GENERATION_STARTED, + operationType: "text", + contextLimit, + memoryEnabled: !!this.memoryManager.getMemory(), + model: modelName, + messageCount: messages?.length || 0, + input: effectiveInput, + }, + ); - methodLogger.debug("Starting agent llm call"); + // Call hooks + await this.getMergedHooks(options).onStart?.({ agent: this, context: oc }); - methodLogger.debug("[LLM] - Generating text", { - messages: messages.map((msg) => ({ - role: msg.role, - content: msg.content, - })), - maxSteps, - tools: tools ? Object.keys(tools) : [], - }); + // Event tracking now handled by OpenTelemetry spans - // Extract VoltAgent-specific options - const { - userId, - conversationId, - context, // Explicitly exclude to prevent collision with AI SDK's future 'context' field - parentAgentId, - parentOperationContext, - hooks, - feedback: _feedback, - maxSteps: userMaxSteps, - tools: userTools, - output, - providerOptions, - ...aiSDKOptions - } = options || {}; + // Setup abort signal listener + this.setupAbortSignalListener(oc); - const forcedToolChoice = oc.systemContext.get(FORCED_TOOL_CHOICE_CONTEXT_KEY) as - | ToolChoice> - | undefined; - applyForcedToolChoice(aiSDKOptions, forcedToolChoice); + methodLogger.debug("Starting agent llm call"); - const llmSpan = this.createLLMSpan(oc, { - operation: "generateText", - modelName, - isStreaming: false, - messages, - tools, - providerOptions, - callOptions: { - temperature: aiSDKOptions?.temperature ?? this.temperature, - maxOutputTokens: aiSDKOptions?.maxOutputTokens ?? this.maxOutputTokens, - topP: aiSDKOptions?.topP, - stop: aiSDKOptions?.stop ?? options?.stop, - }, - }); - const finalizeLLMSpan = this.createLLMSpanFinalizer(llmSpan); + methodLogger.debug("[LLM] - Generating text", { + messages: messages.map((msg) => ({ + role: msg.role, + content: msg.content, + })), + maxSteps, + tools: tools ? Object.keys(tools) : [], + }); - let result!: GenerateTextResult; - try { - result = await oc.traceContext.withSpan(llmSpan, () => - generateText({ - model, - messages, - tools, - // Default values - temperature: this.temperature, - maxOutputTokens: this.maxOutputTokens, - maxRetries: 3, - stopWhen: options?.stopWhen ?? this.stopWhen ?? stepCountIs(maxSteps), - // User overrides from AI SDK options - ...aiSDKOptions, - // Structured output if provided + // Extract VoltAgent-specific options + const { + userId, + conversationId, + context, // Explicitly exclude to prevent collision with AI SDK's future 'context' field + parentAgentId, + parentOperationContext, + hooks, + feedback: _feedback, + maxSteps: userMaxSteps, + tools: userTools, output, - // Provider-specific options providerOptions, - // VoltAgent controlled (these should not be overridden) - abortSignal: oc.abortController.signal, - onStepFinish: this.createStepHandler(oc, options), - }), - ); - } catch (error) { - finalizeLLMSpan(SpanStatusCode.ERROR, { message: (error as Error).message }); - throw error; - } + ...aiSDKOptions + } = options || {}; - const resolvedProviderUsage = result.usage - ? await Promise.resolve(result.usage) - : undefined; - finalizeLLMSpan(SpanStatusCode.OK, { - usage: resolvedProviderUsage, - finishReason: result.finishReason, - }); + const forcedToolChoice = oc.systemContext.get(FORCED_TOOL_CHOICE_CONTEXT_KEY) as + | ToolChoice> + | undefined; + applyForcedToolChoice(aiSDKOptions, forcedToolChoice); - const { toolCalls: aggregatedToolCalls, toolResults: aggregatedToolResults } = - this.collectToolDataFromResult(result); + const { result, modelName: effectiveModelName } = await this.executeWithModelFallback({ + oc, + operation: "generateText", + options, + run: async ({ + model: resolvedModel, + modelName: resolvedModelName, + modelId, + maxRetries, + modelIndex, + attempt, + isLastAttempt: _isLastAttempt, + isLastModel: _isLastModel, + }) => { + const llmSpan = this.createLLMSpan(oc, { + operation: "generateText", + modelName: resolvedModelName, + isStreaming: false, + messages, + tools, + providerOptions, + callOptions: { + temperature: aiSDKOptions?.temperature ?? this.temperature, + maxOutputTokens: aiSDKOptions?.maxOutputTokens ?? this.maxOutputTokens, + topP: aiSDKOptions?.topP, + stop: aiSDKOptions?.stop ?? options?.stop, + maxRetries, + modelIndex, + attempt, + modelId, + }, + }); + const finalizeLLMSpan = this.createLLMSpanFinalizer(llmSpan); - this.recordStepResults(result.steps, oc); + try { + const response = await oc.traceContext.withSpan(llmSpan, () => + generateText({ + model: resolvedModel, + messages, + tools, + // Default values + temperature: this.temperature, + maxOutputTokens: this.maxOutputTokens, + stopWhen: options?.stopWhen ?? this.stopWhen ?? stepCountIs(maxSteps), + // User overrides from AI SDK options + ...aiSDKOptions, + maxRetries: 0, + // Structured output if provided + output, + // Provider-specific options + providerOptions, + // VoltAgent controlled (these should not be overridden) + abortSignal: oc.abortController.signal, + onStepFinish: this.createStepHandler(oc, options), + }), + ); + + const resolvedProviderUsage = response.usage + ? await Promise.resolve(response.usage) + : undefined; + finalizeLLMSpan(SpanStatusCode.OK, { + usage: resolvedProviderUsage, + finishReason: response.finishReason, + }); + + return response; + } catch (error) { + finalizeLLMSpan(SpanStatusCode.ERROR, { message: (error as Error).message }); + throw error; + } + }, + }); - if (!shouldDeferPersist) { - await persistQueue.flush(buffer, oc); - } + addModelAttributesToSpan( + oc.traceContext.getRootSpan(), + effectiveModelName, + options, + this.maxOutputTokens, + this.temperature, + ); - const usageInfo = convertUsage(result.usage); - const finalText = await executeOutputGuardrails({ - output: result.text, - operationContext: oc, - guardrails: guardrailSet.output, - operation: "generateText", - agent: this, - metadata: { - usage: usageInfo, - finishReason: result.finishReason ?? null, - warnings: result.warnings ?? null, - }, - }); + const { toolCalls: aggregatedToolCalls, toolResults: aggregatedToolResults } = + this.collectToolDataFromResult(result); - await this.getMergedHooks(options).onEnd?.({ - conversationId: oc.conversationId || "", - agent: this, - output: { - text: finalText, - usage: usageInfo, - providerResponse: result.response, - finishReason: result.finishReason, - warnings: result.warnings, - context: oc.context, - }, - error: undefined, - context: oc, - }); + const usageInfo = convertUsage(result.usage); + const middlewareText = await runOutputMiddlewares( + result.text, + oc, + middlewareSet.output as NormalizedOutputMiddleware[], + "generateText", + this, + middlewareRetryCount, + { + usage: usageInfo, + finishReason: result.finishReason ?? null, + warnings: result.warnings ?? null, + }, + ); - // Log successful completion with usage details - const providerUsage = result.usage; - const tokenInfo = providerUsage ? `${providerUsage.totalTokens} tokens` : "no usage data"; - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_COMPLETE, - `Text generation completed (${tokenInfo})`, - ), - { - event: LogEvents.AGENT_GENERATION_COMPLETED, - duration: Date.now() - startTime, - finishReason: result.finishReason, - usage: result.usage, - toolCalls: aggregatedToolCalls.length, - text: finalText, - }, - ); + this.recordStepResults(result.steps, oc); - // Add usage to span - this.setTraceContextUsage(oc.traceContext, result.usage); - oc.traceContext.setOutput(finalText); - oc.traceContext.setFinishReason(result.finishReason); + if (!shouldDeferPersist) { + await persistQueue.flush(buffer, oc); + } - // Check if stopped by maxSteps - if (result.steps && result.steps.length >= maxSteps) { - oc.traceContext.setStopConditionMet(result.steps.length, maxSteps); - } + const finalText = await executeOutputGuardrails({ + output: middlewareText, + operationContext: oc, + guardrails: guardrailSet.output, + operation: "generateText", + agent: this, + metadata: { + usage: usageInfo, + finishReason: result.finishReason ?? null, + warnings: result.warnings ?? null, + }, + }); - // Set output in operation context - oc.output = finalText; + await this.getMergedHooks(options).onEnd?.({ + conversationId: oc.conversationId || "", + agent: this, + output: { + text: finalText, + usage: usageInfo, + providerResponse: result.response, + finishReason: result.finishReason, + warnings: result.warnings, + context: oc.context, + }, + error: undefined, + context: oc, + }); - this.enqueueEvalScoring({ - oc, - output: finalText, - operation: "generateText", - metadata: { - finishReason: result.finishReason, - usage: result.usage ? JSON.parse(safeStringify(result.usage)) : undefined, - toolCalls: aggregatedToolCalls, - }, - }); + // Log successful completion with usage details + const providerUsage = result.usage; + const tokenInfo = providerUsage + ? `${providerUsage.totalTokens} tokens` + : "no usage data"; + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_COMPLETE, + `Text generation completed (${tokenInfo})`, + ), + { + event: LogEvents.AGENT_GENERATION_COMPLETED, + duration: Date.now() - startTime, + finishReason: result.finishReason, + usage: result.usage, + toolCalls: aggregatedToolCalls.length, + text: finalText, + }, + ); - // Close span after scheduling scorers - oc.traceContext.end("completed"); + // Add usage to span + this.setTraceContextUsage(oc.traceContext, result.usage); + oc.traceContext.setOutput(finalText); + oc.traceContext.setFinishReason(result.finishReason); - if (feedbackPromise) { - feedbackMetadata = await feedbackPromise; - } + // Check if stopped by maxSteps + if (result.steps && result.steps.length >= maxSteps) { + oc.traceContext.setStopConditionMet(result.steps.length, maxSteps); + } - if (feedbackMetadata) { - buffer.addMetadataToLastAssistantMessage({ feedback: feedbackMetadata }); - } + // Set output in operation context + oc.output = finalText; - if (shouldDeferPersist) { - await persistQueue.flush(buffer, oc); - } + this.enqueueEvalScoring({ + oc, + output: finalText, + operation: "generateText", + metadata: { + finishReason: result.finishReason, + usage: result.usage ? JSON.parse(safeStringify(result.usage)) : undefined, + toolCalls: aggregatedToolCalls, + }, + }); - return cloneGenerateTextResultWithContext(result, { - text: finalText, - context: oc.context, - toolCalls: aggregatedToolCalls, - toolResults: aggregatedToolResults, - feedback: feedbackMetadata, - }); + // Close span after scheduling scorers + oc.traceContext.end("completed"); + + if (feedbackPromise) { + feedbackMetadata = await feedbackPromise; + } + + if (feedbackMetadata) { + buffer.addMetadataToLastAssistantMessage({ feedback: feedbackMetadata }); + } + + if (shouldDeferPersist) { + await persistQueue.flush(buffer, oc); + } + + return cloneGenerateTextResultWithContext(result, { + text: finalText, + context: oc.context, + toolCalls: aggregatedToolCalls, + toolResults: aggregatedToolResults, + feedback: feedbackMetadata, + }); + } catch (error) { + if (this.shouldRetryMiddleware(error, middlewareRetryCount, maxMiddlewareRetries)) { + const retryError = error as { + middlewareId?: string; + metadata?: unknown; + message?: string; + }; + await this.getMergedHooks(options).onRetry?.({ + agent: this, + context: oc, + operation: "generateText", + source: "middleware", + middlewareId: retryError.middlewareId ?? null, + retryCount: middlewareRetryCount, + maxRetries: maxMiddlewareRetries, + reason: retryError.message, + metadata: retryError.metadata, + }); + methodLogger.warn(`[Agent:${this.name}] - Middleware requested retry`, { + operation: "generateText", + retryCount: middlewareRetryCount, + maxMiddlewareRetries, + middlewareId: retryError.middlewareId ?? null, + reason: retryError.message ?? "middleware retry", + metadata: + retryError.metadata !== undefined + ? safeStringify(retryError.metadata) + : undefined, + }); + this.storeMiddlewareRetryFeedback(oc, retryError.message, retryError.metadata); + middlewareRetryCount += 1; + continue; + } + throw error; + } + } } catch (error) { // Check if this is a BailError (subagent early termination via abort) if (isBailError(error as Error)) { @@ -931,6 +1056,9 @@ export class Agent { return await oc.traceContext.withSpan(rootSpan, async () => { const methodLogger = oc.logger; // Extract logger with executionId const guardrailSet = this.resolveGuardrailSets(options); + const middlewareSet = this.resolveMiddlewareSets(options); + const maxMiddlewareRetries = this.resolveMiddlewareRetries(options); + let middlewareRetryCount = 0; const buffer = this.getConversationBuffer(oc); const persistQueue = this.getMemoryPersistQueue(oc); const scheduleFeedbackPersist = (metadata: AgentFeedbackMetadata | null) => { @@ -962,8 +1090,56 @@ export class Agent { } let effectiveInput: typeof input = input; try { + while (true) { + try { + effectiveInput = await runInputMiddlewares( + input, + oc, + middlewareSet.input, + "streamText", + this, + middlewareRetryCount, + ); + break; + } catch (error) { + if (this.shouldRetryMiddleware(error, middlewareRetryCount, maxMiddlewareRetries)) { + const retryError = error as { + middlewareId?: string; + metadata?: unknown; + message?: string; + }; + await this.getMergedHooks(options).onRetry?.({ + agent: this, + context: oc, + operation: "streamText", + source: "middleware", + middlewareId: retryError.middlewareId ?? null, + retryCount: middlewareRetryCount, + maxRetries: maxMiddlewareRetries, + reason: retryError.message, + metadata: retryError.metadata, + }); + methodLogger.warn(`[Agent:${this.name}] - Middleware requested retry`, { + operation: "streamText", + retryCount: middlewareRetryCount, + maxMiddlewareRetries, + middlewareId: retryError.middlewareId ?? null, + reason: retryError.message ?? "middleware retry", + metadata: + retryError.metadata !== undefined + ? safeStringify(retryError.metadata) + : undefined, + }); + this.storeMiddlewareRetryFeedback(oc, retryError.message, retryError.metadata); + middlewareRetryCount += 1; + continue; + } + throw error; + } + } + effectiveInput = await executeInputGuardrails( - input, + effectiveInput, oc, guardrailSet.input, "streamText", @@ -972,13 +1148,11 @@ export class Agent { // No need to initialize stream collection anymore - we'll use UIMessageStreamWriter - const { messages, uiMessages, model, tools, maxSteps } = await this.prepareExecution( + const { messages, uiMessages, modelName, tools, maxSteps } = await this.prepareExecution( effectiveInput, oc, options, ); - - const modelName = this.getModelName(model); const contextLimit = options?.contextLimit; // Add model attributes to root span if TraceContext exists @@ -1062,270 +1236,363 @@ export class Agent { let guardrailPipeline: GuardrailPipeline | null = null; let sanitizedTextPromise!: PromiseLike; - const llmSpan = this.createLLMSpan(oc, { + const { result, modelName: effectiveModelName } = await this.executeWithModelFallback({ + oc, operation: "streamText", - modelName, - isStreaming: true, - messages, - tools, - providerOptions, - callOptions: { - temperature: aiSDKOptions?.temperature ?? this.temperature, - maxOutputTokens: aiSDKOptions?.maxOutputTokens ?? this.maxOutputTokens, - topP: aiSDKOptions?.topP, - stop: aiSDKOptions?.stop ?? options?.stop, - }, - }); - const finalizeLLMSpan = this.createLLMSpanFinalizer(llmSpan); - - const result = streamText({ - model, - messages, - tools, - // Default values - temperature: this.temperature, - maxOutputTokens: this.maxOutputTokens, - maxRetries: 3, - stopWhen: options?.stopWhen ?? this.stopWhen ?? stepCountIs(maxSteps), - // User overrides from AI SDK options - ...aiSDKOptions, - // Structured output if provided - output, - // Provider-specific options - providerOptions, - // VoltAgent controlled (these should not be overridden) - abortSignal: oc.abortController.signal, - onStepFinish: this.createStepHandler(oc, options), - onError: async (errorData) => { - // Handle nested error structure from OpenAI and other providers - // The error might be directly the error or wrapped in { error: ... } - const actualError = (errorData as any)?.error || errorData; - - // Check if this is a BailError (subagent early termination) - // This is not a real error - it's a signal that execution should stop - if (isBailError(actualError)) { - methodLogger.info("Stream aborted due to subagent bail (not an error)", { - agentName: actualError.agentName, - event: LogEvents.AGENT_GENERATION_COMPLETED, - }); + options, + run: async ({ + model: resolvedModel, + modelName: resolvedModelName, + modelId, + maxRetries, + modelIndex, + attempt, + isLastAttempt, + isLastModel, + }) => { + const attemptState: { hasOutput: boolean; lastError?: unknown } = { + hasOutput: false, + }; + const llmSpan = this.createLLMSpan(oc, { + operation: "streamText", + modelName: resolvedModelName, + isStreaming: true, + messages, + tools, + providerOptions, + callOptions: { + temperature: aiSDKOptions?.temperature ?? this.temperature, + maxOutputTokens: aiSDKOptions?.maxOutputTokens ?? this.maxOutputTokens, + topP: aiSDKOptions?.topP, + stop: aiSDKOptions?.stop ?? options?.stop, + maxRetries, + modelIndex, + attempt, + modelId, + }, + }); + const finalizeLLMSpan = this.createLLMSpanFinalizer(llmSpan); - // Don't log as error, don't call error hooks - // onFinish will be called and will handle span ending with correct finish reason - return; - } + const streamResult = streamText({ + model: resolvedModel, + messages, + tools, + // Default values + temperature: this.temperature, + maxOutputTokens: this.maxOutputTokens, + stopWhen: options?.stopWhen ?? this.stopWhen ?? stepCountIs(maxSteps), + // User overrides from AI SDK options + ...aiSDKOptions, + maxRetries: 0, + // Structured output if provided + output, + // Provider-specific options + providerOptions, + // VoltAgent controlled (these should not be overridden) + abortSignal: oc.abortController.signal, + onStepFinish: this.createStepHandler(oc, options), + onError: async (errorData) => { + // Handle nested error structure from OpenAI and other providers + // The error might be directly the error or wrapped in { error: ... } + const actualError = (errorData as any)?.error || errorData; + attemptState.lastError = actualError; + + // Check if this is a BailError (subagent early termination) + // This is not a real error - it's a signal that execution should stop + if (isBailError(actualError)) { + methodLogger.info("Stream aborted due to subagent bail (not an error)", { + agentName: actualError.agentName, + event: LogEvents.AGENT_GENERATION_COMPLETED, + }); + + // Don't log as error, don't call error hooks + // onFinish will be called and will handle span ending with correct finish reason + return; + } - resolveFeedbackDeferred(null); + const fallbackEligible = this.shouldFallbackOnError(actualError); + const retryEligible = fallbackEligible && this.isRetryableError(actualError); + const canRetry = retryEligible && !isLastAttempt; + const canFallback = fallbackEligible && !isLastModel; + const shouldAttemptRecovery = !attemptState.hasOutput && (canRetry || canFallback); + const recoveryMessage = canRetry + ? "[LLM] Stream error before output; retry pending" + : canFallback + ? "[LLM] Stream error before output; fallback pending" + : attemptState.hasOutput + ? "[LLM] Stream error after output; recovery skipped" + : "[LLM] Stream error before output; recovery skipped"; + + if (!shouldAttemptRecovery) { + resolveFeedbackDeferred(null); + } - // Log the error - methodLogger.error("Stream error occurred", { - error: actualError, - agentName: this.name, - modelName, - }); + // Log the error + methodLogger.error("Stream error occurred", { + error: actualError, + agentName: this.name, + modelName: resolvedModelName, + attempt, + maxRetries, + }); - finalizeLLMSpan(SpanStatusCode.ERROR, { message: (actualError as Error)?.message }); + methodLogger.debug(recoveryMessage, { + operation: "streamText", + modelName: resolvedModelName, + fallbackEligible, + retryEligible, + canRetry, + canFallback, + hasOutput: attemptState.hasOutput, + attempt, + maxRetries, + isLastAttempt, + isLastModel, + isRetryable: (actualError as any)?.isRetryable, + statusCode: (actualError as any)?.statusCode, + errorName: (actualError as Error)?.name, + errorMessage: (actualError as Error)?.message, + }); - // History update removed - using OpenTelemetry only + finalizeLLMSpan(SpanStatusCode.ERROR, { message: (actualError as Error)?.message }); - // Event tracking now handled by OpenTelemetry spans + // History update removed - using OpenTelemetry only - // Call error hooks if they exist - this.getMergedHooks(options).onError?.({ - agent: this, - error: actualError as Error, - context: oc, - }); + // Event tracking now handled by OpenTelemetry spans - // Close OpenTelemetry span with error status - oc.traceContext.end("error", actualError as Error); - - // Don't re-throw - let the error be part of the stream - // The onError callback should return void for AI SDK compatibility - // Ensure spans are flushed on error - // Uses waitUntil if available to avoid blocking - await flushObservability( - this.getObservability(), - oc.logger ?? this.logger, - this.observabilityAuthWarningState, - "streamText:onError", - ); - }, - onFinish: async (finalResult) => { - const providerUsage = finalResult.usage - ? await Promise.resolve(finalResult.usage) - : undefined; - finalizeLLMSpan(SpanStatusCode.OK, { - usage: providerUsage, - finishReason: finalResult.finishReason, - }); + if (shouldAttemptRecovery) { + await flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamText:onError", + ); + return; + } - if (!shouldDeferPersist) { - await persistQueue.flush(buffer, oc); - } + // Call error hooks if they exist + this.getMergedHooks(options).onError?.({ + agent: this, + error: actualError as Error, + context: oc, + }); - // History update removed - using OpenTelemetry only + // Close OpenTelemetry span with error status + oc.traceContext.end("error", actualError as Error); + + // Don't re-throw - let the error be part of the stream + // The onError callback should return void for AI SDK compatibility + // Ensure spans are flushed on error + // Uses waitUntil if available to avoid blocking + await flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamText:onError", + ); + }, + onFinish: async (finalResult) => { + const providerUsage = finalResult.usage + ? await Promise.resolve(finalResult.usage) + : undefined; + finalizeLLMSpan(SpanStatusCode.OK, { + usage: providerUsage, + finishReason: finalResult.finishReason, + }); - // Event tracking now handled by OpenTelemetry spans + if (!shouldDeferPersist) { + await persistQueue.flush(buffer, oc); + } - // Add usage to span - this.setTraceContextUsage(oc.traceContext, finalResult.totalUsage); + // History update removed - using OpenTelemetry only + + // Event tracking now handled by OpenTelemetry spans + + // Add usage to span + this.setTraceContextUsage(oc.traceContext, finalResult.totalUsage); + + const usage = convertUsage(finalResult.totalUsage); + let finalText: string; + + // Check if we aborted due to subagent bail (early termination) + const bailedResult = oc.systemContext.get("bailedResult") as + | { agentName: string; response: string } + | undefined; + + if (bailedResult) { + // Use the bailed result instead of the supervisor's output + methodLogger.info("Using bailed subagent result as final output", { + event: LogEvents.AGENT_GENERATION_COMPLETED, + agentName: bailedResult.agentName, + bailed: true, + }); + + // Apply guardrails to bailed result + if (guardrailSet.output.length > 0) { + finalText = await executeOutputGuardrails({ + output: bailedResult.response, + operationContext: oc, + guardrails: guardrailSet.output, + operation: "streamText", + agent: this, + metadata: { + usage, + finishReason: "bail" as any, + warnings: finalResult.warnings ?? null, + }, + }); + } else { + finalText = bailedResult.response; + } + } else if (guardrailPipeline) { + finalText = await sanitizedTextPromise; + } else if (guardrailSet.output.length > 0) { + finalText = await executeOutputGuardrails({ + output: finalResult.text, + operationContext: oc, + guardrails: guardrailSet.output, + operation: "streamText", + agent: this, + metadata: { + usage, + finishReason: finalResult.finishReason ?? null, + warnings: finalResult.warnings ?? null, + }, + }); + } else { + finalText = finalResult.text; + } - const usage = convertUsage(finalResult.totalUsage); - let finalText: string; + const guardrailedResult = + guardrailSet.output.length > 0 + ? { ...finalResult, text: finalText } + : finalResult; - // Check if we aborted due to subagent bail (early termination) - const bailedResult = oc.systemContext.get("bailedResult") as - | { agentName: string; response: string } - | undefined; + oc.traceContext.setOutput(finalText); - if (bailedResult) { - // Use the bailed result instead of the supervisor's output - methodLogger.info("Using bailed subagent result as final output", { - event: LogEvents.AGENT_GENERATION_COMPLETED, - agentName: bailedResult.agentName, - bailed: true, - }); + this.recordStepResults(finalResult.steps, oc); - // Apply guardrails to bailed result - if (guardrailSet.output.length > 0) { - finalText = await executeOutputGuardrails({ - output: bailedResult.response, - operationContext: oc, - guardrails: guardrailSet.output, - operation: "streamText", + // Set finish reason - override to "stop" if bailed (not "error") + if (bailedResult) { + oc.traceContext.setFinishReason("stop" as any); + } else { + oc.traceContext.setFinishReason(finalResult.finishReason); + } + + // Check if stopped by maxSteps + const steps = finalResult.steps; + if (steps && steps.length >= maxSteps) { + oc.traceContext.setStopConditionMet(steps.length, maxSteps); + } + + // Set output in operation context + oc.output = finalText; + // Call hooks with standardized output (stream finish result) + await this.getMergedHooks(options).onEnd?.({ + conversationId: oc.conversationId || "", agent: this, - metadata: { + output: { + text: finalText, usage, - finishReason: "bail" as any, - warnings: finalResult.warnings ?? null, + providerResponse: finalResult.response, + finishReason: finalResult.finishReason, + warnings: finalResult.warnings, + context: oc.context, }, + error: undefined, + context: oc, }); - } else { - finalText = bailedResult.response; - } - } else if (guardrailPipeline) { - finalText = await sanitizedTextPromise; - } else if (guardrailSet.output.length > 0) { - finalText = await executeOutputGuardrails({ - output: finalResult.text, - operationContext: oc, - guardrails: guardrailSet.output, - operation: "streamText", - agent: this, - metadata: { - usage, - finishReason: finalResult.finishReason ?? null, - warnings: finalResult.warnings ?? null, - }, - }); - } else { - finalText = finalResult.text; - } - const guardrailedResult = - guardrailSet.output.length > 0 ? { ...finalResult, text: finalText } : finalResult; + // Call user's onFinish if it exists + if (userOnFinish) { + await userOnFinish(guardrailedResult); + } - oc.traceContext.setOutput(finalText); + const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_COMPLETE, + `Text generation completed (${tokenInfo})`, + ), + { + event: LogEvents.AGENT_GENERATION_COMPLETED, + duration: Date.now() - startTime, + finishReason: finalResult.finishReason, + usage: finalResult.usage, + toolCalls: finalResult.toolCalls?.length || 0, + text: finalText, + }, + ); - this.recordStepResults(finalResult.steps, oc); + this.enqueueEvalScoring({ + oc, + output: finalText, + operation: "streamText", + metadata: { + finishReason: finalResult.finishReason, + usage: finalResult.totalUsage + ? JSON.parse(safeStringify(finalResult.totalUsage)) + : undefined, + toolCalls: finalResult.toolCalls, + }, + }); - // Set finish reason - override to "stop" if bailed (not "error") - if (bailedResult) { - oc.traceContext.setFinishReason("stop" as any); - } else { - oc.traceContext.setFinishReason(finalResult.finishReason); - } + finalizeLLMSpan(SpanStatusCode.OK, { + usage: finalResult.totalUsage, + finishReason: finalResult.finishReason, + }); - // Check if stopped by maxSteps - const steps = finalResult.steps; - if (steps && steps.length >= maxSteps) { - oc.traceContext.setStopConditionMet(steps.length, maxSteps); - } + oc.traceContext.end("completed"); - // Set output in operation context - oc.output = finalText; - // Call hooks with standardized output (stream finish result) - await this.getMergedHooks(options).onEnd?.({ - conversationId: oc.conversationId || "", - agent: this, - output: { - text: finalText, - usage, - providerResponse: finalResult.response, - finishReason: finalResult.finishReason, - warnings: finalResult.warnings, - context: oc.context, - }, - error: undefined, - context: oc, - }); + feedbackFinalizeRequested = true; - // Call user's onFinish if it exists - if (userOnFinish) { - await userOnFinish(guardrailedResult); - } + if (!feedbackResolved && feedbackDeferred) { + await feedbackDeferred.promise; + } - const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_COMPLETE, - `Text generation completed (${tokenInfo})`, - ), - { - event: LogEvents.AGENT_GENERATION_COMPLETED, - duration: Date.now() - startTime, - finishReason: finalResult.finishReason, - usage: finalResult.usage, - toolCalls: finalResult.toolCalls?.length || 0, - text: finalText, - }, - ); + if (feedbackResolved && feedbackValue) { + scheduleFeedbackPersist(feedbackValue); + } else if (shouldDeferPersist) { + void persistQueue.flush(buffer, oc).catch((error) => { + oc.logger?.debug?.("Failed to persist deferred messages", { error }); + }); + } - this.enqueueEvalScoring({ - oc, - output: finalText, - operation: "streamText", - metadata: { - finishReason: finalResult.finishReason, - usage: finalResult.totalUsage - ? JSON.parse(safeStringify(finalResult.totalUsage)) - : undefined, - toolCalls: finalResult.toolCalls, + // Schedule span flush without blocking the response + void flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamText:onFinish", + ); }, }); - finalizeLLMSpan(SpanStatusCode.OK, { - usage: finalResult.totalUsage, - finishReason: finalResult.finishReason, - }); - - oc.traceContext.end("completed"); - - feedbackFinalizeRequested = true; - - if (!feedbackResolved && feedbackDeferred) { - await feedbackDeferred.promise; - } - - if (feedbackResolved && feedbackValue) { - scheduleFeedbackPersist(feedbackValue); - } else if (shouldDeferPersist) { - void persistQueue.flush(buffer, oc).catch((error) => { - oc.logger?.debug?.("Failed to persist deferred messages", { error }); - }); + const probeResult = await this.probeStreamStart(streamResult.fullStream, attemptState); + if (probeResult.status === "error") { + this.discardStream(streamResult.fullStream); + const fallbackEligible = this.shouldFallbackOnError(probeResult.error); + if (!fallbackEligible || isLastModel) { + throw probeResult.error; + } + throw probeResult.error; } - // Schedule span flush without blocking the response - void flushObservability( - this.getObservability(), - oc.logger ?? this.logger, - this.observabilityAuthWarningState, - "streamText:onFinish", - ); + return streamResult; }, }); + if (oc.traceContext) { + addModelAttributesToSpan( + oc.traceContext.getRootSpan(), + effectiveModelName, + options, + this.maxOutputTokens, + this.temperature, + ); + } + // Capture the agent instance for use in helpers type ToUIMessageStreamOptions = Parameters[0]; type ToUIMessageStreamResponseOptions = Parameters< @@ -1475,7 +1742,7 @@ export class Agent { return bailedResult?.response || aiSdkText; }); } else { - // Wrap result.text with custom Promise that checks for bailed result + // Wrap result.text with a bail check // IMPORTANT: Wait for AI SDK text first (stream must complete/abort) // This ensures createStepHandler has processed tool results and set bailedResult sanitizedTextPromise = result.text.then((aiSdkText) => { @@ -1671,7 +1938,7 @@ export class Agent { /** * Generate structured object - * @deprecated Use generateText with Output.object instead. generateObject will be removed in a future release. + * @deprecated — Use generateText with an output setting instead. */ async generateObject( input: string | UIMessage[] | BaseMessage[], @@ -1686,207 +1953,287 @@ export class Agent { const rootSpan = oc.traceContext.getRootSpan(); return await oc.traceContext.withSpan(rootSpan, async () => { const guardrailSet = this.resolveGuardrailSets(options); + const middlewareSet = this.resolveMiddlewareSets(options); + const maxMiddlewareRetries = this.resolveMiddlewareRetries(options); + let middlewareRetryCount = 0; let effectiveInput: typeof input = input; try { - effectiveInput = await executeInputGuardrails( - input, - oc, - guardrailSet.input, - "generateObject", - this, - ); - const { messages, uiMessages, model } = await this.prepareExecution( - effectiveInput, - oc, - options, - ); + while (true) { + try { + if (middlewareRetryCount > 0) { + this.resetOperationAttemptState(oc); + } - const modelName = this.getModelName(model); - const schemaName = schema.description || "unknown"; + effectiveInput = await runInputMiddlewares( + input, + oc, + middlewareSet.input, + "generateObject", + this, + middlewareRetryCount, + ); - // Add model attributes and all options - addModelAttributesToSpan( - rootSpan, - modelName, - options, - this.maxOutputTokens, - this.temperature, - ); + effectiveInput = await executeInputGuardrails( + effectiveInput, + oc, + guardrailSet.input, + "generateObject", + this, + ); + const { messages, uiMessages, modelName } = await this.prepareExecution( + effectiveInput, + oc, + options, + ); + const schemaName = schema.description || "unknown"; - // Add context to span - const contextMap = Object.fromEntries(oc.context.entries()); - if (Object.keys(contextMap).length > 0) { - rootSpan.setAttribute("agent.context", safeStringify(contextMap)); - } + // Add model attributes and all options + addModelAttributesToSpan( + rootSpan, + modelName, + options, + this.maxOutputTokens, + this.temperature, + ); - // Add messages (serialize to JSON string) - rootSpan.setAttribute("agent.messages", safeStringify(messages)); - rootSpan.setAttribute("agent.messages.ui", safeStringify(uiMessages)); + // Add context to span + const contextMap = Object.fromEntries(oc.context.entries()); + if (Object.keys(contextMap).length > 0) { + rootSpan.setAttribute("agent.context", safeStringify(contextMap)); + } - // Add agent state snapshot for remote observability - const agentState = this.getFullState(); - rootSpan.setAttribute("agent.stateSnapshot", safeStringify(agentState)); + // Add messages (serialize to JSON string) + rootSpan.setAttribute("agent.messages", safeStringify(messages)); + rootSpan.setAttribute("agent.messages.ui", safeStringify(uiMessages)); - // Log generation start (object) - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_START, - `Starting object generation with ${modelName}`, - ), - { - event: LogEvents.AGENT_GENERATION_STARTED, - operationType: "object", - schemaName, - model: modelName, - messageCount: messages?.length || 0, - input: effectiveInput, - }, - ); + // Add agent state snapshot for remote observability + const agentState = this.getFullState(); + rootSpan.setAttribute("agent.stateSnapshot", safeStringify(agentState)); - // Call hooks - await this.getMergedHooks(options).onStart?.({ agent: this, context: oc }); + // Log generation start (object) + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_START, + `Starting object generation with ${modelName}`, + ), + { + event: LogEvents.AGENT_GENERATION_STARTED, + operationType: "object", + schemaName, + model: modelName, + messageCount: messages?.length || 0, + input: effectiveInput, + }, + ); - // Event tracking now handled by OpenTelemetry spans + // Call hooks + await this.getMergedHooks(options).onStart?.({ agent: this, context: oc }); - // Extract VoltAgent-specific options - const { - userId, - conversationId, - context, // Explicitly exclude to prevent collision with AI SDK's future 'context' field - parentAgentId, - parentOperationContext, - hooks, - feedback: _feedback, - maxSteps: userMaxSteps, - tools: userTools, - output: _output, - providerOptions, - ...aiSDKOptions - } = options || {}; + // Event tracking now handled by OpenTelemetry spans - const result = await generateObject({ - model, - messages, - schema, - // Default values - maxOutputTokens: this.maxOutputTokens, - temperature: this.temperature, - maxRetries: 3, - // User overrides from AI SDK options - ...aiSDKOptions, - // Provider-specific options - providerOptions, - // VoltAgent controlled - abortSignal: oc.abortController.signal, - }); + // Extract VoltAgent-specific options + const { + userId, + conversationId, + context, // Explicitly exclude to prevent collision with AI SDK's future 'context' field + parentAgentId, + parentOperationContext, + hooks, + feedback: _feedback, + maxSteps: userMaxSteps, + tools: userTools, + output: _output, + providerOptions, + ...aiSDKOptions + } = options || {}; - const usageInfo = convertUsage(result.usage); - const finalObject = await executeOutputGuardrails({ - output: result.object, - operationContext: oc, - guardrails: guardrailSet.output, - operation: "generateObject", - agent: this, - metadata: { - usage: usageInfo, - finishReason: result.finishReason ?? null, - warnings: result.warnings ?? null, - }, - }); + const { result, modelName: effectiveModelName } = await this.executeWithModelFallback({ + oc, + operation: "generateObject", + options, + run: async ({ model: resolvedModel }) => { + return await generateObject({ + model: resolvedModel, + messages, + schema, + // Default values + maxOutputTokens: this.maxOutputTokens, + temperature: this.temperature, + // User overrides from AI SDK options + ...aiSDKOptions, + maxRetries: 0, + // Provider-specific options + providerOptions, + // VoltAgent controlled + abortSignal: oc.abortController.signal, + }); + }, + }); - // Save the object response to memory - if (oc.userId && oc.conversationId) { - // Create UIMessage from the object response - const message: UIMessage = { - id: randomUUID(), - role: "assistant", - parts: [ + addModelAttributesToSpan( + rootSpan, + effectiveModelName, + options, + this.maxOutputTokens, + this.temperature, + ); + + const usageInfo = convertUsage(result.usage); + const middlewareObject = await runOutputMiddlewares>( + result.object, + oc, + middlewareSet.output as NormalizedOutputMiddleware>[], + "generateObject", + this, + middlewareRetryCount, { - type: "text", - text: safeStringify(finalObject), + usage: usageInfo, + finishReason: result.finishReason ?? null, + warnings: result.warnings ?? null, }, - ], - }; + ); + const finalObject = await executeOutputGuardrails({ + output: middlewareObject, + operationContext: oc, + guardrails: guardrailSet.output, + operation: "generateObject", + agent: this, + metadata: { + usage: usageInfo, + finishReason: result.finishReason ?? null, + warnings: result.warnings ?? null, + }, + }); - // Save the message to memory - await this.memoryManager.saveMessage(oc, message, oc.userId, oc.conversationId); + // Save the object response to memory + if (oc.userId && oc.conversationId) { + // Create UIMessage from the object response + const message: UIMessage = { + id: randomUUID(), + role: "assistant", + parts: [ + { + type: "text", + text: safeStringify(finalObject), + }, + ], + }; - // Add step to history - const step: StepWithContent = { - id: randomUUID(), - type: "text", - content: safeStringify(finalObject), - role: "assistant", - usage: usageInfo, - }; - this.addStepToHistory(step, oc); - } + // Save the message to memory + await this.memoryManager.saveMessage(oc, message, oc.userId, oc.conversationId); - // History update removed - using OpenTelemetry only + // Add step to history + const step: StepWithContent = { + id: randomUUID(), + type: "text", + content: safeStringify(finalObject), + role: "assistant", + usage: usageInfo, + }; + this.addStepToHistory(step, oc); + } - // Event tracking now handled by OpenTelemetry spans + // History update removed - using OpenTelemetry only - // Add usage to span - this.setTraceContextUsage(oc.traceContext, result.usage); - oc.traceContext.setOutput(finalObject); + // Event tracking now handled by OpenTelemetry spans - // Set output in operation context - oc.output = finalObject; + // Add usage to span + this.setTraceContextUsage(oc.traceContext, result.usage); + oc.traceContext.setOutput(finalObject); - this.enqueueEvalScoring({ - oc, - output: finalObject, - operation: "generateObject", - metadata: { - finishReason: result.finishReason, - usage: result.usage ? JSON.parse(safeStringify(result.usage)) : undefined, - schemaName, - }, - }); + // Set output in operation context + oc.output = finalObject as unknown as string | object; - oc.traceContext.end("completed"); + this.enqueueEvalScoring({ + oc, + output: finalObject, + operation: "generateObject", + metadata: { + finishReason: result.finishReason, + usage: result.usage ? JSON.parse(safeStringify(result.usage)) : undefined, + schemaName, + }, + }); - // Call hooks - await this.getMergedHooks(options).onEnd?.({ - conversationId: oc.conversationId || "", - agent: this, - output: { - object: finalObject, - usage: usageInfo, - providerResponse: (result as any).response, - finishReason: result.finishReason, - warnings: result.warnings, - context: oc.context, - }, - error: undefined, - context: oc, - }); + oc.traceContext.end("completed"); - // Log successful completion - const usage = result.usage; - const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_COMPLETE, - `Object generation completed (${tokenInfo})`, - ), - { - event: LogEvents.AGENT_GENERATION_COMPLETED, - duration: Date.now() - startTime, - finishReason: result.finishReason, - usage: result.usage, - schemaName, - }, - ); + // Call hooks + await this.getMergedHooks(options).onEnd?.({ + conversationId: oc.conversationId || "", + agent: this, + output: { + object: finalObject, + usage: usageInfo, + providerResponse: (result as any).response, + finishReason: result.finishReason, + warnings: result.warnings, + context: oc.context, + }, + error: undefined, + context: oc, + }); - // Return result with same context reference for consistency - return { - ...result, - object: finalObject, - context: oc.context, - }; + // Log successful completion + const usage = result.usage; + const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_COMPLETE, + `Object generation completed (${tokenInfo})`, + ), + { + event: LogEvents.AGENT_GENERATION_COMPLETED, + duration: Date.now() - startTime, + finishReason: result.finishReason, + usage: result.usage, + schemaName, + }, + ); + + // Return result with same context reference for consistency + return { + ...result, + object: finalObject, + context: oc.context, + }; + } catch (error) { + if (this.shouldRetryMiddleware(error, middlewareRetryCount, maxMiddlewareRetries)) { + const retryError = error as { + middlewareId?: string; + metadata?: unknown; + message?: string; + }; + await this.getMergedHooks(options).onRetry?.({ + agent: this, + context: oc, + operation: "generateObject", + source: "middleware", + middlewareId: retryError.middlewareId ?? null, + retryCount: middlewareRetryCount, + maxRetries: maxMiddlewareRetries, + reason: retryError.message, + metadata: retryError.metadata, + }); + methodLogger.warn(`[Agent:${this.name}] - Middleware requested retry`, { + operation: "generateObject", + retryCount: middlewareRetryCount, + maxMiddlewareRetries, + middlewareId: retryError.middlewareId ?? null, + reason: retryError.message ?? "middleware retry", + metadata: + retryError.metadata !== undefined + ? safeStringify(retryError.metadata) + : undefined, + }); + this.storeMiddlewareRetryFeedback(oc, retryError.message, retryError.metadata); + middlewareRetryCount += 1; + continue; + } + throw error; + } + } } catch (error) { await this.flushPendingMessagesOnError(oc).catch(() => {}); return this.handleError(error as Error, oc, options, startTime); @@ -1905,7 +2252,7 @@ export class Agent { /** * Stream structured object - * @deprecated Use streamText with Output.object instead. streamObject will be removed in a future release. + * @deprecated — Use streamText with an output setting instead. */ async streamObject( input: string | UIMessage[] | BaseMessage[], @@ -1920,23 +2267,72 @@ export class Agent { return await oc.traceContext.withSpan(rootSpan, async () => { const methodLogger = oc.logger; // Extract logger with executionId const guardrailSet = this.resolveGuardrailSets(options); + const middlewareSet = this.resolveMiddlewareSets(options); + const maxMiddlewareRetries = this.resolveMiddlewareRetries(options); + let middlewareRetryCount = 0; let effectiveInput: typeof input = input; try { + while (true) { + try { + effectiveInput = await runInputMiddlewares( + input, + oc, + middlewareSet.input, + "streamObject", + this, + middlewareRetryCount, + ); + break; + } catch (error) { + if (this.shouldRetryMiddleware(error, middlewareRetryCount, maxMiddlewareRetries)) { + const retryError = error as { + middlewareId?: string; + metadata?: unknown; + message?: string; + }; + await this.getMergedHooks(options).onRetry?.({ + agent: this, + context: oc, + operation: "streamObject", + source: "middleware", + middlewareId: retryError.middlewareId ?? null, + retryCount: middlewareRetryCount, + maxRetries: maxMiddlewareRetries, + reason: retryError.message, + metadata: retryError.metadata, + }); + methodLogger.warn(`[Agent:${this.name}] - Middleware requested retry`, { + operation: "streamObject", + retryCount: middlewareRetryCount, + maxMiddlewareRetries, + middlewareId: retryError.middlewareId ?? null, + reason: retryError.message ?? "middleware retry", + metadata: + retryError.metadata !== undefined + ? safeStringify(retryError.metadata) + : undefined, + }); + this.storeMiddlewareRetryFeedback(oc, retryError.message, retryError.metadata); + middlewareRetryCount += 1; + continue; + } + throw error; + } + } + effectiveInput = await executeInputGuardrails( - input, + effectiveInput, oc, guardrailSet.input, "streamObject", this, ); - const { messages, uiMessages, model } = await this.prepareExecution( + const { messages, uiMessages, modelName } = await this.prepareExecution( effectiveInput, oc, options, ); - - const modelName = this.getModelName(model); const schemaName = schema.description || "unknown"; // Add model attributes and all options @@ -1998,187 +2394,268 @@ export class Agent { onFinish: userOnFinish, output: _output, providerOptions, - ...aiSDKOptions - } = options || {}; - - let guardrailObjectPromise!: Promise>; - let resolveGuardrailObject: ((value: z.infer) => void) | undefined; - let rejectGuardrailObject: ((reason: unknown) => void) | undefined; - - const result = streamObject({ - model, - messages, - schema, - // Default values - maxOutputTokens: this.maxOutputTokens, - temperature: this.temperature, - maxRetries: 3, - // User overrides from AI SDK options - ...aiSDKOptions, - // Provider-specific options - providerOptions, - // VoltAgent controlled - abortSignal: oc.abortController.signal, - onError: async (errorData) => { - // Handle nested error structure from OpenAI and other providers - // The error might be directly the error or wrapped in { error: ... } - const actualError = (errorData as any)?.error || errorData; - - // Log the error - methodLogger.error("Stream object error occurred", { - error: actualError, - agentName: this.name, - modelName, - schemaName: schemaName, - }); - - // History update removed - using OpenTelemetry only + ...aiSDKOptions + } = options || {}; - // Event tracking now handled by OpenTelemetry spans + let guardrailObjectPromise!: Promise>; + let resolveGuardrailObject: ((value: z.infer) => void) | undefined; + let rejectGuardrailObject: ((reason: unknown) => void) | undefined; - // Call error hooks if they exist - this.getMergedHooks(options).onError?.({ - agent: this, - error: actualError as Error, - context: oc, - }); + const { result, modelName: effectiveModelName } = await this.executeWithModelFallback({ + oc, + operation: "streamObject", + options, + run: async ({ + model: resolvedModel, + modelName: resolvedModelName, + maxRetries, + attempt, + isLastAttempt, + isLastModel, + }) => { + const attemptState: { hasOutput: boolean; lastError?: unknown } = { + hasOutput: false, + }; + const streamResult = streamObject({ + model: resolvedModel, + messages, + schema, + // Default values + maxOutputTokens: this.maxOutputTokens, + temperature: this.temperature, + // User overrides from AI SDK options + ...aiSDKOptions, + maxRetries: 0, + // Provider-specific options + providerOptions, + // VoltAgent controlled + abortSignal: oc.abortController.signal, + onError: async (errorData) => { + // Handle nested error structure from OpenAI and other providers + // The error might be directly the error or wrapped in { error: ... } + const actualError = (errorData as any)?.error || errorData; + attemptState.lastError = actualError; + + const fallbackEligible = this.shouldFallbackOnError(actualError); + const retryEligible = fallbackEligible && this.isRetryableError(actualError); + const canRetry = retryEligible && !isLastAttempt; + const canFallback = fallbackEligible && !isLastModel; + const shouldAttemptRecovery = !attemptState.hasOutput && (canRetry || canFallback); + const recoveryMessage = canRetry + ? "[LLM] Stream object error before output; retry pending" + : canFallback + ? "[LLM] Stream object error before output; fallback pending" + : attemptState.hasOutput + ? "[LLM] Stream object error after output; recovery skipped" + : "[LLM] Stream object error before output; recovery skipped"; + + // Log the error + methodLogger.error("Stream object error occurred", { + error: actualError, + agentName: this.name, + modelName: resolvedModelName, + schemaName: schemaName, + attempt, + maxRetries, + }); - // Close OpenTelemetry span with error status - oc.traceContext.end("error", actualError as Error); - rejectGuardrailObject?.(actualError); - - // Don't re-throw - let the error be part of the stream - // The onError callback should return void for AI SDK compatibility - // Ensure spans are flushed on error - // Uses waitUntil if available to avoid blocking - await flushObservability( - this.getObservability(), - oc.logger ?? this.logger, - this.observabilityAuthWarningState, - "streamObject:onError", - ); - }, - onFinish: async (finalResult: any) => { - try { - const usageInfo = convertUsage(finalResult.usage as any); - let finalObject = finalResult.object as z.infer; - if (guardrailSet.output.length > 0) { - finalObject = await executeOutputGuardrails({ - output: finalResult.object as z.infer, - operationContext: oc, - guardrails: guardrailSet.output, + methodLogger.debug(recoveryMessage, { operation: "streamObject", - agent: this, - metadata: { - usage: usageInfo, - finishReason: finalResult.finishReason ?? null, - warnings: finalResult.warnings ?? null, - }, + modelName: resolvedModelName, + fallbackEligible, + retryEligible, + canRetry, + canFallback, + hasOutput: attemptState.hasOutput, + attempt, + maxRetries, + isLastAttempt, + isLastModel, + isRetryable: (actualError as any)?.isRetryable, + statusCode: (actualError as any)?.statusCode, + errorName: (actualError as Error)?.name, + errorMessage: (actualError as Error)?.message, }); - resolveGuardrailObject?.(finalObject); - } - - if (oc.userId && oc.conversationId) { - const message: UIMessage = { - id: randomUUID(), - role: "assistant", - parts: [ - { - type: "text", - text: safeStringify(finalObject), - }, - ], - }; - await this.memoryManager.saveMessage(oc, message, oc.userId, oc.conversationId); - - const step: StepWithContent = { - id: randomUUID(), - type: "text", - content: safeStringify(finalObject), - role: "assistant", - usage: usageInfo, - }; - this.addStepToHistory(step, oc); - } + // History update removed - using OpenTelemetry only - // Add usage to span - this.setTraceContextUsage(oc.traceContext, finalResult.usage); - oc.traceContext.setOutput(finalObject); + // Event tracking now handled by OpenTelemetry spans - // Set output in operation context - oc.output = finalObject; + if (shouldAttemptRecovery) { + await flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamObject:onError", + ); + return; + } - await this.getMergedHooks(options).onEnd?.({ - conversationId: oc.conversationId || "", - agent: this, - output: { - object: finalObject, - usage: usageInfo, - providerResponse: finalResult.response, - finishReason: finalResult.finishReason, - warnings: finalResult.warnings, - context: oc.context, - }, - error: undefined, - context: oc, - }); + // Call error hooks if they exist + this.getMergedHooks(options).onError?.({ + agent: this, + error: actualError as Error, + context: oc, + }); - if (userOnFinish) { - const guardrailedResult = - guardrailSet.output.length > 0 - ? { ...finalResult, object: finalObject } - : finalResult; - await userOnFinish(guardrailedResult); - } + // Close OpenTelemetry span with error status + oc.traceContext.end("error", actualError as Error); + rejectGuardrailObject?.(actualError); + + // Don't re-throw - let the error be part of the stream + // The onError callback should return void for AI SDK compatibility + // Ensure spans are flushed on error + // Uses waitUntil if available to avoid blocking + await flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamObject:onError", + ); + }, + onFinish: async (finalResult: any) => { + try { + const usageInfo = convertUsage(finalResult.usage as any); + let finalObject = finalResult.object as z.infer; + if (guardrailSet.output.length > 0) { + finalObject = await executeOutputGuardrails({ + output: finalResult.object as z.infer, + operationContext: oc, + guardrails: guardrailSet.output, + operation: "streamObject", + agent: this, + metadata: { + usage: usageInfo, + finishReason: finalResult.finishReason ?? null, + warnings: finalResult.warnings ?? null, + }, + }); + resolveGuardrailObject?.(finalObject); + } - const usage = finalResult.usage as any; - const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; - methodLogger.debug( - buildAgentLogMessage( - this.name, - ActionType.GENERATION_COMPLETE, - `Object generation completed (${tokenInfo})`, - ), - { - event: LogEvents.AGENT_GENERATION_COMPLETED, - duration: Date.now() - startTime, - finishReason: finalResult.finishReason, - usage: finalResult.usage, - schemaName, - }, - ); + if (oc.userId && oc.conversationId) { + const message: UIMessage = { + id: randomUUID(), + role: "assistant", + parts: [ + { + type: "text", + text: safeStringify(finalObject), + }, + ], + }; + + await this.memoryManager.saveMessage(oc, message, oc.userId, oc.conversationId); + + const step: StepWithContent = { + id: randomUUID(), + type: "text", + content: safeStringify(finalObject), + role: "assistant", + usage: usageInfo, + }; + this.addStepToHistory(step, oc); + } - this.enqueueEvalScoring({ - oc, - output: finalObject, - operation: "streamObject", - metadata: { - finishReason: finalResult.finishReason, - usage: finalResult.usage - ? JSON.parse(safeStringify(finalResult.usage)) - : undefined, - schemaName, - }, - }); + // Add usage to span + this.setTraceContextUsage(oc.traceContext, finalResult.usage); + oc.traceContext.setOutput(finalObject); + + // Set output in operation context + oc.output = finalObject; + + await this.getMergedHooks(options).onEnd?.({ + conversationId: oc.conversationId || "", + agent: this, + output: { + object: finalObject, + usage: usageInfo, + providerResponse: finalResult.response, + finishReason: finalResult.finishReason, + warnings: finalResult.warnings, + context: oc.context, + }, + error: undefined, + context: oc, + }); + + if (userOnFinish) { + const guardrailedResult = + guardrailSet.output.length > 0 + ? { ...finalResult, object: finalObject } + : finalResult; + await userOnFinish(guardrailedResult); + } - oc.traceContext.end("completed"); + const usage = finalResult.usage as any; + const tokenInfo = usage ? `${usage.totalTokens} tokens` : "no usage data"; + methodLogger.debug( + buildAgentLogMessage( + this.name, + ActionType.GENERATION_COMPLETE, + `Object generation completed (${tokenInfo})`, + ), + { + event: LogEvents.AGENT_GENERATION_COMPLETED, + duration: Date.now() - startTime, + finishReason: finalResult.finishReason, + usage: finalResult.usage, + schemaName, + }, + ); + + this.enqueueEvalScoring({ + oc, + output: finalObject, + operation: "streamObject", + metadata: { + finishReason: finalResult.finishReason, + usage: finalResult.usage + ? JSON.parse(safeStringify(finalResult.usage)) + : undefined, + schemaName, + }, + }); + + oc.traceContext.end("completed"); + + // Ensure all spans are exported on finish + // Uses waitUntil if available to avoid blocking + await flushObservability( + this.getObservability(), + oc.logger ?? this.logger, + this.observabilityAuthWarningState, + "streamObject:onFinish", + ); + } catch (error) { + rejectGuardrailObject?.(error); + throw error; + } + }, + }); - // Ensure all spans are exported on finish - // Uses waitUntil if available to avoid blocking - await flushObservability( - this.getObservability(), - oc.logger ?? this.logger, - this.observabilityAuthWarningState, - "streamObject:onFinish", - ); - } catch (error) { - rejectGuardrailObject?.(error); - throw error; + const probeResult = await this.probeStreamStart(streamResult.fullStream, attemptState); + if (probeResult.status === "error") { + this.discardStream(streamResult.fullStream); + const fallbackEligible = this.shouldFallbackOnError(probeResult.error); + if (!fallbackEligible || isLastModel) { + throw probeResult.error; + } + throw probeResult.error; } + + return streamResult; }, }); + addModelAttributesToSpan( + rootSpan, + effectiveModelName, + options, + this.maxOutputTokens, + this.temperature, + ); + if (guardrailSet.output.length > 0) { guardrailObjectPromise = new Promise>((resolve, reject) => { resolveGuardrailObject = resolve; @@ -2252,6 +2729,67 @@ export class Agent { }; } + private resolveMiddlewareSets(options?: { + inputMiddlewares?: InputMiddleware[]; + outputMiddlewares?: OutputMiddleware[]; + }): { + input: NormalizedInputMiddleware[]; + output: NormalizedOutputMiddleware[]; + } { + const optionInput = options?.inputMiddlewares + ? normalizeInputMiddlewareList(options.inputMiddlewares, this.inputMiddlewares.length) + : []; + const optionOutput = options?.outputMiddlewares + ? normalizeOutputMiddlewareList(options.outputMiddlewares, this.outputMiddlewares.length) + : []; + + return { + input: [...this.inputMiddlewares, ...optionInput], + output: [...this.outputMiddlewares, ...optionOutput], + }; + } + + private resolveMiddlewareRetries(options?: BaseGenerationOptions): number { + const optionRetries = options?.maxMiddlewareRetries; + if (typeof optionRetries === "number" && Number.isFinite(optionRetries)) { + return Math.max(0, optionRetries); + } + if (Number.isFinite(this.maxMiddlewareRetries)) { + return Math.max(0, this.maxMiddlewareRetries); + } + return 0; + } + + private storeMiddlewareRetryFeedback( + oc: OperationContext, + reason?: string, + metadata?: unknown, + ): void { + const trimmedReason = typeof reason === "string" ? reason.trim() : ""; + const baseReason = trimmedReason.length > 0 ? trimmedReason : "Middleware requested a retry."; + let feedback = `[Middleware Feedback] ${baseReason} Please retry with the feedback in mind.`; + if (metadata !== undefined) { + feedback = `${feedback}\nMetadata: ${safeStringify(metadata)}`; + } + oc.systemContext.set(MIDDLEWARE_RETRY_FEEDBACK_KEY, feedback); + } + + private consumeMiddlewareRetryFeedback(oc: OperationContext): string | null { + const feedback = oc.systemContext.get(MIDDLEWARE_RETRY_FEEDBACK_KEY); + if (typeof feedback === "string" && feedback.trim().length > 0) { + oc.systemContext.delete(MIDDLEWARE_RETRY_FEEDBACK_KEY); + return feedback; + } + return null; + } + + private shouldRetryMiddleware(error: unknown, retryCount: number, maxRetries: number): boolean { + if (!isMiddlewareAbortError(error)) { + return false; + } + return Boolean(error.retry) && retryCount < maxRetries; + } + /** * Common preparation for all execution methods */ @@ -2262,7 +2800,7 @@ export class Agent { ): Promise<{ messages: BaseMessage[]; uiMessages: UIMessage[]; - model: LanguageModel; + modelName: string; tools: Record; maxSteps: number; }> { @@ -2289,8 +2827,7 @@ export class Agent { // Calculate maxSteps (use provided option or calculate based on subagents) const maxSteps = options?.maxSteps ?? this.calculateMaxSteps(); - // Resolve dynamic values - const model = await this.resolveModel(this.model, oc); + const modelName = this.getModelName(); const dynamicToolList = (await this.resolveValue(this.dynamicTools, oc)) || []; // Merge agent tools with option tools @@ -2303,7 +2840,7 @@ export class Agent { return { messages, uiMessages, - model, + modelName, tools, maxSteps, }; @@ -2453,6 +2990,19 @@ export class Agent { }; } + private resetOperationAttemptState(oc: OperationContext): void { + oc.systemContext.set(BUFFER_CONTEXT_KEY, new ConversationBuffer(undefined, oc.logger)); + oc.systemContext.set( + QUEUE_CONTEXT_KEY, + new MemoryPersistQueue(this.memoryManager, { debounceMs: 200, logger: oc.logger }), + ); + oc.systemContext.delete(STEP_PERSIST_COUNT_KEY); + oc.systemContext.delete("conversationSteps"); + oc.systemContext.delete("bailedResult"); + oc.conversationSteps = []; + oc.output = undefined; + } + private getConversationBuffer(oc: OperationContext): ConversationBuffer { let buffer = oc.systemContext.get(BUFFER_CONTEXT_KEY) as ConversationBuffer | undefined; if (!buffer) { @@ -2626,6 +3176,22 @@ export class Agent { if (maxOutputTokens !== undefined) { attrs["llm.max_output_tokens"] = maxOutputTokens; } + const maxRetries = maybeNumber(callOptions.maxRetries ?? callOptions.max_retries); + if (maxRetries !== undefined) { + attrs["llm.max_retries"] = maxRetries; + } + const modelId = callOptions.modelId ?? callOptions.model_id; + if (typeof modelId === "string" && modelId.length > 0) { + attrs["llm.model_id"] = modelId; + } + const attempt = maybeNumber(callOptions.attempt ?? callOptions.attempt_index); + if (attempt !== undefined) { + attrs["llm.attempt"] = attempt; + } + const modelIndex = maybeNumber(callOptions.modelIndex ?? callOptions.model_index); + if (modelIndex !== undefined) { + attrs["llm.model_index"] = modelIndex; + } const topP = maybeNumber(callOptions.topP); if (topP !== undefined) { attrs["llm.top_p"] = topP; @@ -2917,6 +3483,15 @@ export class Agent { } } + const middlewareRetryFeedback = this.consumeMiddlewareRetryFeedback(oc); + if (middlewareRetryFeedback) { + messages.push({ + id: randomUUID(), + role: "system", + parts: [{ type: "text", text: middlewareRetryFeedback }], + }); + } + const canIUseMemory = options?.userId && options.conversationId; // Load memory context if available (already returns UIMessages) @@ -3544,10 +4119,33 @@ export class Agent { return value; } - /** - * Resolve agent model value (LanguageModel or provider/model string) - */ - private async resolveModel(value: AgentModelValue, oc: OperationContext): Promise { + private getModelCandidates(): AgentModelConfig[] { + if (Array.isArray(this.model)) { + if (this.model.length === 0) { + throw createVoltAgentError("Model list is empty", { code: "MODEL_LIST_EMPTY" }); + } + + return this.model.map((entry) => ({ + id: entry.id, + model: entry.model, + maxRetries: entry.maxRetries, + enabled: entry.enabled ?? true, + })); + } + + return [ + { + model: this.model, + maxRetries: this.maxRetries, + enabled: true, + }, + ]; + } + + private async resolveModelReference( + value: AgentModelConfig["model"], + oc: OperationContext, + ): Promise { const resolved = await this.resolveValue(value, oc); if (typeof resolved === "string") { return await ModelProviderRegistry.getInstance().resolveLanguageModel(resolved); @@ -3555,6 +4153,358 @@ export class Agent { return resolved; } + /** + * Resolve agent model value (LanguageModel or provider/model string) + */ + private async resolveModel(value: AgentModelValue, oc: OperationContext): Promise { + if (Array.isArray(value)) { + const enabledModels = value.filter((entry) => entry.enabled !== false); + if (enabledModels.length === 0) { + throw createVoltAgentError("No enabled models configured", { code: "MODEL_LIST_EMPTY" }); + } + return await this.resolveModelReference(enabledModels[0].model, oc); + } + + return await this.resolveModelReference(value, oc); + } + + private resolveCallMaxRetries( + candidate: AgentModelConfig, + options?: BaseGenerationOptions, + ): number { + const optionRetries = options?.maxRetries; + if (typeof optionRetries === "number" && Number.isFinite(optionRetries)) { + return Math.max(0, optionRetries); + } + if (typeof candidate.maxRetries === "number" && Number.isFinite(candidate.maxRetries)) { + return Math.max(0, candidate.maxRetries); + } + if (Number.isFinite(this.maxRetries)) { + return Math.max(0, this.maxRetries); + } + return DEFAULT_LLM_MAX_RETRIES; + } + + private shouldFallbackOnError(error: unknown): boolean { + if (isBailError(error)) { + return false; + } + + if (error instanceof Error && error.name === "AbortError") { + return false; + } + + if (isVoltAgentError(error)) { + if (error.code === "GUARDRAIL_INPUT_BLOCKED" || error.code === "GUARDRAIL_OUTPUT_BLOCKED") { + return false; + } + if (error.stage === "tool_execution") { + return false; + } + } + + return true; + } + + private isRetryableError(error: unknown): boolean { + const retryable = (error as { isRetryable?: boolean } | undefined)?.isRetryable; + if (typeof retryable === "boolean") { + return retryable; + } + return true; + } + + private async executeWithModelFallback({ + oc, + operation, + options, + run, + }: { + oc: OperationContext; + operation: LLMOperation; + options?: BaseGenerationOptions; + run: (args: { + model: LanguageModel; + modelName: string; + modelId?: string; + maxRetries: number; + modelIndex: number; + isLastModel: boolean; + attempt: number; + isLastAttempt: boolean; + }) => Promise; + }): Promise<{ + result: T; + modelName: string; + modelIndex: number; + maxRetries: number; + }> { + const logger = oc.logger ?? this.logger; + const hooks = this.getMergedHooks(options); + const candidates = this.getModelCandidates().filter((entry) => entry.enabled !== false); + + if (candidates.length === 0) { + throw createVoltAgentError("No enabled models configured", { code: "MODEL_LIST_EMPTY" }); + } + + logger.debug(`[Agent:${this.name}] - Model fallback candidates`, { + operation, + candidates: candidates.map((candidate, index) => ({ + index, + id: candidate.id ?? null, + enabled: candidate.enabled ?? true, + model: + typeof candidate.model === "string" + ? candidate.model + : typeof candidate.model === "function" + ? "dynamic" + : candidate.model?.modelId || "unknown", + maxRetries: candidate.maxRetries ?? null, + })), + }); + + let lastError: unknown; + + for (let index = 0; index < candidates.length; index++) { + const candidate = candidates[index]; + const isLastModel = index === candidates.length - 1; + let resolvedModel: LanguageModel; + let modelName = "unknown"; + + try { + resolvedModel = await this.resolveModelReference(candidate.model, oc); + modelName = this.getModelName(resolvedModel); + } catch (error) { + lastError = error; + if (oc.abortController.signal.aborted) { + throw error; + } + const candidateModelName = + typeof candidate.model === "string" + ? candidate.model + : typeof candidate.model === "function" + ? (candidate.id ?? "dynamic") + : (candidate.model?.modelId ?? candidate.id ?? "unknown"); + const modelMaxRetries = this.resolveCallMaxRetries(candidate, options); + const resolveSpan = this.createLLMSpan(oc, { + operation, + modelName: candidateModelName, + isStreaming: operation === "streamText" || operation === "streamObject", + callOptions: { + maxRetries: modelMaxRetries, + modelIndex: index, + attempt: 1, + modelId: candidate.id, + }, + }); + resolveSpan.setAttribute("llm.model_resolution_failed", true); + const resolveError = error instanceof Error ? error : new Error(String(error)); + resolveSpan.recordException(resolveError); + resolveSpan.setStatus({ code: SpanStatusCode.ERROR, message: resolveError.message }); + resolveSpan.end(); + if (!this.shouldFallbackOnError(error) || isLastModel) { + throw error; + } + const nextCandidate = candidates[index + 1]; + const nextCandidateName = + typeof nextCandidate?.model === "string" + ? nextCandidate.model + : typeof nextCandidate?.model === "function" + ? (nextCandidate?.id ?? "dynamic") + : (nextCandidate?.model?.modelId ?? nextCandidate?.id ?? null); + await hooks.onFallback?.({ + agent: this, + context: oc, + operation, + stage: "resolve", + fromModel: candidateModelName, + fromModelIndex: index, + maxRetries: modelMaxRetries, + error, + nextModel: nextCandidateName, + nextModelIndex: nextCandidate ? index + 1 : undefined, + }); + logger.warn(`[Agent:${this.name}] - Failed to resolve model, falling back`, { + error: safeStringify(error), + modelIndex: index, + operation, + }); + continue; + } + + const maxRetries = this.resolveCallMaxRetries(candidate, options); + let attemptIndex = 0; + + while (attemptIndex <= maxRetries) { + const attempt = attemptIndex + 1; + const isLastAttempt = attemptIndex === maxRetries; + + try { + const result = await run({ + model: resolvedModel, + modelName, + modelId: candidate.id, + maxRetries, + modelIndex: index, + isLastModel, + attempt, + isLastAttempt, + }); + return { result, modelName, modelIndex: index, maxRetries }; + } catch (error) { + lastError = error; + if (oc.abortController.signal.aborted) { + throw error; + } + const fallbackEligible = this.shouldFallbackOnError(error); + const retryEligible = fallbackEligible && this.isRetryableError(error); + const canRetry = retryEligible && !isLastAttempt; + + if (canRetry) { + const retryDelayMs = Math.min(1000 * 2 ** attemptIndex, 10000); + logger.debug(`[Agent:${this.name}] - Model attempt failed, retrying`, { + operation, + modelName, + modelIndex: index, + attempt, + nextAttempt: attempt + 1, + maxRetries, + retryDelayMs, + fallbackEligible, + retryEligible, + isRetryable: (error as any)?.isRetryable, + statusCode: (error as any)?.statusCode, + error: safeStringify(error), + }); + await hooks.onRetry?.({ + agent: this, + context: oc, + operation, + source: "llm", + modelName, + modelIndex: index, + attempt, + nextAttempt: attempt + 1, + maxRetries, + error, + isRetryable: (error as any)?.isRetryable, + statusCode: (error as any)?.statusCode, + }); + + attemptIndex += 1; + if (retryDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, retryDelayMs)); + } + continue; + } + + if (!fallbackEligible || isLastModel) { + logger.debug(`[Agent:${this.name}] - Fallback skipped`, { + operation, + modelName, + modelIndex: index, + attempt, + maxRetries, + fallbackEligible, + retryEligible, + isLastModel, + error: safeStringify(error), + }); + throw error; + } + + const nextCandidate = candidates[index + 1]; + const nextCandidateName = + typeof nextCandidate?.model === "string" + ? nextCandidate.model + : typeof nextCandidate?.model === "function" + ? (nextCandidate?.id ?? "dynamic") + : (nextCandidate?.model?.modelId ?? nextCandidate?.id ?? null); + await hooks.onFallback?.({ + agent: this, + context: oc, + operation, + stage: "execute", + fromModel: modelName, + fromModelIndex: index, + maxRetries, + attempt, + error, + nextModel: nextCandidateName, + nextModelIndex: nextCandidate ? index + 1 : undefined, + }); + logger.warn(`[Agent:${this.name}] - Model failed, trying fallback`, { + error: safeStringify(error), + modelName, + modelIndex: index, + operation, + attempt, + maxRetries, + }); + break; + } + } + } + + throw lastError instanceof Error ? lastError : new Error("Model execution failed"); + } + + private async probeStreamStart( + stream: AsyncIterableStream, + state: { hasOutput: boolean; lastError?: unknown }, + ): Promise<{ status: "ok" } | { status: "error"; error: unknown }> { + const readableStream = stream as ReadableStream; + const reader = + readableStream && typeof readableStream.getReader === "function" + ? readableStream.getReader() + : null; + if (!reader) { + return { status: "ok" }; + } + let sawNonStart = false; + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + if (!value) { + continue; + } + const partType = (value as { type?: string }).type; + if (partType === "start") { + continue; + } + sawNonStart = true; + if (partType === "error") { + const error = + (value as { error?: unknown }).error ?? + state.lastError ?? + new Error("Stream error before output"); + return { status: "error", error }; + } + state.hasOutput = true; + return { status: "ok" }; + } + } catch (error) { + return { status: "error", error: state.lastError ?? error }; + } finally { + try { + await reader.cancel(); + } catch (_) { + // Ignore probe cancellation errors. + } + } + + const error = state.lastError ?? new Error("Stream ended before output"); + return sawNonStart ? { status: "ok" } : { status: "error", error }; + } + + private discardStream(stream: AsyncIterableStream): void { + void consumeStream({ stream, onError: () => {} }).catch(() => {}); + } + /** * Prepare tools with execution context */ @@ -4134,6 +5084,14 @@ export class Agent { await options.hooks?.onStepFinish?.(...args); await this.hooks.onStepFinish?.(...args); }, + onRetry: async (...args) => { + await options.hooks?.onRetry?.(...args); + await this.hooks.onRetry?.(...args); + }, + onFallback: async (...args) => { + await options.hooks?.onFallback?.(...args); + await this.hooks.onFallback?.(...args); + }, onPrepareMessages: options.hooks?.onPrepareMessages || this.hooks.onPrepareMessages, onPrepareModelMessages: options.hooks?.onPrepareModelMessages || this.hooks.onPrepareModelMessages, @@ -4145,6 +5103,10 @@ export class Agent { */ private setupAbortSignalListener(oc: OperationContext): void { if (!oc.abortController) return; + if (oc.systemContext.get(ABORT_LISTENER_ATTACHED_KEY)) { + return; + } + oc.systemContext.set(ABORT_LISTENER_ATTACHED_KEY, true); const signal = oc.abortController.signal; signal.addEventListener("abort", async () => { @@ -4291,6 +5253,20 @@ export class Agent { } return model.modelId || "unknown"; } + if (Array.isArray(this.model)) { + const primary = this.model.find((entry) => entry.enabled !== false) ?? this.model[0]; + if (!primary) { + return "unknown"; + } + const modelValue = primary.model; + if (typeof modelValue === "function") { + return "dynamic"; + } + if (typeof modelValue === "string") { + return modelValue; + } + return modelValue.modelId || "unknown"; + } if (typeof this.model === "function") { return "dynamic"; } diff --git a/packages/core/src/agent/errors/index.ts b/packages/core/src/agent/errors/index.ts index 50fe2bb65..9b5f1e9e8 100644 --- a/packages/core/src/agent/errors/index.ts +++ b/packages/core/src/agent/errors/index.ts @@ -4,6 +4,7 @@ import type { ClientHTTPError } from "./client-http-errors"; export type { VoltAgentError } from "./voltagent-error"; export type { AbortError } from "./abort-error"; export type { BailError } from "./bail-error"; +export type { MiddlewareAbortError, MiddlewareAbortOptions } from "./middleware-abort-error"; export { ToolDeniedError, ClientHTTPError, @@ -12,5 +13,6 @@ export { } from "./client-http-errors"; export { createAbortError, isAbortError } from "./abort-error"; export { createBailError, isBailError } from "./bail-error"; +export { createMiddlewareAbortError, isMiddlewareAbortError } from "./middleware-abort-error"; export { createVoltAgentError, isVoltAgentError } from "./voltagent-error"; export type CancellationError = AbortError | ClientHTTPError; diff --git a/packages/core/src/agent/errors/middleware-abort-error.ts b/packages/core/src/agent/errors/middleware-abort-error.ts new file mode 100644 index 000000000..df32962c2 --- /dev/null +++ b/packages/core/src/agent/errors/middleware-abort-error.ts @@ -0,0 +1,34 @@ +export type MiddlewareAbortOptions = { + retry?: boolean; + metadata?: TMetadata; +}; + +/** + * Error thrown by middleware abort() calls. + */ +export class MiddlewareAbortError extends Error { + name: "MiddlewareAbortError"; + retry?: boolean; + metadata?: TMetadata; + middlewareId?: string; + + constructor(reason: string, options?: MiddlewareAbortOptions, middlewareId?: string) { + super(reason); + this.name = "MiddlewareAbortError"; + this.retry = options?.retry; + this.metadata = options?.metadata; + this.middlewareId = middlewareId; + } +} + +export function createMiddlewareAbortError( + reason: string, + options?: MiddlewareAbortOptions, + middlewareId?: string, +): MiddlewareAbortError { + return new MiddlewareAbortError(reason, options, middlewareId); +} + +export function isMiddlewareAbortError(error: unknown): error is MiddlewareAbortError { + return error instanceof MiddlewareAbortError; +} diff --git a/packages/core/src/agent/hooks/index.ts b/packages/core/src/agent/hooks/index.ts index 7b7bc1e95..6896e0e35 100644 --- a/packages/core/src/agent/hooks/index.ts +++ b/packages/core/src/agent/hooks/index.ts @@ -4,7 +4,7 @@ import type { AgentTool } from "../../tool"; import type { Agent } from "../agent"; import type { AbortError, CancellationError, VoltAgentError } from "../errors"; import type { ToolExecuteOptions, UsageInfo } from "../providers/base/types"; -import type { AgentOperationOutput, OperationContext } from "../types"; +import type { AgentEvalOperationType, AgentOperationOutput, OperationContext } from "../types"; // Argument Object Interfaces (old API restored, adapted for AI SDK types) export interface OnStartHookArgs { @@ -114,6 +114,54 @@ export interface OnStepFinishHookArgs { context: OperationContext; } +export type RetrySource = "llm" | "middleware"; + +export interface OnRetryHookArgsBase { + agent: Agent; + context: OperationContext; + operation: AgentEvalOperationType; + source: RetrySource; +} + +export interface OnRetryLLMHookArgs extends OnRetryHookArgsBase { + source: "llm"; + modelName: string; + modelIndex: number; + attempt: number; + nextAttempt: number; + maxRetries: number; + error: unknown; + isRetryable?: boolean; + statusCode?: number; +} + +export interface OnRetryMiddlewareHookArgs extends OnRetryHookArgsBase { + source: "middleware"; + middlewareId?: string | null; + retryCount: number; + maxRetries: number; + reason?: string; + metadata?: unknown; +} + +export type OnRetryHookArgs = OnRetryLLMHookArgs | OnRetryMiddlewareHookArgs; + +export type FallbackStage = "resolve" | "execute"; + +export interface OnFallbackHookArgs { + agent: Agent; + context: OperationContext; + operation: AgentEvalOperationType; + stage: FallbackStage; + fromModel: string; + fromModelIndex: number; + maxRetries: number; + attempt?: number; + error: unknown; + nextModel?: string | null; + nextModelIndex?: number; +} + // Hook Type Aliases (object-arg style) export type AgentHookOnStart = (args: OnStartHookArgs) => Promise | void; export type AgentHookOnEnd = (args: OnEndHookArgs) => Promise | void; @@ -129,6 +177,8 @@ export type AgentHookOnPrepareModelMessages = ( ) => Promise | OnPrepareModelMessagesHookResult; export type AgentHookOnError = (args: OnErrorHookArgs) => Promise | void; export type AgentHookOnStepFinish = (args: OnStepFinishHookArgs) => Promise | void; +export type AgentHookOnRetry = (args: OnRetryHookArgs) => Promise | void; +export type AgentHookOnFallback = (args: OnFallbackHookArgs) => Promise | void; /** * Type definition for agent hooks using single argument objects. @@ -145,6 +195,8 @@ export type AgentHooks = { // Additional (kept for convenience) onError?: AgentHookOnError; onStepFinish?: AgentHookOnStepFinish; + onRetry?: AgentHookOnRetry; + onFallback?: AgentHookOnFallback; }; /** @@ -161,6 +213,8 @@ const defaultHooks: Required = { onPrepareModelMessages: async (_args: OnPrepareModelMessagesHookArgs) => ({}), onError: async (_args: OnErrorHookArgs) => {}, onStepFinish: async (_args: OnStepFinishHookArgs) => {}, + onRetry: async (_args: OnRetryHookArgs) => {}, + onFallback: async (_args: OnFallbackHookArgs) => {}, }; /** @@ -178,5 +232,7 @@ export function createHooks(hooks: Partial = {}): AgentHooks { onPrepareModelMessages: hooks.onPrepareModelMessages || defaultHooks.onPrepareModelMessages, onError: hooks.onError || defaultHooks.onError, onStepFinish: hooks.onStepFinish || defaultHooks.onStepFinish, + onRetry: hooks.onRetry || defaultHooks.onRetry, + onFallback: hooks.onFallback || defaultHooks.onFallback, }; } diff --git a/packages/core/src/agent/index.ts b/packages/core/src/agent/index.ts index f94ad5a45..6fbadb28f 100644 --- a/packages/core/src/agent/index.ts +++ b/packages/core/src/agent/index.ts @@ -17,8 +17,19 @@ export type { OutputGuardrailStreamArgs, OutputGuardrailStreamResult, OutputGuardrailStreamHandler, + InputMiddleware, + OutputMiddleware, + InputMiddlewareArgs, + OutputMiddlewareArgs, + InputMiddlewareResult, + OutputMiddlewareResult, + MiddlewareFunction, + MiddlewareDefinition, + MiddlewareDirection, + MiddlewareContext, } from "./types"; export type { CreateInputGuardrailOptions, CreateOutputGuardrailOptions } from "./guardrail"; +export type { CreateInputMiddlewareOptions, CreateOutputMiddlewareOptions } from "./middleware"; export { createSensitiveNumberGuardrail, createEmailRedactorGuardrail, @@ -35,3 +46,4 @@ export { createDefaultSafetyGuardrails, } from "./guardrails/defaults"; export { createInputGuardrail, createOutputGuardrail } from "./guardrail"; +export { createInputMiddleware, createOutputMiddleware } from "./middleware"; diff --git a/packages/core/src/agent/middleware.ts b/packages/core/src/agent/middleware.ts new file mode 100644 index 000000000..29163884b --- /dev/null +++ b/packages/core/src/agent/middleware.ts @@ -0,0 +1,347 @@ +import { SpanStatusCode } from "@opentelemetry/api"; +import { safeStringify } from "@voltagent/internal/utils"; +import type { UIMessage } from "ai"; +import { NodeType } from "../utils/node-utils"; +import type { Agent } from "./agent"; +import type { MiddlewareAbortOptions } from "./errors"; +import { createMiddlewareAbortError } from "./errors"; +import type { BaseMessage } from "./providers/base/types"; +import type { + AgentEvalOperationType, + InputMiddleware, + InputMiddlewareArgs, + InputMiddlewareResult, + MiddlewareDefinition, + MiddlewareDirection, + MiddlewareFunction, + MiddlewareFunctionMetadata, + OperationContext, + OutputMiddleware, + OutputMiddlewareArgs, + OutputMiddlewareResult, +} from "./types"; + +export interface NormalizedMiddleware { + id?: string; + name: string; + description?: string; + tags?: string[]; + metadata?: Record; + handler: MiddlewareFunction; +} + +export type NormalizedInputMiddleware = NormalizedMiddleware< + InputMiddlewareArgs, + InputMiddlewareResult +>; +export type NormalizedOutputMiddleware = NormalizedMiddleware< + OutputMiddlewareArgs, + OutputMiddlewareResult +>; + +type EmptyMiddlewareExtras = Record; + +type CreateMiddlewareDefinition< + TArgs, + TResult, + TExtra extends Record = EmptyMiddlewareExtras, +> = Omit, keyof TExtra | "handler"> & + TExtra & { + handler: MiddlewareFunction; + }; + +export type CreateInputMiddlewareOptions = CreateMiddlewareDefinition< + InputMiddlewareArgs, + InputMiddlewareResult +>; + +export type CreateOutputMiddlewareOptions = CreateMiddlewareDefinition< + OutputMiddlewareArgs, + OutputMiddlewareResult +>; + +export function createInputMiddleware(options: CreateInputMiddlewareOptions): InputMiddleware { + return { + id: options.id, + name: options.name, + description: options.description, + tags: options.tags, + metadata: options.metadata, + handler: options.handler, + }; +} + +export function createOutputMiddleware( + options: CreateOutputMiddlewareOptions, +): OutputMiddleware { + return { + id: options.id, + name: options.name, + description: options.description, + tags: options.tags, + metadata: options.metadata, + handler: options.handler, + }; +} + +export function getDefaultMiddlewareName(direction: MiddlewareDirection, index: number): string { + const label = direction === "input" ? "Input" : "Output"; + return `${label} Middleware #${index + 1}`; +} + +export function normalizeMiddlewareDefinition( + middleware: MiddlewareDefinition | MiddlewareFunction, + direction: MiddlewareDirection, + index: number, +): NormalizedMiddleware { + const defaultName = getDefaultMiddlewareName(direction, index); + + if (typeof middleware === "function") { + const handler = middleware as MiddlewareFunction & MiddlewareFunctionMetadata; + return { + id: handler.middlewareId, + name: handler.middlewareName || handler.name || defaultName, + description: handler.middlewareDescription, + tags: handler.middlewareTags, + metadata: undefined, + handler, + }; + } + + if (typeof middleware !== "object" || !middleware) { + throw new Error(`Invalid ${direction} middleware configuration at index ${index}`); + } + + const descriptor = middleware as MiddlewareDefinition; + return { + id: descriptor.id, + name: descriptor.name || defaultName, + description: descriptor.description, + tags: descriptor.tags, + metadata: descriptor.metadata, + handler: descriptor.handler, + }; +} + +export function normalizeInputMiddlewareList( + middlewares: InputMiddleware[], + startIndex = 0, +): NormalizedInputMiddleware[] { + return middlewares.map((middleware, index) => + normalizeMiddlewareDefinition( + middleware, + "input", + startIndex + index, + ), + ); +} + +export function normalizeOutputMiddlewareList( + middlewares: OutputMiddleware[], + startIndex = 0, +): NormalizedOutputMiddleware[] { + return middlewares.map((middleware, index) => + normalizeMiddlewareDefinition, OutputMiddlewareResult>( + middleware, + "output", + startIndex + index, + ), + ) as NormalizedOutputMiddleware[]; +} + +function serializeMiddlewareValue(value: unknown): string { + if (typeof value === "string") { + return value; + } + return safeStringify(value); +} + +export async function runInputMiddlewares( + input: string | UIMessage[] | BaseMessage[], + oc: OperationContext, + middlewares: NormalizedInputMiddleware[], + operation: AgentEvalOperationType, + agent: Agent, + retryCount: number, +): Promise { + if (!middlewares.length) { + return input; + } + + const originalInput = input; + let currentInput = input; + + for (let index = 0; index < middlewares.length; index++) { + const middleware = middlewares[index]; + const span = oc.traceContext.createChildSpan( + `middleware.input.${middleware.id ?? index + 1}`, + "middleware", + { + label: middleware.name, + attributes: { + "middleware.type": NodeType.MIDDLEWARE, + "middleware.direction": "input", + "middleware.operation": operation, + "middleware.index": index, + ...(middleware.id ? { "middleware.id": middleware.id } : {}), + "middleware.name": middleware.name, + ...(middleware.description ? { "middleware.description": middleware.description } : {}), + ...(middleware.tags && middleware.tags.length > 0 + ? { "middleware.tags": safeStringify(middleware.tags) } + : {}), + ...(middleware.metadata + ? { "middleware.metadata": safeStringify(middleware.metadata) } + : {}), + "middleware.retry_count": retryCount, + "middleware.input.original": serializeMiddlewareValue(originalInput), + "middleware.input.current": serializeMiddlewareValue(currentInput), + }, + }, + ); + + try { + const abort = ( + reason?: string, + options?: MiddlewareAbortOptions, + ): never => { + const message = reason ?? `Middleware aborted: ${middleware.name}`; + throw createMiddlewareAbortError(message, options, middleware.id); + }; + + const result = await oc.traceContext.withSpan(span, () => + middleware.handler({ + input: currentInput, + originalInput, + agent, + context: oc, + operation, + retryCount, + abort, + }), + ); + + if (result !== undefined) { + currentInput = result as typeof currentInput; + } + + span.setAttribute("middleware.input.after", serializeMiddlewareValue(currentInput)); + span.setStatus({ code: SpanStatusCode.OK }); + span.end(); + } catch (error) { + if (error instanceof Error) { + span.recordException(error); + span.setStatus({ code: SpanStatusCode.ERROR, message: error.message }); + } else { + span.setStatus({ code: SpanStatusCode.ERROR, message: "Middleware error" }); + } + span.end(); + throw error; + } + } + + if (currentInput !== originalInput) { + oc.traceContext.setInput(currentInput); + } + + return currentInput; +} + +export async function runOutputMiddlewares( + output: TOutput, + oc: OperationContext, + middlewares: NormalizedOutputMiddleware[], + operation: AgentEvalOperationType, + agent: Agent, + retryCount: number, + metadata?: { usage?: unknown; finishReason?: string | null; warnings?: unknown[] | null }, +): Promise { + if (!middlewares.length) { + return output; + } + + const originalOutput = output; + let currentOutput = output; + + for (let index = 0; index < middlewares.length; index++) { + const middleware = middlewares[index]; + const span = oc.traceContext.createChildSpan( + `middleware.output.${middleware.id ?? index + 1}`, + "middleware", + { + label: middleware.name, + attributes: { + "middleware.type": NodeType.MIDDLEWARE, + "middleware.direction": "output", + "middleware.operation": operation, + "middleware.index": index, + ...(middleware.id ? { "middleware.id": middleware.id } : {}), + "middleware.name": middleware.name, + ...(middleware.description ? { "middleware.description": middleware.description } : {}), + ...(middleware.tags && middleware.tags.length > 0 + ? { "middleware.tags": safeStringify(middleware.tags) } + : {}), + ...(middleware.metadata + ? { "middleware.metadata": safeStringify(middleware.metadata) } + : {}), + "middleware.retry_count": retryCount, + "middleware.output.original": serializeMiddlewareValue(originalOutput), + "middleware.output.current": serializeMiddlewareValue(currentOutput), + }, + }, + ); + + if (metadata?.usage !== undefined) { + span.setAttribute("middleware.usage", safeStringify(metadata.usage)); + } + if (metadata?.finishReason !== undefined && metadata.finishReason !== null) { + span.setAttribute("middleware.finish_reason", metadata.finishReason); + } + if (metadata?.warnings && metadata.warnings.length > 0) { + span.setAttribute("middleware.warnings", safeStringify(metadata.warnings)); + } + + try { + const abort = ( + reason?: string, + options?: MiddlewareAbortOptions, + ): never => { + const message = reason ?? `Middleware aborted: ${middleware.name}`; + throw createMiddlewareAbortError(message, options, middleware.id); + }; + + const result = await oc.traceContext.withSpan(span, () => + middleware.handler({ + output: currentOutput, + originalOutput, + agent, + context: oc, + operation, + retryCount, + usage: metadata?.usage as any, + finishReason: metadata?.finishReason ?? null, + warnings: metadata?.warnings ?? null, + abort, + } as OutputMiddlewareArgs), + ); + + if (result !== undefined) { + currentOutput = result as TOutput; + } + + span.setAttribute("middleware.output.after", serializeMiddlewareValue(currentOutput)); + span.setStatus({ code: SpanStatusCode.OK }); + span.end(); + } catch (error) { + if (error instanceof Error) { + span.recordException(error); + span.setStatus({ code: SpanStatusCode.ERROR, message: error.message }); + } else { + span.setStatus({ code: SpanStatusCode.ERROR, message: "Middleware error" }); + } + span.end(); + throw error; + } + } + + return currentOutput; +} diff --git a/packages/core/src/agent/open-telemetry/trace-context.ts b/packages/core/src/agent/open-telemetry/trace-context.ts index d9f2c87d1..7bb2cad57 100644 --- a/packages/core/src/agent/open-telemetry/trace-context.ts +++ b/packages/core/src/agent/open-telemetry/trace-context.ts @@ -164,6 +164,7 @@ export class AgentTraceContext { | "vector" | "agent" | "guardrail" + | "middleware" | "llm" | "summary", options?: { @@ -200,6 +201,7 @@ export class AgentTraceContext { | "vector" | "agent" | "guardrail" + | "middleware" | "llm" | "summary", options?: { diff --git a/packages/core/src/agent/types.ts b/packages/core/src/agent/types.ts index 50badfc9d..520c89fba 100644 --- a/packages/core/src/agent/types.ts +++ b/packages/core/src/agent/types.ts @@ -17,7 +17,7 @@ import type { StreamEvent } from "../utils/streams"; import type { Voice } from "../voice/types"; import type { VoltOpsClient } from "../voltops/client"; import type { Agent } from "./agent"; -import type { CancellationError, VoltAgentError } from "./errors"; +import type { CancellationError, MiddlewareAbortOptions, VoltAgentError } from "./errors"; import type { LLMProvider } from "./providers"; import type { BaseTool } from "./providers"; import type { StepWithContent } from "./providers"; @@ -179,9 +179,33 @@ export type ModelDynamicValue = T | DynamicValue; export type AgentModelReference = LanguageModel | ModelRouterModelId; /** - * Agent model value that can be static or dynamic + * Model fallback configuration for agents. */ -export type AgentModelValue = ModelDynamicValue; +export type AgentModelConfig = { + /** + * Optional stable identifier for the model entry (useful for logging). + */ + id?: string; + /** + * Model reference (static or dynamic). + */ + model: ModelDynamicValue; + /** + * Maximum number of retries for this model before falling back. + * Defaults to the agent's maxRetries. + */ + maxRetries?: number; + /** + * Whether this model is enabled for fallback selection. + * @default true + */ + enabled?: boolean; +}; + +/** + * Agent model value that can be static, dynamic, or a fallback list. + */ +export type AgentModelValue = ModelDynamicValue | AgentModelConfig[]; /** * Enhanced dynamic value for tools that supports static or dynamic values @@ -450,6 +474,71 @@ export type OutputGuardrail = | OutputGuardrailFunction | OutputGuardrailDefinition; +// ----------------------------------------------------------------------------- +// Middleware Types +// ----------------------------------------------------------------------------- + +export type MiddlewareDirection = "input" | "output"; + +export type MiddlewareFunctionMetadata = { + middlewareId?: string; + middlewareName?: string; + middlewareDescription?: string; + middlewareTags?: string[]; +}; + +export type MiddlewareFunction = ((args: TArgs) => TResult | Promise) & + MiddlewareFunctionMetadata; + +export interface MiddlewareDefinition { + id?: string; + name?: string; + description?: string; + tags?: string[]; + metadata?: Record; + handler: MiddlewareFunction; +} + +export interface MiddlewareContext { + agent: Agent; + context: OperationContext; + operation: AgentEvalOperationType; + retryCount: number; +} + +export interface InputMiddlewareArgs extends MiddlewareContext { + input: string | UIMessage[] | BaseMessage[]; + originalInput: string | UIMessage[] | BaseMessage[]; + abort: ( + reason?: string, + options?: MiddlewareAbortOptions, + ) => never; +} + +export type InputMiddlewareResult = string | UIMessage[] | BaseMessage[] | undefined; + +export interface OutputMiddlewareArgs extends MiddlewareContext { + output: TOutput; + originalOutput: TOutput; + usage?: UsageInfo; + finishReason?: string | null; + warnings?: unknown[] | null; + abort: ( + reason?: string, + options?: MiddlewareAbortOptions, + ) => never; +} + +export type OutputMiddlewareResult = TOutput | undefined; + +export type InputMiddleware = + | MiddlewareDefinition + | MiddlewareFunction; + +export type OutputMiddleware = + | MiddlewareDefinition, OutputMiddlewareResult> + | MiddlewareFunction, OutputMiddlewareResult>; + export type AgentSummarizationOptions = { enabled?: boolean; triggerTokens?: number; @@ -493,10 +582,24 @@ export type AgentOptions = { inputGuardrails?: InputGuardrail[]; outputGuardrails?: OutputGuardrail[]; + // Middleware + inputMiddlewares?: InputMiddleware[]; + outputMiddlewares?: OutputMiddleware[]; + /** + * Default retry count for middleware-triggered retries. + * Per-call maxMiddlewareRetries overrides this value. + */ + maxMiddlewareRetries?: number; + // Configuration temperature?: number; maxOutputTokens?: number; maxSteps?: number; + /** + * Default retry count for model calls before falling back. + * Overridden by per-model maxRetries or per-call maxRetries. + */ + maxRetries?: number; feedback?: AgentFeedbackOptions | boolean; /** * Default stop condition for step execution (ai-sdk `stopWhen`). diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 5f56710f3..b57009ddb 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -81,10 +81,15 @@ export { createDefaultSafetyGuardrails, } from "./agent/guardrails/defaults"; export { createInputGuardrail, createOutputGuardrail } from "./agent/guardrail"; +export { createInputMiddleware, createOutputMiddleware } from "./agent/middleware"; export type { CreateInputGuardrailOptions, CreateOutputGuardrailOptions, } from "./agent/guardrail"; +export type { + CreateInputMiddlewareOptions, + CreateOutputMiddlewareOptions, +} from "./agent/middleware"; // Observability exports export { VoltAgentObservability } from "./observability"; @@ -166,6 +171,7 @@ export type { AgentOptions, AgentSummarizationOptions, AgentModelReference, + AgentModelConfig, AgentModelValue, AgentFeedbackOptions, AgentFeedbackMetadata, @@ -207,10 +213,25 @@ export type { InputGuardrailResult, OutputGuardrailArgs, OutputGuardrailResult, + InputMiddleware, + OutputMiddleware, + InputMiddlewareArgs, + OutputMiddlewareArgs, + InputMiddlewareResult, + OutputMiddlewareResult, + MiddlewareFunction, + MiddlewareDefinition, + MiddlewareDirection, + MiddlewareContext, } from "./agent/types"; -export type { VoltAgentError, AbortError } from "./agent/errors"; +export type { + VoltAgentError, + AbortError, + MiddlewareAbortError, + MiddlewareAbortOptions, +} from "./agent/errors"; export { ToolDeniedError, ClientHTTPError } from "./agent/errors"; -export { isAbortError, isVoltAgentError } from "./agent/errors"; +export { isAbortError, isMiddlewareAbortError, isVoltAgentError } from "./agent/errors"; export type { AgentHooks } from "./agent/hooks"; export * from "./types"; export * from "./utils"; diff --git a/packages/core/src/utils/node-utils.ts b/packages/core/src/utils/node-utils.ts index 3efba84ee..9826d1bf0 100644 --- a/packages/core/src/utils/node-utils.ts +++ b/packages/core/src/utils/node-utils.ts @@ -9,6 +9,7 @@ export enum NodeType { MESSAGE = "message", OUTPUT = "output", GUARDRAIL = "guardrail", + MIDDLEWARE = "middleware", RETRIEVER = "retriever", VECTOR = "vector", EMBEDDING = "embedding", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0749bb726..279985b09 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1686,6 +1686,43 @@ importers: specifier: ^5.8.2 version: 5.9.2 + examples/with-middleware: + dependencies: + '@ai-sdk/openai': + specifier: ^3.0.0 + version: 3.0.12(zod@3.25.76) + '@voltagent/cli': + specifier: ^0.1.21 + version: link:../../packages/cli + '@voltagent/core': + specifier: ^2.1.2 + version: link:../../packages/core + '@voltagent/libsql': + specifier: ^2.0.2 + version: link:../../packages/libsql + '@voltagent/logger': + specifier: ^2.0.2 + version: link:../../packages/logger + '@voltagent/server-hono': + specifier: ^2.0.3 + version: link:../../packages/server-hono + ai: + specifier: ^6.0.0 + version: 6.0.3(zod@3.25.76) + zod: + specifier: ^3.25.76 + version: 3.25.76 + devDependencies: + '@types/node': + specifier: ^24.2.1 + version: 24.6.2 + tsx: + specifier: ^4.19.3 + version: 4.20.4 + typescript: + specifier: ^5.8.2 + version: 5.9.3 + examples/with-nestjs: dependencies: '@nestjs/common': @@ -2471,6 +2508,43 @@ importers: specifier: ^5.8.2 version: 5.9.2 + examples/with-retries-fallback: + dependencies: + '@ai-sdk/openai': + specifier: ^3.0.0 + version: 3.0.12(zod@3.25.76) + '@voltagent/cli': + specifier: ^0.1.21 + version: link:../../packages/cli + '@voltagent/core': + specifier: ^2.1.2 + version: link:../../packages/core + '@voltagent/libsql': + specifier: ^2.0.2 + version: link:../../packages/libsql + '@voltagent/logger': + specifier: ^2.0.2 + version: link:../../packages/logger + '@voltagent/server-hono': + specifier: ^2.0.3 + version: link:../../packages/server-hono + ai: + specifier: ^6.0.0 + version: 6.0.3(zod@3.25.76) + zod: + specifier: ^3.25.76 + version: 3.25.76 + devDependencies: + '@types/node': + specifier: ^24.2.1 + version: 24.6.2 + tsx: + specifier: ^4.19.3 + version: 4.20.4 + typescript: + specifier: ^5.8.2 + version: 5.9.3 + examples/with-retrieval: dependencies: '@voltagent/cli': @@ -4350,15 +4424,15 @@ packages: zod: 3.25.76 dev: false - /@ai-sdk/anthropic@3.0.1(zod@4.2.1): + /@ai-sdk/anthropic@3.0.1(zod@4.3.5): resolution: {integrity: sha512-MOiwKs76ilEmau/WRMnGWlheTUoB+cbvXCse+SAtpW5ATLreInsuYlspLABn12Dxu3w1Xzke1dT+tmEnxhy9SA==} engines: {node: '>=18'} peerDependencies: zod: ^3.25.76 || ^4.1.8 dependencies: '@ai-sdk/provider': 3.0.0 - '@ai-sdk/provider-utils': 4.0.1(zod@4.2.1) - zod: 4.2.1 + '@ai-sdk/provider-utils': 4.0.1(zod@4.3.5) + zod: 4.3.5 dev: false /@ai-sdk/azure@1.3.25(zod@3.25.76): @@ -4581,15 +4655,15 @@ packages: zod: 3.25.76 dev: false - /@ai-sdk/google@3.0.1(zod@4.2.1): + /@ai-sdk/google@3.0.1(zod@4.3.5): resolution: {integrity: sha512-gh7i4lEvd1CElmefkq7+RoUhNkhP2OTshzVxSt7/Vh2AV5wTPLhduKJMg1c7SFwErytqffO3el/M/LlfCsqzEw==} engines: {node: '>=18'} peerDependencies: zod: ^3.25.76 || ^4.1.8 dependencies: '@ai-sdk/provider': 3.0.0 - '@ai-sdk/provider-utils': 4.0.1(zod@4.2.1) - zod: 4.2.1 + '@ai-sdk/provider-utils': 4.0.1(zod@4.3.5) + zod: 4.3.5 dev: false /@ai-sdk/groq@1.2.9(zod@3.25.76): @@ -4709,26 +4783,26 @@ packages: zod: 3.25.76 dev: false - /@ai-sdk/openai@3.0.1(zod@4.2.1): - resolution: {integrity: sha512-P+qxz2diOrh8OrpqLRg+E+XIFVIKM3z2kFjABcCJGHjGbXBK88AJqmuKAi87qLTvTe/xn1fhZBjklZg9bTyigw==} + /@ai-sdk/openai@3.0.12(zod@3.25.76): + resolution: {integrity: sha512-zqLWEKuaKnjXhu7xCw1jgz/+yTbd3F7EtgU4T2Q8BAo8OJC5wZv14l+kwM7Jai7M1/2Y2T/zBkrfiIu+7NsvfQ==} engines: {node: '>=18'} peerDependencies: zod: ^3.25.76 || ^4.1.8 dependencies: - '@ai-sdk/provider': 3.0.0 - '@ai-sdk/provider-utils': 4.0.1(zod@4.2.1) - zod: 4.2.1 + '@ai-sdk/provider': 3.0.4 + '@ai-sdk/provider-utils': 4.0.8(zod@3.25.76) + zod: 3.25.76 dev: false - /@ai-sdk/openai@3.0.12(zod@3.25.76): + /@ai-sdk/openai@3.0.12(zod@4.3.5): resolution: {integrity: sha512-zqLWEKuaKnjXhu7xCw1jgz/+yTbd3F7EtgU4T2Q8BAo8OJC5wZv14l+kwM7Jai7M1/2Y2T/zBkrfiIu+7NsvfQ==} engines: {node: '>=18'} peerDependencies: zod: ^3.25.76 || ^4.1.8 dependencies: '@ai-sdk/provider': 3.0.4 - '@ai-sdk/provider-utils': 4.0.8(zod@3.25.76) - zod: 3.25.76 + '@ai-sdk/provider-utils': 4.0.8(zod@4.3.5) + zod: 4.3.5 dev: false /@ai-sdk/perplexity@1.1.9(zod@3.25.76): @@ -4860,7 +4934,7 @@ packages: eventsource-parser: 3.0.6 zod: 3.25.76 - /@ai-sdk/provider-utils@4.0.1(zod@4.2.1): + /@ai-sdk/provider-utils@4.0.1(zod@4.3.5): resolution: {integrity: sha512-de2v8gH9zj47tRI38oSxhQIewmNc+OZjYIOOaMoVWKL65ERSav2PYYZHPSPCrfOeLMkv+Dyh8Y0QGwkO29wMWQ==} engines: {node: '>=18'} peerDependencies: @@ -4869,21 +4943,21 @@ packages: '@ai-sdk/provider': 3.0.0 '@standard-schema/spec': 1.1.0 eventsource-parser: 3.0.6 - zod: 4.2.1 - dev: false + zod: 4.3.5 - /@ai-sdk/provider-utils@4.0.1(zod@4.3.5): - resolution: {integrity: sha512-de2v8gH9zj47tRI38oSxhQIewmNc+OZjYIOOaMoVWKL65ERSav2PYYZHPSPCrfOeLMkv+Dyh8Y0QGwkO29wMWQ==} + /@ai-sdk/provider-utils@4.0.8(zod@3.25.76): + resolution: {integrity: sha512-ns9gN7MmpI8vTRandzgz+KK/zNMLzhrriiKECMt4euLtQFSBgNfydtagPOX4j4pS1/3KvHF6RivhT3gNQgBZsg==} engines: {node: '>=18'} peerDependencies: zod: ^3.25.76 || ^4.1.8 dependencies: - '@ai-sdk/provider': 3.0.0 + '@ai-sdk/provider': 3.0.4 '@standard-schema/spec': 1.1.0 eventsource-parser: 3.0.6 - zod: 4.3.5 + zod: 3.25.76 + dev: false - /@ai-sdk/provider-utils@4.0.8(zod@3.25.76): + /@ai-sdk/provider-utils@4.0.8(zod@4.3.5): resolution: {integrity: sha512-ns9gN7MmpI8vTRandzgz+KK/zNMLzhrriiKECMt4euLtQFSBgNfydtagPOX4j4pS1/3KvHF6RivhT3gNQgBZsg==} engines: {node: '>=18'} peerDependencies: @@ -4892,7 +4966,7 @@ packages: '@ai-sdk/provider': 3.0.4 '@standard-schema/spec': 1.1.0 eventsource-parser: 3.0.6 - zod: 3.25.76 + zod: 4.3.5 dev: false /@ai-sdk/provider@1.1.3: @@ -5071,12 +5145,12 @@ packages: resolution: {integrity: sha512-z4VIND5xkPNgKUpdncPhJmZ45fMrKgZUVus3p9wkbVXteia7MNQch4mw6ma/hjmeTcAW4Ym7m7HCy3gFhwWArQ==} engines: {node: '>=18'} dependencies: - '@ai-sdk/anthropic': 3.0.1(zod@4.2.1) - '@ai-sdk/google': 3.0.1(zod@4.2.1) - '@ai-sdk/openai': 3.0.1(zod@4.2.1) + '@ai-sdk/anthropic': 3.0.1(zod@4.3.5) + '@ai-sdk/google': 3.0.1(zod@4.3.5) + '@ai-sdk/openai': 3.0.12(zod@4.3.5) '@ai-sdk/provider': 3.0.0 - '@ai-sdk/provider-utils': 4.0.1(zod@4.2.1) - zod: 4.2.1 + '@ai-sdk/provider-utils': 4.0.1(zod@4.3.5) + zod: 4.3.5 dev: false /@alloc/quick-lru@5.2.0: @@ -10853,11 +10927,11 @@ packages: ai: ^6.0.0 dependencies: '@ai-sdk/provider': 3.0.0 - '@ai-sdk/provider-utils': 4.0.1(zod@4.2.1) + '@ai-sdk/provider-utils': 4.0.1(zod@4.3.5) '@sap-ai-sdk/orchestration': 2.5.0 ai: 6.0.3(zod@3.25.76) - zod: 4.2.1 - zod-to-json-schema: 3.25.1(zod@4.2.1) + zod: 4.3.5 + zod-to-json-schema: 3.25.1(zod@4.3.5) transitivePeerDependencies: - debug - supports-color @@ -11366,7 +11440,7 @@ packages: validate-npm-package-name: 5.0.1 yaml: 2.8.1 yargs: 17.7.2 - zod: 4.2.1 + zod: 4.3.5 dev: true /@netlify/dev-utils@4.1.3: @@ -19689,10 +19763,10 @@ packages: js-yaml: 4.1.0 linear-sum-assignment: 1.0.7 mustache: 4.2.0 - openai: 5.23.2(zod@4.2.1) + openai: 5.23.2(zod@4.3.5) ts-pattern: 5.8.0 vite-node: 3.2.4(@types/node@24.2.1)(jiti@2.5.1)(tsx@4.20.4) - zod: 4.2.1 + zod: 4.3.5 transitivePeerDependencies: - '@types/node' - encoding @@ -19743,7 +19817,7 @@ packages: tailwind-merge: 3.4.0 tailwindcss: 4.1.14 tw-animate-css: 1.4.0 - zod: 4.2.1 + zod: 4.3.5 transitivePeerDependencies: - '@azure/app-configuration' - '@azure/cosmos' @@ -32589,7 +32663,7 @@ packages: dependencies: zod: 3.25.76 - /openai@5.23.2(zod@4.2.1): + /openai@5.23.2(zod@4.3.5): resolution: {integrity: sha512-MQBzmTulj+MM5O8SKEk/gL8a7s5mktS9zUtAkU257WjvobGc9nKcBuVwjyEEcb9SI8a8Y2G/mzn3vm9n1Jlleg==} hasBin: true peerDependencies: @@ -32601,7 +32675,7 @@ packages: zod: optional: true dependencies: - zod: 4.2.1 + zod: 4.3.5 dev: true /openapi-types@12.1.3: @@ -40031,12 +40105,12 @@ packages: dependencies: zod: 3.25.76 - /zod-to-json-schema@3.25.1(zod@4.2.1): + /zod-to-json-schema@3.25.1(zod@4.3.5): resolution: {integrity: sha512-pM/SU9d3YAggzi6MtR4h7ruuQlqKtad8e9S0fmxcMi+ueAK5Korys/aWcV9LIIHTVbj01NdzxcnXSN+O74ZIVA==} peerDependencies: zod: ^3.25 || ^4 dependencies: - zod: 4.2.1 + zod: 4.3.5 dev: false /zod-validation-error@3.5.3(zod@3.25.76): @@ -40062,9 +40136,6 @@ packages: resolution: {integrity: sha512-AvvthqfqrAhNH9dnfmrfKzX5upOdjUVJYFqNSlkmGf64gRaTzlPwz99IHYnVs28qYAybvAlBV+H7pn0saFY4Ig==} dev: false - /zod@4.2.1: - resolution: {integrity: sha512-0wZ1IRqGGhMP76gLqz8EyfBXKk0J2qo2+H3fi4mcUP/KtTocoX08nmIAHl1Z2kJIZbZee8KOpBCSNPRgauucjw==} - /zod@4.3.5: resolution: {integrity: sha512-k7Nwx6vuWx1IJ9Bjuf4Zt1PEllcwe7cls3VNzm4CQ1/hgtFUK2bRNG3rvnpPUhFjmqJKAKtjV576KnUkHocg/g==} diff --git a/website/docs/agents/hooks.md b/website/docs/agents/hooks.md index 48488f4c2..06d2fa038 100644 --- a/website/docs/agents/hooks.md +++ b/website/docs/agents/hooks.md @@ -345,6 +345,40 @@ onEnd: async ({ agent, output, error, conversationId, context }) => { }; ``` +### `onRetry` + +- **Triggered:** When VoltAgent schedules a retry for an LLM attempt or a middleware abort. +- **Argument Object (`OnRetryHookArgs`):** Includes `source`, `operation`, `context`, and retry details. +- **Use Cases:** Retry metrics, alerts, tracking retry sources. + +```ts +onRetry: async (args) => { + if (args.source === "llm") { + console.log( + `LLM retry ${args.nextAttempt}/${args.maxRetries + 1} for ${args.modelName} (${args.operation})` + ); + } else { + console.log( + `Middleware retry ${args.retryCount + 1}/${args.maxRetries + 1} for ${args.middlewareId ?? "unknown"}` + ); + } +}; +``` + +### `onFallback` + +- **Triggered:** When VoltAgent selects the next model candidate. +- **Argument Object (`OnFallbackHookArgs`):** Includes `stage`, `fromModel`, `fromModelIndex`, `nextModel`, and `error`. +- **Use Cases:** Tracking model failover, audit logs, fallback diagnostics. + +```ts +onFallback: async ({ stage, fromModel, nextModel, operation }) => { + console.warn( + `Fallback (${stage}) from ${fromModel} to ${nextModel ?? "next"} during ${operation}` + ); +}; +``` + ### `onToolStart` - **Triggered:** Before an agent executes a tool. @@ -399,15 +433,15 @@ onHandoff: async ({ agent, sourceAgent }) => { - **Async Execution:** Hooks can be `async` functions. VoltAgent awaits completion before proceeding. Long-running operations in hooks add latency to agent response time. - **Error Handling:** Errors thrown inside hooks may interrupt agent execution. Use `try...catch` within hooks or design them to be reliable. - **Hook Merging:** When hooks are passed to both the Agent constructor and a method call: - - Most hooks (`onStart`, `onEnd`, `onError`, `onHandoff`, `onToolStart`, `onToolEnd`, `onStepFinish`) execute both: method-level first, then agent-level. + - Most hooks (`onStart`, `onEnd`, `onError`, `onHandoff`, `onToolStart`, `onToolEnd`, `onStepFinish`, `onRetry`, `onFallback`) execute both: method-level first, then agent-level. - Message hooks (`onPrepareMessages`, `onPrepareModelMessages`) do not merge: method-level replaces agent-level entirely. ## Additional Hooks -Two additional hooks exist for advanced use cases: - - **`onError`**: Called when an error occurs during agent execution. Receives `{ agent: Agent, error: Error, context: OperationContext }`. - **`onStepFinish`**: Called after each step in multi-step agent execution. Receives `{ agent: Agent, step: any, context: OperationContext }`. +- **`onRetry`**: Called when VoltAgent schedules a retry. Receives `{ source, operation, ... }` with retry metadata. +- **`onFallback`**: Called when VoltAgent selects the next model candidate. Receives `{ stage, fromModel, nextModel, error, ... }`. ## Common Use Cases diff --git a/website/docs/agents/middleware.md b/website/docs/agents/middleware.md new file mode 100644 index 000000000..65939f8ec --- /dev/null +++ b/website/docs/agents/middleware.md @@ -0,0 +1,105 @@ +--- +title: Middleware +slug: /agents/middleware +--- + +# Agent Middleware + +Middleware runs before guardrails and hooks on each agent call. Use it to read or modify input and output, or to stop a call with a typed error. + +## Execution Order + +`generateText` and `generateObject`: + +1. Input middlewares +2. Input guardrails +3. Hooks (`onStart`) +4. Model call (with retries and fallback) +5. Output middlewares +6. Output guardrails +7. Hooks (`onEnd`) + +`streamText` and `streamObject`: + +1. Input middlewares +2. Input guardrails +3. Stream starts + +Output middlewares do not run in streaming calls. + +## Configure Middleware + +Agent-level middlewares run before per-call middlewares. Per-call middlewares are appended to the list for that call. + +```ts +import { Agent, createInputMiddleware, createOutputMiddleware } from "@voltagent/core"; + +const normalizeInput = createInputMiddleware({ + name: "NormalizeInput", + handler: ({ input }) => { + if (typeof input !== "string") return input; + return input.trim(); + }, +}); + +const requireSignature = createOutputMiddleware({ + name: "RequireSignature", + handler: ({ output, abort }) => { + if (!output.includes("-- Support")) { + abort("Missing signature", { retry: true, metadata: { reason: "signature" } }); + } + return output; + }, +}); + +const agent = new Agent({ + name: "Support", + instructions: "Answer support questions with short, direct replies.", + model: "openai/gpt-4o-mini", + maxMiddlewareRetries: 1, + inputMiddlewares: [normalizeInput], + outputMiddlewares: [requireSignature], +}); +``` + +Per-call middleware: + +```ts +await agent.generateText("hi", { + inputMiddlewares: [({ input }) => (typeof input === "string" ? `Customer: ${input}` : input)], +}); +``` + +## Middleware API + +Input middleware receives: + +- `input`: current input value +- `originalInput`: input from the caller +- `agent`: agent instance +- `context`: operation context +- `operation`: operation name (for example `generateText`) +- `retryCount`: current middleware retry count +- `abort(reason?, options?)`: stop the call or request a retry + +Output middleware receives: + +- `output`: current output value +- `originalOutput`: output from the model +- `usage`, `finishReason`, `warnings`: model metadata when available +- `agent`, `context`, `operation`, `retryCount`, `abort(...)` + +Return a new value to replace the input or output; return `void` to keep the current value. + +## Retry Behavior + +Middleware retries are controlled by `maxMiddlewareRetries`. The default is `0`. + +- Call `abort("reason", { retry: true })` to request a retry. +- Each retry restarts the full attempt: input middleware, guardrails, hooks, model selection, model retries/fallback, output middleware. +- The retry reason is added as a system message for the next attempt. +- `retryCount` starts at `0` and increments after each middleware-triggered retry. +- If `maxMiddlewareRetries` is exceeded, the middleware error is returned to the caller. +- In streaming calls, retry can only happen before the stream starts, so only input middlewares can trigger it. + +Middleware retries are separate from model retries. Model retries use `maxRetries` and are evaluated per model call. For model fallback configuration, see [Retries and Fallbacks](/docs/agents/retries-fallback). diff --git a/website/docs/agents/overview.md b/website/docs/agents/overview.md index 09eb758b7..d13a81196 100644 --- a/website/docs/agents/overview.md +++ b/website/docs/agents/overview.md @@ -40,6 +40,9 @@ const agent = new Agent({ The `instructions` property defines the agent's behavior. The `model` can be an ai-sdk `LanguageModel` or a `provider/model` string resolved by VoltAgent. See [Model Router & Registry](/docs/getting-started/model-router) for details. +To configure fallback lists and per-model retries, see [Retries and Fallbacks](/docs/agents/retries-fallback). +To run pre-guardrail input/output handlers, see [Middleware](/docs/agents/middleware). + ## Using Agents: Direct Method Calls Agents have two core methods for generating responses: diff --git a/website/docs/agents/providers.md b/website/docs/agents/providers.md index 213783d12..ab7459698 100644 --- a/website/docs/agents/providers.md +++ b/website/docs/agents/providers.md @@ -40,7 +40,7 @@ const agent = new Agent({ ## Provider Selection -Choose any ai-sdk provider package that fits your needs (OpenAI, Anthropic, Google, Groq, Mistral, Vertex, Bedrock, etc.), or use model strings. +Use any ai-sdk provider package you have configured (OpenAI, Anthropic, Google, Groq, Mistral, Vertex, Bedrock, etc.), or use model strings. For installation and an up-to-date model matrix, see: **[Providers & Models](/docs/getting-started/providers-models)** @@ -48,3 +48,9 @@ For installation and an up-to-date model matrix, see: For how model strings are resolved and how environment variables map, see: **[Model Router & Registry](/docs/getting-started/model-router)** + +## Retries and Fallbacks + +For retry counts and fallback lists, see: + +**[Retries and Fallbacks](/docs/agents/retries-fallback)** diff --git a/website/docs/agents/retries-fallback.md b/website/docs/agents/retries-fallback.md new file mode 100644 index 000000000..944fa2965 --- /dev/null +++ b/website/docs/agents/retries-fallback.md @@ -0,0 +1,203 @@ +--- +title: Retries and Fallbacks +slug: /agents/retries-fallback +--- + +# Retries and Fallbacks + +VoltAgent supports per-call retries and ordered model fallback lists. Retries are handled by VoltAgent with exponential backoff. After retries are exhausted for a model, VoltAgent attempts the next model in the list. + +## Configure Retries + +You can set a default retry count at the agent level, then override it per model or per call. + +```ts +import { Agent } from "@voltagent/core"; + +const agent = new Agent({ + name: "Support", + instructions: "Resolve tickets quickly.", + model: "openai/gpt-4o-mini", + maxRetries: 3, +}); + +// Per-call override +await agent.generateText("Summarize this ticket", { maxRetries: 0 }); +``` + +- `maxRetries` is the number of retry attempts for a single model call (total attempts = `maxRetries + 1`). +- Priority order: per-call `maxRetries` > per-model `maxRetries` > agent `maxRetries` > default `3`. +- `0` disables retries for that call. +- Retry delay uses exponential backoff: 1s, 2s, 4s, 8s (max 10s). +- Errors with `isRetryable: false` skip retries; fallback can still run. + +## Configure Fallbacks + +Provide a list of models. VoltAgent tries them in order until one succeeds. +Each entry can have its own retry count and can be toggled on/off. + +```ts +import { Agent } from "@voltagent/core"; +import { anthropic } from "@ai-sdk/anthropic"; + +const agent = new Agent({ + name: "FallbackAgent", + instructions: "Be concise.", + model: [ + { id: "primary", model: "openai/gpt-4o-mini", maxRetries: 2 }, + { id: "secondary", model: anthropic("claude-3-5-sonnet"), maxRetries: 1 }, + { id: "tertiary", model: "google/gemini-2.0-flash", enabled: true }, + ], +}); +``` + +Notes: + +- The first enabled model is the primary. +- `maxRetries` falls back to the agent default if omitted. +- `enabled: false` lets you keep a model configured but temporarily disabled. + +## How Fallback Works + +VoltAgent tries each enabled model in order: + +1. Resolve the model (dynamic or static). +2. Execute the call with that model, retrying up to `maxRetries` on error. +3. If retries are exhausted, move to the next model. + +Fallback **does not** trigger for: + +- Abort or bail errors. +- Guardrail blocks. +- Tool execution errors. + +Streaming behavior: + +- Retries and fallback happen only if the stream fails before the first output chunk. +- If a stream fails after output starts, the error is surfaced to the caller and the stream is not restarted. + +## Middleware Retries + +Middleware retries are separate from model retries and fallback. A middleware can call `abort("reason", { retry: true })` to restart the full attempt. + +- Controlled by `maxMiddlewareRetries` (agent-level or per-call). +- Each retry reruns the full pipeline: middleware, guardrails, hooks, model selection, model retries, and fallback. +- In streaming calls, only input middlewares can trigger a retry before the stream starts. + +See [Middleware](/docs/agents/middleware) for configuration and API details. + +## Real-World Scenarios + +### 1. Provider Outage During a Traffic Spike + +The primary provider returns 5xx errors during a traffic spike. Configure a fallback provider with its own retry count. + +```ts +const agent = new Agent({ + name: "LaunchAssistant", + instructions: "Handle onboarding questions.", + model: [ + { model: "openai/gpt-4o-mini", maxRetries: 1 }, + { model: "anthropic/claude-3-5-sonnet", maxRetries: 2 }, + ], +}); +``` + +Expected behavior: + +- Primary errors are retried up to `maxRetries`. +- After retries, the secondary provider is attempted. + +### 2. Rate Limits During Peak Hours + +The primary model returns 429 during peak hours. Set `maxRetries: 0` on the primary and allow retries on the fallback. + +```ts +const agent = new Agent({ + name: "PeakTrafficAgent", + instructions: "Answer pricing questions.", + model: [ + { model: "openai/gpt-4o-mini", maxRetries: 0 }, + { model: "groq/llama-3.3-70b-versatile", maxRetries: 2 }, + ], +}); +``` + +Expected behavior: + +- The primary fails without retry delay. +- The fallback is attempted and can retry if needed. + +### 3. Timeouts in One Region + +Requests in a region start timing out. Configure a regional primary with a global fallback. + +```ts +const agent = new Agent({ + name: "RegionalAgent", + instructions: "Support users in APAC.", + model: [ + { model: "google/gemini-2.0-flash", maxRetries: 1 }, + { model: "openai/gpt-4o-mini", maxRetries: 1 }, + ], +}); +``` + +Expected behavior: + +- Timeouts on the primary move the call to the fallback after retries. +- The fallback handles calls that the primary fails. + +### 4. Compliance or Data Residency Constraints + +Some users require a specific provider for data residency. Select the primary model dynamically and keep a fallback entry. + +```ts +const agent = new Agent({ + name: "ComplianceAgent", + instructions: "Handle regulated requests.", + model: [ + { + model: async ({ context }) => { + const region = (context.get("region") as string) || "us"; + return region === "eu" ? "mistral/mistral-large-latest" : "openai/gpt-4o-mini"; + }, + maxRetries: 1, + }, + { model: "anthropic/claude-3-5-sonnet", maxRetries: 1 }, + ], +}); +``` + +Expected behavior: + +- The model selection uses `context` values on each call. +- If the selected model fails, the fallback is attempted. + +### 5. Cost Control with Fallback + +Use a lower-cost primary model and a second model for error cases. + +```ts +const agent = new Agent({ + name: "CostAwareAgent", + instructions: "Handle receipts and billing questions.", + model: [ + { model: "openai/gpt-4o-mini", maxRetries: 1 }, + { model: "openai/gpt-4o", maxRetries: 0 }, + ], +}); +``` + +Expected behavior: + +- The primary handles successful calls. +- The fallback is only used when the primary errors. + +## Troubleshooting Tips + +- If fallbacks never trigger, check for guardrail blocks or tool errors. +- If retries are skipped, check whether the provider marks the error as `isRetryable: false`. +- If all configured models are disabled, VoltAgent throws `MODEL_LIST_EMPTY`. +- For long streaming sessions, consider application-level retry on stream errors. +- Keep the fallback list short to avoid long end-to-end latency when providers are degraded. diff --git a/website/docs/agents/subagents.md b/website/docs/agents/subagents.md index c4cdc3ce9..1cd15bd94 100644 --- a/website/docs/agents/subagents.md +++ b/website/docs/agents/subagents.md @@ -210,8 +210,8 @@ const result = await supervisor.streamText("Process data"); // result contains error message like "Error in DataProcessor: Stream failed" ``` -:::info Native Retry Support -VoltAgent uses the AI SDK's native retry mechanism (default: 3 attempts). Setting `throwOnStreamError: true` is useful for custom error handling or logging at a higher level, not for implementing retry logic. +:::info Retry Behavior +VoltAgent retries model calls based on `maxRetries`. `throwOnStreamError` controls how stream errors are surfaced; it does not change retry or fallback behavior. ::: **Silent Errors - Custom Messaging:** diff --git a/website/docs/workflows/steps/and-agent.md b/website/docs/workflows/steps/and-agent.md index ba3ea412e..61f068d6d 100644 --- a/website/docs/workflows/steps/and-agent.md +++ b/website/docs/workflows/steps/and-agent.md @@ -1,4 +1,4 @@ -# andAgent +# AndAgent > Add AI to your workflow. Get structured, typed responses from language models. @@ -92,6 +92,64 @@ By default, the step result replaces the workflow data with the agent output. If ) ``` +## Middleware and Retries + +`andAgent` passes the config object to `agent.generateText`, so you can use +`inputMiddlewares`, `outputMiddlewares`, `maxMiddlewareRetries`, and `maxRetries` +per step. + +```ts +import { Agent, createInputMiddleware, createOutputMiddleware } from "@voltagent/core"; +import { createWorkflowChain } from "@voltagent/core"; +import { z } from "zod"; + +const redactInput = createInputMiddleware({ + name: "RedactPII", + handler: ({ input }) => { + if (typeof input !== "string") return input; + return input.replace(/\b\d{16}\b/g, "[redacted]"); + }, +}); + +const requireSource = createOutputMiddleware({ + name: "RequireSource", + handler: ({ output, abort }) => { + if (!output.includes("source:")) { + abort("Missing source in summary", { retry: true }); + } + return output; + }, +}); + +const agent = new Agent({ + name: "Assistant", + model: "openai/gpt-4o-mini", + instructions: "Summarize documents with sources.", +}); + +createWorkflowChain({ + id: "doc-summary", + input: z.object({ text: z.string() }), +}).andAgent(async ({ data }) => `Summarize and include source lines: ${data.text}`, agent, { + schema: z.object({ + summary: z.string(), + sources: z.array(z.string()), + }), + inputMiddlewares: [redactInput], + outputMiddlewares: [requireSource], + maxMiddlewareRetries: 1, + maxRetries: 2, +}); +``` + +Retry behavior: + +- `maxRetries` controls LLM retries for the selected model (total attempts = `maxRetries + 1`). +- `maxMiddlewareRetries` controls retries triggered by `abort(..., { retry: true })`. +- A middleware retry restarts the step: middlewares, guardrails, hooks, model selection, and fallback. +- Output middleware runs on the model text and can trigger retries. It does not change the parsed + object returned by `andAgent`. + ## Common Patterns ### Text Analysis diff --git a/website/sidebars.ts b/website/sidebars.ts index 643dd390f..8eeb652ec 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -86,9 +86,11 @@ const sidebars: SidebarsConfig = { }, "agents/a2a/a2a-server", "agents/hooks", + "agents/middleware", "agents/message-types", "agents/multi-modal", "agents/providers", + "agents/retries-fallback", "agents/subagents", "agents/voice", "agents/context",