做资源网站盈利点,asp.net mvc 网站开发,建设银行网银网站无法访问,稳定的网站服务器租用Python中使用Oracle向量数据库实现文本检索系统 代码分析 在本文中,我们将深入分析一个使用Oracle向量数据库实现文本检索系统的Python代码,并基于相同的技术生成一个新的示例。这个系统允许我们存储文档及其嵌入向量,并执行相似性搜索。
代码分析
让我们逐步分析原始代码的主… 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: strvalidator(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_embeddingsself._create_table()def _create_connection_pool(self, config: OracleConfig):return oracledb.create_pool(userconfig.user,passwordconfig.password,dsnf{config.host}:{config.port}/{config.database},min1,max5,increment1)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,arraysizearraysize,inconverterself._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,arraysizecursor.arraysize,outconverterself._array_to_numpy)def _numpy_to_array(self, value):return array.array(f, value)def _array_to_numpy(self, value):return np.array(value, dtypenp.float32)def _create_table(self):with self._get_cursor() as cur:cur.execute(fCREATE 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, dtypenp.float32))for text, emb, meta in zip(texts, embeddings, metadata)]with self._get_cursor() as cur:cur.executemany(fINSERT 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, dtypenp.float32)with self._get_cursor() as cur:cur.execute(fSELECT 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(hostlocalhost,port1521,useryour_username,passwordyour_password,databaseyour_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_k2)for result in results:print(fText: {result[text]})print(fSimilarity: {result[similarity]:.4f})print(---)这个新的示例代码实现了一个简化版的文本嵌入存储系统,使用Oracle向量数据库。它包含以下主要功能:
使用Pydantic进行配置验证创建和管理Oracle连接池使用上下文管理器处理数据库连接处理numpy数组和Oracle向量类型之间的转换实现添加文本和嵌入的方法实现基于向量相似度的搜索方法
这个示例展示了如何使用Oracle向量数据库来存储和检索文本嵌入,可以作为构建更复杂的文本检索或推荐系统的基础。
在实际应用中,你可能需要添加错误处理、日志记录、性能优化等功能。
