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

 更新时间:2024年07月04日 10:39:21   作者:engchina  
在本文中,我们将深入分析一个使用Oracle向量数据库实现文本检索系统的Python代码,并基于相同的技术生成一个新的示例,这个系统允许我们存储文档及其嵌入向量,并执行相似性搜索,感兴趣的朋友可以参考下
(福利推荐:【腾讯云】服务器最新限时优惠活动,云服务器1核2G仅99元/年、2核4G仅768元/3年,立即抢购>>>:9i0i.cn/qcloud

(福利推荐:你还在原价购买阿里云服务器?现在阿里云0.8折限时抢购活动来啦!4核8G企业云服务器仅2998元/3年,立即抢购>>>:9i0i.cn/aliyun

代码分析

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

  1. 导入必要的库:

    • 使用oracledb连接Oracle数据库
    • 使用numpy处理向量
    • 使用pydantic进行配置验证
    • 使用flaskredis进行Web应用程序集成
  2. 定义OracleVectorConfig类:

    • 使用Pydantic模型验证Oracle连接配置
  3. 创建OracleVector类:

    • 实现向量数据库的核心功能
    • 使用contextmanager管理数据库连接
    • 实现CRUD操作和向量搜索
  4. 实现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向量数据库。它包含以下主要功能:

  1. 使用Pydantic进行配置验证
  2. 创建和管理Oracle连接池
  3. 使用上下文管理器处理数据库连接
  4. 处理numpy数组和Oracle向量类型之间的转换
  5. 实现添加文本和嵌入的方法
  6. 实现基于向量相似度的搜索方法

这个示例展示了如何使用Oracle向量数据库来存储和检索文本嵌入,可以作为构建更复杂的文本检索或推荐系统的基础。

在实际应用中,你可能需要添加错误处理、日志记录、性能优化等功能。

到此这篇关于Python使用Oracle向量数据库实现文本检索系统的文章就介绍到这了,更多相关Python Oracle文本检索系统内容请搜索程序员之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持程序员之家!

相关文章

最新评论


http://www.vxiaotou.com