django-vectorstore-indexed-model

可插拔多后端向量数据库的Django数据模型应用。

安装

pip install django-vectorstore-indexed-model

依赖说明

• 数据模型对于向量数据库的操作深度依赖openai-simple-vectorstore。该库支持redis-search、milvus、pgvector、elasticsearch、sqlite-vec等多种后端引擎。相关配置详见该项目文档。
• 依赖版本要求：openai-simple-vectorstore>=0.2.0（filterable_metadata_fields 自定义过滤字段机制自该版本起可用）。
• 后端引擎通过环境变量OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB（别名：VECTOR_DB、VECTOR_DB_TYPE、OPENAI_SIMPLE_VECTORSTORE_BACKEND）选择，默认值为redis。

使用

1. 配置后端引擎

通过环境变量指定向量数据库后端，示例（启动服务前设置）：

# 指定后端引擎（默认 redis）
export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB=redis          # redis-search（默认）
# export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB=milvus      # milvus
# export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB=pgvector    # pgvector
# export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB=elasticsearch
# export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB=sqlite-vec

# 各后端连接配置（以 redis 为例）
export OPENAI_SIMPLE_VECTORSTORE_REDIS_STACK_URL="redis://127.0.0.1:6379"

多后端并行写入

当业务数据（文档、QA 等）发生变更触发索引更新时，可同时向多个后端引擎推送索引。通过环境变量 OPENAI_SIMPLE_VECTORSTORE_VECTOR_DBS（逗号分隔，别名 VECTOR_DBS）指定：

# 同时写入 redis-search 与 milvus 两个后端
export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DBS="redis,milvus"

• 未配置多后端时，退化为单一后端（由 OPENAI_SIMPLE_VECTORSTORE_VECTOR_DB 或其别名决定，默认 redis）。
• 多后端模式下，upsert_index / delete_index 会把同一份数据并行写入 / 删除到每个后端。
• uid 记录：单后端时 vectorstore_uids 保持为扁平列表 ["idx:page"]；多后端时按后端分组为 {"redis": ["idx:page"], "milvus": ["idx:page"]}。

索引默认名称可通过环境变量 VECTORSTORE_INDEX_NAME 配置（默认 default）：

export VECTORSTORE_INDEX_NAME=my_index

2. 定义数据模型

继承 WithVectorStoreIndex（抽象基类），并结合模型自身的启用/删除状态字段实现索引开关。

app/models.py

from typing import List
from django.db import models
from django_vectorstore_indexed_model.models import WithVectorStoreIndex
from django_model_helper.models import WithEnabledStatusFields
from django_model_helper.models import WithDeletedStatusFields


class QA(WithEnabledStatusFields, WithDeletedStatusFields, WithVectorStoreIndex):
    enable_auto_vectorstore_index = False

    kb = models.CharField(max_length=64)
    question = models.CharField(max_length=128)
    answer = models.TextField()

    def get_enable_vectorstore_index_flag(self) -> bool:
        """判断是否需要创建索引（返回 True 建索引，False 删除索引）。"""
        if not self.enabled:
            return False
        if self.deleted:
            return False
        return True

    def get_vectorstore_index_names(self):
        # 返回该记录需要建立的向量库索引名列表；
        # 返回单个元素的 list 表示单索引，返回多个元素表示在多个索引（库）中同时建立。
        return [self.kb]

    def get_vectorstore_index_contents(self) -> List[str]:
        # 向量数据库有索引长度的限制，长文档需要分片；
        # 这里返回分片后的内容列表，每一片会作为一条独立向量写入。
        return [f"问题：{self.question}\n参考答案：{self.answer}"]

3. 触发索引（建 / 删 / 更新）

update_vectorstore_index 会根据 get_enable_vectorstore_index_flag 自动决定写入还是删除：

qa = QA(kb="kb-001", question="你是谁？", answer="我是你的机器人助理！")
qa.save()

# 触发索引：写入向量并记录返回的 uid
qa.update_vectorstore_index(save=True)

# uid 记录：
#   单后端：扁平列表 ['kb-001:<page_id>', ...]
#   多后端：{backend: [uids]}，例如 {'redis': ['kb-001:<page_id>'], 'milvus': [...]}
print(qa.vectorstore_uids)
print(qa.vectorstore_updated)   # True：索引成功

4. 检索

使用工厂 create_vector_store() 获取与配置一致的向量库实例（无需关心后端），配合 Pydantic schema 读取结构化结果：

from openai_simple_vectorstore import create_vector_store
from django_vectorstore_indexed_model.schemas import Document

# 单后端：直接获取与配置一致的实例
vs = create_vector_store()

# 检索并重排
docs = vs.similarity_search_and_rerank(
    "你是谁",
    index_name="kb-001",
    document_schema=Document,
)
for doc in docs:
    print(doc.content, doc.id, doc.app_label, doc.model_name)

# 多后端：遍历每个后端的实例分别检索
for backend, vs in qa.get_vectorstore_instances():
    docs = vs.similarity_search_and_rerank(
        "你是谁",
        index_name="kb-001",
        document_schema=Document,
    )
    print(backend, docs)

注意：检索的入口仍是引擎（create_vector_store() / get_vectorstore_instances()）。 update_vectorstore_index 只是把索引写入配置的所有后端；要查询某个后端，直接对该后端实例调用检索方法即可。

注意：milvus 系列后端的索引名会被自动归一化。milvus collection 名只允许 [A-Za-z0-9_] 且首字符须为字母/下划线，因此写入 milvus 时索引名中的 :、-、. 等会替换为 _（如 faq:kb-uuid → faq_kb_uuid），redis 等其它后端则保留原始名。写入、删除、读取走同一归一化逻辑，同一后端内一致；但检索时若索引名含非法字符，需对 milvus 用归一化后的名称（例见下方示例）。多后端混用时，同一逻辑索引名在 redis 与 milvus 上可能不同，请按后端分别取用。

按需读取 / 删除时，建议按后端路由（uid 归属各自的后端）：

# 单后端：直接对企业主后端实例操作
vs = qa.get_vectorstore_instance()

# 多后端：每个后端的 uid 用各自对应的接口读取 / 删除
for backend, vs in qa.get_vectorstore_instances():
    uids = qa.vectorstore_uids.get(backend, []) if isinstance(
        qa.vectorstore_uids, dict
    ) else qa.vectorstore_uids

    # 按 uid 读取单条元数据（含嵌入向量）
    for uid in uids:
        item = vs.get_item(uid)
        print(backend, item)

    # 删除该后端上的这些 uid
    vs.delete_many(uids)

# 清空某个索引（示例：redis 后端）
vs.flush("kb-001")

如需按业务字段过滤，可自定义继承 Document 的 schema 并填充 metadata：

from typing import Optional
from django_vectorstore_indexed_model.schemas import Document


class QASchema(Document):
    type: Optional[str] = None
    kb: Optional[str] = None

自定义过滤字段

哪些自定义元数据字段可以过滤，由业务侧白名单式声明（默认全部不可过滤）：

• 在模型上覆写 get_vectorstore_filterable_fields()，返回可过滤字段名列表。
• 声明的字段会被当作索引 tag 参与过滤；未声明的自定义元数据字段一律序列化为 blob 存储、不可过滤（过滤未声明字段会报错）。

class QA(WithVectorStoreIndex, models.Model):
    ...
    def get_vectorstore_filterable_fields(self):
        return ["type"]   # 仅 type 这一业务字段可过滤

    def get_vectorstore_index_metadata(self):
        return {
            "app_label": self._meta.app_label,
            "model_name": self._meta.model_name,
            "id": self.id,
            "type": self.get_type(),
        }

声明后即可在检索时按该字段过滤：

vs = qa.get_vectorstore_instance()
docs = vs.similarity_search_and_rerank(
    "你是谁",
    index_name="kb-001",
    document_schema=QASchema,
    filters={"type": "faq"},   # 仅命中 type=faq 的数据
)

说明：filterable_metadata_fields 会透传给 create_vector_store()，对单后端 get_vectorstore_instance() 与多后端 get_vectorstore_instances() 均生效。内置的 kb_id / doc_id / page_id / category 四个字段始终可过滤，无需声明。

多后端并行写入

配置多个后端后，业务数据变更触发 update_vectorstore_index() 时，会同时把同一份索引写入每个后端引擎：

# 同时写入 redis-search 与 milvus
export OPENAI_SIMPLE_VECTORSTORE_VECTOR_DBS="redis,milvus"

qa = QA(kb="kb-001", question="你是谁？", answer="我是你的机器人助理！")
qa.save()
qa.update_vectorstore_index(save=True)

# 多后端时 vectorstore_uids 按后端分组：
print(qa.vectorstore_uids)
# {'redis': ['kb-001:<page_id>'], 'milvus': ['kb-001:<page_id>']}

删除时，delete_index() 会自动按后端名路由，把各后端的 uid 分别删除：

qa.vectorstore_uids = {"redis": ["kb-001:p"], "milvus": ["kb-001:p"]}
qa.delete_index(save=True)      # redis 与 milvus 各自的 uid 都会被删除

说明：

