feat: add TwelveLabs nodes (Pegasus video loader + Marengo embeddings)

[mohit-twelvelabs]

Hi! I'm Mohit, I work at TwelveLabs (@mohit-twelvelabs).

What this adds

Two new opt-in nodes backed by the TwelveLabs video understanding platform, plus a shared API credential:

• TwelveLabs Video (Document Loader) — analyzes a publicly accessible video URL with the Pegasus model and loads the generated description as a Document. Supports a custom prompt, model name, max tokens, a TextSplitter, and the standard metadata / omit-metadata-keys options used by the other loaders.
• TwelveLabs Embedding (Embeddings) — exposes Marengo's multimodal embeddings (a shared 512-dim space across video, image, audio and text) as a standard LangChain Embeddings implementation, so text queries can be embedded against video embeddings stored in any vector store.
• TwelveLabs API credential — single API key field, typed password per the credential guidelines in CONTRIBUTING.

Why it helps Flowise

Flowise has rich text/image embedding and loader coverage but no video understanding provider. These nodes let users build flows that describe/Q&A over videos (Pegasus) and do cross-modal semantic search (Marengo) without leaving Flowise.

Opt-in / non-breaking

Both are new nodes auto-discovered from packages/components/nodes. Nothing existing is changed, no defaults touched, and no new dependencies — they call the public TwelveLabs REST API (https://api.twelvelabs.io/v1.3) using the axios already in the package and @langchain/core's Embeddings base.

How it was tested

• Co-located unit tests for both nodes (*.test.ts), all passing — they mock axios and assert request wiring, the 512-dim vector path, and error handling.
• pnpm build (tsc + gulp) for flowise-components succeeds; both nodes and their icons land in dist.
• Prettier + ESLint clean on the changed files.
• Verified against the live API: a Marengo marengo3.0 text embed returns a 512-dim vector, and a Pegasus analysis of a public sample video returns generated text via the /analyze/tasks submit + poll flow the loader implements.

You can grab a free API key at https://twelvelabs.io — there's a generous free tier.

[gemini-code-assist]

Code Review

This pull request introduces TwelveLabs integration, adding a TwelveLabs API credential, a Video Document Loader utilizing the Pegasus model, and a Marengo-based text embedding node. The review feedback highlights several opportunities for improvement: ensuring compatibility and robustness of multipart requests by using the form-data package and passing explicit headers, parallelizing document embeddings using Promise.all to resolve a performance bottleneck, utilizing standard Flowise helper functions for metadata and output handling, and wrapping the long-polling loop in a try/catch block to handle transient network errors while failing fast on invalid API responses.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Relying on the global FormData can cause runtime errors on older Node.js versions (Node < 18) or boundary issues with Axios. Importing the form-data package ensures compatibility and robust multipart requests across all environments.

Suggested change

	import axios from 'axios'
	import { Embeddings, EmbeddingsParams } from '@langchain/core/embeddings'
	import axios from 'axios'
	import FormData from 'form-data'
	import { Embeddings, EmbeddingsParams } from '@langchain/core/embeddings'

[gemini-code-assist]

When using the form-data package, you should explicitly pass the form headers (including the boundary) to Axios to ensure the multipart request is correctly formatted.

        const form = new FormData()
        form.append('model_name', this.model)
        form.append('text', text.replace(/\n/g, ' '))

        const response = await this.caller.call(async () => {
            const res = await axios.post<EmbedResponse>(`${this.baseUrl}/embed`, form, {
                headers: {
                    ...form.getHeaders(),
                    'x-api-key': this.apiKey as string
                }
            })
            return res.data
        })

[gemini-code-assist]

Embedding documents sequentially in a loop can be a major performance bottleneck. Since this.caller handles concurrency and rate limits automatically, we can use Promise.all to embed all documents in parallel.

    async embedDocuments(documents: string[]): Promise<number[][]> {
        return Promise.all(documents.map((document) => this.embedText(document)))
    }

[gemini-code-assist]

Importing the standard handleDocumentLoaderMetadata and handleDocumentLoaderOutput helper functions from src/utils avoids duplicating code and ensures consistent behavior across all document loaders in Flowise.

import axios from 'axios'
import { Document } from '@langchain/core/documents'
import { TextSplitter } from '@langchain/textsplitters'
import { ICommonObject, IDocument, INode, INodeData, INodeOutputsValue, INodeParams } from '../../../src/Interface'
import {
    getCredentialData,
    getCredentialParam,
    handleDocumentLoaderMetadata,
    handleDocumentLoaderOutput
} from '../../../src/utils'

[gemini-code-assist]

Use the standard handleDocumentLoaderMetadata and handleDocumentLoaderOutput helper functions to parse metadata, omit keys, and format the output consistently with other document loaders.

        let docs: Document[] = []
        const baseMetadata = { source: videoUrl, model: modelName }

        if (textSplitter) {
            const splitDocs = await textSplitter.createDocuments([analysis])
            docs.push(...splitDocs.map((doc) => new Document({ pageContent: doc.pageContent, metadata: { ...doc.metadata, ...baseMetadata } })))
        } else {
            docs.push(new Document({ pageContent: analysis, metadata: baseMetadata }))
        }

        docs = handleDocumentLoaderMetadata(docs, _omitMetadataKeys, metadata)

        return handleDocumentLoaderOutput(docs, output)

[gemini-code-assist]

During long polling (which can take up to 10 minutes), a single transient network error or API glitch will crash the entire flow. Wrapping the status check in a try/catch block makes the polling logic significantly more resilient. Additionally, if the API response is missing the expected analysis data, we should throw an error rather than silently returning an empty string, promoting fail-fast behavior.

        const deadline = Date.now() + params.timeoutSec * 1000
        while (Date.now() < deadline) {
            try {
                const { data: status } = await axios.get<AnalyzeTask>(`${TWELVELABS_API_BASE}/analyze/tasks/${taskId}`, { headers })
                if (status.status === 'ready') {
                    if (!status.result?.data) {
                        throw new Error('TwelveLabs analysis task returned invalid or missing data')
                    }
                    return status.result.data
                }
                if (status.status === 'failed') {
                    throw new Error('TwelveLabs analysis task failed')
                }
            } catch (error: any) {
                if (error.message === 'TwelveLabs analysis task failed' || error.message === 'TwelveLabs analysis task returned invalid or missing data') {
                    throw error
                }
                console.error(`Error polling TwelveLabs task status: ${error.message || error}`)
            }
            await new Promise((resolve) => setTimeout(resolve, 5000))
        }

References

1. When handling potentially invalid data from external sources (like an API response), prefer throwing an error for invalid input types rather than silently returning a default or empty value. This promotes fail-fast behavior.