达梦向量数据开发入门手册

dmSQLAlchemy 向量数据扩展旨在为 Python 开发者提供无缝的向量化存储与检索体验。它支持将文本或图像通过预训练模型转换为向量，并存储在达梦数据库中。

1.1. 开发入门

当你需要自定义表名或增加业务列（如分类 category、作者 author）时，应使用此模式。

1.1.1. 环境准备

dmpython、dmSQLAIchemy需要使用对应数据库软件下方**/dmdbms/drivers/python目录**下的文件进行安装，然后准备适配的python版本；

# 达梦数据库驱动（需从本地安装）

pip install "./python/dmPython/"
pip install "./python/dmSQLAlchemy/dmSQLAlchemy2.0/" # 选择合适版本

numpy: 用来处理python中的list类型与数据库内向量类型的转换；
Sentence-Transformers: 用于文本向量化生成的模型支持；
torch、torchvision、pillow: 用于支持图像向量化生成的模型支持；

1.1.2. 向量映射

与普通创建列相同，在SQLAlchemy中连接DM数据库创建向量类型指定参数主要为dim，format，以下是详细参数介绍。

驱动/库	安装参考命令
达梦数据库驱动	`pip install "./python/dmPython/"`
dmSQLAlchemy	`pip install "./python/dmSQLAlchemy/dmSQLAlchemy2.0/"`

与普通列相同，向量列的插入数据时使用list类型即可，例如插入向量列数据可以写为[1, 2, 3, 4]

1.1.3. 定义向量表列类型

使用 dmSQLAlchemy 提供的 VECTOR 类型定义列。
示例：

from dmSQLAlchemy import VECTOR
from sqlalchemy import Column, Integer, Identity

class ItemModel(Base):
    __tablename__ = "test_vector"
    id = Column(Integer, Identity(), primary_key=True)
    # 向量参数：dim(维度), format(类型: int8/float32/float64/binary)
    embedding = Column(VECTOR(dim=5, format='float64'))

1.1.4. 快速上手示例

"""
达梦向量数据 - ORM方式向量基础操作演示
"""

# 1. 导入依赖
from dmSQLAlchemy import VECTOR, VectorAdaptor
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Integer, Identity, Column, inspect, create_engine


# 2. 配置数据库连接
Base = declarative_base()
# 连接字符串格式: dm+dmPython://<用户名>:<密码>@<IP地址>:<端口>
engine = create_engine('dm+dmPython://SYSDBA:SYSDBA@127.0.0.1:5236')
# 向量适配器: 用于创建和管理向量索引
adaptor = VectorAdaptor(engine)
# 会话工厂和会话实例
Session = sessionmaker(bind=engine)
session = Session()
# 数据库检查器: 用于获取表结构信息
inspector = inspect(engine)

# 3. 定义 ORM 模型
class ItemModel(Base):
    """向量数据模型"""
    __tablename__ = "test_vector"
    # 主键: 自增整数
    id = Column(Integer, Identity(), primary_key=True)
    # 向量字段: dim=维度, format=数据类型(INT8/FLOAT32/FLOAT64)
    embedding = Column(VECTOR(dim=5, format='float64'))

# 4. 创建表
# 删除旧表（如果存在）
Base.metadata.drop_all(engine)
# 创建新表
Base.metadata.create_all(engine)

# 5. 创建向量索引
# IVF 索引: 适合大规模向量搜索
# metric_name: 距离度量方式 (cosine/l2/dot) 
adaptor.create_vector_ivf_index(
    ItemModel.embedding, 
    metric_name="cosine", 
    index_name="ivf_index"
)

# 6. 插入向量数据
# embedding 接受 Python 列表，自动转换为向量格式
session.add(ItemModel(embedding=[1, 2, 3, 4, 5]))
session.add(ItemModel(embedding=[2, 1, 3, 4, 5]))
session.add(ItemModel(embedding=[5, 2, 1, 4, 3]))

# 7. 重建向量索引
# 插入数据后需要重建索引以保证搜索效率
adaptor.rebuild_vector_ivf_index(
    ItemModel.embedding, 
    metric_name="cosine", 
    index_name="ivf_index"
)

