Commit d23c7269 by yanzhicong Committed by GitHub

feat: add openGauss DataVec as vector database backend (#6666)

* feat: add openGauss DataVec as vector database backend

Add openGauss with DataVec extension as a new vector storage option alongside PGVector and Milvus. Includes vector DB controller, Docker Compose deployment configs (CN/Global), deploy generation scripts, and test templates.

* test: add opengauss vectorDB integration entry

* test: adjust vector env template for opengauss run

* fix: ts

---------

Co-authored-by: archer <545436317@qq.com>
parent 546b2a89
...@@ -18,7 +18,8 @@ const VectorEnum = { ...@@ -18,7 +18,8 @@ const VectorEnum = {
milvus: 'milvus', milvus: 'milvus',
zilliz: 'zilliz', zilliz: 'zilliz',
ob: 'ob', ob: 'ob',
seekdb: 'seekdb' seekdb: 'seekdb',
opengauss: 'opengauss'
}; };
// make sure the cwd // make sure the cwd
...@@ -110,6 +111,12 @@ init_sql: ...@@ -110,6 +111,12 @@ init_sql:
`, `,
extra: `` extra: ``
}, },
opengauss: {
db: '',
config: `\
OPENGAUSS_URL: postgresql://gaussdb:FastGPT@123@opengauss:5432/fastgpt`,
extra: ''
},
}; };
/** /**
...@@ -155,6 +162,9 @@ const replace = (source, region, vec) => { ...@@ -155,6 +162,9 @@ const replace = (source, region, vec) => {
const seekdb = fs.readFileSync(path.join(process.cwd(), 'templates', 'vector', 'seekdb.txt')); const seekdb = fs.readFileSync(path.join(process.cwd(), 'templates', 'vector', 'seekdb.txt'));
vector.seekdb.db = String(seekdb); vector.seekdb.db = String(seekdb);
const opengauss = fs.readFileSync(path.join(process.cwd(), 'templates', 'vector', 'opengauss.txt'));
vector.opengauss.db = String(opengauss);
} }
const generateDevFile = async () => { const generateDevFile = async () => {
...@@ -226,6 +236,14 @@ const generateProdFile = async () => { ...@@ -226,6 +236,14 @@ const generateProdFile = async () => {
fs.promises.writeFile( fs.promises.writeFile(
path.join(process.cwd(), 'docker', 'global', 'docker-compose.seekdb.yml'), path.join(process.cwd(), 'docker', 'global', 'docker-compose.seekdb.yml'),
replace(template, 'global', VectorEnum.seekdb) replace(template, 'global', VectorEnum.seekdb)
),
fs.promises.writeFile(
path.join(process.cwd(), 'docker', 'cn', 'docker-compose.opengauss.yml'),
replace(template, 'cn', VectorEnum.opengauss)
),
fs.promises.writeFile(
path.join(process.cwd(), 'docker', 'global', 'docker-compose.opengauss.yml'),
replace(template, 'global', VectorEnum.opengauss)
) )
]); ]);
......
vectorDB:
image: opengauss/opengauss:7.0.0-RC1
container_name: opengauss
restart: always
privileged: true
networks:
- fastgpt
environment:
# 这里的配置只有首次运行生效。修改后,重启镜像是不会生效的。需要把持久化数据删除再重启,才有效果
- GS_USERNAME=gaussdb # 默认会创建 gaussdb 用户
- GS_PASSWORD=FastGPT@123 # 密码必须包含大写、小写、数字和特殊字符,且长度不少于8位
- GS_DB=fastgpt # 默认会创建 postgres 数据库,这里以 fastgpt 为例
volumes:
- ./opengauss/data:/var/lib/opengauss
healthcheck:
test: ['CMD-SHELL', 'su - omm -c "gsql -d postgres -p 5432 -c \"SELECT 1\""']
interval: 10s
timeout: 5s
retries: 10
start_period: 30s
\ No newline at end of file
...@@ -2,6 +2,7 @@ export const DatasetVectorDbName = 'fastgpt'; ...@@ -2,6 +2,7 @@ export const DatasetVectorDbName = 'fastgpt';
export const DatasetVectorTableName = 'modeldata'; export const DatasetVectorTableName = 'modeldata';
export const PG_ADDRESS = process.env.PG_URL; export const PG_ADDRESS = process.env.PG_URL;
export const OPENGAUSS_ADDRESS = process.env.OPENGAUSS_URL;
export const OCEANBASE_ADDRESS = process.env.OCEANBASE_URL; export const OCEANBASE_ADDRESS = process.env.OCEANBASE_URL;
export const SEEKDB_ADDRESS = process.env.SEEKDB_URL; export const SEEKDB_ADDRESS = process.env.SEEKDB_URL;
export const MILVUS_ADDRESS = process.env.MILVUS_ADDRESS; export const MILVUS_ADDRESS = process.env.MILVUS_ADDRESS;
......
...@@ -2,10 +2,17 @@ ...@@ -2,10 +2,17 @@
import { PgVectorCtrl } from './pg'; import { PgVectorCtrl } from './pg';
import { ObVectorCtrl } from './oceanbase'; import { ObVectorCtrl } from './oceanbase';
import { SeekVectorCtrl } from './seekdb'; import { SeekVectorCtrl } from './seekdb';
import { OpenGaussVectorCtrl } from './opengauss';
import { getVectorsByText } from '../../core/ai/embedding'; import { getVectorsByText } from '../../core/ai/embedding';
import type { VectorControllerType, InsertVectorControllerPropsType } from './type'; import type { VectorControllerType, InsertVectorControllerPropsType } from './type';
import { type EmbeddingModelItemType } from '@fastgpt/global/core/ai/model.schema'; import { type EmbeddingModelItemType } from '@fastgpt/global/core/ai/model.schema';
import { MILVUS_ADDRESS, PG_ADDRESS, OCEANBASE_ADDRESS, SEEKDB_ADDRESS } from './constants'; import {
MILVUS_ADDRESS,
PG_ADDRESS,
OPENGAUSS_ADDRESS,
OCEANBASE_ADDRESS,
SEEKDB_ADDRESS
} from './constants';
import { MilvusCtrl } from './milvus'; import { MilvusCtrl } from './milvus';
import { import {
setRedisCache, setRedisCache,
...@@ -23,6 +30,7 @@ const getVectorObj = (): VectorControllerType => { ...@@ -23,6 +30,7 @@ const getVectorObj = (): VectorControllerType => {
if (OCEANBASE_ADDRESS) return new ObVectorCtrl({ type: 'oceanbase' }); if (OCEANBASE_ADDRESS) return new ObVectorCtrl({ type: 'oceanbase' });
if (PG_ADDRESS) return new PgVectorCtrl(); if (PG_ADDRESS) return new PgVectorCtrl();
if (MILVUS_ADDRESS) return new MilvusCtrl(); if (MILVUS_ADDRESS) return new MilvusCtrl();
if (OPENGAUSS_ADDRESS) return new OpenGaussVectorCtrl();
return new PgVectorCtrl(); return new PgVectorCtrl();
}; };
......
import { delay } from '@fastgpt/global/common/system/utils';
import { getLogger, LogCategories } from '../../logger';
import { Pool } from 'pg';
import type { QueryResultRow } from 'pg';
import { OPENGAUSS_ADDRESS } from '../constants';
const logger = getLogger(LogCategories.INFRA.VECTOR);
export const connectOg = async (): Promise<Pool> => {
if (global.pgClient) {
return global.pgClient;
}
const pool = new Pool({
connectionString: OPENGAUSS_ADDRESS,
max: Number(process.env.DB_MAX_LINK || 30),
min: 15,
keepAlive: true,
idleTimeoutMillis: 1800000,
connectionTimeoutMillis: 30000,
query_timeout: 60000,
statement_timeout: 90000,
idle_in_transaction_session_timeout: 60000,
allowExitOnIdle: false,
application_name: 'fastgpt-vector-db'
});
global.pgClient = pool;
global.pgClient.on('error', async (err) => {
logger.error('openGauss pool error', { error: err });
});
global.pgClient.on('connect', async () => {
logger.info('openGauss pool connected');
});
global.pgClient.on('remove', async () => {
logger.warn('openGauss connection removed from pool');
});
try {
await global.pgClient.connect();
return global.pgClient;
} catch (error) {
logger.error('openGauss connection failed', { error });
global.pgClient?.removeAllListeners();
global.pgClient?.end();
global.pgClient = null;
await delay(1000);
logger.warn('openGauss reconnecting after failure');
return connectOg();
}
};
type WhereProps = (string | [string, string | number])[];
type GetProps = {
fields?: string[];
where?: WhereProps;
order?: { field: string; mode: 'DESC' | 'ASC' | string }[];
limit?: number;
offset?: number;
};
type DeleteProps = {
where: WhereProps;
};
type ValuesProps = { key: string; value?: string | number }[];
type UpdateProps = {
values: ValuesProps;
where: WhereProps;
};
type InsertProps = {
values: ValuesProps[];
};
class OgClass {
private getWhereStr(where?: WhereProps) {
return where
? `WHERE ${where
.map((item) => {
if (typeof item === 'string') {
return item;
}
const val = typeof item[1] === 'number' ? item[1] : `'${String(item[1])}'`;
return `${item[0]}=${val}`;
})
.join(' ')}`
: '';
}
private getUpdateValStr(values: ValuesProps) {
return values
.map((item) => {
const val =
typeof item.value === 'number'
? item.value
: `'${String(item.value).replace(/\'/g, '"')}'`;
return `${item.key}=${val}`;
})
.join(',');
}
private getInsertValStr(values: ValuesProps[]) {
return values
.map(
(items) =>
`(${items
.map((item) =>
typeof item.value === 'number'
? item.value
: `'${String(item.value).replace(/\'/g, '"')}'`
)
.join(',')})`
)
.join(',');
}
async query<T extends QueryResultRow = any>(sql: string) {
const og = await connectOg();
const start = Date.now();
return og.query<T>(sql).then((res) => {
const time = Date.now() - start;
if (time > 1000) {
const safeSql = sql.replace(/'\[[^\]]*?\]'/g, "'[x]'");
logger.warn('openGauss slow query detected', {
level: 'slow-2',
durationMs: time,
sql: safeSql
});
} else if (time > 300) {
const safeSql = sql.replace(/'\[[^\]]*?\]'/g, "'[x]'");
logger.warn('openGauss slow query detected', {
level: 'slow-1',
durationMs: time,
sql: safeSql
});
}
return res;
});
}
async select<T extends QueryResultRow = any>(table: string, props: GetProps) {
const sql = `SELECT ${
!props.fields || props.fields?.length === 0 ? '*' : props.fields?.join(',')
}
FROM ${table}
${this.getWhereStr(props.where)}
${
props.order
? `ORDER BY ${props.order.map((item) => `${item.field} ${item.mode}`).join(',')}`
: ''
}
LIMIT ${props.limit || 10} OFFSET ${props.offset || 0}
`;
return this.query<T>(sql);
}
async count(table: string, props: GetProps) {
const sql = `SELECT COUNT(${props?.fields?.[0] || '*'})
FROM ${table}
${this.getWhereStr(props.where)}
`;
return this.query(sql).then((res) => Number(res.rows[0]?.count || 0));
}
async delete(table: string, props: DeleteProps) {
const sql = `DELETE FROM ${table} ${this.getWhereStr(props.where)}`;
return this.query(sql);
}
async update(table: string, props: UpdateProps) {
if (props.values.length === 0) {
return {
rowCount: 0
};
}
const sql = `UPDATE ${table} SET ${this.getUpdateValStr(props.values)} ${this.getWhereStr(
props.where
)}`;
return this.query(sql);
}
async insert(table: string, props: InsertProps) {
if (props.values.length === 0) {
return {
rowCount: 0,
rows: []
};
}
const fields = props.values[0].map((item) => item.key).join(',');
const sql = `INSERT INTO ${table} (${fields}) VALUES ${this.getInsertValStr(
props.values
)} RETURNING id`;
return this.query<{ id: string }>(sql);
}
}
export const OgClient = new OgClass();
/* openGauss DataVec vector crud */
import { DatasetVectorTableName } from '../constants';
import { OgClient, connectOg } from './controller';
import type { VectorControllerType } from '../type';
import dayjs from 'dayjs';
import { getLogger, LogCategories } from '../../logger';
const logger = getLogger(LogCategories.INFRA.VECTOR);
export class OpenGaussVectorCtrl implements VectorControllerType {
constructor() {}
init = async () => {
try {
await connectOg();
await OgClient.query(`
CREATE TABLE IF NOT EXISTS ${DatasetVectorTableName} (
id BIGSERIAL PRIMARY KEY,
vector VECTOR(1536) NOT NULL,
team_id VARCHAR(50) NOT NULL,
dataset_id VARCHAR(50) NOT NULL,
collection_id VARCHAR(50) NOT NULL,
createtime TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
`);
await OgClient.query(
`CREATE INDEX CONCURRENTLY IF NOT EXISTS vector_index ON ${DatasetVectorTableName} USING hnsw (vector vector_ip_ops) WITH (m = 32, ef_construction = 128);`
);
await OgClient.query(
`CREATE INDEX CONCURRENTLY IF NOT EXISTS team_dataset_collection_index ON ${DatasetVectorTableName} USING btree(team_id, dataset_id, collection_id);`
);
await OgClient.query(
`CREATE INDEX CONCURRENTLY IF NOT EXISTS create_time_index ON ${DatasetVectorTableName} USING btree(createtime);`
);
logger.info('openGauss DataVec vector initialization completed');
} catch (error) {
logger.error('openGauss DataVec vector initialization failed', { error });
}
};
insert: VectorControllerType['insert'] = async (props) => {
const { teamId, datasetId, collectionId, vectors } = props;
const values = vectors.map((vector) => [
{ key: 'vector', value: `[${vector}]` },
{ key: 'team_id', value: String(teamId) },
{ key: 'dataset_id', value: String(datasetId) },
{ key: 'collection_id', value: String(collectionId) }
]);
const { rowCount, rows } = await OgClient.insert(DatasetVectorTableName, {
values
});
if (rowCount === 0) {
return Promise.reject('insertDatasetData: no insert');
}
return {
insertIds: rows.map((row) => row.id)
};
};
delete: VectorControllerType['delete'] = async (props) => {
const { teamId } = props;
const teamIdWhere = `team_id='${String(teamId)}' AND`;
const where = await (() => {
if ('id' in props && props.id) return `${teamIdWhere} id=${props.id}`;
if ('datasetIds' in props && props.datasetIds) {
const datasetIdWhere = `dataset_id IN (${props.datasetIds
.map((id) => `'${String(id)}'`)
.join(',')})`;
if ('collectionIds' in props && props.collectionIds) {
return `${teamIdWhere} ${datasetIdWhere} AND collection_id IN (${props.collectionIds
.map((id) => `'${String(id)}'`)
.join(',')})`;
}
return `${teamIdWhere} ${datasetIdWhere}`;
}
if ('idList' in props && Array.isArray(props.idList)) {
if (props.idList.length === 0) return;
return `${teamIdWhere} id IN (${props.idList.map((id) => String(id)).join(',')})`;
}
return Promise.reject('deleteDatasetData: no where');
})();
if (!where) return;
await OgClient.delete(DatasetVectorTableName, {
where: [where]
});
};
embRecall: VectorControllerType['embRecall'] = async (props) => {
const { teamId, datasetIds, vector, limit, forbidCollectionIdList, filterCollectionIdList } =
props;
// Get forbid collection
const formatForbidCollectionIdList = (() => {
if (!filterCollectionIdList) return forbidCollectionIdList;
const list = forbidCollectionIdList
.map((id) => String(id))
.filter((id) => !filterCollectionIdList.includes(id));
return list;
})();
const forbidCollectionSql =
formatForbidCollectionIdList.length > 0
? `AND collection_id NOT IN (${formatForbidCollectionIdList.map((id) => `'${id}'`).join(',')})`
: '';
// Filter by collectionId
const formatFilterCollectionId = (() => {
if (!filterCollectionIdList) return;
return filterCollectionIdList
.map((id) => String(id))
.filter((id) => !forbidCollectionIdList.includes(id));
})();
const filterCollectionIdSql = formatFilterCollectionId
? `AND collection_id IN (${formatFilterCollectionId.map((id) => `'${id}'`).join(',')})`
: '';
// Empty data
if (formatFilterCollectionId && formatFilterCollectionId.length === 0) {
return { results: [] };
}
const results: any = await OgClient.query(
`BEGIN;
SET LOCAL hnsw.ef_search = ${global.systemEnv?.hnswEfSearch || 100};
SELECT id, collection_id, vector <#> '[${vector}]' AS score
FROM ${DatasetVectorTableName}
WHERE dataset_id IN (${datasetIds.map((id) => `'${String(id)}'`).join(',')})
${filterCollectionIdSql}
${forbidCollectionSql}
ORDER BY score LIMIT ${limit};
COMMIT;`
);
const rows = results?.[results.length - 2]?.rows as {
id: string;
collection_id: string;
score: number;
}[];
if (!Array.isArray(rows)) {
return {
results: []
};
}
return {
results: rows.map((item) => ({
id: String(item.id),
collectionId: item.collection_id,
score: item.score * -1
}))
};
};
getVectorDataByTime: VectorControllerType['getVectorDataByTime'] = async (start, end) => {
const { rows } = await OgClient.query<{
id: string;
team_id: string;
dataset_id: string;
}>(`SELECT id, team_id, dataset_id
FROM ${DatasetVectorTableName}
WHERE createtime BETWEEN '${dayjs(start).format('YYYY-MM-DD HH:mm:ss')}' AND '${dayjs(
end
).format('YYYY-MM-DD HH:mm:ss')}';
`);
return rows.map((item) => ({
id: String(item.id),
teamId: item.team_id,
datasetId: item.dataset_id
}));
};
getVectorCount: VectorControllerType['getVectorCount'] = async (props) => {
const { teamId, datasetId, collectionId } = props;
// Build where conditions dynamically
const whereConditions: any[] = [];
if (teamId) {
whereConditions.push(['team_id', String(teamId)]);
}
if (datasetId) {
if (whereConditions.length > 0) whereConditions.push('and');
whereConditions.push(['dataset_id', String(datasetId)]);
}
if (collectionId) {
if (whereConditions.length > 0) whereConditions.push('and');
whereConditions.push(['collection_id', String(collectionId)]);
}
// If no conditions provided, count all
const total = await OgClient.count(DatasetVectorTableName, {
where: whereConditions.length > 0 ? whereConditions : undefined
});
return total;
};
}
...@@ -21,6 +21,7 @@ declare global { ...@@ -21,6 +21,7 @@ declare global {
// Vector // Vector
VECTOR_VQ_LEVEL: string; VECTOR_VQ_LEVEL: string;
PG_URL: string; PG_URL: string;
OPENGAUSS_URL: string;
OCEANBASE_URL: string; OCEANBASE_URL: string;
SEEKDB_URL: string; SEEKDB_URL: string;
MILVUS_ADDRESS: string; MILVUS_ADDRESS: string;
......
...@@ -108,7 +108,7 @@ MONGODB_URI="mongodb://myusername:mypassword@localhost:27017/fastgpt?authSource= ...@@ -108,7 +108,7 @@ MONGODB_URI="mongodb://myusername:mypassword@localhost:27017/fastgpt?authSource=
# 日志库 # 日志库
MONGODB_LOG_URI="mongodb://myusername:mypassword@localhost:27017/fastgpt?authSource=admin&directConnection=true" MONGODB_LOG_URI="mongodb://myusername:mypassword@localhost:27017/fastgpt?authSource=admin&directConnection=true"
# 向量库优先级: pg > oceanbase > milvus # 向量库优先级: pg > oceanbase > milvus > opengauss
# 向量量化等级: PG 支持 32/16,OceanBase 支持 32/8/1 # 向量量化等级: PG 支持 32/16,OceanBase 支持 32/8/1
VECTOR_VQ_LEVEL=32 VECTOR_VQ_LEVEL=32
...@@ -119,6 +119,8 @@ PG_URL=postgresql://username:password@localhost:5432/postgres ...@@ -119,6 +119,8 @@ PG_URL=postgresql://username:password@localhost:5432/postgres
# Milvus 向量库连接参数 # Milvus 向量库连接参数
# MILVUS_ADDRESS= # MILVUS_ADDRESS=
# MILVUS_TOKEN= # MILVUS_TOKEN=
# openGauss 向量库连接参数
# OPENGAUSS_URL=postgresql://gaussdb:FastGPT@123@localhost:5432/fastgpt
# ==================== 域名与前端 ==================== # ==================== 域名与前端 ====================
# 页面地址,用于自动补全相对路径资源的 domain(注意结尾不要带 /) # 页面地址,用于自动补全相对路径资源的 domain(注意结尾不要带 /)
......
...@@ -8,3 +8,5 @@ SEEKDB_URL=mysql://root:seekdbpassword@127.0.0.1:6003/mysql ...@@ -8,3 +8,5 @@ SEEKDB_URL=mysql://root:seekdbpassword@127.0.0.1:6003/mysql
# Milvus vector database connection # Milvus vector database connection
MILVUS_ADDRESS=http://localhost:6002 MILVUS_ADDRESS=http://localhost:6002
MILVUS_TOKEN= MILVUS_TOKEN=
# openGauss vector database connection
OPENGAUSS_URL=postgresql://gaussdb:FastGPT@123@localhost:5432/fastgpt
\ No newline at end of file
...@@ -18,6 +18,7 @@ cp test/.env.test.template test/.env.test.local ...@@ -18,6 +18,7 @@ cp test/.env.test.template test/.env.test.local
| `PG_URL` | PostgreSQL + pgvector 连接串 | PgVectorCtrl | | `PG_URL` | PostgreSQL + pgvector 连接串 | PgVectorCtrl |
| `OCEANBASE_URL` | Oceanbase 连接串(后续) | ObVectorCtrl | | `OCEANBASE_URL` | Oceanbase 连接串(后续) | ObVectorCtrl |
| `MILVUS_ADDRESS` | Milvus 地址(后续) | MilvusCtrl | | `MILVUS_ADDRESS` | Milvus 地址(后续) | MilvusCtrl |
| `OPENGAUSS_URL` | openGauss DataVec 连接串 | OpenGaussVectorCtrl |
未设置对应环境变量时,该驱动的集成测试会**整体跳过**,不会报错。 未设置对应环境变量时,该驱动的集成测试会**整体跳过**,不会报错。
......
import { describe, vi } from 'vitest';
import { createVectorDBTestSuite } from '../testSuites';
// Unmock vector controllers for integration tests
vi.unmock('@fastgpt/service/common/vectorDB/opengauss');
vi.unmock('@fastgpt/service/common/vectorDB/constants');
import { OpenGaussVectorCtrl } from '@fastgpt/service/common/vectorDB/opengauss';
const isEnabled = Boolean(process.env.OPENGAUSS_URL);
describe.skipIf(!isEnabled)('OpenGauss Vector Integration', () => {
const vectorCtrl = new OpenGaussVectorCtrl();
createVectorDBTestSuite(vectorCtrl);
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment