1_项目设计

1. 项目背景

• 本项目希望完成一个面向私有知识库问答场景的 Agent 系统。涉及到：
  • Agent
  • RAG
  • Tool
  • Knowledge Base
  • Conversation Memory

2. 系统设计

2.1 用户角色

• 系统有两类用户，需要考虑用户、权限、会话、消息、知识库

1）普通用户

主要完成：

• 登录
• 聊天
• 知识库问答
• 工具调用
• 查看历史会话
• 删除自己的会话

2）管理员

额外功能：

• 上传知识库文档
• 查看知识库
• 删除文档

2.2 数据流

• 系统围绕两条链路运行

1）用户问答链路

用户问题
   ↓
Agent
   ├── 普通问题 → LLM
   │
   ├── 天气问题 → Weather Tool
   │
   └── 知识问题 → Knowledge Tool
                         ↓
                        RAG
                         ↓
                       检索
                         ↓
                      Evidence
                         ↓
                        LLM
                         ↓
                       Answer

2）文档入库链路

管理员上传文档
        ↓
     文档解析
        ↓
     文档分块
        ↓
      Embedding
        ↓
      知识库存储
        ↓
后续供 RAG 检索

2.3 数据库设计

2.3.1 向量数据库设计

• 向量数据库作为系统里面的一个功能，需要提前设计好，他还会影响后面的数据库设计。

1）文档分块策略

• 采用三级分块策略：

L1 大块
 └── L2 中块
      └── L3 小块

• 检索使用小的 L3，因为小块更容易精准匹配问题。但是最终回答可能需要更完整的上下文，因此要能够根据小块索引到大块（父块），只需要向量化L3，这个决定直接影响后面的数据库设计。
• 举例： 假设三级分块策略是：

L1: chunk_size=2400, overlap=400
L2: chunk_size=1600, overlap=200
L3: chunk_size=800,  overlap=100

我有一个文档，以页为单位进行切分，假设第一页长度是4000，它会被切成两个L1：（0-2400）（2000-4000）。然后再把每个L1切成L2，得到：（0-1600）（1400-2400）（2000-3600）（3400-4000）。然后每个L2再切成L3。这里涉及到两个字段：chunk_id和chunk_idx。这个例子切分后得到的顺序：

L1-0
  L2-0
    L3-0
    L3-1
  L2-1
    L3-2
    L3-3
L1-1
  L2-2
    L3-4
    L3-5

这两个字段的编号逻辑：

L1-0
chunk_id  = VGGT.pdf::p0::l1::0
chunk_idx = 0

L2-0
chunk_id  = VGGT.pdf::p0::l2::0
chunk_idx = 1

L3-0
chunk_id  = VGGT.pdf::p0::l3::0
chunk_idx = 2

L3-1
chunk_id  = VGGT.pdf::p0::l3::1
chunk_idx = 3

L2-1
chunk_id  = VGGT.pdf::p0::l2::1
chunk_idx = 4

L3-2
chunk_id  = VGGT.pdf::p0::l3::2
chunk_idx = 5

L3-3
chunk_id  = VGGT.pdf::p0::l3::3
chunk_idx = 6

L1-1
chunk_id  = VGGT.pdf::p0::l1::1
chunk_idx = 7
...

2）检索策略

• 采用Hybrid Retrieval：

Dense Retrieval
       +
BM25 Retrieval
       ↓
      RRF

3）存储结构

• 使用Milvus作为向量数据库，结合项目背景，设计如下Collection字段：

字段	类型	作用
id	INT64	主键
dense_embedding	FLOAT_VECTOR	语义向量，1024 维
sparse_embedding	SPARSE_FLOAT_VECTOR	BM25 使用的稀疏表示
text	VARCHAR(65535)	L3 Chunk 原始文本
filename	VARCHAR(255)	来源文件名
file_type	VARCHAR(50)	PDF、Word、Excel 等
file_path	VARCHAR(1024)	原始文件路径
page_number	INT64	所在页
chunk_idx	INT64	Chunk 顺序
chunk_id	VARCHAR(512)	当前 Chunk 唯一 ID
parent_chunk_id	VARCHAR(512)	对应 L2 Chunk ID
root_chunk_id	VARCHAR(512)	对应 L1 Chunk ID
chunk_level	INT64	Chunk 层级，这里正常为 3

2.3.2 关系数据库设计

1）用户表 users

字段	类型	约束/作用
id	Integer	主键
username	String(100)	唯一、索引、不能为空
password_hash	String(255)	保存密码 Hash，不保存明文密码
role	String(20)	用户角色，默认 user
created_at	DateTime	创建时间

2）会话表 chat_sessions

字段	类型	约束/作用
id	Integer	主键
user_id	Integer	外键
session_id	String(120)	前端/业务使用的会话 ID
metadata_json	JSON	会话扩展信息
updated_at	DateTime	最近更新时间
created_at	DateTime	创建时间

• 同时建立唯一约束：

(user_id, session_id) UNIQUE

原因是：不同用户可以存在相同的 session_id，但是同一个用户不能存在两个相同 session。

3）聊天消息表 chat_messages

每个 Session 包含多条 Message：

Session
 ├── HumanMessage
 ├── AIMessage
 ├── HumanMessage
 └── AIMessage

字段	类型	约束/作用
id	Integer	主键
session_ref_id	Integer	外键，指向 chat_sessions.id
message_type	String(20)	human、ai、system 等
content	Text	消息正文
timestamp	DateTime	消息时间
rag_trace	JSON，可空	保存该 AI 回答的 RAG 执行轨迹

• 当 Session 被删除时，对应消息也会被删除

