diff --git a/backend/.env.example b/backend/.env.example index 9358bb9..1066965 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -7,3 +7,9 @@ AUTH_ALLOWED_ORIGINS=http://localhost:5173,http://127.0.0.1:5173,http://localhos # 默认要求 HTTPS；仅本机 HTTP 联调可显式改为 false。 AUTH_COOKIE_SECURE=true AUTH_SESSION_SECONDS=28800 +# 可选 CPU 模型；先在独立环境准备固定修订缓存，再显式启用。API 永不下载权重。 +EMBEDDING_ENABLED=false +EMBEDDING_MODEL=small +EMBEDDING_CACHE_DIR= +EMBEDDING_BATCH_SIZE=8 +EMBEDDING_THREADS=2 diff --git a/backend/app/api/routes/retrieval.py b/backend/app/api/routes/retrieval.py index 0e3b870..52e21a7 100644 --- a/backend/app/api/routes/retrieval.py +++ b/backend/app/api/routes/retrieval.py @@ -8,6 +8,7 @@ from app.db.session import get_db from app.models import RetrievalRun from app.schemas.catalog import Page from app.schemas.retrieval import RetrievalRequest, RetrievalResponse +from app.schemas.vectors import VectorRetrievalResponse from app.services.auth import require_admin from app.services.retrieval import retrieve @@ -20,7 +21,11 @@ def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)): return retrieve(db, payload) -@router.get("/runs", response_model=Page[RetrievalResponse], dependencies=[Depends(require_admin)]) +@router.get( + "/runs", + response_model=Page[RetrievalResponse | VectorRetrievalResponse], + dependencies=[Depends(require_admin)], +) def list_runs( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), @@ -41,7 +46,9 @@ def list_runs( @router.get( - "/runs/{run_id}", response_model=RetrievalResponse, dependencies=[Depends(require_admin)] + "/runs/{run_id}", + response_model=RetrievalResponse | VectorRetrievalResponse, + dependencies=[Depends(require_admin)], ) def read_run(run_id: UUID, db: Session = Depends(get_db)): row = db.get(RetrievalRun, str(run_id)) diff --git a/backend/app/api/routes/vectors.py b/backend/app/api/routes/vectors.py new file mode 100644 index 0000000..0f09f8c --- /dev/null +++ b/backend/app/api/routes/vectors.py @@ -0,0 +1,106 @@ +from uuid import UUID + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.db.session import get_db +from app.models.vectors import EmbeddingSpace, VectorPublication, VectorVersion +from app.schemas.catalog import Page +from app.schemas.indexing import PublicationRead, PublishIndex +from app.schemas.vectors import ( + ActiveVectors, + BuildVectors, + SpaceRead, + VectorRead, + VectorRequest, + VectorRetrievalResponse, +) +from app.services.auth import CurrentSession, require_admin +from app.services.catalog import page_result +from app.services.embedding_runtime import get_embedding_provider +from app.services.vectors import ( + active_vectors, + build_vectors, + publish_vectors, + retrieve_vectors, + vector_or_404, +) + +router = APIRouter( + prefix="/vector-indexes", tags=["vector-indexes"], dependencies=[Depends(require_admin)] +) +retrieval_router = APIRouter(tags=["retrieval"]) + + +@router.post("", response_model=VectorRead, status_code=201) +def build( + payload: BuildVectors, + current: CurrentSession = Depends(require_admin), + db: Session = Depends(get_db), + provider=Depends(get_embedding_provider), +): + return build_vectors(db, payload.text_index_version_id, current.user.id, provider) + + +@router.get("", response_model=Page[VectorRead]) +def versions( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + return page_result( + db, + select(VectorVersion).order_by(VectorVersion.created_at.desc(), VectorVersion.id), + page, + page_size, + ) + + +@router.get("/active", response_model=ActiveVectors) +def active(db: Session = Depends(get_db)): + return active_vectors(db) + + +@router.get("/spaces", response_model=Page[SpaceRead]) +def spaces( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + return page_result(db, select(EmbeddingSpace).order_by(EmbeddingSpace.id), page, page_size) + + +@router.get("/publications", response_model=Page[PublicationRead]) +def publications( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + db: Session = Depends(get_db), +): + return page_result( + db, select(VectorPublication).order_by(VectorPublication.generation.desc()), page, page_size + ) + + +@router.get("/{version_id}", response_model=VectorRead) +def read_version(version_id: UUID, db: Session = Depends(get_db)): + return vector_or_404(db, str(version_id)) + + +@router.post("/{version_id}/publish", response_model=ActiveVectors) +def publish( + version_id: UUID, + payload: PublishIndex, + current: CurrentSession = Depends(require_admin), + db: Session = Depends(get_db), +): + return publish_vectors(db, version_id, payload.expected_generation, current.user.id) + + +@retrieval_router.post("/retrieval/vector", response_model=VectorRetrievalResponse) +def search( + payload: VectorRequest, + db: Session = Depends(get_db), + provider=Depends(get_embedding_provider), +): + return retrieve_vectors(db, payload, provider) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 7475f97..1bc66bd 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -1,7 +1,8 @@ from pathlib import Path +from typing import Literal from urllib.parse import urlsplit -from pydantic import Field, model_validator +from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -16,6 +17,11 @@ class Settings(BaseSettings): ) auth_cookie_secure: bool = True auth_session_seconds: int = Field(28800, ge=300, le=604800) + embedding_enabled: bool = False + embedding_model: Literal["small", "base"] = "small" + embedding_cache_dir: Path | None = None + embedding_batch_size: int = Field(8, ge=1, le=32) + embedding_threads: int = Field(2, ge=1, le=4) model_config = SettingsConfigDict( env_file=Path(__file__).resolve().parents[2] / ".env", extra="ignore" ) @@ -28,6 +34,11 @@ class Settings(BaseSettings): def auth_origin_list(self) -> list[str]: return [x.strip() for x in self.auth_allowed_origins.split(",") if x.strip()] + @field_validator("embedding_cache_dir", mode="before") + @classmethod + def empty_embedding_cache_uses_default(cls, value): + return None if isinstance(value, str) and not value.strip() else value + @model_validator(mode="after") def validate_auth_settings(self): if not self.auth_cookie_secure and self.app_env not in {"development", "test"}: diff --git a/backend/app/main.py b/backend/app/main.py index 7a5f2b7..076f906 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,7 +8,17 @@ from sqlalchemy import text from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError from sqlalchemy.orm.exc import StaleDataError -from app.api.routes import auth, chat, documents, health, indexes, knowledge, retrieval, sources +from app.api.routes import ( + auth, + chat, + documents, + health, + indexes, + knowledge, + retrieval, + sources, + vectors, +) from app.core.config import settings from app.db.session import engine from app.services.auth import dummy_hash @@ -21,7 +31,7 @@ def create_app(db_engine=engine) -> FastAPI: try: with db_engine.connect() as connection: revision = connection.scalar(text("SELECT version_num FROM alembic_version")) - if revision != "0005_chunk_indexes": + if revision != "0006_vectors": raise RuntimeError( "数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head" ) @@ -36,7 +46,7 @@ def create_app(db_engine=engine) -> FastAPI: finally: db_engine.dispose() - app = FastAPI(title=settings.app_name, version="0.5.0", lifespan=lifespan) + app = FastAPI(title=settings.app_name, version="0.6.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origin_list, @@ -48,6 +58,8 @@ def create_app(db_engine=engine) -> FastAPI: health.router, auth.router, indexes.router, + vectors.router, + vectors.retrieval_router, chat.router, knowledge.router, sources.router, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index b215e2e..d502882 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -3,13 +3,22 @@ from app.models.catalog import AuditEvent, Document, Source from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion from app.models.knowledge import KnowledgeItem from app.models.retrieval import RetrievalRun +from app.models.vectors import ( + ChunkVector, + EmbeddingSpace, + VectorPublication, + VectorState, + VectorVersion, +) __all__ = [ "AuditEvent", "AuthRateLimit", "AuthSession", "Chunk", + "ChunkVector", "Document", + "EmbeddingSpace", "IndexPublication", "IndexState", "IndexVersion", @@ -17,4 +26,7 @@ __all__ = [ "RetrievalRun", "Source", "User", + "VectorPublication", + "VectorState", + "VectorVersion", ] diff --git a/backend/app/models/indexing.py b/backend/app/models/indexing.py index fa012e0..a602ba6 100644 --- a/backend/app/models/indexing.py +++ b/backend/app/models/indexing.py @@ -50,6 +50,7 @@ class Chunk(Base): __tablename__ = "chunks" __table_args__ = ( UniqueConstraint("index_version_id", "knowledge_id", "ordinal", name="uq_chunk_position"), + UniqueConstraint("id", "index_version_id", name="uq_chunk_text_scope"), CheckConstraint("start_offset >= 0 AND end_offset > start_offset", name="ck_chunk_offsets"), ) id: Mapped[str] = mapped_column(String(36), primary_key=True) diff --git a/backend/app/models/vectors.py b/backend/app/models/vectors.py new file mode 100644 index 0000000..4fa2b3d --- /dev/null +++ b/backend/app/models/vectors.py @@ -0,0 +1,112 @@ +"""Versioned vector spaces, complete snapshots and independent publication history.""" + +from datetime import datetime + +from pgvector.sqlalchemy import Vector +from sqlalchemy import ( + JSON, + CheckConstraint, + DateTime, + ForeignKey, + ForeignKeyConstraint, + Integer, + String, + UniqueConstraint, + func, +) +from sqlalchemy.dialects.postgresql.base import ischema_names +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.session import Base + +# PostgreSQL reflects a qualified name when public is intentionally absent from search_path. +# Retain real type comparison during Alembic inspection in isolated schemas. +ischema_names.setdefault("public.vector", Vector) + + +class EmbeddingSpace(Base): + __tablename__ = "embedding_spaces" + __table_args__ = ( + UniqueConstraint("id", "dimension", name="uq_embedding_space_dimension"), + CheckConstraint("dimension BETWEEN 1 AND 2000", name="ck_embedding_space_dimension"), + ) + id: Mapped[str] = mapped_column(String(64), primary_key=True) + dimension: Mapped[int] = mapped_column(Integer, nullable=False) + spec: Mapped[dict] = mapped_column(JSON, nullable=False) + + +class VectorVersion(Base): + __tablename__ = "vector_versions" + __table_args__ = ( + ForeignKeyConstraint( + ["space_id", "dimension"], + ["embedding_spaces.id", "embedding_spaces.dimension"], + name="fk_vector_version_space", + ), + UniqueConstraint( + "id", "text_index_version_id", "space_id", "dimension", name="uq_vector_version_scope" + ), + CheckConstraint( + "status IN ('ready', 'published', 'superseded')", name="ck_vector_version_status" + ), + CheckConstraint("chunk_count BETWEEN 0 AND 2000", name="ck_vector_version_count"), + ) + id: Mapped[str] = mapped_column(String(36), primary_key=True) + text_index_version_id: Mapped[str] = mapped_column(ForeignKey("index_versions.id")) + space_id: Mapped[str] = mapped_column(String(64), nullable=False) + dimension: Mapped[int] = mapped_column(Integer, nullable=False) + chunk_count: Mapped[int] = mapped_column(Integer, nullable=False) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="ready") + created_by: Mapped[int] = mapped_column(ForeignKey("users.id")) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + +class ChunkVector(Base): + __tablename__ = "chunk_vectors" + __table_args__ = ( + ForeignKeyConstraint( + ["version_id", "text_index_version_id", "space_id", "dimension"], + [ + "vector_versions.id", + "vector_versions.text_index_version_id", + "vector_versions.space_id", + "vector_versions.dimension", + ], + name="fk_chunk_vector_scope", + ), + ForeignKeyConstraint( + ["chunk_id", "text_index_version_id"], + ["chunks.id", "chunks.index_version_id"], + name="fk_chunk_vector_text", + ), + CheckConstraint("public.vector_dims(embedding) = dimension", name="ck_vector_dimension"), + CheckConstraint( + "public.vector_norm(embedding) BETWEEN 0.9999 AND 1.0001", name="ck_vector_unit_norm" + ), + ) + version_id: Mapped[str] = mapped_column(String(36), primary_key=True) + chunk_id: Mapped[str] = mapped_column(String(36), primary_key=True) + text_index_version_id: Mapped[str] = mapped_column(String(36), nullable=False) + space_id: Mapped[str] = mapped_column(String(64), nullable=False) + dimension: Mapped[int] = mapped_column(Integer, nullable=False) + embedding = mapped_column(Vector(), nullable=False) + + +class VectorState(Base): + __tablename__ = "vector_state" + __table_args__ = (CheckConstraint("id = 1", name="ck_vector_state_singleton"),) + id: Mapped[int] = mapped_column(primary_key=True) + active_version_id: Mapped[str | None] = mapped_column(ForeignKey("vector_versions.id")) + generation: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + +class VectorPublication(Base): + __tablename__ = "vector_publications" + generation: Mapped[int] = mapped_column(primary_key=True) + previous_version_id: Mapped[str | None] = mapped_column(ForeignKey("vector_versions.id")) + version_id: Mapped[str] = mapped_column(ForeignKey("vector_versions.id"), nullable=False) + actor_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False) + published_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) diff --git a/backend/app/schemas/vectors.py b/backend/app/schemas/vectors.py new file mode 100644 index 0000000..37cc306 --- /dev/null +++ b/backend/app/schemas/vectors.py @@ -0,0 +1,88 @@ +from datetime import datetime +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + +from app.schemas.catalog import Category, Input +from app.schemas.indexing import TimestampRead +from app.schemas.retrieval import RetrievalHit + + +class EmbeddingSpecRead(BaseModel): + model: str + revision: str + dimension: int + token_limit: int + query_instruction: str + implementation: str + checkpoint_format: str + + +class SpaceRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: str + dimension: int + spec: EmbeddingSpecRead + + +class BuildVectors(Input): + text_index_version_id: UUID + + +class VectorRead(TimestampRead): + id: UUID + text_index_version_id: UUID + space_id: str + dimension: int + chunk_count: int + status: Literal["ready", "published", "superseded"] + created_by: int + created_at: datetime + published_at: datetime | None + + +class ActiveVectors(BaseModel): + generation: int + text_generation: int + catalog_revision: int + is_current: bool + active_version: VectorRead | None + + +class VectorRequest(Input): + query: str = Field(min_length=1, max_length=500) + mode: Literal["vector"] = "vector" + top_k: int = Field(5, ge=1, le=20) + category: Category | None = None + + +class VectorHit(RetrievalHit): + chunk_id: UUID + chunk_ordinal: int + + +class VectorRetrievalConfig(BaseModel): + scorer: Literal["pgvector-cosine-exact-v1"] = "pgvector-cosine-exact-v1" + ranking_unit: Literal["chunk"] = "chunk" + space: SpaceRead + + +class VectorRetrievalResponse(BaseModel): + run_id: UUID + query: str + mode: Literal["vector"] = "vector" + status: Literal["ok", "no_evidence"] + top_k: int + category: str | None + vector_version_id: UUID + text_index_version_id: UUID + generation: int + text_generation: int + corpus_fingerprint: str + corpus_size: int + total_hits: int + results: list[VectorHit] + elapsed_ms: float + created_at: datetime + config: VectorRetrievalConfig diff --git a/backend/app/services/embedding_runtime.py b/backend/app/services/embedding_runtime.py new file mode 100644 index 0000000..81cc015 --- /dev/null +++ b/backend/app/services/embedding_runtime.py @@ -0,0 +1,49 @@ +"""Offline, lazy CPU runtime; one encoder call per process, with a fast busy response.""" + +from functools import lru_cache +from threading import Lock + +from fastapi import HTTPException + +from app.core.config import settings +from app.services.embedding import LOCAL_MODELS, EmbeddingSpec, LocalEmbeddingProvider + + +class EmbeddingRuntime: + def __init__(self, spec: EmbeddingSpec, cache_dir, batch_size: int, threads: int): + self.spec = spec + self.cache_dir, self.batch_size, self.threads = cache_dir, batch_size, threads + self._provider = None + self._lock = Lock() + + def encode(self, texts: list[str], *, query: bool = False) -> list[list[float]]: + if not self._lock.acquire(blocking=False): + raise HTTPException(503, "本地编码器忙，请稍后重试") + try: + if self._provider is None: + self._provider = LocalEmbeddingProvider( + self.spec, + cache_dir=self.cache_dir, + batch_size=self.batch_size, + threads=self.threads, + offline=True, + ) + return self._provider.encode(texts, query=query) + finally: + self._lock.release() + + +@lru_cache(maxsize=1) +def runtime(spec, cache_dir, batch_size, threads): + return EmbeddingRuntime(spec, cache_dir, batch_size, threads) + + +def get_embedding_provider(): + if not settings.embedding_enabled: + raise HTTPException(503, "本地 Embedding 未启用，请先完成模型缓存和运行环境配置") + return runtime( + LOCAL_MODELS[settings.embedding_model], + settings.embedding_cache_dir, + settings.embedding_batch_size, + settings.embedding_threads, + ) diff --git a/backend/app/services/vectors.py b/backend/app/services/vectors.py new file mode 100644 index 0000000..028c008 --- /dev/null +++ b/backend/app/services/vectors.py @@ -0,0 +1,353 @@ +"""Generate outside publication locks; recheck reviewed evidence before persistence/query.""" + +import hashlib +from dataclasses import asdict +from time import perf_counter +from uuid import UUID, uuid4 + +from fastapi import HTTPException +from sqlalchemy import Float, func, select +from sqlalchemy.orm import Session + +from app.models.catalog import utc_now +from app.models.indexing import Chunk, IndexVersion +from app.models.retrieval import RetrievalRun +from app.models.vectors import ( + ChunkVector, + EmbeddingSpace, + VectorPublication, + VectorState, + VectorVersion, +) +from app.schemas.vectors import ( + ActiveVectors, + SpaceRead, + VectorHit, + VectorRead, + VectorRequest, + VectorRetrievalConfig, + VectorRetrievalResponse, +) +from app.services.embedding import EmbeddingError, EmbeddingProvider, validate_vectors +from app.services.indexing import locked_state, split_text +from app.services.retrieval import corpus_fingerprint, load_corpus + +MAX_VECTOR_CHUNKS = 2000 + + +def vector_state(db: Session) -> VectorState: + # Always acquire index_state first: the same ordering as catalog/text publication. + state = db.scalar( + select(VectorState) + .where(VectorState.id == 1) + .with_for_update() + .execution_options(populate_existing=True) + ) + if state is None: + raise HTTPException(503, "向量状态未初始化，请检查迁移") + return state + + +def vector_or_404(db: Session, version_id: str) -> VectorVersion: + version = db.get(VectorVersion, version_id) + if version is None: + raise HTTPException(404, "向量版本不存在") + return version + + +def current_text(db: Session, state): + version = db.get(IndexVersion, state.active_version_id) if state.active_version_id else None + if version is None or version.catalog_revision != state.catalog_revision: + raise HTTPException(409, "尚无当前有效的文本发布版本，请构建并发布") + corpus = load_corpus(db, None) + if version.corpus_fingerprint != corpus_fingerprint(corpus): + raise HTTPException(409, "审核依据已变化，请重新构建并发布文本版本") + return version, corpus + + +def encode(provider: EmbeddingProvider, texts: list[str], *, query=False): + try: + spec = provider.spec + result = provider.encode(texts, query=query) + if provider.spec != spec: + raise EmbeddingError("编码期间模型身份改变") + validate_vectors(result, len(texts), spec) + return result + except HTTPException: + raise + except EmbeddingError as exc: + raise HTTPException(422, "编码校验失败，请核对文本 token 上限、向量维度和归一化") from exc + except Exception as exc: + # Optional dependency/cache failures must not expose server paths or library tracebacks. + raise HTTPException(503, "本地模型不可用，请检查可选运行依赖和固定版本缓存") from exc + + +def active_vectors(db: Session) -> ActiveVectors: + state = locked_state(db) + vector = vector_state(db) + version = db.get(VectorVersion, vector.active_version_id) if vector.active_version_id else None + current = False + if version and version.text_index_version_id == state.active_version_id: + try: + text_version, corpus = current_text(db, state) + checked_chunks(db, text_version, corpus) + complete(db, version) + current = version.chunk_count == text_version.chunk_count + except HTTPException as exc: + if exc.status_code != 409: + raise + return ActiveVectors( + generation=vector.generation, + text_generation=state.generation, + catalog_revision=state.catalog_revision, + is_current=current, + active_version=VectorRead.model_validate(version) if version else None, + ) + + +def checked_chunks(db, text_version, corpus): + if text_version.chunk_count > MAX_VECTOR_CHUNKS: + raise HTTPException(503, "超过同步向量构建容量（2000 chunks）") + chunks = list( + db.scalars( + select(Chunk) + .where(Chunk.index_version_id == text_version.id) + .order_by(Chunk.knowledge_id, Chunk.ordinal) + .limit(MAX_VECTOR_CHUNKS + 1) + ) + ) + if len(chunks) != text_version.chunk_count: + raise HTTPException(409, "文本切分快照不完整，请重新构建") + expected = [ + ( + item["knowledge_id"], + ordinal, + start, + end, + body, + hashlib.sha256(body.encode()).hexdigest(), + {k: v for k, v in item.items() if k != "content"}, + ) + for item in corpus + for ordinal, (start, end, body) in enumerate( + split_text( + item["content"], text_version.config["chunk_size"], text_version.config["overlap"] + ) + ) + ] + actual = [ + ( + c.knowledge_id, + c.ordinal, + c.start_offset, + c.end_offset, + c.text, + c.text_sha256, + c.provenance, + ) + for c in chunks + ] + if actual != expected: + raise HTTPException(409, "文本切分与审核语料不一致，请重新构建") + return [(c.id, c.text) for c in chunks] + + +def build_vectors(db: Session, text_version_id: UUID, actor_id: int, provider: EmbeddingProvider): + state = locked_state(db) + text_version, corpus = current_text(db, state) + if str(text_version_id) != text_version.id: + raise HTTPException(409, "只能对当前已发布的文本版本构建向量") + inputs = checked_chunks(db, text_version, corpus) + token = (state.generation, state.catalog_revision, text_version.id) + spec = provider.spec + db.commit() # No publication/catalog row locks or open read transaction during model work. + vectors = encode(provider, [body for _, body in inputs]) if inputs else [] + if provider.spec != spec: + raise HTTPException(409, "模型身份已变化，请重试") + state = locked_state(db) + new_text, new_corpus = current_text(db, state) + if token != (state.generation, state.catalog_revision, state.active_version_id): + raise HTTPException(409, "编码期间文本或审核版本已变化，请重新构建") + if checked_chunks(db, new_text, new_corpus) != inputs: + raise HTTPException(409, "编码期间文本切分快照已变化，请重新构建") + space = db.get(EmbeddingSpace, spec.space_id) + if space is None: + db.add(EmbeddingSpace(id=spec.space_id, dimension=spec.dimension, spec=asdict(spec))) + db.flush() + elif space.spec != asdict(spec) or space.dimension != spec.dimension: + raise HTTPException(409, "已保存向量空间配置不一致") + version = VectorVersion( + id=str(uuid4()), + text_index_version_id=str(text_version_id), + space_id=spec.space_id, + dimension=spec.dimension, + chunk_count=len(inputs), + created_by=actor_id, + status="ready", + ) + db.add(version) + db.flush() + db.add_all( + [ + ChunkVector( + version_id=version.id, + chunk_id=chunk_id, + text_index_version_id=version.text_index_version_id, + space_id=spec.space_id, + dimension=spec.dimension, + embedding=embedding, + ) + for (chunk_id, _), embedding in zip(inputs, vectors, strict=True) + ] + ) + result = VectorRead.model_validate(version) + db.commit() + return result + + +def complete(db, version): + count = db.scalar( + select(func.count()).select_from(ChunkVector).where(ChunkVector.version_id == version.id) + ) + if count != version.chunk_count: + raise HTTPException(409, "向量快照不完整，请重新构建") + + +def publish_vectors(db: Session, version_id: UUID, expected_generation: int, actor_id: int): + state = locked_state(db) + vector = vector_state(db) + if vector.generation != expected_generation: + raise HTTPException(409, "向量发布代数已变化，请刷新后重试") + version = vector_or_404(db, str(version_id)) + text_version, corpus = current_text(db, state) + if version.text_index_version_id != text_version.id: + raise HTTPException(409, "该向量版本不属于当前文本发布版本") + checked_chunks(db, text_version, corpus) + complete(db, version) + if version.chunk_count != text_version.chunk_count: + raise HTTPException(409, "向量与文本切分数量不一致") + if vector.active_version_id == version.id: + return active_vectors(db) + previous = vector.active_version_id + if previous: + db.get(VectorVersion, previous).status = "superseded" + vector.active_version_id = version.id + vector.generation += 1 + version.status, version.published_at = "published", utc_now() + db.add( + VectorPublication( + generation=vector.generation, + previous_version_id=previous, + version_id=version.id, + actor_id=actor_id, + published_at=version.published_at, + ) + ) + result = ActiveVectors( + generation=vector.generation, + text_generation=state.generation, + catalog_revision=state.catalog_revision, + is_current=True, + active_version=VectorRead.model_validate(version), + ) + db.commit() + return result + + +def retrieval_scope(db, provider): + state = locked_state(db) + vector = vector_state(db) + text_version, corpus = current_text(db, state) + version = db.get(VectorVersion, vector.active_version_id) if vector.active_version_id else None + if version is None or version.text_index_version_id != text_version.id: + raise HTTPException(409, "尚无匹配当前文本版本的向量发布，请先构建并发布") + checked_chunks(db, text_version, corpus) + space = db.get(EmbeddingSpace, version.space_id) + if ( + space.id != provider.spec.space_id + or space.spec != asdict(provider.spec) + or space.dimension != provider.spec.dimension + or version.dimension != provider.spec.dimension + ): + raise HTTPException( + 409, "当前编码器与已发布向量空间不一致，请使用同一配置重建或切换运行配置" + ) + complete(db, version) + if version.chunk_count != text_version.chunk_count: + raise HTTPException(409, "向量与文本切分数量不一致") + token = (state.generation, state.catalog_revision, vector.generation, version.id) + return state, vector, text_version, version, space, token + + +def retrieve_vectors(db: Session, payload: VectorRequest, provider: EmbeddingProvider): + started = perf_counter() + *_, token = retrieval_scope(db, provider) + db.commit() + query_vector = encode(provider, [payload.query], query=True)[0] + state, vector, text_version, version, space, current = retrieval_scope(db, provider) + if token != current: + raise HTTPException(409, "查询编码期间发布版本或审核依据已变化，请重试") + # Extension-qualified operator works even with an isolated business schema search_path. + distance = ChunkVector.embedding.op("OPERATOR(public.<=>)", return_type=Float)(query_vector) + statement = ( + select(Chunk, distance.label("distance")) + .join(ChunkVector, ChunkVector.chunk_id == Chunk.id) + .where( + ChunkVector.version_id == version.id, + ChunkVector.space_id == space.id, + ChunkVector.text_index_version_id == text_version.id, + ChunkVector.dimension == space.dimension, + ) + ) + if payload.category is not None: + statement = statement.where(Chunk.provenance["category"].as_string() == payload.category) + total = db.scalar(select(func.count()).select_from(statement.subquery())) + rows = db.execute( + statement.order_by(distance, Chunk.knowledge_id, Chunk.ordinal, Chunk.id).limit( + payload.top_k + ) + ) + hits = [ + VectorHit( + **chunk.provenance, + rank=rank, + chunk_id=chunk.id, + chunk_ordinal=chunk.ordinal, + snippet=chunk.text, + snippet_start=chunk.start_offset, + snippet_end=chunk.end_offset, + score=round(max(-1.0, min(1.0, 1.0 - float(dist))), 8), + matched_terms=[], + ) + for rank, (chunk, dist) in enumerate(rows, 1) + ] + result = VectorRetrievalResponse( + run_id=uuid4(), + query=payload.query, + status="ok" if hits else "no_evidence", + top_k=payload.top_k, + category=payload.category, + vector_version_id=version.id, + text_index_version_id=text_version.id, + generation=vector.generation, + text_generation=state.generation, + corpus_fingerprint=text_version.corpus_fingerprint, + corpus_size=total, + total_hits=total, + results=hits, + elapsed_ms=round((perf_counter() - started) * 1000, 3), + created_at=utc_now(), + config=VectorRetrievalConfig(space=SpaceRead.model_validate(space)), + ) + db.add( + RetrievalRun( + id=str(result.run_id), + query=result.query, + mode="vector", + corpus_fingerprint=result.corpus_fingerprint, + response=result.model_dump(mode="json"), + created_at=result.created_at, + ) + ) + db.commit() + return result diff --git a/backend/migrations/versions/0006_vectors.py b/backend/migrations/versions/0006_vectors.py new file mode 100644 index 0000000..c28b563 --- /dev/null +++ b/backend/migrations/versions/0006_vectors.py @@ -0,0 +1,122 @@ +"""Persist dimension-checked vector snapshots with composite scope constraints.""" + +import sqlalchemy as sa +from alembic import op + +revision = "0006_vectors" +down_revision = "0005_chunk_indexes" +branch_labels = None +depends_on = None + + +class ExtensionVector(sa.types.UserDefinedType): + """Qualify extension type without adding public business tables to search_path.""" + + def get_col_spec(self, **kw): + return "public.vector" + + +def upgrade(): + schema = op.get_bind().scalar( + sa.text("SELECT extnamespace::regnamespace::text FROM pg_extension WHERE extname='vector'") + ) + if schema != "public": + raise RuntimeError("vector 扩展须安装在 public；请由数据库管理员核对，迁移不会移动共享扩展") + op.create_unique_constraint("uq_chunk_text_scope", "chunks", ["id", "index_version_id"]) + op.create_table( + "embedding_spaces", + sa.Column("id", sa.String(64), primary_key=True), + sa.Column("dimension", sa.Integer(), nullable=False), + sa.Column("spec", sa.JSON(), nullable=False), + sa.UniqueConstraint("id", "dimension", name="uq_embedding_space_dimension"), + sa.CheckConstraint("dimension BETWEEN 1 AND 2000", name="ck_embedding_space_dimension"), + ) + op.create_table( + "vector_versions", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column( + "text_index_version_id", + sa.String(36), + sa.ForeignKey("index_versions.id"), + nullable=False, + ), + sa.Column("space_id", sa.String(64), nullable=False), + sa.Column("dimension", sa.Integer(), nullable=False), + sa.Column("chunk_count", sa.Integer(), nullable=False), + sa.Column("status", sa.String(20), nullable=False), + sa.Column("created_by", sa.Integer(), sa.ForeignKey("users.id"), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + sa.Column("published_at", sa.DateTime(timezone=True)), + sa.ForeignKeyConstraint( + ["space_id", "dimension"], + ["embedding_spaces.id", "embedding_spaces.dimension"], + name="fk_vector_version_space", + ), + sa.UniqueConstraint( + "id", "text_index_version_id", "space_id", "dimension", name="uq_vector_version_scope" + ), + sa.CheckConstraint( + "status IN ('ready', 'published', 'superseded')", name="ck_vector_version_status" + ), + sa.CheckConstraint("chunk_count BETWEEN 0 AND 2000", name="ck_vector_version_count"), + ) + op.create_table( + "chunk_vectors", + sa.Column("version_id", sa.String(36), primary_key=True), + sa.Column("chunk_id", sa.String(36), primary_key=True), + sa.Column("text_index_version_id", sa.String(36), nullable=False), + sa.Column("space_id", sa.String(64), nullable=False), + sa.Column("dimension", sa.Integer(), nullable=False), + sa.Column("embedding", ExtensionVector(), nullable=False), + sa.ForeignKeyConstraint( + ["version_id", "text_index_version_id", "space_id", "dimension"], + [ + "vector_versions.id", + "vector_versions.text_index_version_id", + "vector_versions.space_id", + "vector_versions.dimension", + ], + name="fk_chunk_vector_scope", + ), + sa.ForeignKeyConstraint( + ["chunk_id", "text_index_version_id"], + ["chunks.id", "chunks.index_version_id"], + name="fk_chunk_vector_text", + ), + sa.CheckConstraint("public.vector_dims(embedding) = dimension", name="ck_vector_dimension"), + sa.CheckConstraint( + "public.vector_norm(embedding) BETWEEN 0.9999 AND 1.0001", name="ck_vector_unit_norm" + ), + ) + op.create_table( + "vector_state", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("active_version_id", sa.String(36), sa.ForeignKey("vector_versions.id")), + sa.Column("generation", sa.Integer(), nullable=False), + sa.CheckConstraint("id = 1", name="ck_vector_state_singleton"), + ) + op.execute(sa.text("INSERT INTO vector_state (id, generation) VALUES (1, 0)")) + op.create_table( + "vector_publications", + sa.Column("generation", sa.Integer(), primary_key=True), + sa.Column("previous_version_id", sa.String(36), sa.ForeignKey("vector_versions.id")), + sa.Column("version_id", sa.String(36), sa.ForeignKey("vector_versions.id"), nullable=False), + sa.Column("actor_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False), + sa.Column( + "published_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + ) + + +def downgrade(): + for table in [ + "vector_publications", + "vector_state", + "chunk_vectors", + "vector_versions", + "embedding_spaces", + ]: + op.drop_table(table) + op.drop_constraint("uq_chunk_text_scope", "chunks", type_="unique") diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index 82eede6..8df6a1e 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -16,7 +16,7 @@ def test_startup_requires_migration(database): def test_startup_rejects_previous_schema(database): - migrate(database, "0004_auth_sessions") + migrate(database, "0005_chunk_indexes") with pytest.raises(RuntimeError, match="alembic upgrade head"): with TestClient(create_app(database)): pass @@ -40,10 +40,7 @@ def test_upgrade_preserves_original_rows_and_repeated_upgrade(database): assert row["collected_at"] is None assert row["verification_status"] == "pending" assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2 - assert ( - connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0005_chunk_indexes" - ) + assert connection.scalar(text("SELECT version_num FROM alembic_version")) == "0006_vectors" assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == [] migrate(database, "0001_baseline", downgrade=True) with database.connect() as connection: @@ -98,3 +95,29 @@ def test_chunk_index_downgrade_preserves_users_sessions_knowledge_and_retrieval( assert client.get("/api/auth/me").status_code == 200 assert client.get("/api/indexes").json()["total"] == 0 assert client.get("/api/indexes/active").json()["generation"] == 0 + + +def test_vector_downgrade_retains_text_publications_and_prior_retrieval_runs( + client, migrated_database +): + from test_retrieval import add_item + from test_vectors import ToyProvider, text_version, vector_build, vector_publish, vector_search + + from app.services.embedding_runtime import get_embedding_provider + + client.app.dependency_overrides[get_embedding_provider] = ToyProvider + add_item(client) + text_ = text_version(client) + version = vector_build(client, text_) + vector_publish(client, version) + history = vector_search(client).json() + migrate(migrated_database, "0005_chunk_indexes", downgrade=True) + with migrated_database.connect() as c: + assert c.scalar(text("SELECT active_version_id FROM index_state")) == text_["id"] + assert c.scalar(text("SELECT count(*) FROM chunks")) == 1 + assert c.scalar(text("SELECT count(*) FROM users")) == 1 + assert c.scalar(text("SELECT count(*) FROM index_publications")) == 1 + assert c.scalar(text("SELECT id FROM retrieval_runs")) == history["run_id"] + migrate(migrated_database) + assert client.get("/api/vector-indexes/active").json()["generation"] == 0 + assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history diff --git a/backend/tests/test_runtime.py b/backend/tests/test_runtime.py index 6f51efe..1b62eb8 100644 --- a/backend/tests/test_runtime.py +++ b/backend/tests/test_runtime.py @@ -17,7 +17,7 @@ from app.services.auth import PASSWORD_HASHER @contextmanager -def server(engine, *, authenticate=True): +def server(engine, *, authenticate=True, environment_overrides=None): with socket.socket() as socket_: socket_.bind(("127.0.0.1", 0)) port = socket_.getsockname()[1] @@ -26,6 +26,7 @@ def server(engine, *, authenticate=True): "DATABASE_URL": engine.url.render_as_string(hide_password=False), "APP_ENV": "test", "AUTH_COOKIE_SECURE": "false", + **(environment_overrides or {}), } if authenticate: with Session(engine) as db: diff --git a/backend/tests/test_vector_runtime.py b/backend/tests/test_vector_runtime.py new file mode 100644 index 0000000..c90846b --- /dev/null +++ b/backend/tests/test_vector_runtime.py @@ -0,0 +1,91 @@ +"""Explicit opt-in: actual pinned model -> HTTP -> PostgreSQL -> HTTP process restart.""" + +import os + +import pytest +from sqlalchemy import text +from test_runtime import server + +from app.services.embedding import SMALL_ZH + + +@pytest.mark.skipif( + os.environ.get("RUN_LOCAL_EMBEDDING_TEST") != "1", + reason="requires optional CPU runtime and offline pinned model cache", +) +def test_real_local_model_vectors_retrieval_and_http_restart(migrated_database): + options = { + "EMBEDDING_ENABLED": "true", + "EMBEDDING_MODEL": "small", + "EMBEDDING_CACHE_DIR": os.environ["EMBEDDING_CACHE_DIR"], + "HF_HUB_OFFLINE": "1", + "HF_HUB_DISABLE_TELEMETRY": "1", + "HF_HUB_DISABLE_IMPLICIT_TOKEN": "1", + "TOKENIZERS_PARALLELISM": "false", + } + query = {"query": "合成测试中交通安排应查阅哪份资料？"} + with server(migrated_database, environment_overrides=options) as client: + ids = [] + for topic, body in [ + ("交通", "【合成测试】占位交通资料介绍测试公交线路与班次的查询入口。"), + ("住宿", "【合成测试】占位住宿资料介绍测试房型与入住须知。"), + ("审核", "【合成测试】占位审核资料说明测试出处核验流程。"), + ]: + item = client.post( + "/api/knowledge", + json={ + "title": f"{topic}合成测试", + "content": body, + "source_url": "https://example.com/", + "collected_at": "2026-10-16T00:00:00Z", + }, + ).json() + response = client.post( + f"/api/knowledge/{item['id']}/review", + json={ + "expected_version": item["version"], + "verification_status": "verified", + "reviewer": "test-reviewer", + "verification_notes": "仅为模型/HTTP/数据库持久化测试，不是事实资料", + }, + ) + assert response.status_code == 200 + ids.append(item["id"]) + text_ = client.post("/api/indexes", json={"chunk_size": 200, "overlap": 20}).json() + assert ( + client.post( + f"/api/indexes/{text_['id']}/publish", json={"expected_generation": 0} + ).status_code + == 200 + ) + built = client.post( + "/api/vector-indexes", json={"text_index_version_id": text_["id"]}, timeout=60 + ) + assert built.status_code == 201, built.text + version = built.json() + assert version["space_id"] == SMALL_ZH.space_id and version["dimension"] == 512 + published = client.post( + f"/api/vector-indexes/{version['id']}/publish", json={"expected_generation": 0} + ) + assert published.status_code == 200 + active = published.json() + response = client.post("/api/retrieval/vector", json=query, timeout=60) + assert response.status_code == 200, response.text + first = response.json() + assert {h["knowledge_id"] for h in first["results"]} == set(ids) + assert all(-1 <= h["score"] <= 1 for h in first["results"]) + assert first["config"]["space"]["spec"]["revision"] == SMALL_ZH.revision + with migrated_database.connect() as c: + assert ( + c.scalar( + text("SELECT count(*) FROM chunk_vectors WHERE public.vector_dims(embedding)=512") + ) + == 3 + ) + assert c.scalar(text("SELECT count(*) FROM vector_publications")) == 1 + with server(migrated_database, environment_overrides=options) as client: + assert client.get("/api/vector-indexes/active").json() == active + assert client.get(f"/api/retrieval/runs/{first['run_id']}").json() == first + response = client.post("/api/retrieval/vector", json=query, timeout=60) + assert response.status_code == 200, response.text + assert response.json()["results"] == first["results"] diff --git a/backend/tests/test_vectors.py b/backend/tests/test_vectors.py new file mode 100644 index 0000000..6b67f6d --- /dev/null +++ b/backend/tests/test_vectors.py @@ -0,0 +1,552 @@ +"""Toy vectors test database mathematics and isolation, never model/domain quality.""" + +from dataclasses import replace +from uuid import UUID, uuid4 + +import pytest +from fastapi import HTTPException +from sqlalchemy import func, select, text, update +from sqlalchemy.exc import DBAPIError, OperationalError +from sqlalchemy.orm import Session +from test_indexing import build, publish +from test_retrieval import add_item + +from app.models.indexing import Chunk +from app.models.retrieval import RetrievalRun +from app.models.vectors import ChunkVector, EmbeddingSpace, VectorPublication, VectorVersion +from app.services import vectors +from app.services.embedding import EmbeddingError, EmbeddingSpec +from app.services.embedding_runtime import EmbeddingRuntime, get_embedding_provider +from app.services.indexing import locked_state, publish_index + + +class ToyProvider: + spec = EmbeddingSpec("test-only-coordinate-fixture", "a" * 40, 3) + callback = None + + def encode(self, texts, *, query=False): + if self.callback: + self.callback(query) + return [[1.0, 0.0, 0.0] if query or "alpha" in t else [0.0, 1.0, 0.0] for t in texts] + + +@pytest.fixture +def provider(client): + value = ToyProvider() + client.app.dependency_overrides[get_embedding_provider] = lambda: value + return value + + +def text_version(client, **kwargs): + version = build(client, chunk_size=100, overlap=0, **kwargs) + generation = client.get("/api/indexes/active").json()["generation"] + assert publish(client, version, generation).status_code == 200 + return version + + +def vector_build(client, version): + response = client.post("/api/vector-indexes", json={"text_index_version_id": version["id"]}) + assert response.status_code == 201, response.text + return response.json() + + +def vector_publish(client, version, generation=0): + return client.post( + f"/api/vector-indexes/{version['id']}/publish", json={"expected_generation": generation} + ) + + +def vector_search(client, **kwargs): + return client.post("/api/retrieval/vector", json={"query": "alpha query", **kwargs}) + + +def test_exact_cosine_original_chunks_category_and_mixed_history(client, provider): + first = add_item(client, content="alpha 占位正文", category="交通") + second = add_item(client, content="beta 占位正文", category="其他") + third = add_item(client, content="gamma 占位正文", category="其他") + add_item(client, status="pending", content="alpha 待审核不应进入") + text_ = text_version(client) + version = vector_build(client, text_) + assert ( + version["status"] == "ready" and version["dimension"] == 3 and version["chunk_count"] == 3 + ) + assert vector_search(client).status_code == 409 + assert vector_publish(client, version).json()["is_current"] + result = vector_search(client).json() + assert result["mode"] == "vector" and result["total_hits"] == 3 + assert [h["knowledge_id"] for h in result["results"]] == [ + first["id"], + second["id"], + third["id"], + ] + assert [h["score"] for h in result["results"]] == [1.0, 0.0, 0.0] + assert result["config"]["space"]["id"] == provider.spec.space_id + assert result["text_index_version_id"] == text_["id"] + assert result["vector_version_id"] == version["id"] + assert result["corpus_fingerprint"] == text_["corpus_fingerprint"] + assert result["elapsed_ms"] > 0 + assert result["results"][0]["snippet"] == first["content"] + assert result["results"][0]["snippet_start"] == 0 + assert result["results"][0]["snippet_end"] == len(first["content"]) + assert result["results"][0]["source_url"] == "https://example.com/" + filtered = vector_search(client, category="其他", top_k=1).json() + assert filtered["corpus_size"] == 2 and filtered["results"][0]["knowledge_id"] == second["id"] + empty = vector_search(client, category="不存在分类").json() + assert empty["status"] == "no_evidence" and empty["results"] == [] + keyword = client.post("/api/retrieval", json={"query": "alpha"}).json() + history = client.get("/api/retrieval/runs?page_size=100").json() + assert {r["mode"] for r in history["items"]} == {"keyword", "vector"} + assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result + assert client.get(f"/api/retrieval/runs/{keyword['run_id']}").json() == keyword + client.cookies.clear() + assert vector_search(client).status_code == 200 + assert client.get("/api/retrieval/runs").status_code == 401 + + +def test_multiple_chunks_keep_original_offsets_and_order(client, provider): + item = add_item(client, content="alpha 占位正文。" * 45) + version = vector_build(client, text_version(client)) + vector_publish(client, version) + hits = vector_search(client, top_k=20).json()["results"] + assert len(hits) > 1 + assert [h["chunk_ordinal"] for h in hits] == list(range(len(hits))) + assert len({h["chunk_id"] for h in hits}) == len(hits) + for h in hits: + assert h["snippet"] == item["content"][h["snippet_start"] : h["snippet_end"]] + + +def test_same_dimension_different_revision_is_a_separate_space(client, provider, migrated_database): + add_item(client) + text_ = text_version(client) + first = vector_build(client, text_) + vector_publish(client, first) + old_space = provider.spec.space_id + provider.spec = replace(provider.spec, revision="b" * 40) + assert vector_search(client).status_code == 409 + second = vector_build(client, text_) + assert second["space_id"] != old_space and second["dimension"] == first["dimension"] + assert vector_publish(client, second, 1).status_code == 200 + assert vector_search(client).json()["config"]["space"]["id"] == provider.spec.space_id + assert client.get("/api/vector-indexes/spaces").json()["total"] == 2 + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 2 + assert db.scalar(select(func.count()).select_from(ChunkVector)) == 2 + + +@pytest.mark.parametrize("bad", ["dimension", "zero", "nan", "count", "overflow", "unavailable"]) +def test_invalid_or_failed_encode_never_persists_partial_vectors( + client, provider, migrated_database, bad +): + add_item(client, content="alpha 占位") + text_ = text_version(client) + original = vector_build(client, text_) + vector_publish(client, original) + before = client.get("/api/vector-indexes/active").json() + + def fail(texts, **kwargs): + if bad == "overflow": + raise EmbeddingError("token limit") + if bad == "unavailable": + raise OSError("private-path-must-not-appear") + return { + "dimension": [[1, 0]], + "zero": [[0, 0, 0]], + "nan": [[float("nan"), 0, 0]], + "count": [], + }[bad] + + provider.encode = fail + response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}) + assert response.status_code == (503 if bad == "unavailable" else 422) + assert "private-path" not in response.text + assert client.get("/api/vector-indexes/active").json() == before + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(VectorVersion)) == 1 + assert db.scalar(select(func.count()).select_from(ChunkVector)) == 1 + + +def test_build_and_query_encode_release_locks_but_recheck_catalog( + client, provider, migrated_database +): + item = add_item(client, content="alpha 占位") + text_ = text_version(client) + original = vector_build(client, text_) + vector_publish(client, original) + + def mutate(query): + with Session(migrated_database) as other: + other.execute(text("SET LOCAL lock_timeout = '100ms'")) + state = locked_state(other) # Must succeed during both document and query inference. + state.catalog_revision += 1 + other.execute( + text("UPDATE knowledge_items SET content=:value WHERE id=:id"), + {"value": "修订占位", "id": item["id"]}, + ) + other.commit() + + provider.callback = mutate + response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}) + assert response.status_code == 409 + assert not client.get("/api/vector-indexes/active").json()["is_current"] + assert vector_search(client).status_code == 409 + # Restore an active text/vector snapshot, then race during query encoding too. + provider.callback = None + text_ = text_version(client) + replacement = vector_build(client, text_) + vector_publish(client, replacement, 1) + provider.callback = mutate + assert vector_search(client).status_code == 409 + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(VectorVersion)) == 2 + assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0 + + +def test_query_rejects_publication_changed_during_encode(client, provider, migrated_database): + add_item(client) + text_ = text_version(client) + first, second = vector_build(client, text_), vector_build(client, text_) + vector_publish(client, first) + + def switch(query): + if query: + with Session(migrated_database) as other: + vectors.publish_vectors(other, UUID(second["id"]), 1, second["created_by"]) + + provider.callback = switch + assert vector_search(client).status_code == 409 + assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == second["id"] + + +def test_text_switch_during_build_discards_new_vectors(client, provider, migrated_database): + add_item(client) + first = text_version(client) + replacement = build(client) + + def switch(query): + with Session(migrated_database) as other: + publish_index(other, UUID(replacement["id"]), 1, replacement["created_by"]) + + provider.callback = switch + assert ( + client.post("/api/vector-indexes", json={"text_index_version_id": first["id"]}).status_code + == 409 + ) + assert client.get("/api/vector-indexes").json()["total"] == 0 + + +def test_out_of_band_evidence_change_fails_closed(client, provider, migrated_database): + item = add_item(client) + text_ = text_version(client) + version = vector_build(client, text_) + vector_publish(client, version) + with migrated_database.begin() as c: + c.execute( + text("UPDATE knowledge_items SET content='SQL 更新占位' WHERE id=:id"), + {"id": item["id"]}, + ) + assert not client.get("/api/vector-indexes/active").json()["is_current"] + assert vector_search(client).status_code == 409 + assert vector_publish(client, version, 1).status_code == 409 + + +def test_optimistic_publication_audit_and_rollback(client, provider, migrated_database): + add_item(client) + text_ = text_version(client) + first, second = vector_build(client, text_), vector_build(client, text_) + assert vector_publish(client, first).json()["generation"] == 1 + assert vector_publish(client, second).status_code == 409 + assert vector_publish(client, first, 1).json()["generation"] == 1 + with migrated_database.begin() as c: + c.execute( + text( + "ALTER TABLE vector_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID" + ) + ) + assert vector_publish(client, second, 1).status_code == 409 + assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == first["id"] + assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "published" + assert client.get(f"/api/vector-indexes/{second['id']}").json()["status"] == "ready" + assert client.get("/api/vector-indexes/publications").json()["total"] == 1 + with migrated_database.begin() as c: + c.execute(text("ALTER TABLE vector_publications DROP CONSTRAINT test_reject_publication")) + assert vector_publish(client, second, 1).json()["generation"] == 2 + assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "superseded" + + +def test_vector_insert_failure_rolls_back_entire_build(client, provider, migrated_database): + add_item(client) + text_ = text_version(client) + with migrated_database.begin() as c: + c.execute( + text( + "ALTER TABLE chunk_vectors ADD CONSTRAINT test_reject_vector CHECK (false) NOT VALID" + ) + ) + assert ( + client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code + == 409 + ) + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(VectorVersion)) == 0 + assert db.scalar(select(func.count()).select_from(ChunkVector)) == 0 + assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 0 + + +@pytest.mark.parametrize( + "column,value", + [ + ("dimension", 2), + ("space_id", "0" * 64), + ("text_index_version_id", str(uuid4())), + ("embedding", [1, 0]), + ("embedding", [2, 0, 0]), + ("embedding", [float("nan"), 0, 0]), + ], +) +def test_postgresql_rejects_cross_scope_dimensions_and_nonunit_values( + client, provider, migrated_database, column, value +): + add_item(client) + version = vector_build(client, text_version(client)) + with Session(migrated_database) as db: + with pytest.raises(DBAPIError): + db.execute( + update(ChunkVector) + .where(ChunkVector.version_id == version["id"]) + .values(**{column: value}) + ) + db.commit() + db.rollback() + row = db.scalar(select(ChunkVector)) + assert row.space_id == provider.spec.space_id and row.dimension == 3 + + +def test_same_width_cross_space_and_cross_text_chunks_are_rejected_by_database( + client, provider, migrated_database +): + add_item(client) + original_text = text_version(client) + first = vector_build(client, original_text) + provider.spec = replace(provider.spec, revision="c" * 40) + second = vector_build(client, original_text) + newer_text = text_version(client) + with Session(migrated_database) as db: + other_chunk = db.scalar(select(Chunk.id).where(Chunk.index_version_id == newer_text["id"])) + for fields in [{"space_id": second["space_id"]}, {"chunk_id": other_chunk}]: + with pytest.raises(DBAPIError): + db.execute( + update(ChunkVector) + .where(ChunkVector.version_id == first["id"]) + .values(**fields) + ) + db.commit() + db.rollback() + + +def test_incomplete_vector_set_cannot_be_published_or_queried(client, provider, migrated_database): + add_item(client) + version = vector_build(client, text_version(client)) + vector_publish(client, version) + with migrated_database.begin() as c: + c.execute(text("DELETE FROM chunk_vectors")) + assert vector_publish(client, version, 1).status_code == 409 + assert vector_search(client).status_code == 409 + + +def test_only_current_text_and_bounded_complete_snapshots_can_build( + client, provider, migrated_database, monkeypatch +): + assert vector_search(client).status_code == 409 + add_item(client) + unpublished = build(client) + assert ( + client.post( + "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]} + ).status_code + == 409 + ) + text_ = text_version(client) + assert ( + client.post( + "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]} + ).status_code + == 409 + ) + monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0) + assert ( + client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code + == 503 + ) + monkeypatch.undo() + with migrated_database.begin() as c: + c.execute(text("DELETE FROM chunks")) + assert ( + client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code + == 409 + ) + + +def test_empty_reviewed_corpus_can_be_published_without_fabricated_results(client, provider): + version = vector_build(client, text_version(client)) + assert version["chunk_count"] == 0 + assert vector_publish(client, version).json()["is_current"] + result = vector_search(client).json() + assert ( + result["status"] == "no_evidence" and result["results"] == [] and result["corpus_size"] == 0 + ) + + +def test_metadata_and_build_require_admin_and_csrf(client, provider): + add_item(client) + version = vector_build(client, text_version(client)) + assert ( + client.post( + "/api/vector-indexes", + json={"text_index_version_id": version["text_index_version_id"]}, + headers={"X-CSRF-Token": ""}, + ).status_code + == 403 + ) + client.cookies.clear() + for method, path in [ + ("GET", "/api/vector-indexes"), + ("GET", "/api/vector-indexes/active"), + ("GET", "/api/vector-indexes/spaces"), + ("GET", "/api/vector-indexes/publications"), + ("GET", f"/api/vector-indexes/{version['id']}"), + ("POST", "/api/vector-indexes"), + ("POST", f"/api/vector-indexes/{version['id']}/publish"), + ]: + assert client.request(method, path, json={}).status_code == 401 + assert ( + client.post( + "/api/auth/register", json={"username": "vector_user", "password": "test-password-123"} + ).status_code + == 201 + ) + auth = client.post( + "/api/auth/login", json={"username": "vector_user", "password": "test-password-123"} + ).json() + client.headers["X-CSRF-Token"] = auth["csrf_token"] + assert client.get("/api/vector-indexes").status_code == 403 + assert client.post("/api/vector-indexes", json={}).status_code == 403 + + +def test_vector_publication_serializes_with_catalog_and_other_publishers( + client, provider, migrated_database +): + add_item(client) + text_ = text_version(client) + first, second = vector_build(client, text_), vector_build(client, text_) + with Session(migrated_database) as owner, Session(migrated_database) as other: + locked_state(owner) + other.execute(text("SET LOCAL lock_timeout = '100ms'")) + with pytest.raises(OperationalError) as blocked: + vectors.publish_vectors(other, UUID(first["id"]), 0, first["created_by"]) + assert blocked.value.orig.sqlstate == "55P03" + other.rollback() + vectors.publish_vectors(owner, UUID(first["id"]), 0, first["created_by"]) + with pytest.raises(HTTPException) as stale: + vectors.publish_vectors(other, UUID(second["id"]), 0, second["created_by"]) + assert stale.value.status_code == 409 + other.rollback() + with Session(migrated_database) as db: + assert db.scalar(select(func.count()).select_from(VectorPublication)) == 1 + + +def test_lazy_runtime_is_offline_and_does_not_wait_when_busy(monkeypatch): + from app.services import embedding_runtime + + calls = [] + + def load(spec, **kwargs): + calls.append(kwargs) + return ToyProvider() + + monkeypatch.setattr(embedding_runtime, "LocalEmbeddingProvider", load) + runtime = EmbeddingRuntime(ToyProvider.spec, None, 8, 2) + assert calls == [] + assert runtime.encode(["alpha"]) == [[1, 0, 0]] + assert calls[0]["offline"] is True + runtime.encode(["alpha"], query=True) + assert len(calls) == 1 + with runtime._lock: + with pytest.raises(HTTPException) as busy: + runtime.encode(["alpha"]) + assert busy.value.status_code == 503 + + +def test_disabled_runtime_is_explicit_and_keyword_baseline_remains_available(client): + assert vector_search(client).status_code == 503 + assert client.post("/api/retrieval", json={"query": "那拉提"}).status_code == 200 + + +def test_different_width_spaces_coexist_and_queries_never_compare_across_widths(client, provider): + add_item(client) + text_ = text_version(client) + first = vector_build(client, text_) + vector_publish(client, first) + provider.spec = replace(provider.spec, dimension=2) + provider.encode = lambda texts, **kwargs: [[1, 0] for _ in texts] + second = vector_build(client, text_) + assert vector_publish(client, second, 1).status_code == 200 + result = vector_search(client) + assert result.status_code == 200, result.text + assert result.json()["total_hits"] == 1 and result.json()["results"][0]["score"] == 1 + assert client.get("/api/vector-indexes/spaces").json()["total"] == 2 + + +def test_new_text_publication_requires_corresponding_vectors_and_keeps_history(client, provider): + add_item(client) + old_text = text_version(client) + old = vector_build(client, old_text) + vector_publish(client, old) + history = vector_search(client).json() + new_text = text_version(client) + active = client.get("/api/vector-indexes/active").json() + assert not active["is_current"] and active["active_version"]["id"] == old["id"] + assert vector_search(client).status_code == 409 + assert vector_publish(client, old, 1).status_code == 409 + replacement = vector_build(client, new_text) + assert vector_publish(client, replacement, 1).json()["is_current"] + assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history + assert client.get("/api/vector-indexes/publications").json()["total"] == 2 + + +def test_snapshot_text_corruption_cannot_be_encoded_published_or_served( + client, provider, migrated_database +): + add_item(client) + text_ = text_version(client) + version = vector_build(client, text_) + vector_publish(client, version) + with migrated_database.begin() as c: + c.execute(text("UPDATE chunks SET text='篡改占位'")) + assert not client.get("/api/vector-indexes/active").json()["is_current"] + assert ( + client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code + == 409 + ) + assert vector_publish(client, version, 1).status_code == 409 + assert vector_search(client).status_code == 409 + + +def test_request_shape_pagination_and_missing_vector_version(client, provider): + assert ( + client.post( + "/api/vector-indexes", json={"text_index_version_id": str(uuid4()), "embedding": [1, 0]} + ).status_code + == 422 + ) + assert vector_search(client, top_k=21).status_code == 422 + assert vector_search(client, mode="keyword").status_code == 422 + assert vector_search(client, query=" ").status_code == 422 + assert client.get("/api/vector-indexes?page=0").status_code == 422 + assert client.get(f"/api/vector-indexes/{uuid4()}").status_code == 404 + assert client.get("/api/vector-indexes/spaces").json()["total"] == 0 + + +def test_blank_optional_cache_path_is_not_the_working_directory(): + from app.core.config import Settings + + assert Settings(embedding_cache_dir="", _env_file=None).embedding_cache_dir is None diff --git a/docs/LOCAL_EMBEDDING_EVALUATION.md b/docs/LOCAL_EMBEDDING_EVALUATION.md index 27d8eb8..00ca451 100644 --- a/docs/LOCAL_EMBEDDING_EVALUATION.md +++ b/docs/LOCAL_EMBEDDING_EVALUATION.md @@ -74,4 +74,6 @@ export HF_HUB_DISABLE_XET=1 将 Provider 接入已发布的文本版本，完成 pgvector 存储与严格空间隔离、向量检索及 RRF，再使用真实审核语料评测。模型编码耗时应在发布锁外运行，发布前重新核对目录/语料版本；过期或失败时保留旧版本。当前没有新增向量数据库字段、检索 API 或论文结论。 +以上描述 PR #9 独立评估阶段的范围；后续向量存储与精确检索集成另见 [VECTOR_API.md](VECTOR_API.md)，不改变本页已记录的合成评估结果。 + 依据：[BGE small 官方模型说明](https://huggingface.co/BAAI/bge-small-zh-v1.5)、[BGE base 官方文件](https://huggingface.co/BAAI/bge-base-zh-v1.5/tree/f03589ceff5aac7111bd60cfc7d497ca17ecac65)、[Transformers 模型加载参数](https://huggingface.co/docs/transformers/main_classes/model#transformers.PreTrainedModel.from_pretrained)、[PyTorch CPU 安装](https://pytorch.org/get-started/locally/)、[Hugging Face 固定修订下载](https://huggingface.co/docs/huggingface_hub/guides/download)。 diff --git a/docs/VECTOR_API.md b/docs/VECTOR_API.md new file mode 100644 index 0000000..0b5f9f0 --- /dev/null +++ b/docs/VECTOR_API.md @@ -0,0 +1,116 @@ +# T07 向量快照、发布与精确检索 + +本分支 `feat/backend-vector-retrieval` 组合待审 #8 的文本版本与 #9 的固定版本 CPU Provider，继承 #7/#6。合并到 `dev` 前须按依赖审阅；不自行批准或合并。双方阶段文档均保留，独立阶段差异从组合基线 `bd0b738` 起查看。 + +交付 pgvector 存储、空间约束、独立发布、精确余弦检索与历史。关键词 `POST /api/retrieval` 保持原请求/分数/条目粒度；向量使用新的 `POST /api/retrieval/vector`，粒度为 chunk。本阶段尚未实现 RRF、ANN、RAG、持久化后台任务或正式领域评测。 + +## 配置与迁移 + +执行 `alembic upgrade head` 到 `0006_vectors`。迁移增加 `embedding_spaces`、`vector_versions`、`chunk_vectors`、`vector_state` 和 `vector_publications`，以及 chunks 的复合唯一约束。保留账号、目录、原文版本及既有检索日志，不预置模型或向量。 + +扩展类型/函数/距离算子显式引用 `public.vector`，业务表仍放在连接的当前 Schema，不把 public 加入测试搜索路径。已有 vector 扩展须位于 public；其他安装位置会明确报错，迁移不会移动共享扩展。新数据库可先由有权限的数据库管理员执行 `CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public`。 + +默认 `EMBEDDING_ENABLED=false`，普通 API 启动不导入 PyTorch 或加载权重。模型相关请求返回 503，不回退占位向量或伪造空结果；关键词和管理元数据接口可继续使用。启用前按 [本地评估说明](LOCAL_EMBEDDING_EVALUATION.md) 安装独立 CPU 环境、缓存固定修订；该运行环境还需安装 `backend/requirements.txt` 的 API 依赖。 + +| 变量 | 默认值 | 作用 | +|---|---|---| +| `EMBEDDING_ENABLED` | false | 显式启用本地编码 | +| `EMBEDDING_MODEL` | small | 仅 small/base；完整修订、维度等由固定 Spec 决定 | +| `EMBEDDING_CACHE_DIR` | 未设置 | 固定模型缓存目录；空值使用库默认缓存 | +| `EMBEDDING_BATCH_SIZE` | 8 | 1–32 | +| `EMBEDDING_THREADS` | 2 | 1–4 个 CPU 线程 | + +开发资源优先暂用 small，正式模型仍待已核验领域资料/问题集评测。API 始终只从本机固定修订缓存读取，禁止在请求中下载权重，不接收模型路径、任意仓库名或用户向量。CPU Provider 首次使用才加载，单进程同时只允许一个编码调用，繁忙立即返回 503；当前建议单 worker 本机运行，不能据此宣称已实现跨进程队列或生产并发容量。 + +Linux 仓库根目录示例（先完成模型准备及数据库迁移；本机 Cookie/Origin 配置仍按认证说明设置）： + +```bash +task_vector_root=$(git rev-parse --absolute-git-dir) +uv pip install --python "$task_vector_root/embedding-eval-venv/bin/python" \ + -r backend/requirements.txt +export EMBEDDING_ENABLED=true EMBEDDING_MODEL=small +export EMBEDDING_CACHE_DIR="$task_vector_root/embedding-model-cache" +export HF_HUB_OFFLINE=1 HF_HUB_DISABLE_TELEMETRY=1 HF_HUB_DISABLE_IMPLICIT_TOKEN=1 +export TOKENIZERS_PARALLELISM=false +cd backend +"$task_vector_root/embedding-eval-venv/bin/python" -m uvicorn app.main:app --host 127.0.0.1 +``` + +环境、模型权重和实际日志只留本机，不提交。普通 API 环境不自动增加 PyTorch 依赖。 + +## 构建与发布协议 + +1. 通过 `/api/indexes` 构建并发布当前有效审核语料的文本版本。建议开发先用 `chunk_size=200, overlap=20`；实际 tokenizer 仍须检查，字符数不保证 token 数。 +2. 管理员向 `POST /api/vector-indexes` 提交 `{"text_index_version_id":"<已发布文本版本 UUID>"}`。服务仅接受当前有效的文本发布版本；整份快照最多 2000 chunks，超限返回 503，不截断语料。 +3. 服务短暂锁住目录/文本状态，复核实际审核语料指纹、原文窗口、偏移、哈希、出处和完整数量；复制输入后提交并释放事务。模型加载与正文编码全部在锁外运行。 +4. 编码成功且向量数量/维度/有限数/L2 范数通过后，再取得同一状态锁，重新校验文本代数、目录修订、当前指针、语料指纹及实际 chunks。变化则 409，整批不入库。成功只保存 ready 向量版本，不自动发布。 +5. 管理员向 `POST /api/vector-indexes/{id}/publish` 提交 `{"expected_generation":0}`。向量代数独立于文本代数；原子切换当前指针、前后版本状态和发布历史。旧代数、过期文本或不完整向量集均拒绝。相同当前版本和当前代数为幂等读取。 + +上述构建同步执行，没有后台任务 ID 或自动重试。超过模型 token 上限返回 422，应重新切分并发布文本，再重建；不会静默截断。编码不可用/繁忙为 503；数据库约束或版本竞争为 409。失败不会改写旧指针或留下部分向量版本。空的有效审核语料允许生成/发布 0 chunks 的版本；随后检索如实返回 `no_evidence`。 + +目录变化会使既有文本和向量版本过期；文本切换即使内容相同也要求对应新文本版本的向量。旧指针和日志保留为历史，公开检索对过期状态返回 409，不继续使用旧审核依据。切换模型配置后，若当前空间不一致，也须重新构建/发布相应空间或恢复匹配的运行配置。 + +## 数据库空间隔离 + +`space_id` 为固定 Spec 的 SHA-256，包含模型名、完整 revision、维度、token 上限、查询指令、编码实现（CLS/L2/CPU FP32）及权重格式。同维度的不同修订/指令/配置仍是不同空间。 + +数据库同时约束： + +- 版本的 `(space_id, dimension)` 必须引用同一 EmbeddingSpace。 +- 每条向量的 `(version_id, text_index_version_id, space_id, dimension)` 必须引用同一向量版本。 +- `(chunk_id, text_index_version_id)` 必须引用对应文本版本中的 chunk。 +- `public.vector_dims` 必须等于声明维度；`public.vector_norm` 须在 `0.9999–1.0001`，pgvector 拒绝非有限元素。 + +使用无固定维度的 vector 列容纳独立空间，查询显式限定版本、空间、文本版本和维度，然后使用余弦距离。当前为有限语料精确扫描，没有 HNSW/IVFFlat；模型空间和向量发布均不由客户端参数任意指定。 + +## 接口与字段交接 + +`/api/vector-indexes` 下所有读写要求 admin；写入沿用当前 Cookie、CSRF 和明确 Origin。元数据不要求加载模型。分页使用 `{items,total,page,page_size}`，`page_size` 最大 100。 + +| 方法与路径 | 请求/用途 | 响应 | +|---|---|---| +| POST `/api/vector-indexes` | `text_index_version_id` | 201，VectorVersion | +| GET `/api/vector-indexes` | page/page_size | 版本分页 | +| GET `/api/vector-indexes/{id}` | UUID | VectorVersion，缺失 404 | +| GET `/api/vector-indexes/active` | 当前元数据 | generation/text_generation/catalog_revision/is_current/active_version | +| GET `/api/vector-indexes/spaces` | page/page_size | 空间 id/dimension/spec | +| POST `/api/vector-indexes/{id}/publish` | expected_generation | 当前向量元数据 | +| GET `/api/vector-indexes/publications` | page/page_size | 代数、前后向量版本、actor_id、UTC 时间 | +| POST `/api/retrieval/vector` | 下方请求 | VectorRetrievalResponse | + +VectorVersion 字段：`id/text_index_version_id/space_id/dimension/chunk_count/status/created_by/created_at/published_at`，状态为 ready/published/superseded；时间统一 UTC ISO 8601。发布历史的 `version_id` 是向量版本，不能与文本发布历史混用。 + +公开检索请求示例： + +```json +{"query":"待检索问题","mode":"vector","top_k":5,"category":"交通"} +``` + +`mode` 可省略，固定为 vector；query 1–500 字符，top_k 1–20，category 可省略。未知字段拒绝。流程在查询编码前后均复核当前发布/审核状态和完整向量集；编码时不持有目录发布锁，期间任何发布代数变化都会返回 409，重试使用新快照。 + +响应记录：`run_id/query/mode/status/top_k/category/vector_version_id/text_index_version_id/generation/text_generation/corpus_fingerprint/corpus_size/total_hits/results/elapsed_ms/created_at/config`。`config` 固定 scorer=`pgvector-cosine-exact-v1`、ranking_unit=`chunk`，并保存完整空间 Spec。 + +每个 result 沿用关键词出处字段，再加 `chunk_id/chunk_ordinal`；snippet 为完整原文 chunk，snippet_start/end 是原正文的 Unicode 字符偏移（左闭右开），matched_terms 为空数组。score 为 `1 - cosine_distance`，不是正确率或置信度，也未设领域无依据阈值。按实际距离、knowledge_id、chunk_ordinal、chunk_id 排序；同一知识可出现多个 chunks，不能把结果数量当作命中的知识数。 + +`corpus_size/total_hits` 为分类筛选后的候选 chunk 数，不是超过置信阈值的数量；corpus_fingerprint 是整个已发布文本语料指纹，分类另行保存。耗时包括两次审核/快照检查、查询编码（首次可含模型加载）、SQL 距离排序与片段生成，不含最终日志提交及网络。不同于关键词基线的条目粒度，正式对照前必须统一评测单位。 + +成功和零候选均保存 `retrieval_runs` 不可变快照。既有历史接口现在返回 keyword/vector 判别联合，仍限 admin；不改写过去结果，也不因历史可读取就代表资料仍有效。前端调用层已同步 `services/vectors.ts` 与历史联合类型，页面和浏览器仍待接入验收。 + +## 回滚与验证 + +备份后 `alembic downgrade 0005_chunk_indexes` 删除本阶段空间、向量、向量状态和发布历史，保留文本 chunks/发布、账号、知识及所有检索结果快照。切回对应旧版后端前须处理其历史接口只认识 keyword 响应的兼容性；重升后向量指针为空，需要重新构建，不会复原被删权重结果。当前 API 启动会拒绝旧 Schema。 + +默认测试用明确的三维坐标 fixture 验证 SQL 数学、空间隔离、并发与回滚，绝不代表实际模型质量。真实模型贯通用例显式启用，在独立模型环境运行： + +```bash +task_vector_root=$(git rev-parse --absolute-git-dir) +uv pip install --python "$task_vector_root/embedding-eval-venv/bin/python" \ + -r backend/requirements-dev.txt pgserver +RUN_LOCAL_EMBEDDING_TEST=1 EMBEDDING_CACHE_DIR="$task_vector_root/embedding-model-cache" \ + "$task_vector_root/embedding-eval-venv/bin/python" -m pytest -q \ + backend/tests/test_vector_runtime.py +``` + +该用例用合成测试文字，验证真实 small 的 512 维编码、HTTP 构建/发布/查询、pgvector 持久化及新进程重启后的空间、发布、命中和历史一致；不是领域分数或生产性能实验。Docker 容器、前端浏览器、RRF 和领域验收保持待办。 + +实现依据：[pgvector 精确查询与余弦算子](https://github.com/pgvector/pgvector#querying)、[混合维度存储约束](https://github.com/pgvector/pgvector#can-i-store-vectors-with-different-dimensions-in-the-same-column)、[SQLAlchemy 集成](https://github.com/pgvector/pgvector-python#sqlalchemy)。 diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 9d7c9bc..d597640 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -1,4 +1,5 @@ import { apiRequest } from './services/session' +import type { VectorRetrievalResult } from './services/vectors' export const getHealth = () => apiRequest<{status:string}>('/api/health') export const askQuestion = (question:string) => apiRequest<{answer:string;status:string;sources:{title:string;source_url?:string|null}[]}>('/api/chat',{method:'POST',body:JSON.stringify({question})}) @@ -14,5 +15,6 @@ export function reviewKnowledge(id:number,payload:{expected_version:number;verif export type RetrievalHit={rank:number;knowledge_id:number;knowledge_version:number;document_id:number|null;document_version:number|null;source_id:number|null;source_version:number|null;source_name:string|null;source_url:string;title:string;category:string;snippet:string;snippet_start:number;snippet_end:number;score:number;matched_terms:string[]} export type RetrievalResult={run_id:string;query:string;mode:'keyword';status:'ok'|'no_evidence';top_k:number;category:string|null;query_terms:string[];corpus_size:number;corpus_fingerprint:string;total_hits:number;results:RetrievalHit[];elapsed_ms:number;created_at:string;config:{tokenizer:string;scorer:string;k1:number;b:number;title_weight:number;stopwords_version:string}} export function retrieveKnowledge(payload:{query:string;mode?:'keyword';top_k?:number;category?:string}){return apiRequest('/api/retrieval',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)})} -export function getRetrievalRun(runId:string){return apiRequest(`/api/retrieval/runs/${encodeURIComponent(runId)}`)} -export function getRetrievalRuns(page=1,pageSize=20){return apiRequest<{items:RetrievalResult[];total:number;page:number;page_size:number}>(`/api/retrieval/runs?page=${page}&page_size=${pageSize}`)} +export type RetrievalRunResult = RetrievalResult | VectorRetrievalResult +export function getRetrievalRun(runId:string){return apiRequest(`/api/retrieval/runs/${encodeURIComponent(runId)}`)} +export function getRetrievalRuns(page=1,pageSize=20){return apiRequest<{items:RetrievalRunResult[];total:number;page:number;page_size:number}>(`/api/retrieval/runs?page=${page}&page_size=${pageSize}`)} diff --git a/frontend/src/services/vectors.ts b/frontend/src/services/vectors.ts new file mode 100644 index 0000000..35de688 --- /dev/null +++ b/frontend/src/services/vectors.ts @@ -0,0 +1,18 @@ +import { apiRequest } from './session' +import type { RetrievalHit } from '../api' +import type { IndexPublication } from './indexes' + +type Page = { items: T[]; total: number; page: number; page_size: number } +export type EmbeddingSpec = { model: string; revision: string; dimension: number; token_limit: number; query_instruction: string; implementation: string; checkpoint_format: string } +export type EmbeddingSpace = { id: string; dimension: number; spec: EmbeddingSpec } +export type VectorVersion = { id: string; text_index_version_id: string; space_id: string; dimension: number; chunk_count: number; status: 'ready' | 'published' | 'superseded'; created_by: number; created_at: string; published_at: string | null } +export type ActiveVectors = { generation: number; text_generation: number; catalog_revision: number; is_current: boolean; active_version: VectorVersion | null } +export type VectorRetrievalResult = { run_id: string; query: string; mode: 'vector'; status: 'ok' | 'no_evidence'; top_k: number; category: string | null; vector_version_id: string; text_index_version_id: string; generation: number; text_generation: number; corpus_fingerprint: string; corpus_size: number; total_hits: number; results: (RetrievalHit & { chunk_id: string; chunk_ordinal: number })[]; elapsed_ms: number; created_at: string; config: { scorer: 'pgvector-cosine-exact-v1'; ranking_unit: 'chunk'; space: EmbeddingSpace } } +export const buildVectorIndex = (textIndexVersionId: string) => apiRequest('/api/vector-indexes', { method: 'POST', body: JSON.stringify({ text_index_version_id: textIndexVersionId }) }, true) +export const getVectorVersions = (page = 1, pageSize = 20) => apiRequest>(`/api/vector-indexes?page=${page}&page_size=${pageSize}`) +export const getVectorVersion = (id: string) => apiRequest(`/api/vector-indexes/${encodeURIComponent(id)}`) +export const getActiveVectors = () => apiRequest('/api/vector-indexes/active') +export const getEmbeddingSpaces = (page = 1, pageSize = 20) => apiRequest>(`/api/vector-indexes/spaces?page=${page}&page_size=${pageSize}`) +export const publishVectorIndex = (id: string, expectedGeneration: number) => apiRequest(`/api/vector-indexes/${encodeURIComponent(id)}/publish`, { method: 'POST', body: JSON.stringify({ expected_generation: expectedGeneration }) }, true) +export const getVectorPublications = (page = 1, pageSize = 20) => apiRequest>(`/api/vector-indexes/publications?page=${page}&page_size=${pageSize}`) +export const retrieveVectors = (payload: { query: string; mode?: 'vector'; top_k?: number; category?: string }) => apiRequest('/api/retrieval/vector', { method: 'POST', body: JSON.stringify(payload) })