diff --git a/.changeset/fuzzy-pandas-shout.md b/.changeset/fuzzy-pandas-shout.md new file mode 100644 index 000000000..acd681efc --- /dev/null +++ b/.changeset/fuzzy-pandas-shout.md @@ -0,0 +1,5 @@ +--- +"@voltagent/redis": minor +--- + +feat: add @voltagent/redis memory storage adapter diff --git a/docs/structure.md b/docs/structure.md index d449b86aa..8960689a3 100644 --- a/docs/structure.md +++ b/docs/structure.md @@ -30,6 +30,7 @@ Core packages, AI provider integrations, and utilities: - **libsql** - LibSQL database integration - **logger** - Universal logger implementation - **postgres** - PostgreSQL database integration +- **redis** - Redis memory storage integration - **sdk** - JavaScript/TypeScript SDK for VoltAgent API - **server-core** - Core server handlers, schemas, and business logic - **server-hono** - Hono-based server implementation with API routes diff --git a/packages/redis/README.md b/packages/redis/README.md new file mode 100644 index 000000000..b2a548dff --- /dev/null +++ b/packages/redis/README.md @@ -0,0 +1,47 @@ +# @voltagent/redis + +Redis memory storage adapter for [VoltAgent](https://voltagent.dev) agents. Stores conversations, messages, working memory, and workflow state in Redis for low-latency retrieval. + +## Installation + +```bash +npm install @voltagent/redis +``` + +## Usage + +```ts +import { Agent, Memory } from "@voltagent/core"; +import { RedisMemoryAdapter } from "@voltagent/redis"; + +const memory = new Memory({ + storage: new RedisMemoryAdapter({ + // Connection string or ioredis options object + connection: process.env.REDIS_URL || "redis://localhost:6379", + // Optional: prefix for all Redis keys (default: "voltagent") + keyPrefix: "voltagent", + }), +}); + +const agent = new Agent({ + name: "Assistant", + model: "openai/gpt-4o-mini", + memory, +}); +``` + +## Options + +| Option | Type | Description | +| ------------ | ------------------ | -------------------------------------------------- | +| `connection` | `string \| object` | Redis connection string or ioredis options object | +| `keyPrefix` | `string` | Prefix for all Redis keys (default: `"voltagent"`) | +| `debug` | `boolean` | Enable debug logging (default: `false`) | + +## Persistence + +Redis is an in-memory store. If you need durability, configure RDB snapshots and/or AOF on your Redis instance, or use a managed Redis provider. See the [documentation](https://voltagent.dev/docs/agents/memory/redis/) for data modeling details and persistence guidance. + +## License + +MIT diff --git a/packages/redis/package.json b/packages/redis/package.json new file mode 100644 index 000000000..035df9ece --- /dev/null +++ b/packages/redis/package.json @@ -0,0 +1,55 @@ +{ + "name": "@voltagent/redis", + "description": "VoltAgent Redis - Redis Memory provider integration for VoltAgent", + "version": "0.1.0", + "dependencies": { + "@voltagent/internal": "^1.0.2", + "ioredis": "^5.6.1" + }, + "devDependencies": { + "@vitest/coverage-v8": "^3.2.4", + "@voltagent/core": "^2.4.4", + "ai": "^6.0.0" + }, + "exports": { + ".": { + "import": { + "types": "./dist/index.d.mts", + "default": "./dist/index.mjs" + }, + "require": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + } + } + }, + "files": [ + "dist" + ], + "license": "MIT", + "main": "dist/index.js", + "module": "dist/index.mjs", + "peerDependencies": { + "@voltagent/core": "^2.0.0", + "ai": "^6.0.0" + }, + "publishConfig": { + "access": "public" + }, + "repository": { + "type": "git", + "url": "https://github.com/VoltAgent/voltagent.git", + "directory": "packages/redis" + }, + "scripts": { + "attw": "attw --pack", + "build": "tsup", + "dev": "tsup --watch", + "lint": "biome check .", + "lint:fix": "biome check . --write", + "publint": "publint --strict", + "test": "vitest", + "test:coverage": "vitest run --coverage" + }, + "types": "dist/index.d.ts" +} diff --git a/packages/redis/src/index.ts b/packages/redis/src/index.ts new file mode 100644 index 000000000..e6ed66bf7 --- /dev/null +++ b/packages/redis/src/index.ts @@ -0,0 +1,10 @@ +/** + * Redis Storage Adapter for VoltAgent + * + * Provides low-latency storage for conversations and messages + * using Redis with Memory V2 architecture + */ + +// Export Memory Adapter +export { RedisMemoryAdapter } from "./memory-adapter"; +export type { RedisMemoryOptions } from "./memory-adapter"; diff --git a/packages/redis/src/memory-adapter.spec.ts b/packages/redis/src/memory-adapter.spec.ts new file mode 100644 index 000000000..3069c3be1 --- /dev/null +++ b/packages/redis/src/memory-adapter.spec.ts @@ -0,0 +1,742 @@ +/** + * Unit tests for Redis Memory Storage Adapter + * Tests all functionality using an in-memory ioredis mock + */ + +import { ConversationAlreadyExistsError, ConversationNotFoundError } from "@voltagent/core"; +import type { ConversationStepRecord, WorkflowStateEntry } from "@voltagent/core"; +import type { UIMessage } from "ai"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { RedisMemoryAdapter } from "./memory-adapter"; + +// In-memory ioredis replacement +const { FakeRedis } = vi.hoisted(() => { + class FakePipeline { + private commands: Array<{ name: string; args: any[] }> = []; + + constructor(private redis: FakeRedis) {} + + private record(name: string, args: any[]) { + this.commands.push({ name, args }); + return this; + } + + get(...args: any[]) { + return this.record("get", args); + } + set(...args: any[]) { + return this.record("set", args); + } + del(...args: any[]) { + return this.record("del", args); + } + zadd(...args: any[]) { + return this.record("zadd", args); + } + zrem(...args: any[]) { + return this.record("zrem", args); + } + hset(...args: any[]) { + return this.record("hset", args); + } + hdel(...args: any[]) { + return this.record("hdel", args); + } + sadd(...args: any[]) { + return this.record("sadd", args); + } + srem(...args: any[]) { + return this.record("srem", args); + } + + async exec(): Promise> { + if (this.redis.failNextExec) { + this.redis.failNextExec = false; + // Fail before executing anything so no partial state is applied + return this.commands.map(() => [new Error("simulated pipeline failure"), null]); + } + + const results: Array<[Error | null, unknown]> = []; + for (const command of this.commands) { + try { + const result = await (this.redis as any)[command.name](...command.args); + results.push([null, result]); + } catch (error) { + results.push([error as Error, null]); + } + } + return results; + } + } + + class FakeRedis { + static instances: FakeRedis[] = []; + + strings = new Map(); + hashes = new Map>(); + zsets = new Map>(); + sets = new Map>(); + + failNextExec = false; + quitCalled = false; + status: "ready" | "end" = "ready"; + + constructor(public connection?: unknown) { + FakeRedis.instances.push(this); + } + + on() { + return this; + } + + pipeline() { + return new FakePipeline(this); + } + + async get(key: string) { + return this.strings.get(key) ?? null; + } + + async set(key: string, value: string, ...flags: any[]) { + if (flags.includes("NX") && this.strings.has(key)) { + return null; + } + this.strings.set(key, value); + return "OK"; + } + + async del(...keys: string[]) { + let removed = 0; + for (const key of keys) { + if (this.strings.delete(key)) removed++; + if (this.hashes.delete(key)) removed++; + if (this.zsets.delete(key)) removed++; + if (this.sets.delete(key)) removed++; + } + return removed; + } + + private zset(key: string) { + let zset = this.zsets.get(key); + if (!zset) { + zset = new Map(); + this.zsets.set(key, zset); + } + return zset; + } + + async zadd(key: string, score: number, member: string) { + this.zset(key).set(member, Number(score)); + return 1; + } + + async zrem(key: string, ...members: string[]) { + const zset = this.zsets.get(key); + if (!zset) return 0; + let removed = 0; + for (const member of members) { + if (zset.delete(member)) removed++; + } + return removed; + } + + private sortedMembers(key: string, reverse: boolean) { + const zset = this.zsets.get(key); + if (!zset) return []; + const entries = [...zset.entries()].sort((a, b) => a[1] - b[1] || a[0].localeCompare(b[0])); + if (reverse) entries.reverse(); + return entries.map(([member]) => member); + } + + private sliceRange(members: string[], start: number, stop: number) { + const length = members.length; + const from = start < 0 ? Math.max(length + start, 0) : start; + const to = stop < 0 ? length + stop : stop; + return members.slice(from, to + 1); + } + + async zrange(key: string, start: number, stop: number) { + return this.sliceRange(this.sortedMembers(key, false), start, stop); + } + + async zrevrange(key: string, start: number, stop: number) { + return this.sliceRange(this.sortedMembers(key, true), start, stop); + } + + async hset(key: string, field: string, value: string) { + let hash = this.hashes.get(key); + if (!hash) { + hash = new Map(); + this.hashes.set(key, hash); + } + hash.set(field, value); + return 1; + } + + async hgetall(key: string) { + return Object.fromEntries(this.hashes.get(key) ?? new Map()); + } + + async hdel(key: string, ...fields: string[]) { + const hash = this.hashes.get(key); + if (!hash) return 0; + let removed = 0; + for (const field of fields) { + if (hash.delete(field)) removed++; + } + return removed; + } + + async sadd(key: string, ...members: string[]) { + let set = this.sets.get(key); + if (!set) { + set = new Set(); + this.sets.set(key, set); + } + for (const member of members) set.add(member); + return members.length; + } + + async srem(key: string, ...members: string[]) { + const set = this.sets.get(key); + if (!set) return 0; + let removed = 0; + for (const member of members) { + if (set.delete(member)) removed++; + } + return removed; + } + + async smembers(key: string) { + return [...(this.sets.get(key) ?? new Set())]; + } + + async quit() { + if (this.status === "end") { + throw new Error("Connection is closed."); + } + this.status = "end"; + this.quitCalled = true; + return "OK"; + } + } + + return { FakeRedis }; +}); + +vi.mock("ioredis", () => ({ + Redis: FakeRedis, + default: FakeRedis, +})); + +// ============================================================================ +// Test Helpers +// ============================================================================ + +const createAdapter = () => + new RedisMemoryAdapter({ connection: "redis://localhost:6379", keyPrefix: "test" }); + +const lastInstance = () => FakeRedis.instances[FakeRedis.instances.length - 1]; + +const createConversationInput = (overrides = {}) => ({ + id: "conv-1", + resourceId: "resource-1", + userId: "user-1", + title: "Test Conversation", + metadata: {}, + ...overrides, +}); + +const createMessage = (id: string, role: "user" | "assistant" = "user"): UIMessage => ({ + id, + role, + parts: [{ type: "text", text: `message ${id}` }], +}); + +const createWorkflowState = (overrides: Partial = {}): WorkflowStateEntry => ({ + id: "exec-1", + workflowId: "workflow-1", + workflowName: "Test Workflow", + status: "running", + createdAt: new Date("2025-01-01T00:00:00.000Z"), + updatedAt: new Date("2025-01-01T00:00:00.000Z"), + ...overrides, +}); + +const createStep = (overrides: Partial = {}): ConversationStepRecord => ({ + id: "step-1", + conversationId: "conv-1", + userId: "user-1", + agentId: "agent-1", + operationId: "op-1", + stepIndex: 0, + type: "text", + role: "assistant", + createdAt: "2025-01-01T00:00:00.000Z", + ...overrides, +}); + +describe.sequential("RedisMemoryAdapter", () => { + let adapter: RedisMemoryAdapter; + + beforeEach(() => { + FakeRedis.instances = []; + adapter = createAdapter(); + }); + + // ========================================================================== + // Conversations + // ========================================================================== + + describe("conversations", () => { + it("creates and retrieves a conversation", async () => { + const created = await adapter.createConversation(createConversationInput()); + + expect(created.id).toBe("conv-1"); + expect(created.createdAt).toBe(created.updatedAt); + + const fetched = await adapter.getConversation("conv-1"); + expect(fetched).toEqual(created); + }); + + it("throws ConversationAlreadyExistsError when creating a duplicate", async () => { + await adapter.createConversation(createConversationInput()); + + await expect(adapter.createConversation(createConversationInput())).rejects.toBeInstanceOf( + ConversationAlreadyExistsError, + ); + }); + + it("returns null for a missing conversation", async () => { + await expect(adapter.getConversation("missing")).resolves.toBeNull(); + }); + + it("throws ConversationNotFoundError when updating a missing conversation", async () => { + await expect(adapter.updateConversation("missing", { title: "x" })).rejects.toBeInstanceOf( + ConversationNotFoundError, + ); + }); + + it("queries conversations by resource, user, ordering and pagination", async () => { + await adapter.createConversation( + createConversationInput({ id: "c1", title: "Alpha", userId: "user-1" }), + ); + await adapter.createConversation( + createConversationInput({ id: "c2", title: "Beta", userId: "user-1" }), + ); + await adapter.createConversation( + createConversationInput({ + id: "c3", + title: "Gamma", + userId: "user-2", + resourceId: "resource-2", + }), + ); + + expect((await adapter.getConversations("resource-1")).map((c) => c.id)).toEqual(["c2", "c1"]); + + expect((await adapter.getConversationsByUserId("user-1")).map((c) => c.id)).toEqual([ + "c2", + "c1", + ]); + + const byTitle = await adapter.queryConversations({ + userId: "user-1", + orderBy: "title", + orderDirection: "ASC", + }); + expect(byTitle.map((c) => c.id)).toEqual(["c1", "c2"]); + + const paged = await adapter.queryConversations({ limit: 1, offset: 1 }); + expect(paged).toHaveLength(1); + + await expect(adapter.countConversations({ userId: "user-1" })).resolves.toBe(2); + await expect(adapter.countConversations({})).resolves.toBe(3); + }); + + it("updates a conversation and reindexes when resourceId changes", async () => { + await adapter.createConversation(createConversationInput()); + + const updated = await adapter.updateConversation("conv-1", { + title: "Renamed", + resourceId: "resource-2", + }); + expect(updated.title).toBe("Renamed"); + expect(updated.resourceId).toBe("resource-2"); + + expect(await adapter.getConversations("resource-1")).toHaveLength(0); + expect((await adapter.getConversations("resource-2")).map((c) => c.id)).toEqual(["conv-1"]); + }); + + it("deletes a conversation along with its messages and indexes", async () => { + await adapter.createConversation(createConversationInput()); + await adapter.addMessage(createMessage("m1"), "user-1", "conv-1"); + + await adapter.deleteConversation("conv-1"); + + await expect(adapter.getConversation("conv-1")).resolves.toBeNull(); + await expect(adapter.getMessages("user-1", "conv-1")).resolves.toEqual([]); + await expect(adapter.countConversations({})).resolves.toBe(0); + }); + + it("propagates pipeline failures instead of ignoring them", async () => { + lastInstance().failNextExec = true; + + await expect(adapter.createConversation(createConversationInput())).rejects.toThrow( + "simulated pipeline failure", + ); + + // The failed pipeline must not leave partial index state behind + await expect(adapter.countConversations({})).resolves.toBe(0); + }); + }); + + // ========================================================================== + // Messages + // ========================================================================== + + describe("messages", () => { + beforeEach(async () => { + await adapter.createConversation(createConversationInput()); + }); + + it("throws ConversationNotFoundError when adding to a missing conversation", async () => { + await expect( + adapter.addMessage(createMessage("m1"), "user-1", "missing"), + ).rejects.toBeInstanceOf(ConversationNotFoundError); + }); + + it("adds and retrieves messages with createdAt metadata", async () => { + await adapter.addMessage(createMessage("m1"), "user-1", "conv-1"); + await adapter.addMessage(createMessage("m2", "assistant"), "user-1", "conv-1"); + + const messages = await adapter.getMessages("user-1", "conv-1"); + + expect(messages.map((m) => m.id)).toEqual(["m1", "m2"]); + expect(messages[0].metadata?.createdAt).toBeInstanceOf(Date); + }); + + it("batches addMessages into a single pipeline", async () => { + const instance = lastInstance(); + const pipelineSpy = vi.spyOn(instance, "pipeline"); + + await adapter.addMessages( + [createMessage("m1"), createMessage("m2"), createMessage("m3")], + "user-1", + "conv-1", + ); + + expect(pipelineSpy).toHaveBeenCalledTimes(1); + expect((await adapter.getMessages("user-1", "conv-1")).map((m) => m.id)).toEqual([ + "m1", + "m2", + "m3", + ]); + }); + + it("filters messages by role and limit, returning the latest in chronological order", async () => { + await adapter.addMessages( + [ + createMessage("m1"), + createMessage("m2", "assistant"), + createMessage("m3"), + createMessage("m4", "assistant"), + ], + "user-1", + "conv-1", + ); + + const assistants = await adapter.getMessages("user-1", "conv-1", { + roles: ["assistant"], + }); + expect(assistants.map((m) => m.id)).toEqual(["m2", "m4"]); + + const latestTwo = await adapter.getMessages("user-1", "conv-1", { limit: 2 }); + expect(latestTwo.map((m) => m.id)).toEqual(["m3", "m4"]); + }); + + it("does not return messages belonging to another user", async () => { + await adapter.addMessage(createMessage("m1"), "user-2", "conv-1"); + + await expect(adapter.getMessages("user-1", "conv-1")).resolves.toEqual([]); + await expect(adapter.getMessages("user-2", "conv-1")).resolves.toHaveLength(1); + }); + + it("deletes specific messages by id", async () => { + await adapter.addMessages([createMessage("m1"), createMessage("m2")], "user-1", "conv-1"); + + await adapter.deleteMessages(["m1"], "user-1", "conv-1"); + + expect((await adapter.getMessages("user-1", "conv-1")).map((m) => m.id)).toEqual(["m2"]); + }); + + it("clears messages for a single conversation", async () => { + await adapter.addMessage(createMessage("m1"), "user-1", "conv-1"); + + await adapter.clearMessages("user-1", "conv-1"); + + await expect(adapter.getMessages("user-1", "conv-1")).resolves.toEqual([]); + }); + + it("keeps other users' messages and steps when clearing a conversation", async () => { + await adapter.addMessage(createMessage("m1"), "user-1", "conv-1"); + await adapter.addMessage(createMessage("m2"), "user-2", "conv-1"); + await adapter.saveConversationSteps([ + createStep({ id: "step-1", userId: "user-1" }), + createStep({ id: "step-2", userId: "user-2" }), + ]); + + await adapter.clearMessages("user-1", "conv-1"); + + await expect(adapter.getMessages("user-1", "conv-1")).resolves.toEqual([]); + const remaining = await adapter.getMessages("user-2", "conv-1"); + expect(remaining.map((m) => m.id)).toEqual(["m2"]); + await expect(adapter.getConversationSteps("user-1", "conv-1")).resolves.toEqual([]); + const remainingSteps = await adapter.getConversationSteps("user-2", "conv-1"); + expect(remainingSteps.map((s) => s.id)).toEqual(["step-2"]); + }); + + it("clears messages across all conversations of a user", async () => { + await adapter.createConversation(createConversationInput({ id: "conv-2" })); + await adapter.addMessage(createMessage("m1"), "user-1", "conv-1"); + await adapter.addMessage(createMessage("m2"), "user-1", "conv-2"); + + await adapter.clearMessages("user-1"); + + await expect(adapter.getMessages("user-1", "conv-1")).resolves.toEqual([]); + await expect(adapter.getMessages("user-1", "conv-2")).resolves.toEqual([]); + }); + }); + + // ========================================================================== + // Conversation Steps + // ========================================================================== + + describe("conversation steps", () => { + beforeEach(async () => { + await adapter.createConversation(createConversationInput()); + }); + + it("saves and retrieves steps ordered by step index", async () => { + await adapter.saveConversationSteps([ + createStep({ id: "step-2", stepIndex: 1 }), + createStep({ id: "step-1", stepIndex: 0 }), + ]); + + const steps = await adapter.getConversationSteps("user-1", "conv-1"); + expect(steps.map((s) => s.id)).toEqual(["step-1", "step-2"]); + }); + + it("filters steps by operationId, userId and limit", async () => { + await adapter.saveConversationSteps([ + createStep({ id: "step-1", operationId: "op-1", stepIndex: 0 }), + createStep({ id: "step-2", operationId: "op-2", stepIndex: 1 }), + createStep({ id: "step-3", operationId: "op-1", stepIndex: 2, userId: "user-2" }), + ]); + + const filtered = await adapter.getConversationSteps("user-1", "conv-1", { + operationId: "op-1", + }); + expect(filtered.map((s) => s.id)).toEqual(["step-1"]); + + const limited = await adapter.getConversationSteps("user-1", "conv-1", { limit: 1 }); + expect(limited.map((s) => s.id)).toEqual(["step-1"]); + }); + }); + + // ========================================================================== + // Working Memory + // ========================================================================== + + describe("working memory", () => { + it("sets, gets and deletes conversation-scoped working memory", async () => { + await adapter.setWorkingMemory({ + conversationId: "conv-1", + content: "remember this", + scope: "conversation", + }); + + await expect( + adapter.getWorkingMemory({ conversationId: "conv-1", scope: "conversation" }), + ).resolves.toBe("remember this"); + + await adapter.deleteWorkingMemory({ conversationId: "conv-1", scope: "conversation" }); + + await expect( + adapter.getWorkingMemory({ conversationId: "conv-1", scope: "conversation" }), + ).resolves.toBeNull(); + }); + + it("keeps user-scoped working memory separate from conversation scope", async () => { + await adapter.setWorkingMemory({ userId: "user-1", content: "user data", scope: "user" }); + await adapter.setWorkingMemory({ + conversationId: "conv-1", + content: "conversation data", + scope: "conversation", + }); + + await expect(adapter.getWorkingMemory({ userId: "user-1", scope: "user" })).resolves.toBe( + "user data", + ); + await expect( + adapter.getWorkingMemory({ conversationId: "conv-1", scope: "conversation" }), + ).resolves.toBe("conversation data"); + }); + + it("throws when the identifier required by the scope is missing", async () => { + await expect(adapter.getWorkingMemory({ scope: "conversation" })).rejects.toThrow( + "conversationId is required", + ); + await expect(adapter.setWorkingMemory({ content: "x", scope: "user" })).rejects.toThrow( + "userId is required", + ); + await expect(adapter.deleteWorkingMemory({ scope: "user" })).rejects.toThrow( + "userId is required", + ); + }); + }); + + // ========================================================================== + // Workflow State + // ========================================================================== + + describe("workflow state", () => { + it("sets and gets workflow state with dates revived", async () => { + const state = createWorkflowState({ + status: "suspended", + suspension: { suspendedAt: new Date("2025-01-02T00:00:00.000Z"), stepIndex: 2 }, + }); + + await adapter.setWorkflowState("exec-1", state); + const fetched = await adapter.getWorkflowState("exec-1"); + + expect(fetched?.workflowId).toBe("workflow-1"); + expect(fetched?.createdAt).toBeInstanceOf(Date); + expect(fetched?.suspension?.suspendedAt).toBeInstanceOf(Date); + }); + + it("returns null for a missing workflow state", async () => { + await expect(adapter.getWorkflowState("missing")).resolves.toBeNull(); + }); + + it("updates workflow state, refreshing updatedAt and protecting indexed fields", async () => { + await adapter.setWorkflowState("exec-1", createWorkflowState()); + + await adapter.updateWorkflowState("exec-1", { + status: "completed", + workflowId: "other-workflow", + createdAt: new Date("2020-01-01T00:00:00.000Z"), + }); + + const updated = await adapter.getWorkflowState("exec-1"); + expect(updated?.status).toBe("completed"); + // Index-driving fields must not change + expect(updated?.workflowId).toBe("workflow-1"); + expect(updated?.createdAt).toEqual(new Date("2025-01-01T00:00:00.000Z")); + expect(updated?.updatedAt.getTime()).toBeGreaterThan( + new Date("2025-01-01T00:00:00.000Z").getTime(), + ); + }); + + it("throws when updating a missing workflow state", async () => { + await expect(adapter.updateWorkflowState("missing", { status: "completed" })).rejects.toThrow( + "Workflow state missing not found", + ); + }); + + it("queries workflow runs with filters and pagination", async () => { + await adapter.setWorkflowState( + "exec-1", + createWorkflowState({ createdAt: new Date("2025-01-01T00:00:00.000Z"), userId: "user-1" }), + ); + await adapter.setWorkflowState( + "exec-2", + createWorkflowState({ + id: "exec-2", + status: "completed", + createdAt: new Date("2025-01-02T00:00:00.000Z"), + userId: "user-1", + }), + ); + await adapter.setWorkflowState( + "exec-3", + createWorkflowState({ + id: "exec-3", + workflowId: "workflow-2", + createdAt: new Date("2025-01-03T00:00:00.000Z"), + }), + ); + + // Newest first + expect((await adapter.queryWorkflowRuns({})).map((s) => s.id)).toEqual([ + "exec-3", + "exec-2", + "exec-1", + ]); + + expect( + (await adapter.queryWorkflowRuns({ workflowId: "workflow-1" })).map((s) => s.id), + ).toEqual(["exec-2", "exec-1"]); + + expect((await adapter.queryWorkflowRuns({ status: "completed" })).map((s) => s.id)).toEqual([ + "exec-2", + ]); + + expect((await adapter.queryWorkflowRuns({ userId: "user-1" })).map((s) => s.id)).toEqual([ + "exec-2", + "exec-1", + ]); + + const paged = await adapter.queryWorkflowRuns({ limit: 1, offset: 1 }); + expect(paged.map((s) => s.id)).toEqual(["exec-2"]); + }); + + it("tracks suspended workflow states per workflow", async () => { + await adapter.setWorkflowState("exec-1", createWorkflowState({ status: "suspended" })); + await adapter.setWorkflowState( + "exec-2", + createWorkflowState({ id: "exec-2", status: "running" }), + ); + + expect((await adapter.getSuspendedWorkflowStates("workflow-1")).map((s) => s.id)).toEqual([ + "exec-1", + ]); + + // Resuming removes the execution from the suspended index + await adapter.updateWorkflowState("exec-1", { status: "running" }); + await expect(adapter.getSuspendedWorkflowStates("workflow-1")).resolves.toEqual([]); + }); + }); + + // ========================================================================== + // Connection + // ========================================================================== + + describe("connection", () => { + it("disconnects the underlying client", async () => { + await adapter.disconnect(); + expect(lastInstance().quitCalled).toBe(true); + }); + + it("is idempotent when called repeatedly", async () => { + await adapter.disconnect(); + + await expect(adapter.disconnect()).resolves.toBeUndefined(); + await expect(adapter.close()).resolves.toBeUndefined(); + expect(lastInstance().quitCalled).toBe(true); + }); + + it("tolerates concurrent disconnect calls", async () => { + await expect(Promise.all([adapter.disconnect(), adapter.disconnect()])).resolves.toEqual([ + undefined, + undefined, + ]); + }); + + it("rethrows unexpected quit errors", async () => { + vi.spyOn(lastInstance(), "quit").mockRejectedValueOnce(new Error("boom")); + + await expect(adapter.disconnect()).rejects.toThrow("boom"); + }); + }); +}); diff --git a/packages/redis/src/memory-adapter.ts b/packages/redis/src/memory-adapter.ts new file mode 100644 index 000000000..1e6e87be1 --- /dev/null +++ b/packages/redis/src/memory-adapter.ts @@ -0,0 +1,912 @@ +/** + * Redis Storage Adapter for Memory + * Stores conversations and messages in Redis using strings, hashes and sorted sets + */ + +import { randomUUID } from "node:crypto"; +import { ConversationAlreadyExistsError, ConversationNotFoundError } from "@voltagent/core"; +import type { + Conversation, + ConversationQueryOptions, + ConversationStepRecord, + CreateConversationInput, + GetConversationStepsOptions, + GetMessagesOptions, + StorageAdapter, + WorkflowRunQuery, + WorkflowStateEntry, + WorkingMemoryScope, +} from "@voltagent/core"; +import { safeStringify } from "@voltagent/internal"; +import type { UIMessage } from "ai"; +import { type ChainableCommander, Redis, type RedisOptions } from "ioredis"; + +/** + * Redis configuration options for Memory + */ +export interface RedisMemoryOptions { + /** + * Redis connection configuration + * Can be either a connection string (e.g. "redis://localhost:6379") + * or an ioredis options object + */ + connection: string | RedisOptions; + + /** + * Prefix for all Redis keys managed by this adapter + * @default "voltagent" + */ + keyPrefix?: string; + + /** + * Whether to enable debug logging + * @default false + */ + debug?: boolean; +} + +/** + * Message payload stored in Redis (dates are serialized as ISO strings) + */ +type StoredMessage = UIMessage & { + userId: string; + conversationId: string; + createdAt: string; +}; + +/** + * Redis Storage Adapter for Memory + * Low-latency storage for conversations, messages, working memory and workflow state + * + * Key layout (assuming the default "voltagent" prefix): + * - `voltagent:conv:{id}` STRING serialized conversation + * - `voltagent:convs:all` ZSET conversation ids scored by creation time + * - `voltagent:convs:resource:{resourceId}` ZSET conversation ids per resource + * - `voltagent:convs:user:{userId}` ZSET conversation ids per user + * - `voltagent:msgs:{conversationId}` ZSET message ids scored by creation time + * - `voltagent:msgdata:{conversationId}` HASH message id -> serialized message + * - `voltagent:steps:{conversationId}` ZSET step ids scored by creation time + * - `voltagent:stepdata:{conversationId}` HASH step id -> serialized step + * - `voltagent:wm:conv:{conversationId}` STRING working memory (conversation scope) + * - `voltagent:wm:user:{userId}` STRING working memory (user scope) + * - `voltagent:wf:state:{executionId}` STRING serialized workflow state + * - `voltagent:wf:all` ZSET execution ids scored by creation time + * - `voltagent:wf:idx:{workflowId}` ZSET execution ids per workflow + * - `voltagent:wf:suspended:{workflowId}` SET suspended execution ids per workflow + */ +export class RedisMemoryAdapter implements StorageAdapter { + private client: Redis; + private keyPrefix: string; + private debug: boolean; + private lastMessageScore = 0; + + constructor(options: RedisMemoryOptions) { + this.keyPrefix = options.keyPrefix ?? "voltagent"; + this.debug = options.debug ?? false; + + // ioredis accepts both a connection string and an options object; + // the branches are needed because its constructor overloads don't take a union + this.client = + typeof options.connection === "string" + ? new Redis(options.connection) + : new Redis(options.connection); + + // Swallow client error events so they don't crash the process; + // surface them through debug logging instead + this.client.on("error", (error) => { + this.log("Redis client error:", error.message); + }); + + this.log("Redis Memory V2 adapter initialized"); + } + + /** + * Log debug messages + */ + private log(...args: any[]): void { + if (this.debug) { + console.log("[Redis Memory V2]", ...args); + } + } + + /** + * Generate a random ID + */ + private generateId(): string { + return randomUUID(); + } + + /** + * Build a fully prefixed Redis key + */ + private key(...parts: string[]): string { + return [this.keyPrefix, ...parts].join(":"); + } + + /** + * Parse JSON written by this adapter, tolerating corrupted or manually + * modified entries instead of throwing + */ + private safeParse(raw: string, context: string): T | null { + try { + return JSON.parse(raw) as T; + } catch (error) { + this.log(`Failed to parse ${context}:`, error); + return null; + } + } + + /** + * Execute a pipeline and throw if any of its commands failed, + * so partial multi-key writes never go unnoticed + */ + private async execChecked(pipeline: ChainableCommander): Promise<[Error | null, unknown][]> { + const results = await pipeline.exec(); + if (!results) { + return []; + } + for (const [error] of results) { + if (error) { + throw error; + } + } + return results; + } + + /** + * Build the Redis key for working memory entries, validating that the + * identifier required by the scope is present + */ + private workingMemoryKey( + scope: WorkingMemoryScope, + conversationId?: string, + userId?: string, + ): string { + if (scope === "conversation") { + if (!conversationId) { + throw new Error('conversationId is required when working memory scope is "conversation"'); + } + return this.key("wm", "conv", conversationId); + } + if (!userId) { + throw new Error('userId is required when working memory scope is "user"'); + } + return this.key("wm", "user", userId); + } + + /** + * Fetch conversations by id in a single pipeline round trip + */ + private async getConversationsByIds(ids: string[]): Promise { + if (ids.length === 0) { + return []; + } + const pipeline = this.client.pipeline(); + for (const id of ids) { + pipeline.get(this.key("conv", id)); + } + const results = await this.execChecked(pipeline); + + const conversations: Conversation[] = []; + for (const [, raw] of results) { + if (typeof raw !== "string") continue; + const conversation = this.safeParse(raw, "conversation"); + if (conversation) { + conversations.push(conversation); + } + } + return conversations; + } + + /** + * Sort conversations the same way the SQL adapters do + */ + private sortConversations( + conversations: Conversation[], + orderBy: "created_at" | "updated_at" | "title" = "updated_at", + orderDirection: "ASC" | "DESC" = "DESC", + ): Conversation[] { + const sorted = [...conversations].sort((a, b) => { + if (orderBy === "title") { + return a.title.localeCompare(b.title); + } + const aTime = Date.parse(orderBy === "created_at" ? a.createdAt : a.updatedAt); + const bTime = Date.parse(orderBy === "created_at" ? b.createdAt : b.updatedAt); + return aTime - bTime; + }); + if (orderDirection === "DESC") { + sorted.reverse(); + } + return sorted; + } + + // ============================================================================ + // Message Operations + // ============================================================================ + + /** + * Monotonic score for message ordering, so messages written within the + * same millisecond keep their insertion order in the sorted set + */ + private nextMessageScore(time: number): number { + this.lastMessageScore = Math.max(time, this.lastMessageScore + 1); + return this.lastMessageScore; + } + + /** + * Queue the Redis commands needed to persist a single message + */ + private enqueueMessage( + pipeline: ChainableCommander, + message: UIMessage, + userId: string, + conversationId: string, + ): void { + const messageId = message.id || this.generateId(); + const createdAt = new Date(); + const entry: StoredMessage = { + ...message, + id: messageId, + userId, + conversationId, + createdAt: createdAt.toISOString(), + }; + + pipeline.zadd( + this.key("msgs", conversationId), + this.nextMessageScore(createdAt.getTime()), + messageId, + ); + pipeline.hset(this.key("msgdata", conversationId), messageId, safeStringify(entry)); + } + + async addMessage(message: UIMessage, userId: string, conversationId: string): Promise { + const conversation = await this.getConversation(conversationId); + if (!conversation) { + throw new ConversationNotFoundError(conversationId); + } + + const pipeline = this.client.pipeline(); + this.enqueueMessage(pipeline, message, userId, conversationId); + await this.execChecked(pipeline); + + this.log(`Added message to conversation ${conversationId}`); + } + + /** + * Add multiple messages in a single pipeline round trip + */ + async addMessages(messages: UIMessage[], userId: string, conversationId: string): Promise { + const conversation = await this.getConversation(conversationId); + if (!conversation) { + throw new ConversationNotFoundError(conversationId); + } + + if (messages.length === 0) { + return; + } + + const pipeline = this.client.pipeline(); + for (const message of messages) { + this.enqueueMessage(pipeline, message, userId, conversationId); + } + await this.execChecked(pipeline); + + this.log(`Added ${messages.length} messages to conversation ${conversationId}`); + } + + /** + * Get messages with optional filtering + */ + async getMessages( + userId: string, + conversationId: string, + options?: GetMessagesOptions, + ): Promise[]> { + const { limit, before, after, roles } = options || {}; + + // Sorted set order is chronological (score, then member) + const ids = await this.client.zrange(this.key("msgs", conversationId), 0, -1); + const entries = await this.client.hgetall(this.key("msgdata", conversationId)); + + const matched: Array<{ entry: StoredMessage; time: number }> = []; + for (const id of ids) { + const raw = entries[id]; + if (!raw) continue; + const entry = this.safeParse(raw, "message"); + if (!entry) continue; + if (entry.userId !== userId) continue; + if (roles && roles.length > 0 && !roles.includes(entry.role)) continue; + + const time = Date.parse(entry.createdAt); + if (before && time >= before.getTime()) continue; + if (after && time <= after.getTime()) continue; + + matched.push({ entry, time }); + } + + // Most recent first, apply limit, then restore chronological order + matched.reverse(); + const limited = limit && limit > 0 ? matched.slice(0, limit) : matched; + limited.reverse(); + + return limited.map(({ entry }) => ({ + ...entry, + createdAt: new Date(entry.createdAt), + metadata: { + ...(entry.metadata || {}), + createdAt: new Date(entry.createdAt), + }, + })); + } + + /** + * Clear messages for a user, optionally scoped to a single conversation. + * Matching the SQL adapters: when a conversation is given, only messages and + * steps owned by the user are removed; other users' entries in the same + * conversation are kept + */ + async clearMessages(userId: string, conversationId?: string): Promise { + if (conversationId) { + const [messageEntries, stepEntries] = await Promise.all([ + this.client.hgetall(this.key("msgdata", conversationId)), + this.client.hgetall(this.key("stepdata", conversationId)), + ]); + + const messageIds = Object.entries(messageEntries) + .filter(([, raw]) => this.safeParse(raw, "message")?.userId === userId) + .map(([id]) => id); + const stepIds = Object.entries(stepEntries) + .filter( + ([, raw]) => + this.safeParse(raw, "conversation step")?.userId === userId, + ) + .map(([id]) => id); + + const pipeline = this.client.pipeline(); + if (messageIds.length > 0) { + pipeline.zrem(this.key("msgs", conversationId), ...messageIds); + pipeline.hdel(this.key("msgdata", conversationId), ...messageIds); + } + if (stepIds.length > 0) { + pipeline.zrem(this.key("steps", conversationId), ...stepIds); + pipeline.hdel(this.key("stepdata", conversationId), ...stepIds); + } + await this.execChecked(pipeline); + + this.log(`Cleared messages for user ${userId} in conversation ${conversationId}`); + return; + } + + // Clear messages for every conversation owned by the user + const ids = await this.client.zrange(this.key("convs", "user", userId), 0, -1); + const pipeline = this.client.pipeline(); + for (const id of ids) { + pipeline.del(this.key("msgs", id)); + pipeline.del(this.key("msgdata", id)); + pipeline.del(this.key("steps", id)); + pipeline.del(this.key("stepdata", id)); + } + + await this.execChecked(pipeline); + this.log(`Cleared messages for user ${userId}`); + } + + /** + * Delete specific messages by ID for a conversation + * Only messages owned by the given user are removed, matching the SQL adapters + */ + async deleteMessages( + messageIds: string[], + userId: string, + conversationId: string, + ): Promise { + if (messageIds.length === 0) { + return; + } + + const entries = await this.client.hgetall(this.key("msgdata", conversationId)); + const ownedIds = messageIds.filter((id) => { + const raw = entries[id]; + if (!raw) return false; + const entry = this.safeParse(raw, "message"); + return entry?.userId === userId; + }); + + if (ownedIds.length === 0) { + return; + } + + const pipeline = this.client.pipeline(); + pipeline.zrem(this.key("msgs", conversationId), ...ownedIds); + pipeline.hdel(this.key("msgdata", conversationId), ...ownedIds); + await this.execChecked(pipeline); + + this.log(`Deleted ${ownedIds.length} messages from conversation ${conversationId}`); + } + + // ============================================================================ + // Conversation Operations + // ============================================================================ + + /** + * Create a new conversation + */ + async createConversation(input: CreateConversationInput): Promise { + const now = new Date().toISOString(); + const conversation: Conversation = { + id: input.id, + resourceId: input.resourceId, + userId: input.userId, + title: input.title, + metadata: input.metadata || {}, + createdAt: now, + updatedAt: now, + }; + + // Atomic "create if not exists" to avoid duplicate conversations on races + const created = await this.client.set( + this.key("conv", input.id), + safeStringify(conversation), + "NX", + ); + if (created === null) { + throw new ConversationAlreadyExistsError(input.id); + } + + const score = Date.parse(now); + const pipeline = this.client.pipeline(); + pipeline.zadd(this.key("convs", "all"), score, input.id); + pipeline.zadd(this.key("convs", "resource", input.resourceId), score, input.id); + pipeline.zadd(this.key("convs", "user", input.userId), score, input.id); + await this.execChecked(pipeline); + + this.log(`Created conversation ${input.id}`); + return conversation; + } + + /** + * Get a conversation by ID + */ + async getConversation(id: string): Promise { + const raw = await this.client.get(this.key("conv", id)); + if (!raw) { + return null; + } + return this.safeParse(raw, `conversation ${id}`); + } + + /** + * Get conversations by resource ID + */ + async getConversations(resourceId: string): Promise { + const ids = await this.client.zrange(this.key("convs", "resource", resourceId), 0, -1); + const conversations = await this.getConversationsByIds(ids); + return this.sortConversations(conversations); + } + + /** + * Get conversations by user ID + */ + async getConversationsByUserId( + userId: string, + options?: Omit, + ): Promise { + return this.queryConversations({ ...options, userId }); + } + + /** + * Query conversations with filters + */ + async queryConversations(options: ConversationQueryOptions): Promise { + // Pick the narrowest index available for the candidate set + let indexKey: string; + if (options.userId) { + indexKey = this.key("convs", "user", options.userId); + } else if (options.resourceId) { + indexKey = this.key("convs", "resource", options.resourceId); + } else { + indexKey = this.key("convs", "all"); + } + + const ids = await this.client.zrange(indexKey, 0, -1); + let conversations = await this.getConversationsByIds(ids); + + if (options.userId) { + conversations = conversations.filter((c) => c.userId === options.userId); + } + if (options.resourceId) { + conversations = conversations.filter((c) => c.resourceId === options.resourceId); + } + + conversations = this.sortConversations( + conversations, + options.orderBy || "updated_at", + options.orderDirection || "DESC", + ); + + const offset = options.offset ?? 0; + if (options.limit && options.limit > 0) { + return conversations.slice(offset, offset + options.limit); + } + return conversations.slice(offset); + } + + /** + * Count conversations with filters + */ + async countConversations(options: ConversationQueryOptions): Promise { + const conversations = await this.queryConversations({ + userId: options.userId, + resourceId: options.resourceId, + }); + return conversations.length; + } + + /** + * Update a conversation + */ + async updateConversation( + id: string, + updates: Partial>, + ): Promise { + const existing = await this.getConversation(id); + if (!existing) { + throw new ConversationNotFoundError(id); + } + + const updated: Conversation = { + ...existing, + title: updates.title !== undefined ? updates.title : existing.title, + resourceId: updates.resourceId !== undefined ? updates.resourceId : existing.resourceId, + userId: updates.userId !== undefined ? updates.userId : existing.userId, + metadata: updates.metadata !== undefined ? updates.metadata : existing.metadata, + updatedAt: new Date().toISOString(), + }; + + const pipeline = this.client.pipeline(); + pipeline.set(this.key("conv", id), safeStringify(updated)); + + // Keep the secondary indexes in sync when indexed fields change + if (updated.resourceId !== existing.resourceId) { + pipeline.zrem(this.key("convs", "resource", existing.resourceId), id); + pipeline.zadd( + this.key("convs", "resource", updated.resourceId), + Date.parse(updated.createdAt), + id, + ); + } + if (updated.userId !== existing.userId) { + pipeline.zrem(this.key("convs", "user", existing.userId), id); + pipeline.zadd(this.key("convs", "user", updated.userId), Date.parse(updated.createdAt), id); + } + + await this.execChecked(pipeline); + + this.log(`Updated conversation ${id}`); + return updated; + } + + /** + * Delete a conversation and all of its associated data + */ + async deleteConversation(id: string): Promise { + const conversation = await this.getConversation(id); + + const pipeline = this.client.pipeline(); + pipeline.del(this.key("conv", id)); + pipeline.del(this.key("msgs", id)); + pipeline.del(this.key("msgdata", id)); + pipeline.del(this.key("steps", id)); + pipeline.del(this.key("stepdata", id)); + pipeline.del(this.key("wm", "conv", id)); + + if (conversation) { + pipeline.zrem(this.key("convs", "all"), id); + pipeline.zrem(this.key("convs", "resource", conversation.resourceId), id); + pipeline.zrem(this.key("convs", "user", conversation.userId), id); + } + + await this.execChecked(pipeline); + + this.log(`Deleted conversation ${id}`); + } + + // ============================================================================ + // Conversation Steps Operations + // ============================================================================ + + async saveConversationSteps(steps: ConversationStepRecord[]): Promise { + if (steps.length === 0) { + return; + } + + const pipeline = this.client.pipeline(); + for (const step of steps) { + const createdAt = step.createdAt ?? new Date().toISOString(); + pipeline.zadd(this.key("steps", step.conversationId), Date.parse(createdAt), step.id); + pipeline.hset( + this.key("stepdata", step.conversationId), + step.id, + safeStringify({ ...step, createdAt }), + ); + } + await this.execChecked(pipeline); + + this.log(`Saved ${steps.length} conversation steps`); + } + + async getConversationSteps( + userId: string, + conversationId: string, + options?: GetConversationStepsOptions, + ): Promise { + const entries = await this.client.hgetall(this.key("stepdata", conversationId)); + + const steps: ConversationStepRecord[] = []; + for (const raw of Object.values(entries)) { + const step = this.safeParse(raw, "conversation step"); + if (!step) continue; + if (step.userId !== userId) continue; + if (options?.operationId && step.operationId !== options.operationId) continue; + steps.push(step); + } + + // Match the SQL adapters: chronological order by step index + steps.sort( + (a, b) => a.stepIndex - b.stepIndex || Date.parse(a.createdAt) - Date.parse(b.createdAt), + ); + + if (options?.limit && options.limit > 0) { + return steps.slice(0, options.limit); + } + return steps; + } + + // ============================================================================ + // Working Memory Operations + // ============================================================================ + + /** + * Get working memory + */ + async getWorkingMemory(params: { + conversationId?: string; + userId?: string; + scope: WorkingMemoryScope; + }): Promise { + const key = this.workingMemoryKey(params.scope, params.conversationId, params.userId); + return this.client.get(key); + } + + /** + * Set working memory + */ + async setWorkingMemory(params: { + conversationId?: string; + userId?: string; + content: string; + scope: WorkingMemoryScope; + }): Promise { + const key = this.workingMemoryKey(params.scope, params.conversationId, params.userId); + await this.client.set(key, params.content); + this.log(`Set working memory for ${params.scope} scope`); + } + + /** + * Delete working memory + */ + async deleteWorkingMemory(params: { + conversationId?: string; + userId?: string; + scope: WorkingMemoryScope; + }): Promise { + const key = this.workingMemoryKey(params.scope, params.conversationId, params.userId); + await this.client.del(key); + this.log(`Deleted working memory for ${params.scope} scope`); + } + + // ============================================================================ + // Workflow State Operations + // ============================================================================ + + /** + * Revive dates after reading a serialized workflow state + */ + private deserializeWorkflowState(raw: string): WorkflowStateEntry | null { + const parsed = this.safeParse(raw, "workflow state"); + if (!parsed) { + return null; + } + + return { + ...parsed, + createdAt: new Date(parsed.createdAt), + updatedAt: new Date(parsed.updatedAt), + suspension: parsed.suspension + ? { ...parsed.suspension, suspendedAt: new Date(parsed.suspension.suspendedAt) } + : undefined, + cancellation: parsed.cancellation + ? { ...parsed.cancellation, cancelledAt: new Date(parsed.cancellation.cancelledAt) } + : undefined, + }; + } + + /** + * Get workflow state by execution ID + */ + async getWorkflowState(executionId: string): Promise { + const raw = await this.client.get(this.key("wf", "state", executionId)); + if (!raw) { + return null; + } + return this.deserializeWorkflowState(raw); + } + + /** + * Query workflow runs with filters + */ + async queryWorkflowRuns(query: WorkflowRunQuery): Promise { + const indexKey = query.workflowId + ? this.key("wf", "idx", query.workflowId) + : this.key("wf", "all"); + + // Newest first, matching the SQL adapters + const ids = await this.client.zrevrange(indexKey, 0, -1); + if (ids.length === 0) { + return []; + } + + // Batch fetch all candidate states in a single round trip + const pipeline = this.client.pipeline(); + for (const id of ids) { + pipeline.get(this.key("wf", "state", id)); + } + const results = await this.execChecked(pipeline); + + const states: WorkflowStateEntry[] = []; + for (const [, raw] of results) { + if (typeof raw !== "string") continue; + const state = this.deserializeWorkflowState(raw); + if (!state) continue; + if (query.status && state.status !== query.status) continue; + if (query.userId && state.userId !== query.userId) continue; + if (query.from && state.createdAt < query.from) continue; + if (query.to && state.createdAt > query.to) continue; + if (query.metadata) { + const matches = Object.entries(query.metadata).every( + ([key, value]) => state.metadata?.[key] === value, + ); + if (!matches) continue; + } + states.push(state); + } + + const offset = query.offset ?? 0; + if (query.limit && query.limit > 0) { + return states.slice(offset, offset + query.limit); + } + return states.slice(offset); + } + + /** + * Set (upsert) workflow state + */ + async setWorkflowState(executionId: string, state: WorkflowStateEntry): Promise { + const existing = await this.getWorkflowState(executionId); + + const score = state.createdAt.getTime(); + const pipeline = this.client.pipeline(); + pipeline.set(this.key("wf", "state", executionId), safeStringify(state)); + pipeline.zadd(this.key("wf", "all"), score, executionId); + pipeline.zadd(this.key("wf", "idx", state.workflowId), score, executionId); + + if (state.status === "suspended") { + pipeline.sadd(this.key("wf", "suspended", state.workflowId), executionId); + } else { + pipeline.srem(this.key("wf", "suspended", state.workflowId), executionId); + } + + // Clean up stale indexes if the workflow id ever changes for an execution + if (existing && existing.workflowId !== state.workflowId) { + pipeline.zrem(this.key("wf", "idx", existing.workflowId), executionId); + pipeline.srem(this.key("wf", "suspended", existing.workflowId), executionId); + } + + await this.execChecked(pipeline); + + this.log(`Saved workflow state ${executionId}`); + } + + /** + * Update workflow state + */ + async updateWorkflowState( + executionId: string, + updates: Partial, + ): Promise { + const existing = await this.getWorkflowState(executionId); + if (!existing) { + throw new Error(`Workflow state ${executionId} not found`); + } + + const updated: WorkflowStateEntry = { + ...existing, + ...updates, + // Index-driving fields are immutable to keep the Redis indexes consistent + id: existing.id, + workflowId: existing.workflowId, + createdAt: existing.createdAt, + updatedAt: new Date(), + }; + + await this.setWorkflowState(executionId, updated); + } + + /** + * Get suspended workflow states for a workflow + */ + async getSuspendedWorkflowStates(workflowId: string): Promise { + const ids = await this.client.smembers(this.key("wf", "suspended", workflowId)); + if (ids.length === 0) { + return []; + } + + const pipeline = this.client.pipeline(); + for (const id of ids) { + pipeline.get(this.key("wf", "state", id)); + } + const results = await this.execChecked(pipeline); + + const states: WorkflowStateEntry[] = []; + for (const [, raw] of results) { + if (typeof raw !== "string") continue; + const state = this.deserializeWorkflowState(raw); + // Guard against stale index entries + if (state && state.status === "suspended") { + states.push(state); + } + } + + return states.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime()); + } + + // ============================================================================ + // Connection Management + // ============================================================================ + + /** + * Disconnect the Redis client; safe to call more than once + */ + async disconnect(): Promise { + if (this.isEnded()) { + return; + } + + try { + await this.client.quit(); + } catch (error) { + // A concurrent disconnect() may have closed the connection first + if (this.isEnded()) { + return; + } + throw error; + } + + this.log("Redis connection closed"); + } + + /** + * Whether the client connection has been closed + */ + private isEnded(): boolean { + return this.client.status === "end"; + } + + /** + * Alias for disconnect(), matching the close() convention of other adapters + */ + async close(): Promise { + await this.disconnect(); + } +} diff --git a/packages/redis/tsconfig.json b/packages/redis/tsconfig.json new file mode 100644 index 000000000..fb0c6c836 --- /dev/null +++ b/packages/redis/tsconfig.json @@ -0,0 +1,30 @@ +{ + "compilerOptions": { + "target": "es2018", + "lib": ["dom", "dom.iterable", "esnext"], + "module": "esnext", + "moduleResolution": "node", + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "outDir": "./dist", + "rootDir": "./", + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "strictFunctionTypes": true, + "strictBindCallApply": true, + "strictPropertyInitialization": true, + "noImplicitThis": true, + "noUnusedLocals": true, + "noUnusedParameters": true, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "types": ["node", "vitest/globals"] + }, + "include": ["src/**/*.ts", "__tests__/**/*.ts"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/redis/tsup.config.ts b/packages/redis/tsup.config.ts new file mode 100644 index 000000000..0819104fd --- /dev/null +++ b/packages/redis/tsup.config.ts @@ -0,0 +1,19 @@ +import { defineConfig } from "tsup"; +import { markAsExternalPlugin } from "../shared/tsup-plugins/mark-as-external"; + +export default defineConfig({ + entry: ["src/index.ts"], + format: ["cjs", "esm"], + splitting: false, + sourcemap: true, + clean: false, + target: "es2022", + outDir: "dist", + minify: false, + dts: true, + esbuildPlugins: [markAsExternalPlugin], + esbuildOptions(options) { + options.keepNames = true; + return options; + }, +}); diff --git a/packages/redis/vitest.config.mts b/packages/redis/vitest.config.mts new file mode 100644 index 000000000..1c4fd6c9a --- /dev/null +++ b/packages/redis/vitest.config.mts @@ -0,0 +1,16 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["**/__tests__/**/*.test.ts", "**/*.spec.ts"], + exclude: ["**/node_modules/**", "**/*.integration.test.ts"], // Exclude node_modules and integration tests + environment: "node", + coverage: { + provider: "v8", + reporter: ["text", "json", "html"], + include: ["src/**/*.ts"], + exclude: ["src/**/*.d.ts", "src/**/index.ts"], + }, + globals: true, + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fae443158..ba5c07020 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -4252,6 +4252,25 @@ importers: specifier: ^3.2.4 version: 3.2.4(@types/node@24.2.1)(@vitest/ui@1.6.1)(jsdom@22.1.0)(msw@2.11.6) + packages/redis: + dependencies: + '@voltagent/internal': + specifier: ^1.0.2 + version: link:../internal + ioredis: + specifier: ^5.6.1 + version: 5.7.0 + devDependencies: + '@vitest/coverage-v8': + specifier: ^3.2.4 + version: 3.2.4(vitest@3.2.4) + '@voltagent/core': + specifier: ^2.4.4 + version: link:../core + ai: + specifier: ^6.0.0 + version: 6.0.3(zod@4.3.5) + packages/resumable-streams: dependencies: '@voltagent/core': @@ -15446,8 +15465,8 @@ packages: dev: false optional: true - /@oxc-project/types@0.142.0: - resolution: {integrity: sha512-7W+2q5AKQVU36fkaryontrHn3YDt1RyUYXatw9i5H8ocYe2sPKSFB6eS8WNPeRKiN1qAWWZUPm7gwFzJGrccqQ==} + /@oxc-project/types@0.147.0: + resolution: {integrity: sha512-IJ3s6ltHLp45S0bh7phkX+gJO7A1Wuz2EaqpAhb8WjqDwbzMiWKHhyyT42tskaWjEYXtHtVCPpnBJVT9+dcRLg==} dev: true /@oxc-project/types@0.94.0: @@ -17659,8 +17678,17 @@ packages: /@repeaterjs/repeater@3.0.6: resolution: {integrity: sha512-Javneu5lsuhwNCryN+pXH93VPQ8g0dBX7wItHFgYiwQmzE1sVdg5tWHiOgHywzL2W21XQopa7IwIEnNbmeUJYA==} - /@rolldown/binding-android-arm64@1.2.2: - resolution: {integrity: sha512-l7x215OGvo1s52JWmR8U/DAVzEDWBCIbTm28aeJV/WDTSHgcKXaZTuBT0hJMs5NggilfJTW3clZVvd24yfKJxA==} + /@rolldown/binding-android-arm-eabi@1.2.6: + resolution: {integrity: sha512-b+jTcARdTiFLI6jB4a5XjTm0RWd6KcRfQj/I2356fxUZemiho9zQLxo0RtCuMDAyKcLo6cEltkgbQp6d1+sjjQ==} + engines: {node: ^20.19.0 || >=22.12.0} + cpu: [arm] + os: [android] + requiresBuild: true + dev: true + optional: true + + /@rolldown/binding-android-arm64@1.2.6: + resolution: {integrity: sha512-lkWU8ZJaRk9q3CIEY1Tc7vIFALp3Xw5NfGJo2hQg5oIqNgxWi1zI+IiDEK3r70BF5Dzol1tcXsnzsRc8NLhG+Q==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [android] @@ -17668,8 +17696,8 @@ packages: dev: true optional: true - /@rolldown/binding-darwin-arm64@1.2.2: - resolution: {integrity: sha512-9u9Xv6c1AJZT0FfwH5vrMG5Jjcwhc1MlyrPu0XfTqkzsmqfks2M6W/o5XwAJgVVN/jHpqqngC1WevHKKTIUtIA==} + /@rolldown/binding-darwin-arm64@1.2.6: + resolution: {integrity: sha512-dgR56NYnvAszm7Ob1B2/Vn0e8bUQYZH2UjVaMMtMVOCKFSfjhfLmuA/9+O+F+ajUdG6B/bSssrKW6JJYASa8jA==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [darwin] @@ -17677,8 +17705,8 @@ packages: dev: true optional: true - /@rolldown/binding-darwin-x64@1.2.2: - resolution: {integrity: sha512-9W1mbGZAfW3oqd85bhBkmpyHCCzL1TeG/zFFP3vg7b0rlly8cxOcre5nXwz+LHazCwac2MNWgPdPCHndABjpWQ==} + /@rolldown/binding-darwin-x64@1.2.6: + resolution: {integrity: sha512-vpVxFvUCFioJqug7OTvqptkc4yb8UX0AwfDmJpaR/0sWz+BUmqSVAf7c8JkUgnN8YLspb4a/N6NhTyMAmdyQ7Q==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [darwin] @@ -17686,8 +17714,8 @@ packages: dev: true optional: true - /@rolldown/binding-freebsd-x64@1.2.2: - resolution: {integrity: sha512-0p1lhiCSCyaerFwtrdZQUx7NqGk6LQnaRKWX7tFQqwQgvX0rjM15cIkm3pax1UpEakK14C4mOxx/jSqCBdBRqQ==} + /@rolldown/binding-freebsd-x64@1.2.6: + resolution: {integrity: sha512-h1wG6Y6K3JlRswxsI64qQJqBAy4vrLuHgRbc8CZMGSWTOFRY6ghMApM1NKzB2I0n5xV1fjkE18SuVl2QpLeNpA==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [freebsd] @@ -17695,8 +17723,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-arm-gnueabihf@1.2.2: - resolution: {integrity: sha512-e+cOJXrJ2L3zx6YzqPg+f6Wbk3V1cKB8bOhbaYdVYN3DdquzNdRAmrbETz1qnt5yp/c7JNlNjmITiA2cVneQ7w==} + /@rolldown/binding-linux-arm-gnueabihf@1.2.6: + resolution: {integrity: sha512-tbCiqub0q2MVWJKgF5PoAlNWCtQydiOYSLIkd8sByqK/6MMYLJRcSXSYodqYtd0O+Fw7QaVmKKlS4oL94YRZ0w==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm] os: [linux] @@ -17704,8 +17732,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-arm64-gnu@1.2.2: - resolution: {integrity: sha512-JsSMsj6sNat/MuhG5fnBD7QgbtpHKVe30x5/bAVirDHdhoQRXJkF6xc0Jqk8O4fiCUQAzMOoH9wZi3m60c8wtg==} + /@rolldown/binding-linux-arm64-gnu@1.2.6: + resolution: {integrity: sha512-oxK9+baEBPhZG5HB4URY+uU04zJWeZlH6Tb9rB5DK4DF9XR1uXNLXt5Q5ZsugTKayNCNLhkcwz/ye74hRI98dg==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [linux] @@ -17713,8 +17741,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-arm64-musl@1.2.2: - resolution: {integrity: sha512-B5G/zJdHaoJn9vD50eGHWkiWfmq8Uhi3IiLPJTzmZTrAalk1bztUikSXo0qga18ibE0IXboyeMUnhPjhAJ45wQ==} + /@rolldown/binding-linux-arm64-musl@1.2.6: + resolution: {integrity: sha512-muWCk27FVBEZtv0MsK8gnfSmgczA8KQ0uRVJbTABKhkRfQc38aUrcb7fhi3BNiyseFmgcRsoMfQsSNJ+DbZdSw==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [linux] @@ -17722,8 +17750,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-ppc64-gnu@1.2.2: - resolution: {integrity: sha512-6mC/awzKka8W6EoekjegpfGkjz8jXWDX63pqu/HYVpyKtZfu65Jsh4QAH3Kej3CAv/c1oGX7psTmFEbr0mDxLA==} + /@rolldown/binding-linux-ppc64-gnu@1.2.6: + resolution: {integrity: sha512-eWDoSfU7Co2qj3vgB3Dt4lj1mG6CoWbcJQkRMP3XJplyCMtuaq3LHvPFjS9QIPvMGWVadJC04Xiy0IdcVPtnwQ==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [ppc64] os: [linux] @@ -17731,8 +17759,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-s390x-gnu@1.2.2: - resolution: {integrity: sha512-412MX9fJLdA1IK28EZnc8jYv2HRTleOZgfLQumJ5zy7OeJLZlg/CETwFaXjNmGVxG51cFHpKLqb5LKvBC+HsHA==} + /@rolldown/binding-linux-s390x-gnu@1.2.6: + resolution: {integrity: sha512-2bWNjRSIayvupRKxXUY2tWG9fYdoUlTqWywHRvE8Eq3GvuQ+f2HeIkve697fIt+IQs/PV8yFsdWuhp1aJ1PdnA==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [s390x] os: [linux] @@ -17740,8 +17768,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-x64-gnu@1.2.2: - resolution: {integrity: sha512-Q/+HI/ToJafZ1iCqGgVQXUEkIjufHCTF0gBQ2a5o3cg7GJ2h0qyq3nvvSmU+bGda2/7ygXpTY4TM6gO9OhQ0ZA==} + /@rolldown/binding-linux-x64-gnu@1.2.6: + resolution: {integrity: sha512-KekI0gS0wLxe1UBSQSjenBVwou/JkcQPDzBPICGZjxUv9k3RteHDPBQaiOicZUFKRIH2wKEimGwVpnJsbPzu7w==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [linux] @@ -17749,8 +17777,8 @@ packages: dev: true optional: true - /@rolldown/binding-linux-x64-musl@1.2.2: - resolution: {integrity: sha512-ZKp/w41n6wCvxzxQHtQSbuphfX3Y4cCvbjkKHusrLx4lh+JWLTU7StSltO/DKARISzbj368d+qUaCjI8K2wzXw==} + /@rolldown/binding-linux-x64-musl@1.2.6: + resolution: {integrity: sha512-TvtPnfVr+HtyGiDmPK4VWmlNm7QhNNAcK5Q9A7aOXsI8545yCyaoMaicXrFZ72JzeYjaUVk7yT243zT0jzjFKQ==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [linux] @@ -17758,8 +17786,8 @@ packages: dev: true optional: true - /@rolldown/binding-openharmony-arm64@1.2.2: - resolution: {integrity: sha512-pxE6xD4KS3eAROkKK5yrhB9/3+vhlhVGMvlQLbdpzrBGDbKrnzx3RLwPaHvasLo6jgaiBLn7e04Df9C4tYhjmA==} + /@rolldown/binding-openharmony-arm64@1.2.6: + resolution: {integrity: sha512-iOo0VEay2XFhaCcH0sps5XIimkSuOnNaZrf6+ZkoSOQBJPKNU48RkmJv0/lSpipexu5P+ouFgafe5IGr/DiQfg==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [openharmony] @@ -17767,8 +17795,8 @@ packages: dev: true optional: true - /@rolldown/binding-win32-arm64-msvc@1.2.2: - resolution: {integrity: sha512-4MqEue5re+xIZzAWsB8sj0P1kqZySWqIuN4t6QaIO/YA6SFwySOLruvWQFzfmqk8LBK2P30KCSJwf6mJCZZ5/A==} + /@rolldown/binding-win32-arm64-msvc@1.2.6: + resolution: {integrity: sha512-y5NTmmasMS455JlOCO4ZM9krIchv3Mvm1crL1iUPGOPgEzSkves9n0SdC5Sjz6+qWDFhd8/JpfWMH8NSWNHe+A==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [win32] @@ -17776,8 +17804,8 @@ packages: dev: true optional: true - /@rolldown/binding-win32-x64-msvc@1.2.2: - resolution: {integrity: sha512-NweNxxD0Nf9t8v7kodun45Ijp3EIwYY+uydPP6qBEYvfBqhIjN6dZMzlQja3tqX/aLs3F3Uz+AxDpKgRhpOZQg==} + /@rolldown/binding-win32-x64-msvc@1.2.6: + resolution: {integrity: sha512-np8iZSLfXlAD4kWhiyq/u0Yt8oZDtRQ8lGhQaCXo2rl37KNjeU0GjJuwr4P3oeZ++ROfofsKNBqR5LTO8aXyWQ==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [win32] @@ -27092,6 +27120,7 @@ packages: /eslint@9.33.0: resolution: {integrity: sha512-TS9bTNIryDzStCpJN93aC5VRSW3uTx9sClUn4B87pwiCaJh220otoI0X8mJKr+VcPtniMdN8GKjlwgWGUv5ZKA==} engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0} + deprecated: This version is no longer supported. Please see https://eslint.org/version-support for other options. hasBin: true peerDependencies: jiti: '*' @@ -37702,7 +37731,7 @@ packages: resolution: {integrity: sha512-IXgzBWvWQwE6PrDI05OvmXUIruQTcoMDzRsOd5CDvHCVLcLHMTSYvOK5Cm46kWqlV3yAbuSpBZdJ5oP5OUoStg==} dev: false - /rolldown-plugin-dts@0.16.11(rolldown@1.2.2)(typescript@5.9.2): + /rolldown-plugin-dts@0.16.11(rolldown@1.2.6)(typescript@5.9.2): resolution: {integrity: sha512-9IQDaPvPqTx3RjG2eQCK5GYZITo203BxKunGI80AGYicu1ySFTUyugicAaTZWRzFWh9DSnzkgNeMNbDWBbSs0w==} engines: {node: '>=20.18.0'} peerDependencies: @@ -37730,35 +37759,36 @@ packages: dts-resolver: 2.1.2 get-tsconfig: 4.10.1 magic-string: 0.30.19 - rolldown: 1.2.2 + rolldown: 1.2.6 typescript: 5.9.2 transitivePeerDependencies: - oxc-resolver - supports-color dev: true - /rolldown@1.2.2: - resolution: {integrity: sha512-opwpo1tQBAcpSUJDt94B7hhLNGOKjCdE//XXjeLrnx9b83bjnw45tXdg1b09yEw/VLFBJGZpwRULMmOZo7ol+A==} + /rolldown@1.2.6: + resolution: {integrity: sha512-vMM4q3aixf46GiF1Kok8jDPFsEpXgFWGjUHXNkNHNm+Y2adXAG2dbX91jkti3i0ZRsOlcmbuzAz1poObSHCmUA==} engines: {node: ^20.19.0 || >=22.12.0} hasBin: true dependencies: - '@oxc-project/types': 0.142.0 + '@oxc-project/types': 0.147.0 '@rolldown/pluginutils': 1.0.0 optionalDependencies: - '@rolldown/binding-android-arm64': 1.2.2 - '@rolldown/binding-darwin-arm64': 1.2.2 - '@rolldown/binding-darwin-x64': 1.2.2 - '@rolldown/binding-freebsd-x64': 1.2.2 - '@rolldown/binding-linux-arm-gnueabihf': 1.2.2 - '@rolldown/binding-linux-arm64-gnu': 1.2.2 - '@rolldown/binding-linux-arm64-musl': 1.2.2 - '@rolldown/binding-linux-ppc64-gnu': 1.2.2 - '@rolldown/binding-linux-s390x-gnu': 1.2.2 - '@rolldown/binding-linux-x64-gnu': 1.2.2 - '@rolldown/binding-linux-x64-musl': 1.2.2 - '@rolldown/binding-openharmony-arm64': 1.2.2 - '@rolldown/binding-win32-arm64-msvc': 1.2.2 - '@rolldown/binding-win32-x64-msvc': 1.2.2 + '@rolldown/binding-android-arm-eabi': 1.2.6 + '@rolldown/binding-android-arm64': 1.2.6 + '@rolldown/binding-darwin-arm64': 1.2.6 + '@rolldown/binding-darwin-x64': 1.2.6 + '@rolldown/binding-freebsd-x64': 1.2.6 + '@rolldown/binding-linux-arm-gnueabihf': 1.2.6 + '@rolldown/binding-linux-arm64-gnu': 1.2.6 + '@rolldown/binding-linux-arm64-musl': 1.2.6 + '@rolldown/binding-linux-ppc64-gnu': 1.2.6 + '@rolldown/binding-linux-s390x-gnu': 1.2.6 + '@rolldown/binding-linux-x64-gnu': 1.2.6 + '@rolldown/binding-linux-x64-musl': 1.2.6 + '@rolldown/binding-openharmony-arm64': 1.2.6 + '@rolldown/binding-win32-arm64-msvc': 1.2.6 + '@rolldown/binding-win32-x64-msvc': 1.2.6 dev: true /rollup-plugin-inject@3.0.2: @@ -39962,8 +39992,8 @@ packages: empathic: 2.0.0 hookable: 5.5.3 publint: 0.3.12 - rolldown: 1.2.2 - rolldown-plugin-dts: 0.16.11(rolldown@1.2.2)(typescript@5.9.2) + rolldown: 1.2.6 + rolldown-plugin-dts: 0.16.11(rolldown@1.2.6)(typescript@5.9.2) semver: 7.7.2 tinyexec: 1.0.1 tinyglobby: 0.2.15 diff --git a/website/docs/agents/memory/redis.md b/website/docs/agents/memory/redis.md new file mode 100644 index 000000000..c72ed3a15 --- /dev/null +++ b/website/docs/agents/memory/redis.md @@ -0,0 +1,170 @@ +--- +title: Redis Memory +slug: /agents/memory/redis +--- + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +# Redis Memory + +`RedisMemoryAdapter` stores conversations in Redis for low-latency memory retrieval. It is a good fit for short-term conversational memory, working memory, or as a caching layer in front of slower persistent stores. + +## Installation + + + + +```bash +npm install @voltagent/redis +``` + + + + +```bash +yarn add @voltagent/redis +``` + + + + +```bash +pnpm add @voltagent/redis +``` + + + + +## Configuration + +```ts +import { Agent, Memory } from "@voltagent/core"; +import { RedisMemoryAdapter } from "@voltagent/redis"; + +// Using connection string +const memory = new Memory({ + storage: new RedisMemoryAdapter({ + connection: process.env.REDIS_URL!, + // or: "redis://localhost:6379" + }), +}); + +// Using connection object (any ioredis options) +const memory = new Memory({ + storage: new RedisMemoryAdapter({ + connection: { + host: process.env.REDIS_HOST || "localhost", + port: parseInt(process.env.REDIS_PORT || "6379"), + password: process.env.REDIS_PASSWORD, + db: 0, + }, + }), +}); + +const agent = new Agent({ + name: "Assistant", + model: "openai/gpt-4o-mini", + memory, +}); +``` + +### Configuration Options + +| Option | Type | Description | +| ------------ | ------------------ | ------------------------------------------------------------------------------------------------ | +| `connection` | `string \| object` | Connection string or [ioredis options](https://github.com/redis/ioredis#connect-to-redis) object | +| `keyPrefix` | `string` | Prefix for all Redis keys (default: `voltagent`) | +| `debug` | `boolean` | Enable debug logging (default: `false`) | + +## Features + +### Conversation Storage + +- Messages stored per `conversationId` in Redis hashes, ordered by creation time in sorted sets +- All `StorageAdapter` methods supported +- Conversation indexes by resource, user, and globally for efficient lookups +- Supports filtering, pagination, and sorting +- No automatic message pruning - all messages are preserved until deleted. Note that the adapter never removes keys on its own, but a Redis eviction policy (e.g. `maxmemory-policy`) may still evict keys before they are explicitly deleted + +### Working Memory + +Supports both conversation and user-scoped working memory: + +```ts +import { z } from "zod"; + +const memory = new Memory({ + storage: new RedisMemoryAdapter({ + connection: process.env.REDIS_URL!, + }), + workingMemory: { + enabled: true, + scope: "user", // or "conversation" + schema: z.object({ + preferences: z.array(z.string()).optional(), + }), + }, +}); +``` + +Storage: + +- Conversation scope: `${keyPrefix}:wm:conv:{conversationId}` +- User scope: `${keyPrefix}:wm:user:{userId}` + +See [Working Memory](./working-memory.md) for configuration details. + +## Data Modeling + +The adapter maps memory structures to native Redis data types (assuming the default `voltagent` key prefix): + +| Key pattern | Type | Contents | +| --------------------------------------- | ------ | ---------------------------------------- | +| `voltagent:conv:{id}` | STRING | Serialized conversation | +| `voltagent:convs:all` | ZSET | Conversation ids scored by creation time | +| `voltagent:convs:resource:{resourceId}` | ZSET | Conversation ids per resource | +| `voltagent:convs:user:{userId}` | ZSET | Conversation ids per user | +| `voltagent:msgs:{conversationId}` | ZSET | Message ids scored by creation time | +| `voltagent:msgdata:{conversationId}` | HASH | Message id → serialized message | +| `voltagent:steps:{conversationId}` | ZSET | Step ids scored by creation time | +| `voltagent:stepdata:{conversationId}` | HASH | Step id → serialized step | +| `voltagent:wm:conv:{conversationId}` | STRING | Working memory (conversation scope) | +| `voltagent:wm:user:{userId}` | STRING | Working memory (user scope) | +| `voltagent:wf:state:{executionId}` | STRING | Serialized workflow state | +| `voltagent:wf:all` | ZSET | Execution ids scored by creation time | +| `voltagent:wf:idx:{workflowId}` | ZSET | Execution ids per workflow | +| `voltagent:wf:suspended:{workflowId}` | SET | Suspended execution ids per workflow | + +Multi-key writes are batched with Redis pipelines, and duplicate conversation creation is guarded by an atomic `SET ... NX`. + +## Persistence Considerations + +Redis is an in-memory store. By default, data can be lost on restart: + +- **RDB snapshots**: Configure `save` directives in `redis.conf` (e.g. `save 900 1`) for periodic point-in-time snapshots. +- **AOF (Append Only File)**: Enable with `appendonly yes` for durable, per-write persistence. Use `appendfsync everysec` for a good durability/performance balance. +- **Managed Redis**: Providers like AWS ElastiCache, Upstash, or Redis Cloud handle persistence for you. + +If you need durable long-term history, consider pairing Redis with a disk-based adapter such as [PostgreSQL](./postgres.md) or [LibSQL](./libsql.md), or use Redis primarily for working memory and short-term conversations. + +## Memory Limits + +Since Redis is memory-bound, very large or numerous conversation histories can consume significant RAM. Consider: + +- Setting a `maxmemory` policy appropriate to your workload +- Using `keyPrefix` to isolate VoltAgent keys, making them easy to inspect (`SCAN`) or flush +- Periodically deleting old conversations via `deleteConversation` + +## Connection Management + +The adapter holds a single ioredis connection. Close it during graceful shutdown: + +```ts +const storage = new RedisMemoryAdapter({ connection: process.env.REDIS_URL! }); + +process.on("SIGTERM", async () => { + await storage.disconnect(); + process.exit(0); +}); +``` diff --git a/website/sidebars.ts b/website/sidebars.ts index 160d22981..749300d25 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -211,6 +211,7 @@ const sidebars: SidebarsConfig = { "agents/memory/libsql", "agents/memory/cloudflare-d1", "agents/memory/postgres", + "agents/memory/redis", "agents/memory/supabase", ], },