# 8. 向量距离查询
# 计算与查询向量 [5, 2, 1, 4, 3] 的 L1 距离
distance = ItemModel.embedding.l1_distance([5, 2, 1, 4, 3])
# 查询所有记录的 id 和距离值，按距离升序排列
result = session.query(ItemModel.id, distance).all()
for row in result:
    print(row)

# 9. 删除数据
session.query(ItemModel).filter(
    ItemModel.embedding == [5, 2, 1, 4, 3]
).delete()
# 提交事务
session.commit()

# 10. 获取表结构信息
columns = inspector.get_columns('test_vector', schema='SYSDBA')
print(columns)

1.1.5. 索引管理 (VectorAdaptor)

对于大规模向量搜索，必须创建索引以保证性能。

函数名	参数	说明
create_vector_ivf_index	column, skip_existing, metric_name, index_name, percentage_value, num_of_partitions	首次建表后调用，初始化 IVF 索引
rebuild_vector_ivf_index	column, schema_name , index_name, metric_name, target_accuracy, cluster_centers	重要：插入大量数据后必须调用，否则搜不到新数据
create_vector_hnsw_index	column, skip_existing, metric_name, index_name, percentage_value, max_connection, ef_construction	首次建表后调用，初始化 HNSW 索引
rebuild_vector_hnsw_index	column, schema_name , index_name, metric_name, percentage_value, max_connection, ef_construction	重要：插入大量数据后必须调用，否则搜不到新数据

1.1.6. 距离度量函数映

在 ORM 查询中，可以通过列对象直接调用距离计算方法：

SQLAlchemy 方法	达梦内部算法	适用场景
l1_distance	MANHATTAN	曼哈顿距离
l2_distance	EUCLIDEAN	欧几里得距离
cosine_distance	COSINE	余弦相似度（最常用）
inner_product	DOT	内积
inner_product_negative	DOT_NEGATIVE	负内积
hamming_distance	HAMMING	汉明距离

SQLAIchemy不支持但是DM支持的还有VECTOR_DISTANCE

计算两个向量之间的距离。
语法说明：
VECTOR_DISTANCE(
vec1 VECTOR,
vec2 VECTOR,
distance_type VARCHAR
);
参数详解：
1．vec1：参与距离计算的向量或向量字符串。
2．vec2：参与距离计算的向量或向量字符串。
3．distance_type：距离类型（可选参数，支持COSINE、DOT、EUCLIDEAN、HAMMING、MANHATTAN和EUCLIDEAN_SQUARED，默认COSINE）。
举例说明：
例1 计算两个向量之间的余弦距离。

SELECT vector_distance(to_vector('[153.372, 278.869, 158.958, 168.344, 146.378]', 5, FLOAT64), to_vector('[463.037, 256.848, 336.582, 368.122, 268.492]', 5, FLOAT64), COSINE);

1.1.7. 演示案例

"""
达梦向量数据 - ORM方式向量搜索完整演示

功能: 使用 SQLAlchemy ORM 实现完整的文本向量搜索功能
特点: 自定义表结构、向量索引管理、多种距离度量、分类筛选
模型: all-MiniLM-L6-v2 (向量维度: 384)
"""

# 1. 导入依赖
import logging
from typing import List, Optional
from pathlib import Path
from dmSQLAlchemy import VECTOR, VectorAdaptor
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Integer, String, Text, Column, DateTime, create_engine, func

# 2. 全局配置
# 日志配置
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 数据库连接: dm+dmPython://<用户名>:<密码>@<IP地址>:<端口>
CONN_STR = "dm+dmPython://SYSDBA:SYSDBA@127.0.0.1:5236"
# 模型路径
MODEL_PATH = "./models/all-MiniLM-L6-v2"
# 表名和向量维度
TABLE_NAME = "knowledge_docs"
VECTOR_DIM = 384
# ORM 基类
Base = declarative_base()

