企业官网建设流程全解析

用 pgai 构建 FastAPI 语义搜索与 RAG 应用：完整实战指南

【免费下载链接】pgaiA suite of tools to develop RAG, semantic search, and other AI applications more easily with PostgreSQL项目地址: https://gitcode.com/GitHub_Trending/pg/pgai

本文基于 pgai 仓库中的 simple_fastapi_app 示例 编写，带你完整跑通一个基于 PostgreSQL 的 RAG 应用：应用启动时通过pgai.install()安装数据库对象并拉起 Vectorizer 后台 Worker，对 Wikipedia 文章列自动创建向量嵌入，再用 pgvector 实现语义搜索，最后将检索结果作为上下文喂给 LLM 完成 Retrieval Augmented Generation（RAG）。读完后你可以掌握 pgai 的核心工作流——建表、建 vectorizer、跟踪嵌入进度、用余弦距离做检索、以及利用触发器实现"数据变了、嵌入自动更新"，并能将其套用到自己的 FastAPI 业务中。

1. 示例应用概览

该目录包含两个可运行的 FastAPI 应用，演示同一套 pgai 能力，区别仅在于数据访问方式：

• with_psycopg.py：使用psycopg3异步连接池直接写 SQL，语义检索基于 pgvector 的<=>操作符；
• with_sqlalchemy.py：使用 SQLAlchemy ORM，通过pgai.sqlalchemy.vectorizer_relationship把嵌入表自动映射为 ORM 关系。

两个应用的功能流程一致：

1. 启动时安装 pgai 所需的数据库对象，并以后台任务形式运行 Vectorizer Worker；
2. 创建wiki表，从 Hugging Face 的wikimedia/wikipedia数据集流式加载少量英文文章；
3. 为wiki.text列创建 vectorizer，自动生成 384 维嵌入（Ollamaall-minilm模型）；
4. 提供/vectorizer_status、/search、/insert_pgai_article、/rag四个端点。

2. 环境准备与启动

2.1 前置条件

• 一个可连接的 PostgreSQL 数据库（示例默认连接串为postgresql://postgres:postgres@localhost:5432/test，两个应用文件中都写在该常量处，见 with_psycopg.py）；
• 本地或可达的 Ollama 服务（http://localhost:11434），并拉取all-minilm（嵌入）与tinyllama（生成）两个模型；
• Python 依赖：fastapi、psycopg（3.x）、psycopg_pool、pgvector、ollama、datasets、pgai、numpy，SQLAlchemy 版本另需sqlalchemy。

2.2 获取示例代码并启动

在本地克隆仓库后，进入示例目录即可直接运行（对应 README 中通过 curl 下载with_psycopg.py的做法）：

git clone https://gitcode.com/GitHub_Trending/pg/pgai cd pgai/examples/simple_fastapi_app pip install pgai fastapi uvicorn psycopg psycopg-pool pgvector ollama datasets numpy fastapi dev with_psycopg.py

启动后访问http://0.0.0.0:8000/docs即可查看自动生成的 API 文档并在线试调各端点。主要端点为：

• /vectorizer_status：查看嵌入创建进度；
• /search?query=...：语义搜索；
• /insert_pgai_article（POST）：向wiki表插入一条关于 pgai 的文章，用于演示嵌入自动更新；
• /rag?query=...：基于检索上下文的 RAG 问答。

需要注意的适用前提：pgai.install()要求PostgreSQL 15 及以上（源码在 install.py 中显式检查server_version_num < 15并抛出异常），并且会自动执行CREATE EXTENSION IF NOT EXISTS vector（即 pgvector）。

3. 应用启动流程：安装 pgai 并拉起 Worker

示例应用把"数据库初始化 + 后台任务"全部放进 FastAPI 的lifespan生命周期钩子中，这是 with_psycopg.py 的核心：

