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

代码分析

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

1. 导入必要的库:

   • 使用oracledb连接Oracle数据库
   • 使用numpy处理向量
   • 使用pydantic进行配置验证
   • 使用flask和Redis进行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._c编程reate_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):
     www.devze.com   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, 0php.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}")
        javascriptprint("---")

这个新的示例代码实现了一个简化版的文本嵌入存储系统,使用Oracle向量数据库。它包含以下主要功能:

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

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

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

到此这篇关于Python使用Oracle向量数据库实现文本检索系统的文章就介绍到这了,更多相关Python Oracle文本检索系统内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！