ChatSession
   ↓ CASCADE
ChatMessage

• rag_trace 记录：

检索了哪些内容
↓
是否进行了 Grader
↓
是否 Rewrite
↓
最终使用哪些文档

4）父级 Chunk 表 parent_chunks

存 L1/L2

字段	类型	作用
chunk_id	String(512)	主键
text	Text	Parent Chunk 完整文本
filename	String(255)	来源文件
file_type	String(50)	文件类型
file_path	String(1024)	文件路径
page_number	Integer	页码
parent_chunk_id	String(512)	上一级 Chunk ID
root_chunk_id	String(512)	L1 Chunk ID
chunk_level	Integer	1 或 2
chunk_idx	Integer	Chunk 顺序
updated_at	DateTime	更新时间

这就是数据库设计与 Auto-merging RAG 发生联系的地方。

---

2.3.3 Redis 缓存设计

• 频繁读取的数据需要缓存到内存

1）聊天历史

用户每发一条消息，Agent 都可能需要获取当前 Session 的历史：

用户发送消息
 ↓
读取历史消息
 ↓
构造 LLM Context
 ↓
调用 Agent

会产生很多重复数据库访问。因此缓存每个用户的对话记录，key定义为：

supermew:chat_messages:{user_id}:{session_id}

---

2）会话列表

用户打开左侧历史会话列表时，需要查询当前用户的所有session：

chat_sessions:{user_id}

因此定义一个key保存这个用户所有 Session 的摘要：

supermew:chat_sessions:{user_id}

3）Parent Chunk

一次查询查询出来很多的L3，然后为了扩展上下文，需要进一步多次查询L2/L1，如果多个L3都属于同一个L2，那就会把这个L2重复多次查询。因此缓存（key里面的id是L2/L1）：

parent_chunk:{chunk_id}

---

3. 代码实现

3.1 数据存储模块

包括PostgreSQL、Redis 和 Milvus。

3.1.1 文件结构设计

backend/

• env.py：加载项目环境变量。

backend/infra/

• database.py：PostgreSQL 连接与初始化。
  • engine：数据库引擎。
  • SessionLocal：数据库 Session 工厂。
  • Base：ORM 基类。
  • init_db()：初始化数据表。
• cache.py：Redis 缓存封装。
  • RedisCache：统一管理缓存读取、写入和删除。

backend/db/

• models.py：定义 PostgreSQL 数据模型。
  • User：用户信息。
  • ChatSession：会话信息。
  • ChatMessage：聊天消息及 rag_trace。
  • ParentChunk：保存 L1/L2 父级文本块。

backend/indexing/

• milvus_client.py：Milvus 向量数据库封装。
  • MilvusSettings：Milvus 配置。
  • MilvusStore：L3 Chunk 的存储与检索，包含 Dense、BM25 和 Hybrid Retrieval。
• parent_chunk_store.py：L1/L2 Parent Chunk 存储层。
  • ParentChunkStore：负责 PostgreSQL 持久化及 Redis 缓存。

3.1.2 模块实现

backend/

1）env.py

"""从项目根目录加载 .env；由应用入口或独立脚本在 import 其它 backend 模块前调用一次。"""
from pathlib import Path

from dotenv import load_dotenv

PROJECT_ROOT = Path(__file__).resolve().parent.parent
_LOADED = False


def load_env() -> None:
    global _LOADED
    if _LOADED:
        return
    load_dotenv(PROJECT_ROOT / ".env")
    _LOADED = True

backend/infra/

1）database.py

import os
from sqlalchemy import create_engine, event
from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg2://postgres:postgres@localhost:5432/langchain_app",
)

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)


import re
import unicodedata

# 编译非打印 C0/C1 控制字符的正则（保留常规排版字：\t, \n, \r）
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
# 编译零宽字符和不可见格式化控制字符
_INVISIBLE_CHAR_RE = re.compile(r"[\u200b-\u200d\ufeff\u200f\u202a-\u202e]")


def _clean_nul_chars(val):
    """
    递归清除 Python 数据结构中所有可能破坏底层关系驱动、造成 JSON 解析异常或乱码的非标字符。
    1. 规范化 (Normalization)：自动规范化为标准 Unicode NFC 格式。
    2. 清洗不可见及非排版字：抹除零宽空字符、格式强制方向符、非排版控制字符及 PUA 私有使用区方块。
    3. 收敛并擦除代理项：利用 utf-8 ignore 安全剥离任何残损的、孤立的 UTF-16 代理。
    """
    if isinstance(val, str):
        # 1. 规范化为 NFC
        val = unicodedata.normalize("NFC", val)
        # 2. 擦除零宽、不可见控制和 BOM 字符
        val = _INVISIBLE_CHAR_RE.sub("", val)
        # 3. 擦除非排版 C0/C1 字符及 PUA 私有区字符
        val = _CONTROL_CHAR_RE.sub("", val)
        val = re.sub(r"[\ue000-\uf8ff]", "", val)
        # 4. 转换并确保 100% 合规的 UTF-8 (等同于 PG 的 utf8mb4 标准)
        try:
            return val.encode("utf-8", "ignore").decode("utf-8", "ignore")
        except Exception:
            chars = []
            for char in val:
                if 0xD800 <= ord(char) <= 0xDFFF:
                    continue
                chars.append(char)
            return "".join(chars)
    elif isinstance(val, dict):
        return {k: _clean_nul_chars(v) for k, v in val.items()}
    elif isinstance(val, list):
        return [_clean_nul_chars(v) for v in val]
    elif isinstance(val, tuple):
        return tuple(_clean_nul_chars(v) for v in val)
    return val


@event.listens_for(engine, "before_cursor_execute", retval=True)
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    """
    SQLAlchemy 全局引擎监听器：拦截所有底层的 SQL 执行和传入绑定的参数。
    在所有底层驱动执行前自动过滤、擦除所有参数中的 \x00 字符，
    彻底、优雅地解决 PostgreSQL 不允许 VARCHAR/TEXT 写入 NUL (0x00) 字节的异常，
    使业务层不需要在各处手动书写 replace() 代码。
    """
    if parameters is not None:
        if isinstance(parameters, dict):
            for k, v in list(parameters.items()):
                parameters[k] = _clean_nul_chars(v)
        elif isinstance(parameters, list):
            for i, v in enumerate(parameters):
                parameters[i] = _clean_nul_chars(v)
        elif isinstance(parameters, tuple):
            parameters = tuple(_clean_nul_chars(v) for v in parameters)
    return statement, parameters


SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
Base = declarative_base()


def init_db() -> None:
    # Delayed import to avoid circular dependency.
    import backend.db.models  # noqa: F401

    Base.metadata.create_all(bind=engine)

• 先根据 DATABASE_URL 连接 PostgreSQL，再提供 SessionLocal 供后续代码执行数据库读写，并用 Base 统一管理 db/models.py 中定义的数据表；当调用 init_db() 时，会加载这些表定义并在 PostgreSQL 中创建尚不存在的表。

2）cache.py

import json
import os
from typing import Any, Optional

import redis


