• 首页
  • Pgai 向 Python 进军：SQLAlchemy 与 Alembic 的整合

Pgai 向 Python 进军：SQLAlchemy 与 Alembic 的整合

pgai Vectorizer 的发布改变了开发者将向量嵌入整合到其应用程序中的方式。通过一条 SQL 命令，他们 可以自动化嵌入的创建和管理，确保与不断变化的源数据保持同步——传统的手动和耗时的过程已不复存在。

本博客文章将向您展示如何将 pgai Vectorizer 与 Python 无缝集成，Python 是一种广泛使用且受欢迎的编程语言。我们将使用 SQLAlchemy，这是一种 Python 对象关系映射器（ORM）和 SQL 工具包，以及 Alembic，它是用于数据库迁移的工具。

我们将通过一个博客应用程序的示例来引导您，该应用程序包含不同的博客文章，其标题和内容存储在一个表中。我们的目标是设置一个向量化器，以生成和管理博客文章内容的嵌入。

Pgai 向量化器：自动嵌入创建与管理

Pgai 向量化器通过自动化多个步骤简化了嵌入工作流程：

  • 分块和格式化策略，以生成高质量的嵌入
  • 与多个嵌入提供者的集成，包括 OpenAI、Cohere、Ollama 和 LiteLLM
  • 多种嵌入配置，用于对不同嵌入模型进行实验，以找到最适合您用例的嵌入模型
  • 源数据与嵌入之间的自动同步，减少开发开销
  • 让我们深入了解如何将其添加到使用 SQLAlchemy 和 Alembic 的示例设置中！

 

安装和导入 Pgai Python 库

首先，安装支持 SQLAlchemy 的 pgai Python 库：

pip install "pgai[sqlalchemy]"

使用 Alembic 创建一个向量化器

要集成 pgai 向量化器，首先使用 Alembic 定义一个迁移脚本。首先，在你的 env.py 中注册 pgai 的 Alembic 操作：

from pgai.alembic import register_operations

register_operations()

如果您使用 Alembic 的自动生成特性来生成迁移，请确保在您的 env.py 中添加以下代码，以确保在 pgai Alembic 操作下生成和管理的表被排除在自动生成过程之外：

def include_object(object, name, type_, reflected, compare_to):
   if type_ == "table" and name in target_metadata.info.get("pgai_managed_tables", set()):
        return False
    return True


context.configure(
      connection=connection,
      target_metadata=target_metadata,
      include_object=include_object
  )

在这段代码中，首先检查 `type_` 是否等于 `”table”`，并且 `name` 是否在 `target_metadata.info` 中获取的 `”pgai_managed_tables”` 集合里。如果满足这些条件，则返回 `False`，否则返回 `True`。

接着，调用 `context.configure` 方法，传入 `connection`、`target_metadata` 和 `include_object` 参数。

 

假设博客文章存储在名为 blog_posts 的表中，我们创建一个迁移脚本，以使用 OpenAI 的 text-embedding-3-small 作为嵌入模型生成嵌入：

from alembic import op
from pgai.vectorizer.configuration import (
    EmbeddingOpenaiConfig,
    ChunkingCharacterTextSplitterConfig,
    FormattingPythonTemplateConfig
)

def upgrade() -> None:
   op.create_vectorizer(
        source="blog_posts",
        target_table ='blog_posts_embedding_store',
        view_table ='blog_posts_embedding',
        embedding=EmbeddingOpenaiConfig(
            model='text-embedding-3-small',
            dimensions=1536
        ),
        chunking=ChunkingRecursiveCharacterTextSplitterConfig(
            chunk_column='content',
            chunk_size=800,
            chunk_overlap=400,
            separators=['.', ' '] ),
      formatting=FormattingPythonTemplateConfig(template='$title - $chunk') ) 

def downgrade() -> None: 

    op.drop_vectorizer(target_table="blog_posts_embedding_store", drop_all=True)

