Function index

packages/components/src/indexing.ts:257–378

(args: IndexArgs)

 * @returns {Promise<IndexingResult>}
 */
export async function index(args: IndexArgs): Promise<IndexingResult> {
    const { docsSource, recordManager, vectorStore, options } = args
    const { batchSize = 100, cleanup, sourceIdKey, cleanupBatchSize = 1000, forceUpdate = false, vectorStoreName } = options ?? {}

    if (cleanup === 'incremental' && !sourceIdKey) {
        throw new Error("sourceIdKey is required when cleanup mode is incremental. Please provide through 'options.sourceIdKey'.")
    }

    if (vectorStoreName) {
        ;(recordManager as any).namespace = (recordManager as any).namespace + '_' + vectorStoreName
    }

    const docs = _isBaseDocumentLoader(docsSource) ? await docsSource.load() : docsSource

    const sourceIdAssigner = _getSourceIdAssigner(sourceIdKey ?? null)

    const indexStartDt = await recordManager.getTime()
    let numAdded = 0
    let addedDocs: Document[] = []
    let numDeleted = 0
    let numUpdated = 0
    let numSkipped = 0
    let totalKeys = 0

    const batches = _batch<DocumentInterface>(batchSize ?? 100, docs)

    for (const batch of batches) {
        const hashedDocs = _deduplicateInOrder(batch.map((doc) => _HashedDocument.fromDocument(doc)))

        const sourceIds = hashedDocs.map((doc) => sourceIdAssigner(doc))

        if (cleanup === 'incremental') {
            hashedDocs.forEach((_hashedDoc, index) => {
                const source = sourceIds[index]
                if (source === null) {
                    throw new Error('sourceIdKey must be provided when cleanup is incremental')
                }
            })
        }

        const batchExists = await recordManager.exists(hashedDocs.map((doc) => doc.uid))

        const uids: string[] = []
        const docsToIndex: DocumentInterface[] = []
        const docsToUpdate: Array<{ uid: string; docId: string }> = []
        const seenDocs = new Set<string>()
        hashedDocs.forEach((hashedDoc, i) => {
            const docExists = batchExists[i]
            if (docExists) {
                if (forceUpdate) {
                    seenDocs.add(hashedDoc.uid)
                } else {
                    docsToUpdate.push({ uid: hashedDoc.uid, docId: hashedDoc.metadata.docId as string })
                    return
                }
            }
            uids.push(hashedDoc.uid)
            docsToIndex.push(hashedDoc.toDocument())