0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.
数据库工程完全指南：从 SQL 到向量数据库、AI RAG 系统
即使在 AI 时代，数据库工程依然是所有系统的根基。无论 LLM 多么强大，没有安全存储数据、快速检索数据的能力，就无法构建生产系统。尤其随着向量检索和 RAG 系统的出现，数据库工程的角色反而变得更加重要。
本指南系统地梳理了 SQL 高级技巧、PostgreSQL 实战、NoSQL、向量数据库、分布式数据库理论，以及 LLM + DB 集成模式。
1. 关系型数据库核心：SQL 高级技巧
1.1 窗口函数（Window Functions）
窗口函数与聚合函数不同，它不会把行分组折叠，而是在保留每一行上下文的同时进行计算。这在分析型查询中不可或缺。
-- 按部门统计薪资排名和累计总额
SELECT
employee_id,
name,
department,
salary,
RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
SUM(salary) OVER (PARTITION BY department) AS dept_total,
AVG(salary) OVER (
PARTITION BY department
ORDER BY hire_date
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
) AS rolling_avg_3
FROM employees;
常用窗口函数：
ROW_NUMBER()：不重复的序号RANK()：并列时排名相同，下一名跳过DENSE_RANK()：并列时排名相同，下一名连续LAG() / LEAD()：引用前一行/后一行的值FIRST_VALUE() / LAST_VALUE()：窗口内第一个/最后一个值
-- 计算月度营收环比增长率
SELECT
month,
revenue,
LAG(revenue, 1) OVER (ORDER BY month) AS prev_month,
ROUND(
(revenue - LAG(revenue, 1) OVER (ORDER BY month)) * 100.0
/ NULLIF(LAG(revenue, 1) OVER (ORDER BY month), 0),
2
) AS growth_rate_pct
FROM monthly_sales;
1.2 CTE（公用表表达式）
CTE 把复杂查询拆分成若干阶段，从而提升可读性和可复用性。递归 CTE 在遍历层级结构时格外强大。
-- 用递归 CTE 遍历组织架构
WITH RECURSIVE org_tree AS (
-- Base case: 最顶层员工
SELECT
employee_id,
name,
manager_id,
0 AS depth,
name::TEXT AS path
FROM employees
WHERE manager_id IS NULL
UNION ALL
-- Recursive case
SELECT
e.employee_id,
e.name,
e.manager_id,
ot.depth + 1,
ot.path || ' > ' || e.name
FROM employees e
INNER JOIN org_tree ot ON e.manager_id = ot.employee_id
)
SELECT employee_id, name, depth, path
FROM org_tree
ORDER BY path;
-- 用分阶段的 CTE 拆解复杂分析
WITH
top_customers AS (
SELECT customer_id, SUM(amount) AS total_spent
FROM orders
WHERE created_at >= NOW() - INTERVAL '90 days'
GROUP BY customer_id
HAVING SUM(amount) > 1000
),
customer_details AS (
SELECT c.*, tc.total_spent
FROM customers c
JOIN top_customers tc ON c.id = tc.customer_id
),
ranked AS (
SELECT *,
NTILE(4) OVER (ORDER BY total_spent DESC) AS quartile
FROM customer_details
)
SELECT * FROM ranked WHERE quartile = 1;
1.3 执行计划分析（EXPLAIN ANALYZE）
要找到查询性能问题的根本原因，就必须学会阅读执行计划。
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT u.name, COUNT(o.id) AS order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.created_at > '2025-01-01'
GROUP BY u.id, u.name
ORDER BY order_count DESC
LIMIT 10;
执行计划输出解读：
Limit (cost=1234.56..1234.57 rows=10 width=72) (actual time=45.123..45.125 rows=10 loops=1)
-> Sort (cost=1234.56..1259.56 rows=10000 width=72) (actual time=45.120..45.121 rows=10 loops=1)
Sort Key: (count(o.id)) DESC
Sort Method: top-N heapsort Memory: 25kB
-> HashAggregate (cost=876.00..976.00 rows=10000 width=72) (actual time=38.456..42.234 rows=8523 loops=1)
Group Key: u.id
Batches: 1 Memory Usage: 1553kB
-> Hash Left Join (cost=345.00..801.00 rows=15000 width=40) (actual time=5.678..28.901 rows=15000 loops=1)
Hash Cond: (o.user_id = u.id)
-> Seq Scan on orders o (cost=0.00..312.00 rows=15000 width=16) (actual time=0.023..8.456 rows=15000 loops=1)
-> Hash (cost=270.00..270.00 rows=6000 width=32) (actual time=5.234..5.234 rows=6000 loops=1)
Buckets: 8192 Batches: 1 Memory Usage: 358kB
-> Index Scan using idx_users_created_at on users u (cost=0.29..270.00 rows=6000 width=32)
Planning Time: 1.234 ms
Execution Time: 45.456 ms
核心指标：
Seq ScanvsIndex Scan：出现 Seq Scan 时要检查是否需要索引actual rows与预估rows差距很大时，需要更新统计信息（ANALYZE）Buffers: shared hitvsread：检查缓存命中率loops：嵌套循环次数，数值越高性能下降越明显
1.4 索引设计策略
-- 复合索引：把选择度高的列放在前面
CREATE INDEX idx_orders_user_status_date
ON orders (user_id, status, created_at DESC);
-- 部分索引：只对特定条件建立索引（节省空间）
CREATE INDEX idx_active_users
ON users (email)
WHERE deleted_at IS NULL AND status = 'active';
-- 表达式索引：对函数结果建立索引
CREATE INDEX idx_users_lower_email
ON users (LOWER(email));
-- BRIN 索引：对时序数据高效（体积非常小）
CREATE INDEX idx_logs_timestamp_brin
ON application_logs USING BRIN (created_at);
-- GIN 索引：适合数组、JSONB、全文检索
CREATE INDEX idx_products_tags_gin
ON products USING GIN (tags);
1.5 事务与 ACID
-- 设置事务隔离级别
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
BEGIN;
-- 账户转账：保证原子性
UPDATE accounts SET balance = balance - 500 WHERE id = 1;
UPDATE accounts SET balance = balance + 500 WHERE id = 2;
-- 余额校验
DO $$
DECLARE
bal NUMERIC;
BEGIN
SELECT balance INTO bal FROM accounts WHERE id = 1;
IF bal < 0 THEN
RAISE EXCEPTION 'Insufficient funds';
END IF;
END $$;
COMMIT;
不同隔离级别下可能出现的问题：
| 隔离级别 | 脏读（Dirty Read） | 不可重复读（Non-repeatable Read） | 幻读（Phantom Read） |
|---|---|---|---|
| READ UNCOMMITTED | 可能发生 | 可能发生 | 可能发生 |
| READ COMMITTED | 已避免 | 可能发生 | 可能发生 |
| REPEATABLE READ | 已避免 | 已避免 | 可能发生 |
| SERIALIZABLE | 已避免 | 已避免 | 已避免 |
2. PostgreSQL 实战：高级功能
2.1 JSONB 与半结构化数据
PostgreSQL 的 JSONB 以二进制格式存储 JSON，支持快速查询和索引。
-- 创建 JSONB 列并建立 GIN 索引
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'
);
CREATE INDEX idx_products_metadata ON products USING GIN (metadata);
-- JSONB 运算符
-- ->> : 以文本形式提取
-- -> : 以 JSON 形式提取
-- @> : 包含关系检查
-- ? : 键是否存在
SELECT * FROM products
WHERE metadata @> '{"category": "electronics", "in_stock": true}';
SELECT
name,
metadata->>'brand' AS brand,
(metadata->>'price')::NUMERIC AS price,
metadata->'specs'->>'cpu' AS cpu
FROM products
WHERE metadata ? 'discount_pct'
AND (metadata->>'discount_pct')::NUMERIC > 10;
-- JSONB 更新：只修改指定的键
UPDATE products
SET metadata = jsonb_set(
metadata,
'{price}',
'29900'::jsonb
)
WHERE id = 42;
-- JSONB 路径查询
SELECT jsonb_path_query(
metadata,
'$.specs.memory ? (@ > 16)'
) FROM products;
2.2 分区（Table Partitioning）
-- 按时序数据做范围分区
CREATE TABLE events (
id BIGSERIAL,
user_id INT,
event_type TEXT,
payload JSONB,
created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);
-- 按月自动创建分区的函数
CREATE OR REPLACE FUNCTION create_monthly_partition(target_date DATE)
RETURNS VOID AS $$
DECLARE
partition_name TEXT;
start_date DATE;
end_date DATE;
BEGIN
start_date := DATE_TRUNC('month', target_date);
end_date := start_date + INTERVAL '1 month';
partition_name := 'events_' || TO_CHAR(start_date, 'YYYY_MM');
EXECUTE FORMAT(
'CREATE TABLE IF NOT EXISTS %I PARTITION OF events
FOR VALUES FROM (%L) TO (%L)',
partition_name, start_date, end_date
);
END;
$$ LANGUAGE plpgsql;
SELECT create_monthly_partition('2026-03-01');
SELECT create_monthly_partition('2026-04-01');
-- 确认分区裁剪是否生效
EXPLAIN SELECT * FROM events
WHERE created_at BETWEEN '2026-03-01' AND '2026-03-31';
2.3 逻辑复制（Logical Replication）
-- Publisher 端设置
ALTER SYSTEM SET wal_level = logical;
CREATE PUBLICATION app_publication
FOR TABLE users, orders, products
WITH (publish = 'insert, update, delete');
-- Subscriber 端设置订阅
CREATE SUBSCRIPTION app_subscription
CONNECTION 'host=primary-db port=5432 dbname=myapp user=replicator'
PUBLICATION app_publication;
-- 监控复制状态
SELECT
subname,
received_lsn,
latest_end_lsn,
latest_end_time
FROM pg_stat_subscription;
3. NoSQL 数据库实战
3.1 Redis：缓存模式
Redis 是一种内存数据结构存储，广泛应用于缓存、会话管理、消息队列。
import redis
import json
import hashlib
from functools import wraps
from typing import Any, Optional
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# Cache-Aside 模式（Lazy Loading）
def get_user_profile(user_id: int) -> dict:
cache_key = f"user:profile:{user_id}"
# 1. 先查缓存
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# 2. 从 DB 查询
user = db.query("SELECT * FROM users WHERE id = %s", user_id)
# 3. 写入缓存（TTL 1 小时）
r.setex(cache_key, 3600, json.dumps(user))
return user
# Write-Through 模式：写入时同步更新缓存
def update_user_profile(user_id: int, data: dict) -> None:
cache_key = f"user:profile:{user_id}"
# 更新 DB
db.execute("UPDATE users SET ... WHERE id = %s", user_id)
# 同时立即更新缓存
updated = get_user_from_db(user_id)
r.setex(cache_key, 3600, json.dumps(updated))
# Write-Behind（Write-Back）模式：异步写入 DB
class WriteBehindCache:
def __init__(self):
self.dirty_keys_set = "cache:dirty_keys"
def write(self, key: str, value: dict, ttl: int = 3600):
# 立即写入缓存
r.setex(key, ttl, json.dumps(value))
# 加入脏键列表（稍后 flush 到 DB）
r.sadd(self.dirty_keys_set, key)
def flush_to_db(self):
dirty_keys = r.smembers(self.dirty_keys_set)
for key in dirty_keys:
data = r.get(key)
if data:
db.upsert(json.loads(data))
r.srem(self.dirty_keys_set, key)
# 分布式锁（Redlock）
import time
def acquire_lock(lock_name: str, timeout: int = 10) -> Optional[str]:
identifier = str(time.time())
lock_key = f"lock:{lock_name}"
acquired = r.set(lock_key, identifier, nx=True, ex=timeout)
return identifier if acquired else None
def release_lock(lock_name: str, identifier: str) -> bool:
lock_key = f"lock:{lock_name}"
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = r.eval(lua_script, 1, lock_key, identifier)
return bool(result)
Redis 数据结构应用：
# Sorted Set：实时排行榜
def update_score(player: str, score: int):
r.zadd("leaderboard", {player: score})
def get_top_players(n: int = 10):
return r.zrevrange("leaderboard", 0, n-1, withscores=True)
# HyperLogLog：唯一访客数（近似值，内存高效）
def track_visitor(page: str, user_id: str):
r.pfadd(f"visitors:{page}:{today()}", user_id)
def get_unique_visitors(page: str, date: str) -> int:
return r.pfcount(f"visitors:{page}:{date}")
# Pub/Sub：实时通知
import threading
def publisher():
for i in range(10):
r.publish("notifications", json.dumps({"type": "alert", "msg": f"Event {i}"}))
def subscriber():
pubsub = r.pubsub()
pubsub.subscribe("notifications")
for message in pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
print(f"Received: {data}")
3.2 MongoDB：文档建模
from pymongo import MongoClient, ASCENDING, DESCENDING
from datetime import datetime
client = MongoClient('mongodb://localhost:27017/')
db = client['ecommerce']
# 嵌入式文档 vs 引用式设计
# 嵌入式：经常一起被查询的数据
orders_collection = db['orders']
sample_order = {
"_id": "order_12345",
"user_id": "user_67890",
"status": "shipped",
"created_at": datetime.utcnow(),
# 地址采用嵌入式（保留变更历史）
"shipping_address": {
"street": "北京市朝阳区建国路88号",
"city": "北京",
"zip": "100025"
},
# 商品快照采用嵌入式（保留价格变动历史）
"items": [
{"product_id": "p001", "name": "笔记本电脑", "price": 8000, "qty": 1},
{"product_id": "p002", "name": "鼠标", "price": 200, "qty": 2}
],
"total": 8400
}
# Aggregation Pipeline
pipeline = [
# 第 1 步：筛选最近 30 天的订单
{"$match": {
"created_at": {"$gte": datetime(2026, 2, 17)},
"status": {"$in": ["delivered", "shipped"]}
}},
# 第 2 步：展开 items 数组
{"$unwind": "$items"},
# 第 3 步：按商品聚合
{"$group": {
"_id": "$items.product_id",
"product_name": {"$first": "$items.name"},
"total_qty": {"$sum": "$items.qty"},
"total_revenue": {"$sum": {"$multiply": ["$items.price", "$items.qty"]}}
}},
# 第 4 步：按营收排序
{"$sort": {"total_revenue": -1}},
# 第 5 步：取前 10 名
{"$limit": 10},
# 第 6 步：重组结果
{"$project": {
"product_id": "$_id",
"product_name": 1,
"total_qty": 1,
"total_revenue": 1,
"_id": 0
}}
]
top_products = list(orders_collection.aggregate(pipeline))
3.3 Cassandra：宽列（Wide-Column）模型
Cassandra 适合写入密集、地理分布广泛的大规模系统。
-- Cassandra CQL：以查询模式为中心的表设计
-- "先确定查询，再让表去适配它"
-- 为按用户查询时间线设计的表
CREATE TABLE user_timeline (
user_id UUID,
created_at TIMEUUID,
post_id UUID,
content TEXT,
likes COUNTER,
PRIMARY KEY (user_id, created_at)
) WITH CLUSTERING ORDER BY (created_at DESC)
AND compaction = {
'class': 'TimeWindowCompactionStrategy',
'compaction_window_unit': 'DAYS',
'compaction_window_size': 7
};
-- 按标签检索帖子的表（另行反规范化）
CREATE TABLE posts_by_tag (
tag TEXT,
created_at TIMEUUID,
post_id UUID,
user_id UUID,
title TEXT,
PRIMARY KEY (tag, created_at, post_id)
) WITH CLUSTERING ORDER BY (created_at DESC);
4. 向量数据库：AI 时代的核心
4.1 pgvector：在 PostgreSQL 中做向量检索
-- 安装 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
-- 存储嵌入向量的表
CREATE TABLE document_embeddings (
id BIGSERIAL PRIMARY KEY,
content TEXT NOT NULL,
metadata JSONB DEFAULT '{}',
embedding vector(1536), -- OpenAI text-embedding-3-small 维度
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- HNSW 索引（高速 ANN 检索）
CREATE INDEX idx_doc_embeddings_hnsw
ON document_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- 余弦相似度检索
SELECT
id,
content,
metadata,
1 - (embedding <=> '[0.1, 0.2, ...]'::vector) AS similarity
FROM document_embeddings
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 5;
-- 元数据过滤与向量检索结合（Hybrid Search）
SELECT
id,
content,
metadata->>'source' AS source,
1 - (embedding <=> query_embedding) AS similarity
FROM document_embeddings
WHERE
metadata->>'language' = 'ko'
AND metadata->>'category' = 'technical'
AND (embedding <=> query_embedding) < 0.3 -- 阈值过滤
ORDER BY embedding <=> query_embedding
LIMIT 10;
向量距离运算符：
<=>：余弦距离（最适合文本嵌入）<->：L2 欧氏距离（适用于图像特征等）<#>：内积（Inner Product，对归一化向量而言与余弦等价）
4.2 IVFFlat vs HNSW 索引对比
-- IVFFlat：构建速度快、内存效率高
-- 适合大规模数据集的初期构建
CREATE INDEX idx_ivfflat
ON document_embeddings
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100); -- 建议取 sqrt(行数)
-- 调整检索质量
SET ivfflat.probes = 10; -- 检索的簇数量（越高越准确但越慢）
-- HNSW：检索速度快、recall 高
-- 构建时间较长、内存占用较大
CREATE INDEX idx_hnsw
ON document_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
SET hnsw.ef_search = 40; -- 检索时的搜索宽度（越高越准确）
4.3 Pinecone、Weaviate、Milvus 对比
| 特性 | pgvector | Pinecone | Weaviate | Milvus |
|---|---|---|---|---|
| 部署方式 | PostgreSQL 扩展 | SaaS | 自托管 / 云 | 自托管 / 云 |
| 过滤能力 | 完整 SQL 支持 | 元数据过滤 | GraphQL + 过滤 | 丰富的过滤能力 |
| 规模 | 数百万行 | 数亿以上 | 数亿以上 | 数十亿以上 |
| 成本 | 仅 PostgreSQL 成本 | 按用量计费 | 开源免费 | 开源免费 |
| 混合检索 | BM25 + 向量 | 原生支持 | 内置 BM25 + 向量 | 支持丰富 |
| 适用场景 | 已有 PG 基础设施 | 快速原型验证 | 知识图谱 | 超大规模 |
5. 分布式数据库：理论与实战
5.1 CAP 定理
CAP 定理指出，分布式系统无法同时满足以下三点：
- C（Consistency，一致性）：所有节点返回同一份最新数据
- A（Availability，可用性）：每个请求都能收到响应
- P（Partition Tolerance，分区容忍性）：即使发生网络分区也能继续运行
网络分区在现实中随时可能发生，因此实际上只能在 CP 与 AP 之间二选一。
CP 系统（一致性优先）：
- 分区发生时牺牲可用性
- 例如：ZooKeeper、HBase、MongoDB（w:majority）
- 适合：金融交易、库存管理
AP 系统（可用性优先）：
- 分区发生时可能返回非最新数据
- 例如：DynamoDB、Cassandra、CouchDB
- 适合：社交动态、配置信息、DNS
5.2 一致性模型
强一致性（Strong Consistency）
↓ 性能下降，延迟增加
顺序一致性（Sequential Consistency）
↓
因果一致性（Causal Consistency）
↓
最终一致性（Eventual Consistency）
↓ 性能提升，可用性增加
单调读（Monotonic Read）
5.3 分片（Sharding）策略
# Range Sharding：按连续范围切分
def range_shard(user_id: int, num_shards: int = 4) -> int:
shard_size = 250_000_000 # 10 亿用户分成 4 个分片
return min(user_id // shard_size, num_shards - 1)
# Hash Sharding：均匀分布（避免热点）
import hashlib
def hash_shard(key: str, num_shards: int = 8) -> int:
hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
return hash_val % num_shards
# Consistent Hashing：节点增减时最小化再平衡
import bisect
class ConsistentHashRing:
def __init__(self, nodes: list, replicas: int = 150):
self.replicas = replicas
self.ring = {}
self.sorted_keys = []
for node in nodes:
self.add_node(node)
def add_node(self, node: str):
for i in range(self.replicas):
key = self._hash(f"{node}:{i}")
self.ring[key] = node
bisect.insort(self.sorted_keys, key)
def get_node(self, key: str) -> str:
if not self.ring:
return None
hash_val = self._hash(key)
idx = bisect.bisect(self.sorted_keys, hash_val)
if idx == len(self.sorted_keys):
idx = 0
return self.ring[self.sorted_keys[idx]]
def _hash(self, key: str) -> int:
return int(hashlib.md5(key.encode()).hexdigest(), 16)
6. 数据建模
6.1 规范化 vs 反规范化
规范化（到 3NF 为止）：
- 数据冗余最小化
- 防止更新异常
- 写入性能好，读取需要 JOIN
反规范化：
- 优化读取性能
- 允许数据冗余
- 适合 OLAP、数据仓库
-- 规范化的 Schema
CREATE TABLE categories (id SERIAL PRIMARY KEY, name TEXT);
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name TEXT,
category_id INT REFERENCES categories(id)
);
-- 反规范化的 Schema（读取优化）
CREATE TABLE products_denormalized (
id SERIAL PRIMARY KEY,
name TEXT,
category_id INT,
category_name TEXT -- 冗余存储以消除 JOIN
);
6.2 星型模式（数据仓库）
-- 事实表（度量值）
CREATE TABLE fact_sales (
sale_id BIGINT,
date_key INT,
product_key INT,
customer_key INT,
store_key INT,
quantity INT,
unit_price DECIMAL(10,2),
total_amount DECIMAL(10,2)
);
-- 维度表（上下文信息）
CREATE TABLE dim_date (
date_key INT PRIMARY KEY,
full_date DATE,
year INT, quarter INT, month INT, week INT, day_of_week INT
);
CREATE TABLE dim_product (
product_key INT PRIMARY KEY,
product_id TEXT,
name TEXT, category TEXT, brand TEXT, unit_cost DECIMAL(10,2)
);
7. AI 集成：LLM + DB 模式
7.1 用 LangChain + pgvector 搭建 RAG 系统
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_postgres import PGVector
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.schema import Document
# 1. 设置嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 2. 连接 pgvector 向量存储
CONNECTION_STRING = "postgresql+psycopg://user:password@localhost:5432/vectordb"
COLLECTION_NAME = "documents"
vector_store = PGVector(
embeddings=embeddings,
collection_name=COLLECTION_NAME,
connection=CONNECTION_STRING,
use_jsonb=True, # 元数据以 JSONB 存储
)
# 3. 文档切块并存入嵌入
def ingest_documents(file_path: str, metadata: dict):
with open(file_path, 'r', encoding='utf-8') as f:
raw_text = f.read()
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=50,
separators=["\n\n", "\n", ".", "。", " "]
)
chunks = splitter.split_text(raw_text)
documents = [
Document(page_content=chunk, metadata={**metadata, "chunk_index": i})
for i, chunk in enumerate(chunks)
]
ids = vector_store.add_documents(documents)
print(f"Ingested {len(ids)} chunks from {file_path}")
return ids
# 4. 构建 RAG 链
def build_rag_chain(k: int = 5, score_threshold: float = 0.7):
retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"k": k, "score_threshold": score_threshold}
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True,
verbose=True
)
return qa_chain
# 5. 执行查询
rag_chain = build_rag_chain()
result = rag_chain.invoke({"query": "PostgreSQL MVCC 是如何处理并发的？"})
print(result['result'])
for doc in result['source_documents']:
print(f"Source: {doc.metadata.get('source')}, Chunk: {doc.metadata.get('chunk_index')}")
7.2 混合检索：BM25 + 向量检索结合
from sqlalchemy import text
def hybrid_search(
query: str,
query_embedding: list,
k: int = 10,
alpha: float = 0.5 # 0=仅 BM25, 1=仅向量
) -> list:
"""用 RRF（Reciprocal Rank Fusion）合并结果"""
sql = text("""
WITH
vector_search AS (
SELECT id, content, metadata,
ROW_NUMBER() OVER (ORDER BY embedding <=> :embedding) AS rank
FROM document_embeddings
ORDER BY embedding <=> :embedding
LIMIT :k
),
bm25_search AS (
SELECT id, content, metadata,
ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('chinese', content),
plainto_tsquery('chinese', :query)) DESC) AS rank
FROM document_embeddings
WHERE to_tsvector('chinese', content) @@ plainto_tsquery('chinese', :query)
LIMIT :k
),
rrf_scores AS (
SELECT
COALESCE(v.id, b.id) AS id,
COALESCE(v.content, b.content) AS content,
COALESCE(v.metadata, b.metadata) AS metadata,
COALESCE(1.0 / (60 + v.rank), 0) * :alpha +
COALESCE(1.0 / (60 + b.rank), 0) * (1 - :alpha) AS rrf_score
FROM vector_search v
FULL OUTER JOIN bm25_search b ON v.id = b.id
)
SELECT id, content, metadata, rrf_score
FROM rrf_scores
ORDER BY rrf_score DESC
LIMIT :k
""")
with engine.connect() as conn:
results = conn.execute(sql, {
"embedding": str(query_embedding),
"query": query,
"k": k,
"alpha": alpha
})
return [dict(row) for row in results]
7.3 嵌入缓存策略
import hashlib
import json
from typing import Optional
class EmbeddingCache:
def __init__(self, redis_client, ttl: int = 86400 * 7): # 缓存 7 天
self.redis = redis_client
self.ttl = ttl
def _cache_key(self, text: str, model: str) -> str:
content_hash = hashlib.sha256(f"{model}:{text}".encode()).hexdigest()
return f"embedding:{content_hash}"
def get(self, text: str, model: str) -> Optional[list]:
key = self._cache_key(text, model)
cached = self.redis.get(key)
if cached:
return json.loads(cached)
return None
def set(self, text: str, model: str, embedding: list) -> None:
key = self._cache_key(text, model)
self.redis.setex(key, self.ttl, json.dumps(embedding))
def get_or_compute(self, text: str, model: str, compute_fn) -> list:
cached = self.get(text, model)
if cached:
return cached
embedding = compute_fn(text)
self.set(text, model, embedding)
return embedding
测验：实力检验
Q1. B-Tree 索引和 Hash 索引分别适合什么场景？
答案：B-Tree 适合范围查询、排序、LIKE 'prefix%' 检索。Hash 索引只适合等值比较（=）。
说明：B-Tree 把键存储为排序的树结构，因此在 WHERE age > 30、ORDER BY name、BETWEEN 等范围运算上表现出色。Hash 索引通过哈希存储键，因此在 WHERE id = 42 这类精确值匹配上能达到 O(1) 性能，但完全无法用于范围查询或排序。在 PostgreSQL 中，Hash 索引自 v10 加入 WAL 日志支持后才可在生产环境中使用。
Q2. PostgreSQL MVCC（多版本并发控制）如何处理并发？
答案：每个事务开始时都会拍摄一份快照，只能看到该时间点的数据版本。读不会阻塞写，写也不会阻塞读。
说明：PostgreSQL 更新一行数据时不会删除旧行，而是新增一行带有 xmin/xmax 事务 ID 的新版本。每个事务根据自己开始时的 xid 决定能看到哪个版本。旧版本由 VACUUM 进程定期清理（移除 dead tuple）。正因如此，读写冲突被降到最低，从而实现高并发。
Q3. CAP 定理中 CP 系统与 AP 系统的权衡是什么？
答案：CP 在网络分区发生时对部分请求返回错误，以此保证一致性。AP 在分区发生时依然返回响应，但数据可能不是最新的。
说明：CP 系统（ZooKeeper、HBase）在分区状况下拒绝响应或返回错误，以维持一致性，适合金融交易、库存扣减等对准确性要求高的系统。AP 系统（Cassandra、DynamoDB）即使在分区状况下也会尽力返回响应，但节点间数据可能短暂不一致，只保证最终一致性（Eventual Consistency），适合社交媒体动态、通知系统这类可以容忍轻微不一致的场景。
Q4. 向量数据库中 HNSW 算法为何被用于 ANN 检索？
答案：HNSW（Hierarchical Navigable Small World）通过构建多层图，实现了 O(log N) 级别的快速近似最近邻检索，同时兼顾高 recall 与快速查询。
说明：要在数百万个向量中找到精确的最近邻（KNN），需要与所有向量逐一比较，耗时 O(N)。HNSW 分层构建一个每个节点只连接少数邻居的小世界图（small-world graph）。上层用于粗略导航，下层用于精细检索。构建参数 m（每个节点的连接数）与 ef_construction（构建时的搜索宽度）决定索引质量，查询时的 ef_search 则调节 recall 与速度之间的权衡。
Q5. Redis 的 write-through 策略与 write-behind 策略有什么区别？
答案：Write-through 同时更新缓存和 DB，保证一致性，但会产生写入延迟。Write-behind 先更新缓存，DB 写入异步处理，写入速度快，但存在数据丢失风险。
说明：Write-through 在每次写请求中同步更新缓存和 DB，因此数据一致性有保障，但 DB 的写入延迟会直接传导给用户。Write-behind（Write-back）只立即更新缓存，把脏（dirty）数据批量 flush 到 DB。写入吞吐量高、响应速度快，但如果系统在 flush 之前发生故障，缓存中的数据就可能没有反映到 DB 中而丢失。像电商购物车、游戏分数这类写入频繁的场景，适合采用 write-behind。
结语
数据库工程早已超越单纯编写 SQL 的范畴，需要从数据模型设计到索引策略、分布式系统理论，如今还要延伸到向量检索与 AI 集成这样广泛的知识。尤其在 AI 时代，嵌入存储、语义检索、与 LLM 的集成已经成为数据库工程师的核心能力。
希望你能以本指南的内容为基础，将其应用到实际项目中。仅凭从 pgvector 起步的 RAG 系统、Redis 缓存层，以及合理的索引设计，就足以让系统性能提升数十倍。
현재 단락 (1/659)
即使在 AI 时代，数据库工程依然是所有系统的根基。无论 LLM 多么强大，没有安全存储数据、快速检索数据的能力，就无法构建生产系统。尤其随着向量检索和 RAG 系统的出现，数据库工程的角色反而变得更...
작성 글자: 0자원문 글자: 19,182자작성 단락: 0/659