from contextlib import asynccontextmanager from fastapi import FastAPI from pgai.vectorizer import Worker @asynccontextmanager async def lifespan(_app: FastAPI): # 1. 安装 pgai 库到数据库（幂等，已安装则忽略） pgai.install(DB_URL) # 2. 初始化连接池（安装完成后再 open，保证 ai schema 对象已就绪） await pool.open() # 3. 以后台 asyncio 任务运行 Vectorizer Worker worker = Worker(DB_URL) task = asyncio.create_task(worker.run()) # 4. 建表并加载演示数据 await create_wiki_table() if await wiki_table_is_empty(): await load_wiki_articles() # 5. 创建 vectorizer（重复启动时忽略 already exists） await create_vectorizer() yield # 应用在此运行，处理请求 # ---- 关闭阶段：优雅停止 Worker ---- print("gracefully shutting down worker...") await worker.request_graceful_shutdown() try: result = await asyncio.wait_for(task, timeout=20) except asyncio.TimeoutError: print("Worker did not shutdown in time, killing it") app = FastAPI(lifespan=lifespan)

要点解析：

• pgai.install(DB_URL)：同步地把内置的 ai.sql 脚本执行到目标库。从 install.py 的实现看，它会依次做：校验 PG 版本、确保vector扩展存在、读取向量扩展所在 schema、执行安装 SQL；若已安装则捕获DuplicateObject错误并静默通过（strict=False默认行为），因此可以在应用每次启动时安全调用。生产环境如 README 所建议，这一步也可以改用数据库迁移完成，而不是放在应用启动路径里。
• Worker(DB_URL)+asyncio.create_task(worker.run())：Worker 是嵌入创建的实际执行者。从 worker.py 的构造函数可以看到其可调参数：poll_interval（默认 1 分钟轮询间隔）、once（处理完一轮即退出）、vectorizer_ids（只处理指定 vectorizer，缺省为动态模式——自动发现库中所有 vectorizer）、concurrency（并发数）。示例中全部使用默认值，即"动态发现 + 每分钟轮询"。
• Worker 部署位置的取舍：示例为简化起见把 Worker 跑在 FastAPI 进程内；README 明确提示，生产上也可以把它放到独立进程或独立容器中运行，参见 vectorizer worker 文档。
• 优雅关闭：worker.request_graceful_shutdown()会置位内部asyncio.Event（见 worker.py），Worker 在当前批次处理完后退出；asyncio.wait_for(task, timeout=20)给出 20 秒上限兜底。

4. 建表与加载 Wikipedia 演示数据

4.1 创建wiki表

async def create_wiki_table(): async with pool.connection() as conn: async with conn.cursor() as cur: await cur.execute(""" CREATE TABLE IF NOT EXISTS wiki ( id SERIAL PRIMARY KEY, url TEXT NOT NULL, title TEXT NOT NULL, text TEXT NOT NULL ) """) await conn.commit()

表结构很简单：自增主键id+ 三个文本列。其中text将作为 vectorizer 的加载列。

4.2 从 Hugging Face 数据集流式加载文章

async def load_wiki_articles(): # to keep the demo fast, we have some simple limits num_articles = 10 max_text_length = 1000 wiki_dataset = load_dataset("wikimedia/wikipedia", "20231101.en", split="train", streaming=True) async with pool.connection() as conn: async with conn.cursor() as cur: for article in wiki_dataset.take(num_articles): await cur.execute( "INSERT INTO wiki (url, title, text) VALUES (%s, %s, %s)", (article['url'], article['title'], article['text'][:max_text_length]) ) await conn.commit()

这里有两个刻意的"演示性限制"：只取前 10 篇文章、每篇截断到 1000 字符，目的是让 demo 快速可跑。streaming=True让datasets库按迭代器方式逐条拉取数据，避免下载整个数据集；load_wiki_articles只在wiki_table_is_empty()为真时执行，保证重启应用不会重复灌数据。

5. 创建 Vectorizer：一次声明，嵌入自动同步

要让wiki.text可被语义检索，需要为它生成向量嵌入并保持与数据同步，这正是 pgai vectorizer 的职责。示例中的创建代码（with_psycopg.py）：

from pgai.vectorizer import CreateVectorizer from pgai.vectorizer.configuration import ( EmbeddingOllamaConfig, LoadingColumnConfig, ) async def create_vectorizer(): vectorizer_statement = CreateVectorizer( source="wiki", target_table='wiki_embedding_storage', loading=LoadingColumnConfig(column_name='text'), embedding=EmbeddingOllamaConfig( model='all-minilm', dimensions=384, base_url="http://localhost:11434" ) ).to_sql() try: async with pool.connection() as conn: async with conn.cursor() as cur: await cur.execute(vectorizer_statement) await conn.commit() except Exception as e: if "already exists" in str(e): pass # vectorizer 已存在时忽略 else: raise e

