mirror of
https://github.com/farcasclaudiu/Flowise.git
synced 2026-06-28 15:00:57 +03:00
[Prostgres Vector Store] Add PGVector Driver option + Fix null character issue w/ TypeORM Driver (#3367)
* Add PGVector Driver option + Fix null character issue w/ TypeORM Driver * Handle file upload case with PGVector * Cleanup * Fix data filtering for chatflow uploaded files * Add distanceStrategy parameter * Fix query to improve chatflow uploaded files filtering * Ensure PGVector release connections * Await client connected * Make Postgres credentials optionnal when set on env variables * Document env variables in nodes directories * Prevent reuse client * Fix empty metadataFilter * Update CONTRIBUTING.md * Update Postgres.ts --------- Co-authored-by: Henry Heng <henryheng@flowiseai.com>
This commit is contained in:
Jérémy JOURDIN
@@ -0,0 +1,48 @@
|
||||
import { VectorStore } from '@langchain/core/vectorstores'
|
||||
import { getCredentialData, getCredentialParam, ICommonObject, INodeData } from '../../../../src'
|
||||
import { Document } from '@langchain/core/documents'
|
||||
import { Embeddings } from '@langchain/core/embeddings'
|
||||
import { getDatabase, getHost, getPort, getTableName } from '../utils'
|
||||
|
||||
export abstract class VectorStoreDriver {
|
||||
constructor(protected nodeData: INodeData, protected options: ICommonObject) {}
|
||||
|
||||
abstract instanciate(metaDataFilters?: any): Promise<VectorStore>
|
||||
|
||||
abstract fromDocuments(documents: Document[]): Promise<VectorStore>
|
||||
|
||||
protected async adaptInstance(instance: VectorStore, _metaDataFilters?: any): Promise<VectorStore> {
|
||||
return instance
|
||||
}
|
||||
|
||||
getHost() {
|
||||
return getHost(this.nodeData) as string
|
||||
}
|
||||
|
||||
getPort() {
|
||||
return getPort(this.nodeData) as number
|
||||
}
|
||||
|
||||
getDatabase() {
|
||||
return getDatabase(this.nodeData) as string
|
||||
}
|
||||
|
||||
getTableName() {
|
||||
return getTableName(this.nodeData)
|
||||
}
|
||||
|
||||
getEmbeddings() {
|
||||
return this.nodeData.inputs?.embeddings as Embeddings
|
||||
}
|
||||
|
||||
async getCredentials() {
|
||||
const credentialData = await getCredentialData(this.nodeData.credential ?? '', this.options)
|
||||
const user = getCredentialParam('user', credentialData, this.nodeData, process.env.POSTGRES_VECTORSTORE_USER)
|
||||
const password = getCredentialParam('password', credentialData, this.nodeData, process.env.POSTGRES_VECTORSTORE_PASSWORD)
|
||||
|
||||
return {
|
||||
user,
|
||||
password
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
import { VectorStoreDriver } from './Base'
|
||||
import { FLOWISE_CHATID } from '../../../../src'
|
||||
import { DistanceStrategy, PGVectorStore, PGVectorStoreArgs } from '@langchain/community/vectorstores/pgvector'
|
||||
import { Document } from '@langchain/core/documents'
|
||||
import { PoolConfig } from 'pg'
|
||||
import { getContentColumnName } from '../utils'
|
||||
|
||||
export class PGVectorDriver extends VectorStoreDriver {
|
||||
static CONTENT_COLUMN_NAME_DEFAULT: string = 'pageContent'
|
||||
|
||||
protected _postgresConnectionOptions: PoolConfig
|
||||
|
||||
protected async getPostgresConnectionOptions() {
|
||||
if (!this._postgresConnectionOptions) {
|
||||
const { user, password } = await this.getCredentials()
|
||||
const additionalConfig = this.nodeData.inputs?.additionalConfig as string
|
||||
|
||||
let additionalConfiguration = {}
|
||||
|
||||
if (additionalConfig) {
|
||||
try {
|
||||
additionalConfiguration = typeof additionalConfig === 'object' ? additionalConfig : JSON.parse(additionalConfig)
|
||||
} catch (exception) {
|
||||
throw new Error('Invalid JSON in the Additional Configuration: ' + exception)
|
||||
}
|
||||
}
|
||||
|
||||
this._postgresConnectionOptions = {
|
||||
...additionalConfiguration,
|
||||
host: this.getHost(),
|
||||
port: this.getPort(),
|
||||
user: user,
|
||||
password: password,
|
||||
database: this.getDatabase()
|
||||
}
|
||||
}
|
||||
|
||||
return this._postgresConnectionOptions
|
||||
}
|
||||
|
||||
async getArgs(): Promise<PGVectorStoreArgs> {
|
||||
return {
|
||||
postgresConnectionOptions: await this.getPostgresConnectionOptions(),
|
||||
tableName: this.getTableName(),
|
||||
columns: {
|
||||
contentColumnName: getContentColumnName(this.nodeData)
|
||||
},
|
||||
distanceStrategy: (this.nodeData.inputs?.distanceStrategy || 'cosine') as DistanceStrategy
|
||||
}
|
||||
}
|
||||
|
||||
async instanciate(metadataFilters?: any) {
|
||||
return this.adaptInstance(await PGVectorStore.initialize(this.getEmbeddings(), await this.getArgs()), metadataFilters)
|
||||
}
|
||||
|
||||
async fromDocuments(documents: Document[]) {
|
||||
const instance = await this.instanciate()
|
||||
|
||||
await instance.addDocuments(documents)
|
||||
|
||||
return this.adaptInstance(instance)
|
||||
}
|
||||
|
||||
protected async adaptInstance(instance: PGVectorStore, metadataFilters?: any): Promise<PGVectorStore> {
|
||||
const { [FLOWISE_CHATID]: chatId, ...pgMetadataFilter } = metadataFilters || {}
|
||||
|
||||
const baseSimilaritySearchVectorWithScoreFn = instance.similaritySearchVectorWithScore.bind(instance)
|
||||
|
||||
instance.similaritySearchVectorWithScore = async (query, k, filter) => {
|
||||
return await baseSimilaritySearchVectorWithScoreFn(query, k, filter ?? pgMetadataFilter)
|
||||
}
|
||||
|
||||
const basePoolQueryFn = instance.pool.query.bind(instance.pool)
|
||||
|
||||
// @ts-ignore
|
||||
instance.pool.query = async (queryString: string, parameters: any[]) => {
|
||||
if (!instance.client) {
|
||||
instance.client = await instance.pool.connect()
|
||||
}
|
||||
|
||||
const whereClauseRegex = /WHERE ([^\n]+)/
|
||||
let chatflowOr = ''
|
||||
|
||||
// Match chatflow uploaded file and keep filtering on other files:
|
||||
// https://github.com/FlowiseAI/Flowise/pull/3367#discussion_r1804229295
|
||||
if (chatId) {
|
||||
parameters.push({ [FLOWISE_CHATID]: chatId })
|
||||
|
||||
chatflowOr = `OR metadata @> $${parameters.length}`
|
||||
}
|
||||
|
||||
if (queryString.match(whereClauseRegex)) {
|
||||
queryString = queryString.replace(whereClauseRegex, `WHERE (($1) AND NOT (metadata ? '${FLOWISE_CHATID}')) ${chatflowOr}`)
|
||||
} else {
|
||||
const orderByClauseRegex = /ORDER BY (.*)/
|
||||
// Insert WHERE clause before ORDER BY
|
||||
queryString = queryString.replace(
|
||||
orderByClauseRegex,
|
||||
`WHERE (metadata @> '{}' AND NOT (metadata ? '${FLOWISE_CHATID}')) ${chatflowOr}
|
||||
ORDER BY $1
|
||||
`
|
||||
)
|
||||
}
|
||||
|
||||
// Run base function
|
||||
const queryResult = await basePoolQueryFn(queryString, parameters)
|
||||
|
||||
// ensure connection is released
|
||||
instance.client.release()
|
||||
instance.client = undefined
|
||||
|
||||
return queryResult
|
||||
}
|
||||
|
||||
return instance
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
import { DataSourceOptions } from 'typeorm'
|
||||
import { VectorStoreDriver } from './Base'
|
||||
import { FLOWISE_CHATID, ICommonObject } from '../../../../src'
|
||||
import { TypeORMVectorStore, TypeORMVectorStoreArgs, TypeORMVectorStoreDocument } from '@langchain/community/vectorstores/typeorm'
|
||||
import { VectorStore } from '@langchain/core/vectorstores'
|
||||
import { Document } from '@langchain/core/documents'
|
||||
import { Pool } from 'pg'
|
||||
|
||||
export class TypeORMDriver extends VectorStoreDriver {
|
||||
protected _postgresConnectionOptions: DataSourceOptions
|
||||
|
||||
protected async getPostgresConnectionOptions() {
|
||||
if (!this._postgresConnectionOptions) {
|
||||
const { user, password } = await this.getCredentials()
|
||||
const additionalConfig = this.nodeData.inputs?.additionalConfig as string
|
||||
|
||||
let additionalConfiguration = {}
|
||||
|
||||
if (additionalConfig) {
|
||||
try {
|
||||
additionalConfiguration = typeof additionalConfig === 'object' ? additionalConfig : JSON.parse(additionalConfig)
|
||||
} catch (exception) {
|
||||
throw new Error('Invalid JSON in the Additional Configuration: ' + exception)
|
||||
}
|
||||
}
|
||||
|
||||
this._postgresConnectionOptions = {
|
||||
...additionalConfiguration,
|
||||
type: 'postgres',
|
||||
host: this.getHost(),
|
||||
port: this.getPort(),
|
||||
username: user, // Required by TypeORMVectorStore
|
||||
user: user, // Required by Pool in similaritySearchVectorWithScore
|
||||
password: password,
|
||||
database: this.getDatabase()
|
||||
} as DataSourceOptions
|
||||
}
|
||||
return this._postgresConnectionOptions
|
||||
}
|
||||
|
||||
async getArgs(): Promise<TypeORMVectorStoreArgs> {
|
||||
return {
|
||||
postgresConnectionOptions: await this.getPostgresConnectionOptions(),
|
||||
tableName: this.getTableName()
|
||||
}
|
||||
}
|
||||
|
||||
async instanciate(metadataFilters?: any) {
|
||||
return this.adaptInstance(await TypeORMVectorStore.fromDataSource(this.getEmbeddings(), await this.getArgs()), metadataFilters)
|
||||
}
|
||||
|
||||
async fromDocuments(documents: Document[]) {
|
||||
return this.adaptInstance(await TypeORMVectorStore.fromDocuments(documents, this.getEmbeddings(), await this.getArgs()))
|
||||
}
|
||||
|
||||
sanitizeDocuments(documents: Document[]) {
|
||||
// Remove NULL characters which triggers error on PG
|
||||
for (var i in documents) {
|
||||
documents[i].pageContent = documents[i].pageContent.replace(/\0/g, '')
|
||||
}
|
||||
|
||||
return documents
|
||||
}
|
||||
|
||||
protected async adaptInstance(instance: TypeORMVectorStore, metadataFilters?: any): Promise<VectorStore> {
|
||||
const tableName = this.getTableName()
|
||||
|
||||
// Rewrite the method to use pg pool connection instead of the default connection
|
||||
/* Otherwise a connection error is displayed when the chain tries to execute the function
|
||||
[chain/start] [1:chain:ConversationalRetrievalQAChain] Entering Chain run with input: { "question": "what the document is about", "chat_history": [] }
|
||||
[retriever/start] [1:chain:ConversationalRetrievalQAChain > 2:retriever:VectorStoreRetriever] Entering Retriever run with input: { "query": "what the document is about" }
|
||||
[ERROR]: uncaughtException: Illegal invocation TypeError: Illegal invocation at Socket.ref (node:net:1524:18) at Connection.ref (.../node_modules/pg/lib/connection.js:183:17) at Client.ref (.../node_modules/pg/lib/client.js:591:21) at BoundPool._pulseQueue (/node_modules/pg-pool/index.js:148:28) at .../node_modules/pg-pool/index.js:184:37 at process.processTicksAndRejections (node:internal/process/task_queues:77:11)
|
||||
*/
|
||||
instance.similaritySearchVectorWithScore = async (query: number[], k: number, filter?: any) => {
|
||||
return await TypeORMDriver.similaritySearchVectorWithScore(
|
||||
query,
|
||||
k,
|
||||
tableName,
|
||||
await this.getPostgresConnectionOptions(),
|
||||
filter ?? metadataFilters,
|
||||
this.computedOperatorString
|
||||
)
|
||||
}
|
||||
|
||||
instance.delete = async (params: { ids: string[] }): Promise<void> => {
|
||||
const { ids } = params
|
||||
|
||||
if (ids?.length) {
|
||||
try {
|
||||
instance.appDataSource.getRepository(instance.documentEntity).delete(ids)
|
||||
} catch (e) {
|
||||
console.error('Failed to delete')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const baseAddVectorsFn = instance.addVectors.bind(instance)
|
||||
|
||||
instance.addVectors = async (vectors, documents) => {
|
||||
return baseAddVectorsFn(vectors, this.sanitizeDocuments(documents))
|
||||
}
|
||||
|
||||
return instance
|
||||
}
|
||||
|
||||
get computedOperatorString() {
|
||||
const { distanceStrategy = 'cosine' } = this.nodeData.inputs || {}
|
||||
|
||||
switch (distanceStrategy) {
|
||||
case 'cosine':
|
||||
return '<=>'
|
||||
case 'innerProduct':
|
||||
return '<#>'
|
||||
case 'euclidean':
|
||||
return '<->'
|
||||
default:
|
||||
throw new Error(`Unknown distance strategy: ${distanceStrategy}`)
|
||||
}
|
||||
}
|
||||
|
||||
static similaritySearchVectorWithScore = async (
|
||||
query: number[],
|
||||
k: number,
|
||||
tableName: string,
|
||||
postgresConnectionOptions: ICommonObject,
|
||||
filter?: any,
|
||||
distanceOperator: string = '<=>'
|
||||
) => {
|
||||
const embeddingString = `[${query.join(',')}]`
|
||||
let chatflowOr = ''
|
||||
const { [FLOWISE_CHATID]: chatId, ...restFilters } = filter || {}
|
||||
|
||||
const _filter = JSON.stringify(restFilters || {})
|
||||
const parameters: any[] = [embeddingString, _filter, k]
|
||||
|
||||
// Match chatflow uploaded file and keep filtering on other files:
|
||||
// https://github.com/FlowiseAI/Flowise/pull/3367#discussion_r1804229295
|
||||
if (chatId) {
|
||||
parameters.push({ [FLOWISE_CHATID]: chatId })
|
||||
chatflowOr = `OR metadata @> $${parameters.length}`
|
||||
}
|
||||
|
||||
const queryString = `
|
||||
SELECT *, embedding ${distanceOperator} $1 as "_distance"
|
||||
FROM ${tableName}
|
||||
WHERE ((metadata @> $2) AND NOT (metadata ? '${FLOWISE_CHATID}')) ${chatflowOr}
|
||||
ORDER BY "_distance" ASC
|
||||
LIMIT $3;`
|
||||
|
||||
const pool = new Pool(postgresConnectionOptions)
|
||||
|
||||
const conn = await pool.connect()
|
||||
|
||||
const documents = await conn.query(queryString, parameters)
|
||||
|
||||
conn.release()
|
||||
|
||||
const results = [] as [TypeORMVectorStoreDocument, number][]
|
||||
for (const doc of documents.rows) {
|
||||
if (doc._distance != null && doc.pageContent != null) {
|
||||
const document = new Document(doc) as TypeORMVectorStoreDocument
|
||||
document.id = doc.id
|
||||
results.push([document, doc._distance])
|
||||
}
|
||||
}
|
||||
|
||||
return results
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user