从上面的代码中需要注意几点：
  • create_vectorizer 函数创建了 嵌入表 和 连接嵌入和源表的视图。参数 target_table 和 view_table 分别定义了表和视图的名称。
  • 使用函数 ChunkingRecursiveCharacterTextSplitterConfig 来指明 源表中将用于生成嵌入的列的内容。我们还可以定义不同的参数来拆分内容，以适应嵌入模型的上下文窗口。

这些块还附加了额外的信息（即 文章标题），遵循函数 FormattingPythonTemplateConfig 中定义的模式，这是丰富提供给嵌入模型的文本和生成的嵌入的一种方式。

您可以在 Vectorizer 的 SQL API 参考 中找到更多参数。

通过 SQLAlchemy 与嵌入进行交互

Pgai 的 Python 库提供了预配置的 SQLAlchemy 关系，vectorizer_relationship，通过该关系可以完成与向量化器的所有 ORM 交互。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pgai.sqlalchemy import vectorizer_relationship

class Base(DeclarativeBase):
    pass

class BlogPost(Base):
    __tablename__ = "blog_posts"
   id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    content: Mapped[str]

# 为 内容 字段 添加 向量 嵌入
   content_embeddings = vectorizer_relationship(
    dimensions=1536
)

除了维度参数，vectorizer_relationship 还支持更多参数，包括：
  • target_schema：嵌入表的模式；如果未指定，嵌入表将继承父（源表）的模式。
  • target_table：嵌入表的名称。

您可以将此关系视为任何 SQLAlchemy 关系，您可以通过设置 parent_kwargs 参数来 进行配置。

查询嵌入

此关系允许您访问嵌入表中定义的不同嵌入属性，并将嵌入查询与常规 SQL 查询连接，如下所示：


# 同时 对 博客 帖子 和 嵌入 进行 过滤
results = (
    session.query(BlogPost, BlogPost.content_embeddings)
    .join(BlogPost.content_embeddings)
    .filter(BlogPost.title.ilike("%search term%"))
    .all()
)

# 访问 嵌入 属性


for post, embedding in results:
    print(post.title)         
    print(embedding.embedding) # 向量 嵌入
    print(embedding.chunk) # 文本 块
    print(embedding.chunk_seq) # 块的 序列 编号
    print(embedding.embedding_uiid) # 嵌入 的 UUID 编号

语义搜索

通过这种关系，我们可以使用来自pgvector-python库的距离比较器之一cosine_distance进行语义搜索，如下所示：

from sqlalchemy import func, text

similar_posts = (
    session.query(BlogPost.content_embeddings)
    .order_by(
        BlogPost.content_embeddings.embedding.cosine_distance(
                func.ai.openai_embed( # 使用 pgai's embedding functions 
                "text-embedding-3-small",
               "search query",
               text("维度 => 1536") ) ) )
               .limit(5)
               .all() )

让我们来分析一下：

  • 函数 func.ai.openai_embed，指的是 pgai 的 OpenAI 嵌入函数，生成搜索查询的嵌入，然后用于在向量化器生成的嵌入上进行搜索。

您也可以提供自己的查询嵌入，而不是使用pgai的嵌入功能：

similar_posts = (
    session.query(BlogPost.content_embeddings)
    .order_by(
        BlogPost.content_embeddings.embedding.cosine_distance(
            [3, 1, 2,...]
        )
    )
    .limit(5)
    .all()
)

结论

在这篇文章中，我们观察到 pgai Vectorizer 如何通过 Alembic 创建向量化器、查询嵌入以及执行语义搜索，从而简化嵌入生成和管理！这种 Python 集成使开发人员能够使用熟悉的工具，同时以最小的努力实现强大的 AI 驱动功能。

如果您正在构建 AI 应用程序，请探索 Timescale 的完整开源 AI 堆栈，或前往 pgai 的 GitHub 仓库，开始将 AI 工作流引入 PostgreSQL——无需离开您的数据库。

更多

Weibo Weixin

京ICP备17072685号-4 © 2025 QPython.