5.1CreateVectorizer是 SQL 语句构建器

CreateVectorizer是 create_vectorizer.py 中定义的 Python 参数模型，to_sql()会将其渲染为一条SELECT ai.create_vectorizer(...)调用（模板见 config_generator.py）。每个配置段（loading、embedding、chunking、destination等）都继承自 configuration.py 的SQLArgumentMixin，各自映射到一个数据库函数，例如EmbeddingOllamaConfig对应ai.embedding_ollama(...)。也就是说，你在 Python 里写的这段代码最终等价于一条可以直接在 psql 中执行的SELECT ai.create_vectorizer('wiki', loading => ai.loading_column(...), embedding => ai.embedding_ollama(...), ...)——两条路径（Python builder / 原生 SQL）创建的是同一个数据库对象，完整 SQL API 参考见 vectorizer API reference，使用概览见 vectorizer overview。

本例涉及的参数及其含义：

未显式给出的配置项走数据库侧默认值：chunking 默认为none（整行作为一个 chunk）、formatting、indexing 等也有各自默认。若想调整分块策略（比如长文档场景），可传入ChunkingCharacterTextSplitterConfig(chunk_size=..., chunk_overlap=...)等参数，全部可选项定义在 configuration.py 与 create_vectorizer.py。

5.2 幂等处理

应用重启会再次执行create_vectorizer()，因此代码捕获了already exists异常并忽略。这是 demo 里的实用手法；正式项目建议用 Alembic 迁移管理向量器，参见 alembic-integration。

6. 跟踪嵌入创建进度：ai.vectorizer_status视图

Vectorizer 的嵌入创建是异步的：Worker 从数据库工作队列中批量取行、调用嵌入服务、再写回目标表。这样设计的目的是支持批量处理并从嵌入服务的瞬时故障中恢复（详见 README 第 4 步的说明）。要观察进度，示例提供了一个只读端点：

@app.get("/vectorizer_status") async def vectorizer_status(): async with pool.connection() as conn: async with conn.cursor(row_factory=dict_row) as cur: await cur.execute("SELECT * FROM ai.vectorizer_status") return await cur.fetchall()

curl -X 'GET' \ 'http://0.0.0.0:8000/vectorizer_status' \ -H 'accept: application/json'

ai.vectorizer_status是一个由 pgai 安装脚本创建的视图，定义在 ai.sql。从视图定义可以看到它返回的列包括：vectorizer 的id、name、source_table、target_table、检索view名、embedding_column，以及pending_items（来自ai.vectorizer_queue_pending(v.id)，即工作队列中尚未处理的行数）。当pending_items为 0 时，说明该 vectorizer 的存量数据已全部完成嵌入——对本 demo 的 10 篇短文，这个过程通常只需几秒。

从队列的实现还能看到 Worker 的健壮性设计：取任务查询使用FOR UPDATE SKIP LOCKED加pg_try_advisory_xact_lock的组合（见 vectorizer.py），保证多个 Worker 并发安全、互不重复处理同一行；批次大小默认 50 行（文档类加载为 1），可由processing.batch_size覆盖，取值被钳制在 1–2048（见 vectorizer.py）。

7. 基于 pgvector 的语义搜索

7.1 检索视图与<=>余弦距离操作符

Vectoriser 创建后，除了目标表wiki_embedding_storage，还会生成一个把源表列与嵌入"拼"在一起的检索视图——本例即wiki_embedding（view_name默认为<target_table>去掉_storage后缀）。该视图包含wiki表的所有列，外加embedding（向量）和chunk（该条嵌入对应的文本分片）两列。

示例中的检索函数（with_psycopg.py）：