class RedisCache:
    def __init__(self):
        # 用于配置 Redis 连接
        self.redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
        # 给所有key添加前缀
        self.key_prefix = os.getenv("REDIS_KEY_PREFIX", "supermew")
        # 默认缓存过期时间（秒）
        self.default_ttl = int(os.getenv("REDIS_CACHE_TTL_SECONDS", "300"))
        self._client = None

    def _get_client(self):
        # 初始胡类不会连接redis，第一次真正读写时通过本函数建立连接
        if self._client is None:
            self._client = redis.Redis.from_url(self.redis_url, decode_responses=True)
        return self._client

    def _key(self, key: str) -> str:
        # 给所有key加上前缀
        return f"{self.key_prefix}:{key}"

    def get_json(self, key: str) -> Optional[Any]:
        # 读取json后再还原成 Python 对象
        try:
            value = self._get_client().get(self._key(key))
            if not value:
                return None
            return json.loads(value)
        except Exception:
            return None

    def set_json(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        # 把 Python 对象转成 JSON 后写入 Redis
        try:
            payload = json.dumps(value, ensure_ascii=False)
            self._get_client().setex(self._key(key), ttl or self.default_ttl, payload)
        except Exception:
            return

    def delete(self, key: str) -> None:
        # 删除指定的 key
        try:
            self._get_client().delete(self._key(key))
        except Exception:
            return

    def delete_pattern(self, pattern: str) -> None:
        # 按规则批量删除
        try:
            full_pattern = self._key(pattern)
            keys = self._get_client().keys(full_pattern)
            if keys:
                self._get_client().delete(*keys)
        except Exception:
            return

# 创建了一个全局缓存对象，其他模块直接 from backend.infra.cache import cache 使用即可
cache = RedisCache()

• 统一管理 Redis 缓存。需要缓存数据时，把数据转成 JSON 存进去；读取时再取出来恢复；支持删除单个缓存和批量删除缓存。为了避免不同功能的缓存互相冲突，所有 key 前面统一加 supermew:，缓存默认 5 分钟过期。如果 Redis 出问题，直接忽略，不影响主程序继续运行。

backend/db/

1）models.py

from datetime import datetime

from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from backend.infra.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(100), unique=True, index=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[str] = mapped_column(String(20), default="user", nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)

    sessions = relationship("ChatSession", back_populates="user", cascade="all, delete-orphan")


class ChatSession(Base):
    __tablename__ = "chat_sessions"
    __table_args__ = (UniqueConstraint("user_id", "session_id", name="uq_user_session"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
    session_id: Mapped[str] = mapped_column(String(120), nullable=False, index=True)
    metadata_json: Mapped[dict] = mapped_column(JSON, default=dict, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)

    user = relationship("User", back_populates="sessions")
    messages = relationship("ChatMessage", back_populates="session", cascade="all, delete-orphan")


class ChatMessage(Base):
    __tablename__ = "chat_messages"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    session_ref_id: Mapped[int] = mapped_column(ForeignKey("chat_sessions.id", ondelete="CASCADE"), nullable=False, index=True)
    message_type: Mapped[str] = mapped_column(String(20), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    timestamp: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
    rag_trace: Mapped[dict | None] = mapped_column(JSON, nullable=True)

    session = relationship("ChatSession", back_populates="messages")


class ParentChunk(Base):
    __tablename__ = "parent_chunks"

    chunk_id: Mapped[str] = mapped_column(String(512), primary_key=True)
    text: Mapped[str] = mapped_column(Text, nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
    file_type: Mapped[str] = mapped_column(String(50), default="", nullable=False)
    file_path: Mapped[str] = mapped_column(String(1024), default="", nullable=False)
    page_number: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    parent_chunk_id: Mapped[str] = mapped_column(String(512), default="", nullable=False)
    root_chunk_id: Mapped[str] = mapped_column(String(512), default="", nullable=False)
    chunk_level: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    chunk_idx: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)

• 定义关系数据库表结构

backend/indexing/

1）milvus_client.py

Milvus 的统一操作层。后面的代码不直接调用 Milvus，而是通过这个文件完成建库、写入、查询和检索。这个文件需要提供如下功能：

• 保存 Milvus 配置
• 读取配置环境
• 提供一个连接Milvus的客户端，用的时候连接，用完关闭
• 条件过滤处理
• 定义Milvus操作：
  • 操作执行函数用于统一执行
  • Collection 初始化与创建
  • Collection 是否存在检查
  • L3 Chunk 数据写入
  • 条件查询
  • 查询全部数据
  • 根据 chunk_id 查询指定 Chunk
  • Dense 向量检索
  • Dense + BM25 + RRF 混合检索
  • 删除指定数据
  • 删除整个 Collection

保存 Milvus 配置

• 读取环境配置，创建实例

QUERY_MAX_LIMIT = 16384 # 一次查询最多返回多少条数据
T = TypeVar("T") #给后面通用函数做类型标注，本身不参与业务逻辑。

# Milvus 配置
@dataclass(frozen=True) #让这个配置类更方便定义，并且创建后不能随意修改
class MilvusSettings:
    host: str
    port: str
    collection_name: str
    uri: str
    timeout: float

    @classmethod
    def from_env(cls) -> MilvusSettings:
        host = os.getenv("MILVUS_HOST", "localhost")
        port = os.getenv("MILVUS_PORT", "19530")
        collection = os.getenv("MILVUS_COLLECTION", "embeddings_collection")
        timeout = float(os.getenv("MILVUS_TIMEOUT", "30"))
        return cls( # 返回一个实例
            host=host,
            port=port,
            collection_name=collection,
            uri=f"http://{host}:{port}",
            timeout=timeout,
        )

连接Milvus的客户端

@contextmanager
def milvus_client_session(settings: MilvusSettings | None = None) -> Iterator[MilvusClient]:
    """一次 RPC 会话：创建连接，用完后关闭，不缓存 gRPC channel。"""
    cfg = settings or MilvusSettings.from_env()
    client = MilvusClient(uri=cfg.uri, timeout=cfg.timeout)
    try:
        yield client
    finally:
        client.close()

条件过滤处理

def _normalize_filter(filter_expr: str) -> str:
    return filter_expr.strip() if filter_expr.strip() else "id >= 0"

定义Milvus操作

"""Milvus 访问层：无状态 Store + 短生命周期 gRPC 连接（避免长期持有失效 channel）。"""
from __future__ import annotations

import os
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Callable, Iterator, TypeVar

from pymilvus import AnnSearchRequest, DataType, MilvusClient, RRFRanker, Function, FunctionType

QUERY_MAX_LIMIT = 16384 # 一次查询最多返回多少条数据
T = TypeVar("T") #给后面通用函数做类型标注，本身不参与业务逻辑。

# Milvus 配置
@dataclass(frozen=True) #让这个配置类更方便定义，并且创建后不能随意修改
class MilvusSettings:
    host: str
    port: str
    collection_name: str
    uri: str
    timeout: float

    @classmethod
    def from_env(cls) -> MilvusSettings:
        host = os.getenv("MILVUS_HOST", "localhost")
        port = os.getenv("MILVUS_PORT", "19530")
        collection = os.getenv("MILVUS_COLLECTION", "embeddings_collection")
        timeout = float(os.getenv("MILVUS_TIMEOUT", "30"))
        return cls( # 返回一个实例
            host=host,
            port=port,
            collection_name=collection,
            uri=f"http://{host}:{port}",
            timeout=timeout,
        )


@contextmanager
def milvus_client_session(settings: MilvusSettings | None = None) -> Iterator[MilvusClient]:
    """一次 RPC 会话：创建连接，用完后关闭，不缓存 gRPC channel。"""
    cfg = settings or MilvusSettings.from_env()
    client = MilvusClient(uri=cfg.uri, timeout=cfg.timeout)
    try:
        yield client
    finally:
        client.close()


def _normalize_filter(filter_expr: str) -> str:
    return filter_expr.strip() if filter_expr.strip() else "id >= 0"


class MilvusStore:
    """Milvus 集合读写；本身不持有连接，所有 IO 经 milvus_client_session。"""

    def __init__(self, settings: MilvusSettings | None = None):
        self._settings = settings or MilvusSettings.from_env()

    @property
    def collection_name(self) -> str:
        # 当前类对应一个表，取出表名
        return self._settings.collection_name

    def _run(self, operation: Callable[[MilvusClient], T]) -> T:
        # 统一操作函数
        with milvus_client_session(self._settings) as client: #该函数返回一个数据库连接对象
            return operation(client)

    @contextmanager
    def session(self) -> Iterator[MilvusClient]:
        """同一业务流（如整次上传）内复用一条连接，用毕即关。"""
        # 把这个函数又包装一次，后续with调用就不需要传递参数了，在这个with下面可以公用一个连接多次执行
        with milvus_client_session(self._settings) as client:
            yield client

    @staticmethod
    def ensure_collection(client: MilvusClient, collection_name: str, dense_dim: int) -> None:
        # 创建 Collection
        if client.has_collection(collection_name):
            return

        schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
        schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True)
        schema.add_field("dense_embedding", DataType.FLOAT_VECTOR, dim=dense_dim)
        schema.add_field("sparse_embedding", DataType.SPARSE_FLOAT_VECTOR)
        schema.add_field(
            "text",
            DataType.VARCHAR,
            max_length=65535,
            enable_analyzer=True,
            analyzer_params={"type": "chinese"},
            enable_match=True,
        )
        schema.add_field("filename", DataType.VARCHAR, max_length=255)
        schema.add_field("file_type", DataType.VARCHAR, max_length=50)
        schema.add_field("file_path", DataType.VARCHAR, max_length=1024)
        schema.add_field("page_number", DataType.INT64)
        schema.add_field("chunk_idx", DataType.INT64)
        schema.add_field("chunk_id", DataType.VARCHAR, max_length=512)
        schema.add_field("parent_chunk_id", DataType.VARCHAR, max_length=512)
        schema.add_field("root_chunk_id", DataType.VARCHAR, max_length=512)
        schema.add_field("chunk_level", DataType.INT64)

        bm25_function = Function( # 写入text字段后，计算BM25写道sparseembedding字段
            name="text_bm25_emb",
            function_type=FunctionType.BM25,
            input_field_names=["text"],
            output_field_names=["sparse_embedding"],
        )
        schema.add_function(bm25_function)

        index_params = client.prepare_index_params()
        # 创建索引，给dense_embedding字段创建HNSW索引，给sparse_embedding字段创建SPARSE_INVERTED_INDEX索引
        index_params.add_index(
            field_name="dense_embedding",
            index_type="HNSW",
            metric_type="IP", #内积衡量相似度
            params={"M": 16, "efConstruction": 256},
        )
        index_params.add_index(
            field_name="sparse_embedding",
            index_type="SPARSE_INVERTED_INDEX", # 稀疏倒排索引
            metric_type="BM25", # 按照 BM25 分数排序
            params={"drop_ratio_build": 0.2},
        )
        client.create_collection(
            collection_name=collection_name,
            schema=schema,
            index_params=index_params,
        )

    def init_collection(self, dense_dim: int | None = None) -> None:
        # 外部真正调用的初始化入口，创建Collection
        if dense_dim is None:
            dense_dim = int(os.getenv("DENSE_EMBEDDING_DIM", "1024"))

        def _init(client: MilvusClient) -> None:
            self.ensure_collection(client, self.collection_name, dense_dim)

        self._run(_init)

    def insert(self, data: list[dict]):
        # 插入数据
        return self._run(lambda client: client.insert(self.collection_name, data))

    def query(
        self,
        filter_expr: str = "",
        output_fields: list[str] | None = None,
        limit: int = 10000,
        offset: int = 0,
    ):
        # 普通字段查询，根据关键词输出 对应字段
        expr = _normalize_filter(filter_expr)
        fields = output_fields or ["filename", "file_type"]

        def _query(client: MilvusClient):
            return client.query(
                collection_name=self.collection_name,
                filter=expr,
                output_fields=fields,
                limit=min(limit, QUERY_MAX_LIMIT),
                offset=offset,
            )

        return self._run(_query)

    def query_all(self, filter_expr: str = "", output_fields: list[str] | None = None) -> list:
        """分页拉取；单次 session 内完成，避免每页新建连接。"""
        fields = output_fields or ["filename", "file_type"]
        expr = _normalize_filter(filter_expr)

        def _query_all(client: MilvusClient) -> list:
            out: list = []
            offset = 0
            while True:
                batch = client.query(
                    collection_name=self.collection_name,
                    filter=expr,
                    output_fields=fields,
                    limit=QUERY_MAX_LIMIT,
                    offset=offset,
                )
                if not batch:
                    break
                out.extend(batch)
                if len(batch) < QUERY_MAX_LIMIT:
                    break
                offset += len(batch)
            return out

        return self._run(_query_all)

    def get_chunks_by_ids(self, chunk_ids: list[str]) -> list[dict]:
        # 按chunkid查询
        ids = [item for item in chunk_ids if item]
        if not ids:
            return []
        quoted_ids = ", ".join(f'"{item}"' for item in ids)
        return self.query(
            filter_expr=f"chunk_id in [{quoted_ids}]",
            output_fields=[
                "text",
                "filename",
                "file_type",
                "page_number",
                "chunk_id",
                "parent_chunk_id",
                "root_chunk_id",
                "chunk_level",
                "chunk_idx",
            ],
            limit=len(ids),
        )

    def hybrid_retrieve(
        self,
        dense_embedding: list[float],
        query: str,
        top_k: int = 5,
        rrf_k: int = 60,
        filter_expr: str = "",
    ) -> list[dict]:
        output_fields = [
            "text",
            "filename",
            "file_type",
            "page_number",
            "chunk_id",
            "parent_chunk_id",
            "root_chunk_id",
            "chunk_level",
            "chunk_idx",
        ]
        dense_search = AnnSearchRequest( #定义一个向量检索请求，还没执行
            data=[dense_embedding],
            anns_field="dense_embedding",
            param={"metric_type": "IP", "params": {"ef": 64}},
            limit=top_k * 2,
            expr=filter_expr,
        )
        sparse_search = AnnSearchRequest(
            data=[query],
            anns_field="sparse_embedding",
            param={"metric_type": "BM25", "params": {"drop_ratio_search": 0.2}},
            limit=top_k * 2,
            expr=filter_expr,
        )
        reranker = RRFRanker(k=rrf_k)

        def _search(client: MilvusClient):
            return client.hybrid_search(
                collection_name=self.collection_name,
                reqs=[dense_search, sparse_search],
                ranker=reranker,
                limit=top_k,
                output_fields=output_fields,
            )

        results = self._run(_search)
        formatted_results = []
        for hits in results:
            for hit in hits:
                formatted_results.append({
                    "id": hit.get("id"),
                    "text": hit.get("text", ""),
                    "filename": hit.get("filename", ""),
                    "file_type": hit.get("file_type", ""),
                    "page_number": hit.get("page_number", 0),
                    "chunk_id": hit.get("chunk_id", ""),
                    "parent_chunk_id": hit.get("parent_chunk_id", ""),
                    "root_chunk_id": hit.get("root_chunk_id", ""),
                    "chunk_level": hit.get("chunk_level", 0),
                    "chunk_idx": hit.get("chunk_idx", 0),
                    "score": hit.get("distance", 0.0),
                })
        return formatted_results

    def dense_retrieve(
        self,
        dense_embedding: list[float],
        top_k: int = 5,
        filter_expr: str = "",
    ) -> list[dict]:
        # 向量检索
        def _search(client: MilvusClient):
            return client.search(
                collection_name=self.collection_name,
                data=[dense_embedding],
                anns_field="dense_embedding",
                search_params={"metric_type": "IP", "params": {"ef": 64}},
                limit=top_k,
                output_fields=[
                    "text",
                    "filename",
                    "file_type",
                    "page_number",
                    "chunk_id",
                    "parent_chunk_id",
                    "root_chunk_id",
                    "chunk_level",
                    "chunk_idx",
                ],
                filter=filter_expr,
            )

        results = self._run(_search)
        formatted_results = []
        for hits in results:
            for hit in hits:
                formatted_results.append({
                    "id": hit.get("id"),
                    "text": hit.get("entity", {}).get("text", ""),
                    "filename": hit.get("entity", {}).get("filename", ""),
                    "file_type": hit.get("entity", {}).get("file_type", ""),
                    "page_number": hit.get("entity", {}).get("page_number", 0),
                    "chunk_id": hit.get("entity", {}).get("chunk_id", ""),
                    "parent_chunk_id": hit.get("entity", {}).get("parent_chunk_id", ""),
                    "root_chunk_id": hit.get("entity", {}).get("root_chunk_id", ""),
                    "chunk_level": hit.get("entity", {}).get("chunk_level", 0),
                    "chunk_idx": hit.get("entity", {}).get("chunk_idx", 0),
                    "score": hit.get("distance", 0.0),
                })
        return formatted_results

    def delete(self, filter_expr: str):
        # 删除指定条目
        return self._run(
            lambda client: client.delete(collection_name=self.collection_name, filter=filter_expr)
        )

    def has_collection(self) -> bool:
        # 检查表是否存在
        return self._run(lambda client: client.has_collection(self.collection_name))

    def drop_collection(self) -> None:
        # 删除表
        def _drop(client: MilvusClient) -> None:
            if client.has_collection(self.collection_name):
                client.drop_collection(self.collection_name)

        self._run(_drop)


_store: MilvusStore | None = None # 这个变量要么是none要么是milvusstore， 初始化为none

# 获取 MilvusStore 实例，全局变量
def get_milvus_store() -> MilvusStore:
    global _store
    if _store is None:
        _store = MilvusStore()
    return _store

2）parent_chunk_store.py

L1/L2 Parent Chunk 的统一存储层。后面的代码不直接分别操作 PostgreSQL 和 Redis，而是通过这个文件统一完成父级 Chunk 的写入、读取和删除。这个文件主要提供：

1. 数据转换功能
   • 数据库字段转成字典
2. 增删改功能：
   • 增/改 L1/L2 Chunk，同步 Redis
   • 删 L1/L2 Chunk，同步 Redis
   • 查询父级 Chunk，同步 Redis
3. Redis功能：
   • 生成Redis Key

"""父级分块文档存储（用于 Auto-merging Retriever）"""
from datetime import datetime
from typing import List

from backend.infra.cache import cache
from backend.infra.database import SessionLocal
from backend.db.models import ParentChunk


class ParentChunkStore:
    """基于 PostgreSQL + Redis 的父级分块存储。"""

    @staticmethod
    def _to_dict(item: ParentChunk) -> dict:
        # 把 PostgreSQL 查询出来的 `ParentChunk` 数据库对象转换成普通 Python 字典
        return {
            "text": item.text,
            "filename": item.filename,
            "file_type": item.file_type,
            "file_path": item.file_path,
            "page_number": item.page_number,
            "chunk_id": item.chunk_id,
            "parent_chunk_id": item.parent_chunk_id,
            "root_chunk_id": item.root_chunk_id,
            "chunk_level": item.chunk_level,
            "chunk_idx": item.chunk_idx,
        }

    @staticmethod
    def _cache_key(chunk_id: str) -> str:
        # 根据 `chunk_id` 生成 Redis 缓存 Key，这里还没加上前缀
        return f"parent_chunk:{chunk_id}"

    def upsert_documents(self, docs: List[dict]) -> int:
        """写入/更新父级分块，返回写入条数。"""
        if not docs:
            return 0

        db = SessionLocal()
        upserted = 0
        try:
            for doc in docs:
                chunk_id = (doc.get("chunk_id") or "").strip()
                if not chunk_id:
                    continue
                # 先根据 chunk_id 查 是否存在
                record = db.query(ParentChunk).filter(ParentChunk.chunk_id == chunk_id).first()
                payload = {
                    "text": doc.get("text", ""),
                    "filename": doc.get("filename", ""),
                    "file_type": doc.get("file_type", ""),
                    "file_path": doc.get("file_path", ""),
                    "page_number": int(doc.get("page_number", 0) or 0),
                    "parent_chunk_id": doc.get("parent_chunk_id", ""),
                    "root_chunk_id": doc.get("root_chunk_id", ""),
                    "chunk_level": int(doc.get("chunk_level", 0) or 0),
                    "chunk_idx": int(doc.get("chunk_idx", 0) or 0),
                    "updated_at": datetime.utcnow(),
                }
                cache_payload = {
                    "chunk_id": chunk_id,
                    "text": payload["text"],
                    "filename": payload["filename"],
                    "file_type": payload["file_type"],
                    "file_path": payload["file_path"],
                    "page_number": payload["page_number"],
                    "parent_chunk_id": payload["parent_chunk_id"],
                    "root_chunk_id": payload["root_chunk_id"],
                    "chunk_level": payload["chunk_level"],
                    "chunk_idx": payload["chunk_idx"],
                }
                if record: # 如果记录存在，更新它
                    for key, value in payload.items():
                        setattr(record, key, value)
                else:
                    db.add(ParentChunk(chunk_id=chunk_id, **payload))

                cache.set_json(self._cache_key(chunk_id), cache_payload)
                upserted += 1

            db.commit()
        finally:
            db.close()

        return upserted

    def get_documents_by_ids(self, chunk_ids: List[str]) -> List[dict]:
        # 根据 chunk_ids 查询
        if not chunk_ids:
            return []

        ordered_results = {}
        missing_ids = []
        # 先从redis查询
        for chunk_id in chunk_ids:
            key = (chunk_id or "").strip()
            if not key:
                continue
            cached = cache.get_json(self._cache_key(key))
            if cached:
                ordered_results[key] = cached
            else:
                missing_ids.append(key)

        # redis未命中的部分
        if missing_ids:
            db = SessionLocal()
            try:
                rows = db.query(ParentChunk).filter(ParentChunk.chunk_id.in_(missing_ids)).all()
                for row in rows:
                    payload = self._to_dict(row)
                    ordered_results[row.chunk_id] = payload
                    # 未命中的查出来后写入redis
                    cache.set_json(self._cache_key(row.chunk_id), payload)
            finally:
                db.close()

        return [ordered_results[item] for item in chunk_ids if item in ordered_results]

    def delete_by_filename(self, filename: str) -> int:
        """按文件名删除父级分块，返回删除条数。"""
        if not filename:
            return 0

        db = SessionLocal()
        try:
            rows = db.query(ParentChunk).filter(ParentChunk.filename == filename).all()
            chunk_ids = [row.chunk_id for row in rows]
            deleted = len(chunk_ids)
            if deleted > 0:
                db.query(ParentChunk).filter(ParentChunk.filename == filename).delete(synchronize_session=False)
                db.commit()
                for chunk_id in chunk_ids:
                    cache.delete(self._cache_key(chunk_id))
            return deleted
        finally:
            db.close()

3.1.3 模块测试

test/1_test_storage.py

from dataclasses import replace

# 一定要先加载环境变量
from backend.env import load_env

load_env()

from sqlalchemy import inspect

from backend.db.models import User, ChatSession, ChatMessage
from backend.infra.database import SessionLocal, engine, init_db
from backend.infra.cache import cache
from backend.indexing.milvus_client import MilvusSettings, MilvusStore
from backend.indexing.parent_chunk_store import ParentChunkStore


def test_postgresql():
    print("\n=== 1. 测试 PostgreSQL ===")

    # 创建尚不存在的数据表
    init_db()

    tables = set(inspect(engine).get_table_names())
    required_tables = {
        "users",
        "chat_sessions",
        "chat_messages",
        "parent_chunks",
    }

    assert required_tables.issubset(tables)

    print("数据表创建成功：", required_tables)

    # 测试 User → Session → Message
    db = SessionLocal()

    try:
        # 清理可能残留的测试用户
        old_user = (
            db.query(User)
            .filter(User.username == "storage_test_user")
            .first()
        )
        if old_user:
            db.delete(old_user)
            db.commit()

        user = User(
            username="storage_test_user",
            password_hash="test_hash",
            role="user",
        )
        db.add(user)
        db.flush()

        session = ChatSession(
            user_id=user.id,
            session_id="storage_test_session",
            metadata_json={"title": "测试会话"},
        )
        db.add(session)
        db.flush()

        message = ChatMessage(
            session_ref_id=session.id,
            message_type="human",
            content="你好",
        )
        db.add(message)
        db.commit()

        saved_message = (
            db.query(ChatMessage)
            .filter(ChatMessage.session_ref_id == session.id)
            .first()
        )

        assert saved_message is not None
        assert saved_message.content == "你好"

        print("User → ChatSession → ChatMessage 写入成功")

        # 清理测试数据
        db.delete(user)
        db.commit()

    finally:
        db.close()


def test_redis():
    print("\n=== 2. 测试 Redis ===")

    key = "storage_test"

    cache.set_json(
        key,
        {
            "name": "SuperMew",
            "status": "ok",
        },
        ttl=60,
    )

    result = cache.get_json(key)

    assert result is not None
    assert result["name"] == "SuperMew"

    print("Redis 写入/读取成功：", result)

    cache.delete(key)

    assert cache.get_json(key) is None

    print("Redis 删除成功")


def test_milvus():
    print("\n=== 3. 测试 Milvus ===")

    # 不污染正式 Collection，专门创建测试 Collection
    config = MilvusSettings.from_env()

    test_config = replace(
        config,
        collection_name=f"{config.collection_name}_storage_test",
    )

    store = MilvusStore(test_config)

    # 如果上次测试残留则先删除
    if store.has_collection():
        store.drop_collection()

    try:
        # 1. 创建 Collection
        store.init_collection(dense_dim=1024)

        assert store.has_collection()

        print("Collection 创建成功：", store.collection_name)

        # 两个简单的 1024 维测试向量
        vector1 = [0.0] * 1024
        vector1[0] = 1.0

        vector2 = [0.0] * 1024
        vector2[1] = 1.0

        # 2. 写入两个 L3 Chunk
        store.insert([
            {
                "dense_embedding": vector1,
                "text": "VGGT 可以预测相机参数、深度和三维点图。",
                "filename": "storage_test.pdf",
                "file_type": "pdf",
                "file_path": "/tmp/storage_test.pdf",
                "page_number": 0,
                "chunk_idx": 2,
                "chunk_id": "storage_test.pdf::p0::l3::0",
                "parent_chunk_id": "storage_test.pdf::p0::l2::0",
                "root_chunk_id": "storage_test.pdf::p0::l1::0",
                "chunk_level": 3,
            },
            {
                "dense_embedding": vector2,
                "text": "DUSt3R 是一种多视图三维重建方法。",
                "filename": "storage_test.pdf",
                "file_type": "pdf",
                "file_path": "/tmp/storage_test.pdf",
                "page_number": 0,
                "chunk_idx": 3,
                "chunk_id": "storage_test.pdf::p0::l3::1",
                "parent_chunk_id": "storage_test.pdf::p0::l2::0",
                "root_chunk_id": "storage_test.pdf::p0::l1::0",
                "chunk_level": 3,
            },
        ])
        with store.session() as client:
            client.flush(
                collection_name=store.collection_name
            )
        print("L3 Chunk 写入成功")


        # 3. 普通查询
        chunks = store.get_chunks_by_ids([
            "storage_test.pdf::p0::l3::0"
        ])

        print("chunks =", chunks)
        assert len(chunks) == 1

        print("按 chunk_id 查询成功：", chunks[0]["text"])

        # 4. Dense Retrieval
        dense_result = store.dense_retrieve(
            dense_embedding=vector1,
            top_k=1,
        )

        assert len(dense_result) == 1
        assert dense_result[0]["chunk_id"] == \
            "storage_test.pdf::p0::l3::0"

        print("Dense Retrieval 成功：", dense_result[0]["text"])

        # 5. Hybrid Retrieval
        hybrid_result = store.hybrid_retrieve(
            dense_embedding=vector1,
            query="VGGT 相机参数",
            top_k=2,
        )

        assert len(hybrid_result) > 0

        print("Hybrid Retrieval 成功：")
        for item in hybrid_result:
            print(
                item["chunk_id"],
                item["score"],
                item["text"],
            )

        # 6. 删除数据
        store.delete(
            'filename == "storage_test.pdf"'
        )

        # 等待删除操作真正落盘并对查询可见
        with store.session() as client:
            client.flush(
                collection_name=store.collection_name
            )

        chunks = store.get_chunks_by_ids([
            "storage_test.pdf::p0::l3::0"
        ])

        print("删除后 chunks =", chunks)

        assert len(chunks) == 0

        print("Milvus 数据删除成功")

    finally:
        # 测试结束删除测试 Collection
        if store.has_collection():
            store.drop_collection()

        print("测试 Collection 已清理")


def test_parent_chunk_store():
    print("\n=== 4. 测试 ParentChunkStore ===")

    store = ParentChunkStore()

    filename = "parent_store_test.pdf"

    # 避免之前测试残留
    store.delete_by_filename(filename)

    docs = [
        {
            "text": "这是测试文档的 L1 大块。",
            "filename": filename,
            "file_type": "pdf",
            "file_path": "/tmp/parent_store_test.pdf",
            "page_number": 0,
            "chunk_id": f"{filename}::p0::l1::0",
            "parent_chunk_id": "",
            "root_chunk_id": f"{filename}::p0::l1::0",
            "chunk_level": 1,
            "chunk_idx": 0,
        },
        {
            "text": "这是 L1 下面的 L2 中块。",
            "filename": filename,
            "file_type": "pdf",
            "file_path": "/tmp/parent_store_test.pdf",
            "page_number": 0,
            "chunk_id": f"{filename}::p0::l2::0",
            "parent_chunk_id": f"{filename}::p0::l1::0",
            "root_chunk_id": f"{filename}::p0::l1::0",
            "chunk_level": 2,
            "chunk_idx": 1,
        },
    ]

    # 1. 写入
    count = store.upsert_documents(docs)

    assert count == 2

    print("L1/L2 写入成功")

    # 2. 查询
    result = store.get_documents_by_ids([
        f"{filename}::p0::l2::0"
    ])

    assert len(result) == 1
    assert result[0]["chunk_level"] == 2

    print("Parent Chunk 查询成功：", result[0]["text"])

    # 3. 验证 Redis 中已经存在缓存
    redis_key = f"parent_chunk:{filename}::p0::l2::0"

    cached = cache.get_json(redis_key)

    assert cached is not None

    print("Parent Chunk Redis 缓存成功")

    # 4. 删除
    deleted = store.delete_by_filename(filename)

    assert deleted == 2

    print("Parent Chunk 删除成功：", deleted)


if __name__ == "__main__":
    test_postgresql()
    test_redis()
    test_milvus()
    test_parent_chunk_store()

    print("\n==============================")
    print("数据存储模块全部测试通过")
    print("==============================")

• 如果刚insert或delete，下一毫秒就做query可能Milvus 的插入/删除结果还没立即可见，因此测试代码中间做了flush。实际过程中可以在模块实现部分加上flush，但是这样会有额外开销；也可以创建Collection 时增加一致性配置，不过这样查询延迟会更高。考虑到我们的业务场景，管理员上传文档->写入milvus->后续用户再来检索。通常不是“刚 insert 完，下一毫秒就必须 query”，所以没必要让所有正常 RAG 查询都承担 Strong 的额外等待。

赏

谢谢你请我吃糖果