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

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

代码分析

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

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._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("---")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

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

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

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

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