企业官网建设流程全解析

Python后端AI专题20：为什么主项目选 pgvector：表结构、距离与 HNSW

KnowFlow 已经能在内存里写向量，但企业项目还要同时管理租户、知识库、文档状态、来源和事务。主项目选择 PostgreSQL + pgvector，不是因为它在所有规模都最快，而是当前阶段“业务过滤与向量检索同库完成”比再引入一套分布式向量系统更容易保证一致性和权限。

续建决策与指纹答案

为了避免只比较一个模型名，新增完整指纹模块：

@dataclass(frozen=True,slots=True)classIndexPipelineVersion:parser_version:strcleaner_version:strchunk_size:intchunk_overlap:intembedding_model:strembedding_dimensions:intdeffingerprint(self)->str:canonical=json.dumps(asdict(self),ensure_ascii=False,sort_keys=True,separators=(",",":"),)returnhashlib.sha256(canonical.encode("utf-8")).hexdigest()

排序键与紧凑 JSON 保证相同配置序列化稳定。测试证明 chunk_size 或维数改变会得到不同的 64 位十六进制哈希：1 passed in 0.08s。

为什么此时不优先选独立向量库

pgvector 的优势：

• tenant_id/knowledge_base_id/status与向量距离能在一条 SQL 中过滤；
• 文档、chunk 与索引任务可用外键和事务约束；
• 团队已有 PostgreSQL 运维经验；
• 当前数据规模不要求独立集群的水平扩展。

当向量达到数亿、需要跨区域水平扩展、专用稀疏向量或复杂分片时，Qdrant/Milvus/Weaviate 等可能更合适。Provider 合同保留了替换空间，但课程不会为了简历关键词同时运行四个数据库。

距离与相似度不要混用

pgvector 的 cosine distance 越小越近；业务 API 常返回 similarity 越大越近。本项目查询后转换：

score=1.0-float(distance_value)

排序必须仍按distance ASC，不能一边写距离、一边按降序排。索引操作符类也要一致：vector_cosine_ops对应余弦距离；如果查询改成内积或 L2，索引和归一化策略都要重新审查。

HNSW 是近似索引，不是魔法

没有 ANN 索引时，数据库要与大量向量逐一计算。HNSW 构建多层邻接图，加速近邻搜索，代价是更多内存、写入成本和近似误差。参数会影响召回与性能，因此必须用评测集验证，不能只看 SQL 延迟。

模型声明：

Index("ix_document_chunks_embedding_hnsw","embedding",postgresql_using="hnsw",postgresql_ops={"embedding":"vector_cosine_ops"},)

完整知识模型模块

from__future__importannotationsimportuuidfrompgvector.sqlalchemyimportVectorfromsqlalchemyimport(BigInteger,CheckConstraint,ForeignKey,Index,Integer,JSON,String,Text,UniqueConstraint,)fromsqlalchemy.ormimportMapped,mapped_columnfromapp.models.baseimportBase,TimestampMixin,UUIDPrimaryKeyMixinclassKnowledgeBase(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="knowledge_bases"__table_args__=(UniqueConstraint("tenant_id","name"),Index("ix_knowledge_bases_tenant_id","tenant_id"),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)name:Mapped[str]=mapped_column(String(160),nullable=False)description:Mapped[str]=mapped_column(Text,default="",nullable=False)classStoredDocument(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="documents"__table_args__=(UniqueConstraint("knowledge_base_id","content_hash"),CheckConstraint("status IN ('pending_upload','uploaded','indexing','ready','failed')",name="valid_document_status",),Index("ix_documents_tenant_id_kb","tenant_id","knowledge_base_id"),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)knowledge_base_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("knowledge_bases.id",ondelete="CASCADE"),nullable=False)filename:Mapped[str]=mapped_column(String(255),nullable=False)content_type:Mapped[str]=mapped_column(String(100),nullable=False)content_hash:Mapped[str]=mapped_column(String(64),nullable=False)object_key:Mapped[str]=mapped_column(String(500),nullable=False,unique=True)size_bytes:Mapped[int]=mapped_column(BigInteger,nullable=False)status:Mapped[str]=mapped_column(String(24),default="uploaded",nullable=False)error_code:Mapped[str|None]=mapped_column(String(80))classDocumentChunk(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="document_chunks"__table_args__=(UniqueConstraint("document_id","ordinal"),Index("ix_document_chunks_tenant_id_kb","tenant_id","knowledge_base_id"),Index("ix_document_chunks_embedding_hnsw","embedding",postgresql_using="hnsw",postgresql_ops={"embedding":"vector_cosine_ops"},),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)knowledge_base_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("knowledge_bases.id",ondelete="CASCADE"),nullable=False)document_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("documents.id",ondelete="CASCADE"),nullable=False)ordinal:Mapped[int]=mapped_column(Integer,nullable=False)text:Mapped[str]=mapped_column(Text,nullable=False)page:Mapped[int|None]=mapped_column(Integer)section:Mapped[str|None]=mapped_column(String(500))start_char:Mapped[int]=mapped_column(Integer,nullable=False)end_char:Mapped[int]=mapped_column(Integer,nullable=False)embedding_model:Mapped[str]=mapped_column(String(160),nullable=False)embedding:Mapped[list[float]]=mapped_column(Vector(384),nullable=False)source_metadata:Mapped[dict[str,object]]=mapped_column(JSON,default=dict,nullable=False)classIndexJob(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="index_jobs"__table_args__=(CheckConstraint("status IN ('queued','parsing','chunking','embedding','ready','failed')",name="valid_index_job_status",),Index("ix_index_jobs_document_id","document_id"),)document_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("documents.id",ondelete="CASCADE"),nullable=False)status:Mapped[str]=mapped_column(String(24),default="queued",nullable=False)next_batch:Mapped[int]=mapped_column(Integer,default=0,nullable=False)total_chunks:Mapped[int]=mapped_column(Integer,default=0,nullable=False)attempts:Mapped[int]=mapped_column(Integer,default=0,nullable=False)error:Mapped[str|None]=mapped_column(Text)

