当前位置：首页 > news >正文

Python中使用Oracle向量数据库实现文本检索系统

news 2025/7/2 4:28:03

Python中使用Oracle向量数据库实现文本检索系统

- 代码分析

在本文中,我们将深入分析一个使用Oracle向量数据库实现文本检索系统的Python代码,并基于相同的技术生成一个新的示例。这个系统允许我们存储文档及其嵌入向量,并执行相似性搜索。

代码分析

让我们逐步分析原始代码的主要组件和功能:

导入必要的库:
- 使用oracledb连接Oracle数据库
- 使用numpy处理向量
- 使用pydantic进行配置验证
- 使用flask和redis进行Web应用程序集成
定义OracleVectorConfig类:
- 使用Pydantic模型验证Oracle连接配置
创建OracleVector类:
- 实现向量数据库的核心功能
- 使用contextmanager管理数据库连接
- 实现CRUD操作和向量搜索
实现OracleVectorFactory类:
- 用于初始化向量数据库实例

现在,让我们基于相同的技术创建一个新的示例代码:

import array
import json
import uuid
from contextlib import contextmanager
from typing import List, Dict, Anyimport numpy as np
import oracledb
from pydantic import BaseModel, validatorclass OracleConfig(BaseModel):host: strport: intuser: strpassword: strdatabase: str@validator('host', 'user', 'password', 'database')def check_not_empty(cls, v):if not v:raise ValueError("Field cannot be empty")return vclass TextEmbeddingStore:def __init__(self, config: OracleConfig):self.pool = self._create_connection_pool(config)self.table_name = "text_embeddings"self._create_table()def _create_connection_pool(self, config: OracleConfig):return oracledb.create_pool(user=config.user,password=config.password,dsn=f"{config.host}:{config.port}/{config.database}",min=1,max=5,increment=1)@contextmanagerdef _get_cursor(self):conn = self.pool.acquire()conn.inputtypehandler = self._input_type_handlerconn.outputtypehandler = self._output_type_handlercur = conn.cursor()try:yield curfinally:cur.close()conn.commit()conn.close()def _input_type_handler(self, cursor, value, arraysize):if isinstance(value, np.ndarray):return cursor.var(oracledb.DB_TYPE_VECTOR,arraysize=arraysize,inconverter=self._numpy_to_array)def _output_type_handler(self, cursor, metadata):if metadata.type_code is oracledb.DB_TYPE_VECTOR:return cursor.var(metadata.type_code,arraysize=cursor.arraysize,outconverter=self._array_to_numpy)def _numpy_to_array(self, value):return array.array('f', value)def _array_to_numpy(self, value):return np.array(value, dtype=np.float32)def _create_table(self):with self._get_cursor() as cur:cur.execute(f"""CREATE TABLE IF NOT EXISTS {self.table_name} (id VARCHAR2(100) PRIMARY KEY,text CLOB NOT NULL,metadata JSON,embedding VECTOR NOT NULL)""")def add_texts(self, texts: List[str], embeddings: List[List[float]], metadata: List[Dict] = None):if metadata is None:metadata = [{} for _ in texts]values = [(str(uuid.uuid4()), text, json.dumps(meta), np.array(emb, dtype=np.float32))for text, emb, meta in zip(texts, embeddings, metadata)]with self._get_cursor() as cur:cur.executemany(f"INSERT INTO {self.table_name} (id, text, metadata, embedding) VALUES (:1, :2, :3, :4)",values)def search_similar(self, query_vector: List[float], top_k: int = 5) -> List[Dict[str, Any]]:query_vector = np.array(query_vector, dtype=np.float32)with self._get_cursor() as cur:cur.execute(f"""SELECT id, text, metadata, vector_distance(embedding, :1) AS distanceFROM {self.table_name}ORDER BY distanceFETCH FIRST :2 ROWS ONLY""",[query_vector, top_k])results = []for id, text, metadata, distance in cur:results.append({"id": id,"text": text,"metadata": json.loads(metadata),"distance": distance,"similarity": 1 - distance})return results# 使用示例
if __name__ == "__main__":config = OracleConfig(host="localhost",port=1521,user="your_username",password="your_password",database="your_database")store = TextEmbeddingStore(config)# 添加文本和嵌入texts = ["Hello world", "Python programming", "Vector database"]embeddings = [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6], [0.7, 0.8, 0.9]]store.add_texts(texts, embeddings)# 搜索相似文本query_vector = [0.2, 0.3, 0.4]results = store.search_similar(query_vector, top_k=2)for result in results:print(f"Text: {result['text']}")print(f"Similarity: {result['similarity']:.4f}")print("---")