@dataclass class WikiSearchResult: id: int url: str title: str text: str chunk: str distance: float async def _find_relevant_chunks(client: ollama.AsyncClient, query: str, limit: int = 2): response = await client.embed(model="all-minilm", input=query) embedding = np.array(response.embeddings[0]) async with pool.connection() as conn: async with conn.cursor(row_factory=class_row(WikiSearchResult)) as cur: await cur.execute(""" SELECT w.id, w.url, w.title, w.text, w.chunk, w.embedding <=> %s as distance FROM wiki_embedding w ORDER BY distance LIMIT %s """, (embedding, limit)) return await cur.fetchall()

这段代码的关键点：

• 查询向量用与源数据相同的嵌入模型（all-minilm）生成，通过 Ollama SDK 的embed接口获得，并转成numpy数组以便 psycopg 以 pgvector 类型绑定（连接池在创建时用register_vector_async(conn)注册了向量类型适配器，见 with_psycopg.py）；
• embedding <=> %s是 pgvector 定义的余弦距离操作符：距离越小越相似，所以ORDER BY distance LIMIT n即取最相关的 n 个分片；
• 为什么需要 chunking：大段文本要拆成语义自洽的小片段，每个片段单独嵌入，检索时才可能精确命中；vectorizer 在创建嵌入时会自动完成切分，查询侧直接拿到的是最相关的chunk。示例返回结果中同时带出文章全文text与命中的chunk——不同应用可以按需只取其一（README 第 5 步的解释）；
• class_row(WikiSearchResult)是 psycopg 的行工厂，把每行直接映射进 dataclass，/search端点随后用asdict序列化为 JSON。

/search端点本身只有三行：

@app.get("/search") async def search(query: str): client = ollama.AsyncClient(host="http://localhost:11434") results = await _find_relevant_chunks(client, query) return [asdict(result) for result in results]

试试下面的查询——"Properties of Light" 这几个词可能根本不出现在任何文章里，但嵌入捕捉了语义含义，相关段落仍会被排到前面：

curl -X 'GET' \ 'http://0.0.0.0:8000/search?query=Properties%20of%20Light' \ -H 'accept: application/json'

8. 数据变更后嵌入自动更新

语义搜索是独立有用的能力，也是 RAG 的基石组件。示例用一个端点模拟"业务数据变化"——向wiki表插入一条关于 pgai 的文章：

@app.post("/insert_pgai_article") async def insert_pgai_article(): async with pool.connection() as conn: async with conn.cursor() as cur: await cur.execute(""" INSERT INTO wiki (url, title, text) VALUES (%s, %s, %s) """, ( "https://en.wikipedia.org/wiki/Pgai", "pgai - Power your AI applications with PostgreSQL", "pgai is a tool to make developing RAG and other AI applications easier..." )) await conn.commit() return {"message": "Article inserted successfully"}

注意这里没有任何创建嵌入的代码。Vectoriser 在源表上安装了触发器，INSERT提交后新行会进入工作队列，Worker 在下一轮轮询中自动为其生成嵌入；数据更新或删除时，对应嵌入也会被同步更新或清理（Worker 内部"先删旧嵌入再写新嵌入"的处理见 vectorizer.py）。几秒后，用与主题相近的查询再检索，就能看到新条目出现在结果中：

curl -X 'GET' \ 'http://0.0.0.0:8000/search?query=AI%20Tools' \ -H 'accept: application/json'

9. 用 RAG 回答 LLM 没见过的问题

LLM 没有在 pgai 的资料上训练过，靠数据库里的数据才能回答相关问题——这正是 RAG 的价值。/rag端点（with_psycopg.py）把第 7 节的检索结果组织成提示词上下文，再交给 Ollama 上的tinyllama生成回答：

@app.get("/rag") async def rag(query: str) -> Optional[str]: # 1. 初始化 Ollama 客户端并检索相关分片 client = ollama.AsyncClient(host="http://localhost:11434") chunks = await _find_relevant_chunks(client, query) # 2. 把检索到的文章拼成上下文 context = "\n\n".join( f"{chunk.title}:\n{chunk.text}" for chunk in chunks ) logger.debug(f"Context: {context}") # 3. 构造带上下文的提示词 prompt = f"""Question: {query} Please use the following context to provide an accurate response: {context} Answer:""" # 4. 调用 LLM 生成回答 response = await client.generate( model='tinyllama', prompt=prompt, stream=False ) return response['response']

