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, Any
import numpy as np
import oracledb
from pydantic import BaseModel, validator
class OracleConfig(BaseModel):
host: str
port: int
user: str
password: str
database: str
@validator('host', 'user', 'password', 'database')
def check_not_empty(cls, v):
if not v:
raise ValueError("Field cannot be empty")
return v
class 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
)
@contextmanager
def _get_cursor(self):
conn = self.pool.acquire()
conn.inputtypehandler = self._input_type_handler
conn.outputtypehandler = self._output_type_handler
cur = conn.cursor()
try:
yield cur
finally:
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 distance
FROM {self.table_name}
ORDER BY distance
FETCH 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("---")
这个新的示例代码实现了一个简化版的文本嵌入存储系统,使用Oracle向量数据库。它包含以下主要功能:
- 使用Pydantic进行配置验证
- 创建和管理Oracle连接池
- 使用上下文管理器处理数据库连接
- 处理numpy数组和Oracle向量类型之间的转换
- 实现添加文本和嵌入的方法
- 实现基于向量相似度的搜索方法
这个示例展示了如何使用Oracle向量数据库来存储和检索文本嵌入,可以作为构建更复杂的文本检索或推荐系统的基础。
在实际应用中,你可能需要添加错误处理、日志记录、性能优化等功能。