# 3. 演示数据
DEMO_DATA = [
    {"category": "人工智能", "title": "什么是人工智能", "content": "人工智能（AI）是研究、开发用于模拟人的智能的理论、方法、技术及应用系统的技术科学，包括机器人、语言识别、图像识别、自然语言处理等领域。"},
    {"category": "人工智能", "title": "机器学习基础", "content": "机器学习是研究计算机怎样模拟人类学习行为的学科，是人工智能的核心，使计算机具有智能的根本途径。"},
    {"category": "人工智能", "title": "深度学习技术", "content": "深度学习是机器学习的分支，使用多层神经网络对数据进行高层抽象，在图像识别、语音识别、NLP等领域取得突破。"},
    {"category": "数据库", "title": "向量数据介绍", "content": "向量数据专门用于存储和检索向量数据，支持向量相似度搜索，广泛应用于图像检索、文本搜索、推荐系统等场景。"},
    {"category": "数据库", "title": "SQL查询优化", "content": "SQL查询优化是提高数据库性能的重要手段，包括合理使用索引、避免全表扫描、优化JOIN操作等技巧。"},
    {"category": "编程语言", "title": "Python编程入门", "content": "Python是高级编程语言，语法简洁，广泛应用于Web开发、数据分析、人工智能、自动化脚本等领域。"},
    {"category": "编程语言", "title": "Java企业级开发", "content": "Java是面向对象编程语言，具有跨平台、安全性高、稳定性好等特点，Spring框架是企业级开发主流。"},
    {"category": "云计算", "title": "Docker容器技术", "content": "Docker是开源容器化平台，将应用及依赖打包到可移植容器中，广泛应用于持续集成、微服务架构。"},
    {"category": "云计算", "title": "Kubernetes编排", "content": "Kubernetes是容器编排平台，用于自动化部署、扩展和管理容器化应用，提供服务发现、负载均衡等功能。"},
    {"category": "网络安全", "title": "数据加密技术", "content": "数据加密是保护数据安全的重要手段，常见算法包括对称加密（AES）、非对称加密（RSA）、哈希算法（SHA）。"},
]

# 4. 向量生成模型类（单例模式）
class EmbeddingModel:
    """向量生成模型: 使用 sentence-transformers 进行文本向量化"""
    _instance = None
    _model = None
    def __new__(cls, path: str = None):
        """单例模式: 确保只创建一个实例"""
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, path: str = None):
        """加载模型: 如果已加载则跳过"""
        if self._model is not None:
            return
        if path and Path(path).exists():
            from sentence_transformers import SentenceTransformer
            logger.info(f"加载模型: {path}")
            self._model = SentenceTransformer(path)
            self._dim = self._model.get_sentence_embedding_dimension()
            logger.info(f"模型加载成功，维度: {self._dim}")
        else:
            logger.warning(f"模型不存在: {path}，使用模拟向量")
            self._dim = VECTOR_DIM
    
    def encode(self, text: str) -> List[float]:
        """单文本向量化"""
        if self._model:
            return self._model.encode(text, convert_to_numpy=True).tolist()
        import hashlib, numpy as np
        rng = np.random.RandomState(int(hashlib.md5(text.encode()).hexdigest()[:8], 16))
        return rng.randn(self._dim).tolist()
    
    def encode_batch(self, texts: List[str]) -> List[List[float]]:
        """批量文本向量化"""
        if self._model:
            return self._model.encode(texts, convert_to_numpy=True).tolist()
        return [self.encode(t) for t in texts]
    
    @property
    def dimension(self) -> int:
        """获取向量维度"""
        return self._dim

# 5. ORM 模型定义
class KnowledgeDoc(Base):
    """知识文档模型: 包含分类、标题、内容、向量等字段"""
    __tablename__ = TABLE_NAME
    id = Column(Integer, primary_key=True, autoincrement=True)    # 主键
    category = Column(String(50))                                  # 分类
    title = Column(String(200))                                    # 标题
    content = Column(Text)                                         # 内容
    embedding = Column(VECTOR(dim=VECTOR_DIM, format='float64'))  # 向量
    created_at = Column(DateTime, server_default=func.current_timestamp())  # 创建时间
    
    def __repr__(self):
        return f"<KnowledgeDoc(id={self.id}, category='{self.category}', title='{self.title}')>"