迁移先执行CREATE EXTENSION IF NOT EXISTS vector，再由 metadata 建表；从空库验证了 10 张产品表、vector(384)与 HNSW。模型与真实 PostgreSQL 检查当前结果：

........ [100%] 8 passed in 1.01s

本篇最终完整模块：knowledge.py

前面的代码片段用于解释本次改动；下面是本篇结束时可直接核对和替换的磁盘完整版本。

from__future__importannotationsimportuuidfromdatetimeimportdatetimefrompgvector.sqlalchemyimportVectorfromsqlalchemyimportBigInteger,CheckConstraint,DateTime,ForeignKey,Index,Integer,JSON,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_columnfromapp.models.baseimportBase,TimestampMixin,UUIDPrimaryKeyMixinclassKnowledgeBase(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="knowledge_bases"__table_args__=(UniqueConstraint("tenant_id","name"),Index("ix_knowledge_bases_tenant_id","tenant_id"),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)name:Mapped[str]=mapped_column(String(160),nullable=False)description:Mapped[str]=mapped_column(Text,default="",nullable=False)classStoredDocument(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="documents"__table_args__=(UniqueConstraint("knowledge_base_id","content_hash"),CheckConstraint("status IN ('pending_upload','uploaded','indexing','ready','failed')",name="valid_document_status",),Index("ix_documents_tenant_id_kb","tenant_id","knowledge_base_id"),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)knowledge_base_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("knowledge_bases.id",ondelete="CASCADE"),nullable=False)filename:Mapped[str]=mapped_column(String(255),nullable=False)content_type:Mapped[str]=mapped_column(String(100),nullable=False)content_hash:Mapped[str]=mapped_column(String(64),nullable=False)object_key:Mapped[str]=mapped_column(String(500),nullable=False,unique=True)size_bytes:Mapped[int]=mapped_column(BigInteger,nullable=False)status:Mapped[str]=mapped_column(String(24),default="uploaded",nullable=False)error_code:Mapped[str|None]=mapped_column(String(80))classDocumentChunk(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="document_chunks"__table_args__=(UniqueConstraint("document_id","ordinal"),Index("ix_document_chunks_tenant_id_kb","tenant_id","knowledge_base_id"),Index("ix_document_chunks_embedding_hnsw","embedding",postgresql_using="hnsw",postgresql_ops={"embedding":"vector_cosine_ops"},),)tenant_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("tenants.id",ondelete="CASCADE"),nullable=False)knowledge_base_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("knowledge_bases.id",ondelete="CASCADE"),nullable=False)document_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("documents.id",ondelete="CASCADE"),nullable=False)ordinal:Mapped[int]=mapped_column(Integer,nullable=False)text:Mapped[str]=mapped_column(Text,nullable=False)page:Mapped[int|None]=mapped_column(Integer)section:Mapped[str|None]=mapped_column(String(500))start_char:Mapped[int]=mapped_column(Integer,nullable=False)end_char:Mapped[int]=mapped_column(Integer,nullable=False)embedding_model:Mapped[str]=mapped_column(String(160),nullable=False)embedding:Mapped[list[float]]=mapped_column(Vector(384),nullable=False)source_metadata:Mapped[dict[str,object]]=mapped_column(JSON,default=dict,nullable=False)classIndexJob(UUIDPrimaryKeyMixin,TimestampMixin,Base):__tablename__="index_jobs"__table_args__=(CheckConstraint("status IN ('queued','parsing','chunking','embedding','ready','failed')",name="valid_index_job_status",),CheckConstraint("dispatch_status IN ('pending','claimed','dispatched','failed')",name="valid_index_job_dispatch_status",),Index("ix_index_jobs_document_id","document_id"),Index("ix_index_jobs_dispatch_pending","status","dispatch_status","dispatch_lease_until",),)document_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("documents.id",ondelete="CASCADE"),nullable=False)status:Mapped[str]=mapped_column(String(24),default="queued",nullable=False)next_batch:Mapped[int]=mapped_column(Integer,default=0,nullable=False)total_chunks:Mapped[int]=mapped_column(Integer,default=0,nullable=False)attempts:Mapped[int]=mapped_column(Integer,default=0,nullable=False)error:Mapped[str|None]=mapped_column(Text)dispatch_status:Mapped[str]=mapped_column(String(16),default="pending",server_default="pending",nullable=False)dispatch_token:Mapped[str|None]=mapped_column(String(32))dispatch_attempts:Mapped[int]=mapped_column(Integer,default=0,server_default="0",nullable=False)dispatch_lease_until:Mapped[datetime|None]=mapped_column(DateTime(timezone=True))dispatched_at:Mapped[datetime|None]=mapped_column(DateTime(timezone=True))dispatch_error:Mapped[str|None]=mapped_column(String(80))

本篇练习：手算距离并检查 SQL 顺序

给定查询向量(1,0)，A=(0.8,0.6)，B=(1,0)，C=(-1,0)：手算三者余弦相似度与 distance=1-similarity，写出按distance ASC的顺序。然后在真实 pgvector 测试中插入两个租户各一条极相似记录，断言租户 A 查询永远拿不到租户 B，即使 B 距离更小。

下一篇给出计算和完整集成测试，并实现第一个受权限约束的向量检索接口。