调用效果：

curl -X 'GET' \ 'http://0.0.0.0:8000/rag?query=What%20is%20pgai' \ -H 'accept: application/json'

整体链路即：用户问题 → 嵌入 → pgvector 余弦检索 top-k 分片 → 拼接为 prompt 上下文 → LLM 生成答案。检索部分完全由 PostgreSQL 承担，LLM 只负责"基于给定上下文作答"，这就是 RAG 的分工。

10. SQLAlchemy 变体：用vectorizer_relationship声明嵌入关系

with_sqlalchemy.py 演示了用 ORM 表达同样的流程。模型定义里只需加一行vectorizer_relationship（with_sqlalchemy.py）：

class Wiki(Base): __tablename__ = "wiki" id: Mapped[int] = mapped_column(primary_key=True) url: Mapped[str] title: Mapped[str] text: Mapped[str] # 为 text 字段声明向量嵌入关系 text_embeddings = vectorizer_relationship( target_table='wiki_embeddings', dimensions=384 )

从实现（sqlalchemy/init.py）看，vectorizer_relationship（即_Vectorizer描述符）会在 mapper 配置完成后自动：

• 动态生成一个嵌入 ORM 模型（默认表名<源表名>_embedding_store，或显式指定的target_table），含embedding_uuid主键、chunk、embedding（Vector(dimensions)列）、chunk_seq以及回指父模型的parent关系；
• 复制父表主键列并建立ondelete=CASCADE的外键约束；
• 把生成的类挂载为Wiki.text_embeddings_model，并把 relationship 注册为Wiki.text_embeddings，因此可以直接session.query(Wiki).join(Wiki.text_embeddings)。

对应的向量检索代码因此可以完全 ORM 化（with_sqlalchemy.py）：

result = session.query( Wiki, Wiki.text_embeddings.embedding.cosine_distance(embedding).label('distance') ).join(Wiki.text_embeddings).order_by( 'distance' ).limit(limit).all()

其中cosine_distance(...)是 pgvector 对 SQLAlchemy 的封装（pgvector.sqlalchemy.Vector），语义与 psycopg 版本里的<=>相同。创建 vectorizer 的调用参数与 psycopg 版完全一致，只是经由Session.execute(sqlalchemy.text(...))提交。

两个变体怎么选：如果应用本来就用 SQLAlchemy，with_sqlalchemy.py让嵌入表和源表之间的 join/距离计算都留在 ORM 层，代码更统一；如果追求轻量或直接用 SQL，with_psycopg.py的路径更短、也更接近 pgai 的数据库对象本身。

11. 小结与延伸阅读

• 安装与初始化：pgai.install(DB_URL)幂等地安装aischema 下的全部对象（要求 PG 15+，自动装 pgvector），实现见 install.py；
• Worker 生命周期：应用内以asyncio.create_task(worker.run())启动、以request_graceful_shutdown()停止；生产可拆分为独立进程/容器，参考 worker 文档；
• Vectoriser 声明：CreateVectorizer(...).to_sql()是SELECT ai.create_vectorizer(...)的 Python 构建器，参数模型在 configuration.py 与 create_vectorizer.py，SQL 全量参考在 api-reference；
• 进度监控：ai.vectorizer_status视图的pending_items归零即表示存量嵌入完成（视图定义见 ai.sql）；
• 检索：检索视图 + pgvector<=>余弦距离 +ORDER BY ... LIMIT n，分片文本在chunk列；
• 数据同步：触发器 + 工作队列（FOR UPDATE SKIP LOCKED+ advisory lock 保证并发安全）实现嵌入自动创建/更新/删除，队列设计见 vectorizer.py；
• RAG：检索 top-k 分片 → 拼上下文 → LLM 生成，端点代码见 with_psycopg.py。

想进一步扩展时，建议按 vectorizer overview 与 Python 集成文档 了解 chunking、indexing（如 HNSW）等更多配置项，并结合 tests 下的测试用例观察各参数在真实 PostgreSQL 中的行为。

【免费下载链接】pgaiA suite of tools to develop RAG, semantic search, and other AI applications more easily with PostgreSQL项目地址: https://gitcode.com/GitHub_Trending/pg/pgai