# 6. 向量数据客户端类
class VectorDBClient:
    """向量数据客户端: 封装所有数据库操作""" 
    def __init__(self, conn_str: str, model_path: str):
        """初始化: 加载模型、创建引擎、创建适配器"""
        self.model = EmbeddingModel(model_path)
        self.engine = create_engine(conn_str, echo=False)
        self.adaptor = VectorAdaptor(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.session = None
        logger.info(f"客户端初始化完成，向量维度: {self.model.dimension}")
    
    def connect(self):
        """连接数据库"""
        self.session = self.Session()
        return self
    
    def close(self):
        """关闭连接"""
        if self.session:
            self.session.close()
    
    def __enter__(self):
        return self.connect()
    
    def __exit__(self, *args):
        self.close()
    
    def create_table(self, drop: bool = False):
        """创建表和向量索引"""
        if drop:
            Base.metadata.drop_all(self.engine)
        Base.metadata.create_all(self.engine)
        self.adaptor.create_vector_ivf_index(
            KnowledgeDoc.embedding, metric_name="cosine", index_name="ivf_index_doc"
        )
        logger.info(f"表 '{TABLE_NAME}' 创建成功，已创建向量索引")
    
    def insert(self, docs: List[dict]):
        """插入文档: 自动向量化并存储"""
        texts = [f"{d['title']} {d['content']}" for d in docs]
        embeddings = self.model.encode_batch(texts)
        for d, e in zip(docs, embeddings):
            self.session.add(KnowledgeDoc(
                category=d['category'], title=d['title'], content=d['content'], embedding=e
            ))
        self.session.commit()
        self.adaptor.rebuild_vector_ivf_index(
            KnowledgeDoc.embedding, metric_name="cosine", index_name="ivf_index_doc"
        )
        logger.info(f"插入 {len(docs)} 条文档，已重建向量索引")
    
    def search(self, query: str, category: Optional[str] = None, top_k: int = 5) -> List[tuple]:
        """余弦距离搜索: 值越小越相似"""
        query_embedding = self.model.encode(query)
        distance = KnowledgeDoc.embedding.cosine_distance(query_embedding)
        q = self.session.query(
            KnowledgeDoc.id, KnowledgeDoc.category, KnowledgeDoc.title,
            KnowledgeDoc.content, KnowledgeDoc.created_at, distance
        )
        if category:
            q = q.filter(KnowledgeDoc.category == category)
        q = q.filter(KnowledgeDoc.embedding.isnot(None)).order_by(distance.asc()).limit(top_k)
        return q.all()
    
    def search_by_l2(self, query: str, category: Optional[str] = None, top_k: int = 5) -> List[tuple]:
        """L2距离搜索: 值越小越相似"""
        query_embedding = self.model.encode(query)
        distance = KnowledgeDoc.embedding.l2_distance(query_embedding)
        q = self.session.query(
            KnowledgeDoc.id, KnowledgeDoc.category, KnowledgeDoc.title,
            KnowledgeDoc.content, KnowledgeDoc.created_at, distance
        )
        if category:
            q = q.filter(KnowledgeDoc.category == category)
        q = q.filter(KnowledgeDoc.embedding.isnot(None)).order_by(distance.asc()).limit(top_k)
        return q.all()
    
    def search_by_inner_product(self, query: str, category: Optional[str] = None, top_k: int = 5) -> List[tuple]:
        """内积搜索: 使用负内积，值越小越相似"""
        query_embedding = self.model.encode(query)
        distance = KnowledgeDoc.embedding.inner_product_negative(query_embedding)
        q = self.session.query(
            KnowledgeDoc.id, KnowledgeDoc.category, KnowledgeDoc.title,
            KnowledgeDoc.content, KnowledgeDoc.created_at, distance
        )
        if category:
            q = q.filter(KnowledgeDoc.category == category)
        q = q.filter(KnowledgeDoc.embedding.isnot(None)).order_by(distance.asc()).limit(top_k)
        return q.all()
    
    def get_categories(self) -> List[str]:
        """获取所有分类"""
        result = self.session.query(KnowledgeDoc.category).distinct().all()
        return [r[0] for r in result]
    
    def delete_by_id(self, doc_id: int):
        """按ID删除文档"""
        self.session.query(KnowledgeDoc).filter(KnowledgeDoc.id == doc_id).delete()
        self.session.commit()
        logger.info(f"删除文档 id={doc_id}")
    
    def drop_table(self):
        """删除表"""
        Base.metadata.drop_all(self.engine)
        logger.info(f"表 '{TABLE_NAME}' 已删除")
    
    def get_table_info(self):
        """获取表结构信息"""
        from sqlalchemy import inspect
        return inspect(self.engine).get_columns(TABLE_NAME, schema='SYSDBA')

# 7. 辅助函数
def show_results(rows, query: str, metric: str):
    """显示搜索结果"""
    print(f"\n查询: '{query}' | 度量: {metric}")
    print("-" * 70)
    for i, r in enumerate(rows, 1):
        print(f"  {i}. [距离:{r[5]:.4f}] {r[1]} | {r[2]}")
        if r[3]:
            preview = r[3][:50] + "..." if len(r[3]) > 50 else r[3]
            print(f"     {preview}")

# 8. 演示函数
def run_demo():
    """运行完整演示"""
    print("=" * 70)
    print("达梦向量数据 - ORM方式向量搜索演示")
    print("基于 dmSQLAlchemy + SQLAlchemy ORM + all-MiniLM-L6-v2 模型")
    print("=" * 70)
    with VectorDBClient(CONN_STR, MODEL_PATH) as client:
        print("\n[1] 创建表和向量索引...")
        client.create_table(drop=True)
        
        print("\n[2] 插入数据...")
        client.insert(DEMO_DATA)
        
        print("\n[3] 查看表结构...")
        for col in client.get_table_info():
            print(f"    {col['name']}: {col['type']}")
        
        print("\n[4] Cosine距离向量搜索...")
        for q in ["人工智能应用", "数据库优化", "编程语言", "云计算技术"]:
            rows = client.search(q, top_k=2)
            show_results(rows, q, "COSINE")
        
        print("\n[5] L2距离向量搜索...")
        rows = client.search_by_l2("深度学习", top_k=3)
        show_results(rows, "深度学习", "L2")
        
        print("\n[6] 内积距离向量搜索...")
        rows = client.search_by_inner_product("数据安全", top_k=3)
        show_results(rows, "数据安全", "INNER_PRODUCT")
        
        print("\n[7] 按分类筛选...")
        for cat in ["人工智能", "数据库"]:
            rows = client.search("技术", category=cat, top_k=2)
            show_results(rows, f"技术 (筛选:{cat})", "COSINE")
        
        print("\n[8] 获取所有分类...")
        print(f"    分类: {client.get_categories()}")
        
        print("\n[9] 删除文档演示...")
        first_row = client.search("人工智能", top_k=1)
        if first_row:
            client.delete_by_id(first_row[0][0])
            print(f"    已删除 id={first_row[0][0]} 的文档")
        
        print("\n[10] 清理...")
        client.drop_table()
    
    print("\n演示完成！")

# 9. 主程序入口
if __name__ == "__main__":
    run_demo()

1.1.8. 常见问题与提示 (Tips)

模型一致性: 插入数据和查询数据必须使用完全相同的模型，否则计算出的相似度无意义。
模型加载: 在初始化时会加载模型到内存，若模型较大，初始化过程可能需要几秒钟时间。

1.2. 使用VectorWordSeek

1.2.1. 环境准备

在使用 dmSQLAlchemy 向量功能前，请确保已安装以下依赖：

numpy: 用来处理python中的list类型与数据库内向量类型的转换。
dmPython: 达梦数据库 Python 驱动。
dmSQLAlchemy: 达梦 SQLAlchemy 方言包。
dmSQLAIchemy安装
Sentence-Transformers: 用于本地向量化生成的模型支持。

1.2.2. 快速上手示例 (Text-to-Text)

以下展示了如何连接数据库、创建向量表、插入数据并进行余弦相似度查询。

"""
达梦向量数据 - 文本向量搜索演示

功能: 使用 VectorWordSeek 类实现文本向量搜索
模型: all-MiniLM-L6-v2 (向量维度: 384)
"""

# 1. 导入依赖
from dmSQLAlchemy import VectorWordSeek

# 2. 配置连接串
# 连接字符串格式: dm+dmPython://<用户名>:<密码>@<IP地址>:<端口>
conn_str = 'dm+dmPython://SYSDBA:SYSDBA@127.0.0.1:5236'

# 3. 初始化 VectorWordSeek 对象
wordseekmodel = VectorWordSeek(
    table_name='test_wordseek',              # 表名
    connection_str=conn_str,                 # 数据库连接
    model_path='./models/all-MiniLM-L6-v2',  # 模型路径
    drop_if_existing=True,                   # 表存在时删除重建
    engine_args={'echo': False}              # 不打印 SQL
)

# 4. 创建表结构
# 自动创建的表结构: id(VARCHAR), embedding(VECTOR), document(TEXT), meta(JSON), create_time, update_time
wordseekmodel.create_table(True)

# 5. 插入文本数据
# insert() 方法会自动将文本转换为向量并存储
wordseekmodel.insert(texts=['This is a flag', 'This is a notice', 'This is a camp'])

# 6. 向量查询
# query() 方法会自动将查询文本转换为向量并搜索相似结果
# query_vector: 查询文本（列表格式，只能有一个元素）
result = wordseekmodel.query(count=2, DistanceMetric='COSINE', query_vector=['flag'])

# 7. 输出结果
# QueryResult 属性: id, document, metadata, distance
for row in result:
    print(f"ID: {row.id} | Document: {row.document} | Distance: {row.distance}")

1.2.3. WordSeek相关函数介绍

用于处理文本内容的向量化映射。

1.2.3.1. VectorWordSeek (对象初始化)

用途：初始化一个由 Python 到数据库表的 VectorWordSeek 映射对象。

参数名	用途	参数类型	默认值
connection_str	数据库连接串，与 SQLAlchemy 格式相同	str	None
table_name	数据库中对应的表名	str	None
vector_dim	向量列维度（通常根据模型自动调整，可忽略）	int	None
drop_if_existing	初始化时若表已存在，是否自动删除	bool	FALSE
model	显式传入的模型对象，model和model_path，二者传其中一个就可以，但是至少有一个	None	None
model_path	sentence_transformers模型在本地的存储路径，如果还存在torch模型则优先取model的	str	None
engine_args	SQLAlchemy 引擎的其他参数（如 {‘echo’: True}）	dict	None

1.2.3.2. drop_table 函数

用途：手动删除 VectorWordSeek 对象中指定表名的物理表。

1.2.3.3. create_table 函数

用途：在数据库中创建向量表。（VectorWordSeek下的create_table表结构固定）

drop_if_existing (bool): 若设为 True，创建前会强制删除旧表。

表结构说明：

序号	列名	介绍
1	ID	唯一标识符（UUID）
2	EMBEDDING	存储向量数据列（VECTOR类型）
3	DOCUMENT	存储原始文本内容列
4	META	筛选列，保留字段（暂无具体逻辑作用）
5	CREATE_TIME	数据创建时间
6	UPDATE_TIME	数据最后更新时间

1.2.3.4. insert 函数

用途：向指定表中插入文本内容，并由模型自动生成向量。

参数名	用途	参数类型	默认值
texts	需要插入的原始文本内容列表	Iterable[str]	无
metadatas	对应数据的元数据内容	list	无
ids	自定义的 ID 列表（若不填则自动生成）	list	无

1.2.3.5. query 函数

用途：通过向量相似度检索获取最匹配的结果集。

参数名	用途	参数类型	默认值
DistanceMetric	相似度计算方法（如 COSINE, DOT, EUCLIDEAN, HAMMING, MANHATTAN, EUCLIDEAN_SQUARED）	str	无
query_vector	检索的文本内容	str	无
count	返回的最相似结果个数	int	5
filter	额外的 SQL 过滤条件项	dict	None

1.3. 使用VectorImageSeek

VectorImageSeek 是 dmSQLAlchemy 提供的图像向量化映射模块。它允许用户将本地图像路径传入，通过 TorchScript 模型（如 ResNet）提取特征向量，并存储到达梦数据库中，从而实现“以图搜图”的功能。

1.3.1. 环境准备

驱动支持：需安装numpy、 dmPython 和 dmSQLAlchemy。
模型格式：必须提供 TorchScript 格式的模型文件（后缀通常为 .pt 或 .pth），不支持模型目录。
连接串：dm+dmPython://<USER>:<PASS>@<IP>:<PORT>。

1.3.2. 快速上手示例

"""
达梦向量数据 - 图像向量搜索演示

功能: 使用 VectorImageSeek 类实现图像向量搜索
模型: ResNet18 (向量维度: 512)
支持格式: PNG, JPG, JPEG, GIF, BMP, PGM, PBM
"""

# 1. 导入依赖
from dmSQLAlchemy import VectorImageSeek

# 2. 配置连接串
# 连接字符串格式: dm+dmPython://<用户名>:<密码>@<IP地址>:<端口>
conn_str = 'dm+dmPython://SYSDBA:SYSDBA@127.0.0.1:5236'

# 3. 初始化 VectorImageSeek 对象
# VectorImageSeek 使用 TorchScript 模型文件 (.pt 或 .pth)
# 模型要求: 输入 [batch, 3, H, W] 的图像张量，输出特征向量
imageseekmodel = VectorImageSeek(
    table_name='test_imageseek',           # 表名
    connection_str=conn_str,               # 数据库连接
    model_path='./models/resnet18.pt',     # TorchScript 模型文件路径
    drop_if_existing=True,                 # 表存在时删除重建
    engine_args={'echo': False}            # 不打印 SQL
)

# 4. 创建表结构
# 自动创建的表结构: id(VARCHAR), embedding(VECTOR), document(TEXT), meta(JSON), create_time, update_time
imageseekmodel.create_table(True)

# 5. 插入图像数据
# insert() 方法会自动: 加载图像 -> 预处理 -> 提取特征向量 -> 存储
imageseekmodel.insert(paths=[r'img.png'])

# 6. 向量查询
# query() 方法会自动: 加载查询图像 -> 提取特征向量 -> 搜索相似结果
# query_vector: 查询图像路径
result = imageseekmodel.query(count=10, DistanceMetric='COSINE', query_vector=r"img.png")

# 7. 输出结果
# QueryResult 属性: id, document(图像路径), metadata, distance
# distance 值越小越相似
for row in result:
    print(f"ID: {row.id} | Path: {row.document} | Distance: {row.distance}")

1.3.3. ImageSeek相关函数介绍

1.3.3.1. VectorImageSeek (对象初始化)

用途：初始化图像向量对象，建立 Python 与数据库表的映射。

参数名	用途	类型	默认值
connection_str	达梦数据库连接字符串	str	None
table_name	存储图像向量的数据库表名	str	None
model	显式传入的模型对象model和model_path，二者传其中一个就可以，但是至少有一个	None	None
model_path	torch模型在本地的存储路径，如果还存在sentence_transformers模型则优先取model的	str	None
vector_dim	向量维度（通常由模型自动识别，可忽略）	int	None
drop_if_existing	实例化时是否检查并删除同名表	bool	FALSE
engine_args	传递给 SQLAlchemy 引擎的参数（如 echo）	dict	None

1.3.3.2. drop_table 函数

用途：手动删除 VectorImageSeek 对象中指定表名的物理表。

1.3.3.3. create_table 函数

用途：在达梦数据库中创建专门存储图像向量的表。
drop_if_existing (bool): 若设为 True，创建前会强制删除旧表。
表结构说明：

序号	列名	介绍
1	ID	唯一标识符（UUID）
2	EMBEDDING	图像特征向量数据（VECTOR类型）
3	DOCUMENT	存储图片的原始路径
4	META	元数据列（预留，可存储 JSON 信息）
5	CREATE_TIME	数据创建时间
6	UPDATE_TIME	数据最后更新时间

1.3.3.4. insert 函数

**用途：**读取本地图片，提取特征向量并入库。

参数名	用途	类型	默认值
paths	必填。待插入的本地图片路径列表	Iterable[str]	无
metadatas	与图片对应的元数据信息列表	list	无
ids	自定义的 ID 列表（若不填则自动生成）	list	无

1.3.3.5. query 函数

用途：输入一张图片，检索库中最相似的结果。

参数名	用途	类型	默认值
DistanceMetric	距离算法：COSINE (余弦), L2 (欧氏距离), IP (内积)	str	无
query_vector	待检索的图片路径	str	无
count	返回最接近的前 N 条数据	int	5
filter	额外的 SQL 过滤条件字典	dict	None