Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 67 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -410,6 +410,73 @@ const chunksWithEmbeddings = await Promise.all(
await rag.add(ctx, { namespace: "global", chunks });
```

### Reusing stored embeddings

Pass `includeEmbeddings: true` to `rag.listChunks` to include an `embedding` on
each chunk. You can use it as the `query` in `rag.search`, or pass the chunks to
`rag.add` to copy content into another namespace without embedding it again. Use
the same embedding model and dimension for the source and destination.

```ts
// Inside an action:
const { page } = await rag.listChunks(ctx, {
entryId,
paginationOpts: { numItems: 1, cursor: null },
includeEmbeddings: true,
});
const embedding = page[0]?.embedding;
if (embedding) {
const results = await rag.search(ctx, {
namespace: "global",
query: embedding,
});
}
```

Run the paginated copy in an **action** so each page read and insertion batch
gets a separate transaction budget. `rag.add` also accepts a mutation context
for small, pre-embedded entries, but its nested calls then share the enclosing
mutation's read and write limits. An async iterator alone does not avoid those
limits. A single copy still needs to finish within the action runtime limit.

```ts
import { internalAction } from "./_generated/server.js";
import { vEntryId } from "@convex-dev/rag";

export const copyEntry = internalAction({
args: { entryId: vEntryId },
handler: async (ctx, { entryId }) => {
// Copy all pages, preserving chunk order and metadata.
async function* storedChunks() {
let cursor: string | null = null;
while (true) {
const result = await rag.listChunks(ctx, {
entryId,
paginationOpts: { numItems: 50, cursor },
includeEmbeddings: true,
});
yield* result.page;
if (result.isDone) break;
cursor = result.continueCursor;
}
}
return await rag.add(ctx, {
namespace: "user",
chunks: storedChunks(),
});
},
});
```

Embeddings are omitted by default. Including them adds vector reads and response
bytes, so use smaller pages for large dimensions. Stored embeddings are returned
normalized, with importance weighting removed, rather than as an exact copy of
the provider's output. Zero-importance entries have lost their vector direction
and return zeros. For 4096-dimensional embeddings, storage discards the last
coordinate; it is returned as zero. Pending chunks return their original
embeddings. Entry metadata, filters, and importance can be supplied separately
when copying; they are not part of the chunk list.

## Add Entries Asynchronously using File Storage

For large files, you can upload them to file storage, then provide a chunker
Expand Down
188 changes: 186 additions & 2 deletions src/client/index.test.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,11 @@
import { describe, expect, test } from "vitest";
import { RAG } from "./index.js";
import { RAG, vEntryId, type EntryId } from "./index.js";
import { v } from "convex/values";
import { defineSchema } from "convex/server";
import {
defineSchema,
type GenericActionCtx,
type GenericDataModel,
} from "convex/server";
import { defineTestApp } from "convex-test";
import componentTest from "../test.js";
import { openai } from "@ai-sdk/openai";
Expand Down Expand Up @@ -136,12 +140,64 @@ const generate = app.action({
},
});

const seedLargeEntry = app.action({
args: { count: v.number() },
handler: async (ctx, { count }) => {
function* chunks() {
for (let i = 0; i < count; i++) {
yield {
text: `Chunk ${i}`,
metadata: { index: i },
embedding: dummyEmbeddings("A"),
};
}
}
return rag.add(ctx, { namespace: "large-source", chunks: chunks() });
},
});

async function copyStoredEntry(
ctx: Pick<GenericActionCtx<GenericDataModel>, "runQuery" | "runMutation">,
{ entryId, namespace }: { entryId: EntryId; namespace: string },
) {
let pagesRead = 0;
async function* storedChunks() {
let cursor: string | null = null;
while (true) {
const result = await rag.listChunks(ctx, {
entryId,
paginationOpts: { numItems: 50, cursor },
includeEmbeddings: true,
});
pagesRead++;
yield* result.page;
if (result.isDone) break;
cursor = result.continueCursor;
}
}
const result = await rag.add(ctx, { namespace, chunks: storedChunks() });
return { ...result, pagesRead };
}

const copyEntryArgs = { entryId: vEntryId, namespace: v.string() };
const copyEntryAction = app.action({
args: copyEntryArgs,
handler: copyStoredEntry,
});
const copyEntryMutation = app.mutation({
args: copyEntryArgs,
handler: copyStoredEntry,
});

const { api, createTest } = app.defineModules({
fns: {
findEntryByContentHash,
add,
search,
generate,
seedLargeEntry,
copyEntryAction,
copyEntryMutation,
},
});

Expand All @@ -152,6 +208,64 @@ function dummyEmbeddings(text: string) {
}

describe("RAG thick client", () => {
test("large paginated copies require an action to stay within transaction limits", async () => {
const t = createTest({ transactionLimits: true });
// The embedding coordinates alone exceed the default 16 MiB transaction
// budget. Use enough chunks for multiple read pages and insertion batches,
// including a partial final page/batch.
const count = 1501;
expect(count * 1536 * 8).toBeGreaterThan(16 * 1024 * 1024);
const source = await t.action(api.fns.seedLargeEntry, { count });
expect(source.status).toBe("ready");
expect(source.usage).toEqual({ tokens: 0 });

await expect(
t.mutation(api.fns.copyEntryMutation, {
entryId: source.entryId,
namespace: "mutation-copy",
}),
).rejects.toThrow(/Read too much data in a single function execution/);

const copied = await t.action(api.fns.copyEntryAction, {
entryId: source.entryId,
namespace: "action-copy",
});
expect(copied.status).toBe("ready");
expect(copied.usage).toEqual({ tokens: 0 });
expect(copied.pagesRead).toBe(Math.ceil(count / 50));

let cursor: string | null = null;
let chunksRead = 0;
const original = dummyEmbeddings("A");
const magnitude = Math.sqrt(
original.reduce((sum, value) => sum + value ** 2, 0),
);
while (true) {
const result: Awaited<ReturnType<typeof rag.listChunks>> = await t.query(
app.components.rag.chunks.list,
{
entryId: copied.entryId,
order: "asc",
paginationOpts: { numItems: 50, cursor },
includeEmbeddings: true,
},
);
for (const chunk of result.page) {
expect(chunk.order).toBe(chunksRead);
expect(chunk.text).toBe(`Chunk ${chunksRead}`);
expect(chunk.metadata).toEqual({ index: chunksRead });
expect(chunk.state).toBe("ready");
expect(chunk.embedding).toHaveLength(1536);
expect(chunk.embedding![0]).toBeCloseTo(original[0] / magnitude);
expect(chunk.embedding![1535]).toBeCloseTo(original[1535] / magnitude);
chunksRead++;
}
if (result.isDone) break;
cursor = result.continueCursor;
}
expect(chunksRead).toBe(count);
}, 30_000);

test("should add a entry and be able to list it", async () => {
const t = createTest();
const { entryId, status, usage } = await t.mutation(api.fns.add, {
Expand Down Expand Up @@ -193,6 +307,76 @@ describe("RAG thick client", () => {
// expect(result).toBe(1);
});

test("reuses listed embeddings for search and copying without embedding calls", async () => {
const t = createTest();
const { entryId } = await t.mutation(api.fns.add, {
key: "template",
namespace: "templates",
importance: 0.25,
chunks: [
{
text: "A",
metadata: { source: "template" },
embedding: dummyEmbeddings("A"),
},
{ text: "B", metadata: {}, embedding: dummyEmbeddings("B") },
],
});
const { page } = await t.run((ctx) =>
rag.listChunks(ctx, {
entryId,
paginationOpts: { numItems: 10, cursor: null },
includeEmbeddings: true,
}),
);
const chunks = page.map((chunk) => {
expect(chunk.embedding).toHaveLength(1536);
return {
text: chunk.text,
metadata: chunk.metadata ?? {},
embedding: chunk.embedding!,
};
});
const originalSearch = await t.action(api.fns.search, {
namespace: "templates",
embedding: dummyEmbeddings("A"),
});
const reusedSearch = await t.action(api.fns.search, {
namespace: "templates",
embedding: chunks[0].embedding,
});
expect(reusedSearch.usage).toEqual({ tokens: 0 });
expect(reusedSearch.results.map((result) => result.order)).toEqual(
originalSearch.results.map((result) => result.order),
);
reusedSearch.results.forEach((result, i) => {
expect(result.score).toBeCloseTo(originalSearch.results[i].score);
});
const copied = await t.mutation(api.fns.add, {
key: "copy",
namespace: "user",
importance: 0.75,
chunks,
});
expect(copied.status).toBe("ready");
expect(copied.usage).toEqual({ tokens: 0 });
const copiedSearch = await t.action(api.fns.search, {
namespace: "user",
embedding: dummyEmbeddings("A"),
});
expect(copiedSearch.results[0].entryId).toBe(copied.entryId);
expect(copiedSearch.results[0].score).toBeCloseTo(0.75);
const copiedChunks = await t.run((ctx) =>
rag.listChunks(ctx, {
entryId: copied.entryId,
paginationOpts: { numItems: 10, cursor: null },
}),
);
expect(
copiedChunks.page.map(({ text, metadata }) => ({ text, metadata })),
).toEqual(chunks.map(({ text, metadata }) => ({ text, metadata })));
});

test("should be able to re-add an entry with the same key", async () => {
const t = createTest();
const { entryId, status, usage } = await t.mutation(api.fns.add, {
Expand Down
16 changes: 16 additions & 0 deletions src/client/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,11 @@ export class RAG<
* If you don't provide a key, it will always create a new entry.
* If you provide a contentHash, it will deduplicate the entry if it already exists.
* The filterValues you provide can be used later to search for it.
*
* Use an action for large entries or when embeddings need to be generated.
* With an action context, each batch runs in a separate transaction. A mutation
* context is supported for small, pre-embedded entries, but all nested reads
* and writes share the enclosing transaction's limits, even with an iterator.
*/
async add(
ctx: CtxWith<"runMutation">,
Expand Down Expand Up @@ -740,12 +745,23 @@ export class RAG<
paginationOpts: PaginationOptions;
entryId: EntryId;
order?: "desc" | "asc";
/**
* Include embeddings for reuse with `search` or `add` using the same
* embedding model and dimension. Defaults to false.
* Stored embeddings are normalized with importance weighting removed.
* Zero-importance embeddings are all zeros; for 4096 dimensions, the
* discarded last coordinate is returned as zero. Pending chunks return
* their original embeddings.
* Use smaller pages when including these additional reads and bytes.
*/
includeEmbeddings?: boolean;
},
): Promise<PaginationResult<Chunk>> {
return ctx.runQuery(this.component.chunks.list, {
entryId: args.entryId,
paginationOpts: args.paginationOpts,
order: args.order ?? "asc",
includeEmbeddings: args.includeEmbeddings,
});
}

Expand Down
2 changes: 2 additions & 0 deletions src/component/_generated/component.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
"internal",
{
entryId: string;
includeEmbeddings?: boolean;
order: "desc" | "asc";
paginationOpts: {
cursor: string | null;
Expand All @@ -58,6 +59,7 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
continueCursor: string;
isDone: boolean;
page: Array<{
embedding?: Array<number>;
metadata?: Record<string, any>;
order: number;
state: "pending" | "ready" | "replaced";
Expand Down
Loading
Loading