• 多后端配置也可通过 Django settings 提供（OPENAI_SIMPLE_VECTORSTORE_VECTOR_DBS / VECTOR_DBS）。
• 未配置多后端时，一切行为与单一后端完全一致（vectorstore_uids 保持扁平列表），无需改动任何调用代码。
• 检索仍需对各后端实例分别调用（见上文「检索」一节）。

合并多模型知识库到同一索引

当不同的模型（如 QA、Doc）返回相同的 get_vectorstore_index_names() 索引名时， 它们的数据会写入同一个向量索引，检索时即可在一次查询中同时命中多个模型。 每条记录自动写入的 metadata 中带有 app_label 与 model_name，据此可区分命中来自哪个模型。

app/models.py —— 两个模型共用同一索引名 kb：

from django.db import models
from django_vectorstore_indexed_model.models import WithVectorStoreIndex


class QA(WithVectorStoreIndex, models.Model):
    kb = models.CharField(max_length=64)
    question = models.CharField(max_length=128)
    answer = models.TextField()

    def get_enable_vectorstore_index_flag(self) -> bool:
        return True

    def get_vectorstore_index_names(self):
        return [self.kb]                     # 与 Doc 共用同一个索引名

    def get_vectorstore_index_contents(self):
        return [f"问题：{self.question}\n参考答案：{self.answer}"]


class Doc(WithVectorStoreIndex, models.Model):
    kb = models.CharField(max_length=64)     # 与 QA 相同的知识库字段
    title = models.CharField(max_length=128)
    body = models.TextField()

    def get_enable_vectorstore_index_flag(self) -> bool:
        return True

    def get_vectorstore_index_names(self):
        return [self.kb]                     # 与 QA 返回同一个索引名

    def get_vectorstore_index_contents(self):
        return [self.title, self.body]       # 文档分片为多段

写入数据并分别建索引：

qa = QA(kb="kb-001", question="如何重置密码？", answer="在设置页点击忘记密码。")
qa.save(); qa.update_vectorstore_index(save=True)

doc = Doc(kb="kb-001", title="用户手册", body="重置密码：进入设置->账户->修改密码。")
doc.save(); doc.update_vectorstore_index(save=True)

一次检索，同时命中两个模型的数据：

from openai_simple_vectorstore import create_vector_store
from django_vectorstore_indexed_model.schemas import Document

vs = create_vector_store()
docs = vs.similarity_search_and_rerank(
    "如何修改密码",
    index_name="kb-001",          # 指向共享索引
    document_schema=Document,
)

for doc in docs:
    print(
        f"[{doc.model_name}] id={doc.id} {doc.content}"
    )
    # 例如：
    # [doc] id=2 重置密码：进入设置->账户->修改密码。
    # [qa]  id=1 问题：如何重置密码？\n参考答案：在设置页点击忘记密码。

提示：若只需检索某个特定模型的数据，可在 schema 中定义业务字段（如 model_name/type）， 检索后按该字段过滤；或为不同模型定制不同的检索路径。

各数据模型使用独立的索引名

若希望不同模型的数据不混用索引，只需让各模型返回不同的 get_vectorstore_index_names() 即可。 这样每个模型维护自己的向量索引（互不干扰），检索时按各自的索引名分别查询。 最常见的做法是在前缀中带上数据模型标识（如 qa: / doc:），再拼接知识库 id。

app/models.py —— 每个模型使用独立的索引名：

from django.db import models
from django_vectorstore_indexed_model.models import WithVectorStoreIndex


class QA(WithVectorStoreIndex, models.Model):
    kb = models.CharField(max_length=64)
    question = models.CharField(max_length=128)
    answer = models.TextField()

    def get_enable_vectorstore_index_flag(self) -> bool:
        return True

    def get_vectorstore_index_names(self):
        # 返回 List[str]：每个元素是一个独立索引名
        return [f"qa:{self.kb}"]             # 独立索引名：['qa:<kb>']

    def get_vectorstore_index_contents(self):
        return [f"问题：{self.question}\n参考答案：{self.answer}"]


class Doc(WithVectorStoreIndex, models.Model):
    kb = models.CharField(max_length=64)
    title = models.CharField(max_length=128)
    body = models.TextField()

    def get_enable_vectorstore_index_flag(self) -> bool:
        return True

    # 也可返回 list，让同一模型同时写入多个独立索引
    def get_vectorstore_index_names(self):
        return [f"doc:{self.kb}", f"doc-title:{self.kb}"]

    def get_vectorstore_index_contents(self):
        return [self.title, self.body]

分别建索引（各自落到自己的索引中）：

qa = QA(kb="kb-001", question="如何重置密码？", answer="在设置页点击忘记密码。")
qa.save(); qa.update_vectorstore_index(save=True)
# qa 写入索引 qa:kb-001

doc = Doc(kb="kb-001", title="用户手册", body="重置密码：进入设置->账户->修改密码。")
doc.save(); doc.update_vectorstore_index(save=True)
# doc 同时写入 doc:kb-001 与 doc-title:kb-001 两个独立索引

按索引名分别检索：

from openai_simple_vectorstore import create_vector_store
from django_vectorstore_indexed_model.schemas import Document

vs = create_vector_store()

# 只在 QA 的索引中查询
qa_docs = vs.similarity_search_and_rerank(
    "如何重置密码", index_name="qa:kb-001", document_schema=Document,
)
for d in qa_docs:
    print(f"[qa] {d.content}")     # 只包含提问/回答

# 只在 Doc 的索引中查询
doc_docs = vs.similarity_search_and_rerank(
    "如何重置密码", index_name="doc:kb-001", document_schema=Document,
)
for d in doc_docs:
    print(f"[doc] {d.content}")    # 只包含文档内容

提示：多个 index_name 的取舍——

• 多个模型共用一个索引名 → 一次检索同时命中多模型（上一节）。
• 各模型使用独立索引名 → 检索隔离、互不干扰，也便于按模型单独 flush/清理。 可根据业务对检索范围和隔离性的需求灵活选择。

可以重载的方法

快速解决 contents 与 metadatas 一致情况下的单索引或多重索引问题

• get_vectorstore_index_names：返回 List[str] 类型，每个元素为一个索引名；单元素 list 表示单索引，多元素表示在多索引（库）中同时建立。基类实现统一按 list 处理，返回裸字符串也会被自动归一化为单元素 list。
• get_vectorstore_index_contents：返回分片后的内容列表（须实现，否则抛出 NotImplementedError）。
• get_vectorstore_index_metadata：返回单条 metadata 字典；默认含 app_label、model_name、id。
• get_vectorstore_index_metadatas(contents=None)：默认返回每页一份 metadata 的列表（逐页 copy，避免共享同一 dict 引用），content 与 meta 一致时无需重载。

以上方法默认由抽象基类提供实现，可按需在子类中重载。

其他可重载的钩子

• get_enable_vectorstore_index_flag → bool：控制是否启用索引（基类抛出 NotImplementedError，须实现）。
• get_kb_id()：知识库标识，默认 "default"。
• get_doc_id()：文档标识，默认 "default"。
• get_page_id()：分页标识，默认 uuid.uuid4()；多后端写入时同一批页面（及 page_id）在各后端间复用，保证跨后端 page_id 一致。
• get_category()：分类标识，默认 "default"。
• get_vectorstore_instance()：返回向量库实例，默认调用 openai_simple_vectorstore.create_vector_store()，可按需重载。
• get_vectorstore_filterable_fields()：返回本模型可过滤的自定义元数据字段名列表（白名单），默认 []；声明后这些字段会作为索引 tag 参与过滤。
• upsert_index(save=False) / delete_index(save=False)：手动写入 / 删除本次记录的索引。

版本记录

v0.3.0

• 新增：接入多后端向量数据库，摆脱对单一引擎的绑定。

  • 支持通过环境变量切换后端引擎（redis-search / milvus / pgvector / elasticsearch / sqlite-vec 等），默认 redis。
  • 底层统一由 create_vector_store() 按配置创建实例，业务侧无需感知后端差异。

• 新增：支持同时向多个后端写入索引，便于迁移 / 双写 / 容灾。

  • 通过 OPENAI_SIMPLE_VECTORSTORE_VECTOR_DBS 指定多个后端，数据变更时同时向每个引擎推送索引。
  • 写入的 uid 按后端分组记录，删除时按后端路由。

• 新增：支持按业务自定义字段过滤检索。

  • 在模型上覆写 get_vectorstore_filterable_fields() 白名单声明可过滤的自定义元数据字段。
  • 声明的字段作为索引 tag 参与过滤；未声明的自定义字段仅随元数据存储、不可过滤。
  • 单后端 get_vectorstore_instance() 与多后端 get_vectorstore_instances() 均生效。

v0.2.0

• 修正：避免保存数据时触发重复 / 死循环的索引重建。

  • 保存时不再自动重建索引。

• 新增：支持按知识库、文档、类型等字段过滤检索。

  • 匹配最新的 openai-redis-vectorstore，支持字段知识库、文档、类型过滤。

v0.1.1

• 新增：支持 content 与 meta 各不相同的多重索引场景。
  • 添加 WithVectorStoreIndex.get_vectorstore_index_segments 以支持 content 与 meta 各不相同的多重索引。
  • 修正打包时版本号引用问题。

v0.1.0

• 首发：首个可用版本。