saint/border-red-post-nalati
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat: persist reviewed chunks and index publication versions

Browse Source
This commit is contained in:
Kronecker
2026-10-16 08:44:30 +08:00
parent 3f25ffab83
commit 41b3ae0e41
20 changed files with 1027 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/documents.py
+6 -1
View File
@@ -15,8 +15,13 @@ from app.services.catalog import (
record_event,
update_values,
)
from app.services.indexing import catalog_write_guard
router = APIRouter(prefix="/documents", tags=["documents"], dependencies=[Depends(require_admin)])
router = APIRouter(
prefix="/documents",
tags=["documents"],
dependencies=[Depends(require_admin), Depends(catalog_write_guard)],
)
@router.get("", response_model=Page[DocumentRead])
backend/app/api/routes/indexes.py
+91
View File
@@ -0,0 +1,91 @@
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.indexing import Chunk, IndexPublication, IndexVersion
from app.schemas.catalog import Page
from app.schemas.indexing import (
ActiveIndex,
BuildIndex,
ChunkRead,
IndexRead,
PublicationRead,
PublishIndex,
)
from app.services.auth import CurrentSession, require_admin
from app.services.catalog import page_result
from app.services.indexing import active_index, build_index, publish_index, version_or_404
router = APIRouter(prefix="/indexes", tags=["indexes"], dependencies=[Depends(require_admin)])
@router.post("", response_model=IndexRead, status_code=201)
def build(
payload: BuildIndex,
current: CurrentSession = Depends(require_admin),
db: Session = Depends(get_db),
):
return build_index(db, payload, current.user.id)
@router.get("", response_model=Page[IndexRead])
def list_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(IndexVersion).order_by(IndexVersion.created_at.desc(), IndexVersion.id),
page,
page_size,
)
@router.get("/active", response_model=ActiveIndex)
def active(db: Session = Depends(get_db)):
return active_index(db)
@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(IndexPublication).order_by(IndexPublication.generation.desc()), page, page_size
)
@router.get("/{version_id}", response_model=IndexRead)
def read_version(version_id: UUID, db: Session = Depends(get_db)):
return version_or_404(db, str(version_id))
@router.post("/{version_id}/publish", response_model=ActiveIndex)
def publish(
version_id: UUID,
payload: PublishIndex,
current: CurrentSession = Depends(require_admin),
db: Session = Depends(get_db),
):
return publish_index(db, version_id, payload.expected_generation, current.user.id)
@router.get("/{version_id}/chunks", response_model=Page[ChunkRead])
def chunks(
version_id: UUID,
knowledge_id: int | None = Query(None, gt=0),
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
db: Session = Depends(get_db),
):
version_or_404(db, str(version_id))
statement = select(Chunk).where(Chunk.index_version_id == str(version_id))
if knowledge_id is not None:
statement = statement.where(Chunk.knowledge_id == knowledge_id)
return page_result(db, statement.order_by(Chunk.knowledge_id, Chunk.ordinal), page, page_size)
backend/app/api/routes/knowledge.py
+6 -1
View File
@@ -18,8 +18,13 @@ from app.services.catalog import (
resolve_knowledge_links,
update_values,
)
from app.services.indexing import catalog_write_guard
router = APIRouter(prefix="/knowledge", tags=["knowledge"], dependencies=[Depends(require_admin)])
router = APIRouter(
prefix="/knowledge",
tags=["knowledge"],
dependencies=[Depends(require_admin), Depends(catalog_write_guard)],
)
def filtered_query(q="", category=None, verification_status=None, source_id=None, document_id=None):
backend/app/api/routes/sources.py
+6 -1
View File
@@ -15,8 +15,13 @@ from app.services.catalog import (
record_event,
update_values,
)
from app.services.indexing import catalog_write_guard
router = APIRouter(prefix="/sources", tags=["sources"], dependencies=[Depends(require_admin)])
router = APIRouter(
prefix="/sources",
tags=["sources"],
dependencies=[Depends(require_admin), Depends(catalog_write_guard)],
)
@router.get("", response_model=Page[SourceRead])
backend/app/main.py
+4 -3
View File
@@ -8,7 +8,7 @@ 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, knowledge, retrieval, sources
from app.api.routes import auth, chat, documents, health, indexes, knowledge, retrieval, sources
from app.core.config import settings
from app.db.session import engine
from app.services.auth import dummy_hash
@@ -21,7 +21,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 != "0004_auth_sessions":
if revision != "0005_chunk_indexes":
raise RuntimeError(
"数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head"
)
@@ -36,7 +36,7 @@ def create_app(db_engine=engine) -> FastAPI:
finally:
db_engine.dispose()
app = FastAPI(title=settings.app_name, version="0.4.0", lifespan=lifespan)
app = FastAPI(title=settings.app_name, version="0.5.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
@@ -47,6 +47,7 @@ def create_app(db_engine=engine) -> FastAPI:
for router in (
health.router,
auth.router,
indexes.router,
chat.router,
knowledge.router,
sources.router,
backend/app/models/__init__.py
+6 -1
View File
@@ -1,4 +1,6 @@
from app.models.auth import AuthRateLimit, AuthSession, User
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
@@ -6,10 +8,13 @@ __all__ = [
"AuditEvent",
"AuthRateLimit",
"AuthSession",
"Chunk",
"Document",
"IndexPublication",
"IndexState",
"IndexVersion",
"KnowledgeItem",
"RetrievalRun",
"Source",
"User",
]
from app.models.auth import AuthRateLimit, AuthSession, User
backend/app/models/indexing.py
+74
View File
@@ -0,0 +1,74 @@
"""Immutable chunk snapshots and atomic publication metadata (no vector index yet)."""
from datetime import datetime
from sqlalchemy import (
JSON,
CheckConstraint,
DateTime,
ForeignKey,
Integer,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.orm import Mapped, mapped_column
from app.db.session import Base
class IndexVersion(Base):
__tablename__ = "index_versions"
__table_args__ = (
CheckConstraint(
"status IN ('ready', 'published', 'superseded')", name="ck_index_version_status"
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True)
status: Mapped[str] = mapped_column(String(20), nullable=False, default="ready")
config: Mapped[dict] = mapped_column(JSON, nullable=False)
corpus_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
catalog_revision: Mapped[int] = mapped_column(Integer, nullable=False)
item_count: Mapped[int] = mapped_column(Integer, nullable=False)
chunk_count: Mapped[int] = mapped_column(Integer, nullable=False)
created_by: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
class IndexState(Base):
__tablename__ = "index_state"
__table_args__ = (CheckConstraint("id = 1", name="ck_index_state_singleton"),)
id: Mapped[int] = mapped_column(primary_key=True)
active_version_id: Mapped[str | None] = mapped_column(ForeignKey("index_versions.id"))
generation: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
catalog_revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
class Chunk(Base):
__tablename__ = "chunks"
__table_args__ = (
UniqueConstraint("index_version_id", "knowledge_id", "ordinal", name="uq_chunk_position"),
CheckConstraint("start_offset >= 0 AND end_offset > start_offset", name="ck_chunk_offsets"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True)
index_version_id: Mapped[str] = mapped_column(ForeignKey("index_versions.id"), index=True)
knowledge_id: Mapped[int] = mapped_column(ForeignKey("knowledge_items.id"), nullable=False)
ordinal: Mapped[int] = mapped_column(Integer, nullable=False)
start_offset: Mapped[int] = mapped_column(Integer, nullable=False)
end_offset: Mapped[int] = mapped_column(Integer, nullable=False)
text: Mapped[str] = mapped_column(Text, nullable=False)
text_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
provenance: Mapped[dict] = mapped_column(JSON, nullable=False)
class IndexPublication(Base):
__tablename__ = "index_publications"
generation: Mapped[int] = mapped_column(primary_key=True)
previous_version_id: Mapped[str | None] = mapped_column(ForeignKey("index_versions.id"))
version_id: Mapped[str] = mapped_column(ForeignKey("index_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()
)
backend/app/schemas/indexing.py
+89
View File
@@ -0,0 +1,89 @@
from datetime import datetime, timezone
from typing import Literal
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from app.schemas.catalog import Input
class BuildIndex(Input):
chunk_size: int = Field(500, ge=100, le=2000)
overlap: int = Field(80, ge=0, le=1000)
@model_validator(mode="after")
def bounded_overlap(self):
if self.overlap > self.chunk_size // 2:
raise ValueError("重叠不得超过切分长度的一半")
return self
class PublishIndex(Input):
expected_generation: int = Field(ge=0)
class IndexConfig(BuildIndex):
splitter: Literal["unicode-window-v1"]
class ChunkProvenance(BaseModel):
knowledge_id: int
knowledge_version: int
document_id: int | None
document_version: int | None
source_id: int | None
source_version: int | None
source_name: str | None
source_url: str
title: str
category: str
class TimestampRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
@field_validator("created_at", "published_at", check_fields=False)
@classmethod
def utc_timestamp(cls, value):
return value.astimezone(timezone.utc) if value is not None else None
class IndexRead(TimestampRead):
id: UUID
status: Literal["ready", "published", "superseded"]
config: IndexConfig
corpus_fingerprint: str
catalog_revision: int
item_count: int
chunk_count: int
created_by: int
created_at: datetime
published_at: datetime | None
class ActiveIndex(BaseModel):
generation: int
catalog_revision: int
is_current: bool
active_version: IndexRead | None
class ChunkRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: UUID
index_version_id: UUID
knowledge_id: int
ordinal: int
start_offset: int
end_offset: int
text: str
text_sha256: str
provenance: ChunkProvenance
class PublicationRead(TimestampRead):
generation: int
previous_version_id: UUID | None
version_id: UUID
actor_id: int
published_at: datetime
backend/app/services/indexing.py
+147
View File
@@ -0,0 +1,147 @@
"""Bounded chunk snapshots and serialized, optimistic publication."""
import hashlib
from uuid import UUID, uuid4, uuid5
from fastapi import Depends, HTTPException, Request
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.catalog import utc_now
from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion
from app.schemas.indexing import ActiveIndex, BuildIndex, IndexRead
from app.services.retrieval import corpus_fingerprint, load_corpus
MAX_CHUNKS = 20_000
def locked_state(db: Session) -> IndexState:
state = db.scalar(
select(IndexState)
.where(IndexState.id == 1)
.with_for_update()
.execution_options(populate_existing=True)
)
if state is None:
raise HTTPException(503, "索引状态尚未初始化，请检查数据库迁移")
return state
def catalog_write_guard(request: Request, db: Session = Depends(get_db)):
if request.method in {"POST", "PATCH", "PUT", "DELETE"}:
# Always before catalog/parent row locks; failures roll back this revision as well.
state = locked_state(db)
state.catalog_revision += 1
def version_or_404(db: Session, version_id: str) -> IndexVersion:
version = db.get(IndexVersion, version_id)
if version is None:
raise HTTPException(404, "索引版本不存在")
return version
def split_text(content: str, size: int, overlap: int):
"""Offsets are original Python Unicode character offsets, end-exclusive."""
for start in range(0, len(content), size - overlap):
end = min(len(content), start + size)
yield start, end, content[start:end]
if end == len(content):
break
def build_index(db: Session, payload: BuildIndex, actor_id: int) -> IndexRead:
state = locked_state(db)
corpus = load_corpus(db, None)
version_id = uuid4()
chunks = []
for item in corpus:
provenance = {key: value for key, value in item.items() if key != "content"}
for ordinal, (start, end, content) in enumerate(
split_text(item["content"], payload.chunk_size, payload.overlap)
):
if len(chunks) >= MAX_CHUNKS:
raise HTTPException(503, "切分数量超过本阶段容量，请增加切分长度或减少重叠")
chunks.append(
Chunk(
id=str(uuid5(version_id, f"{item['knowledge_id']}:{ordinal}")),
index_version_id=str(version_id),
knowledge_id=item["knowledge_id"],
ordinal=ordinal,
start_offset=start,
end_offset=end,
text=content,
text_sha256=hashlib.sha256(content.encode()).hexdigest(),
provenance=provenance,
)
)
version = IndexVersion(
id=str(version_id),
status="ready",
config={"splitter": "unicode-window-v1", **payload.model_dump()},
corpus_fingerprint=corpus_fingerprint(corpus),
catalog_revision=state.catalog_revision,
item_count=len(corpus),
chunk_count=len(chunks),
created_by=actor_id,
)
db.add(version)
db.flush()
db.add_all(chunks)
result = IndexRead.model_validate(version)
db.commit()
return result
def active_index(db: Session, state: IndexState | None = None) -> ActiveIndex:
state = state or locked_state(db)
version = db.get(IndexVersion, state.active_version_id) if state.active_version_id else None
current = bool(version and version.catalog_revision == state.catalog_revision)
if current:
current = version.corpus_fingerprint == corpus_fingerprint(load_corpus(db, None))
return ActiveIndex(
generation=state.generation,
catalog_revision=state.catalog_revision,
is_current=current,
active_version=IndexRead.model_validate(version) if version else None,
)
def publish_index(db: Session, version_id: UUID, expected_generation: int, actor_id: int):
state = locked_state(db)
if state.generation != expected_generation:
raise HTTPException(409, "已发布版本发生变化，请刷新后重试")
version = version_or_404(db, str(version_id))
if (
version.catalog_revision != state.catalog_revision
or version.corpus_fingerprint != corpus_fingerprint(load_corpus(db, None))
):
raise HTTPException(409, "资料或审核依据已变化，请重新构建后发布")
if state.active_version_id == version.id:
return active_index(db, state)
previous = state.active_version_id
if previous:
db.get(IndexVersion, previous).status = "superseded"
state.active_version_id = version.id
state.generation += 1
version.status = "published"
version.published_at = utc_now()
db.add(
IndexPublication(
generation=state.generation,
previous_version_id=previous,
version_id=version.id,
actor_id=actor_id,
published_at=version.published_at,
)
)
# Freeze the response before commit expires ORM objects and releases the lock.
result = ActiveIndex(
generation=expected_generation + 1,
catalog_revision=version.catalog_revision,
is_current=True,
active_version=IndexRead.model_validate(version),
)
db.commit()
return result
backend/app/services/retrieval.py
+7 -3
View File
@@ -151,15 +151,19 @@ def load_corpus(db: Session, category: str | None) -> list[dict]:
return corpus
def corpus_fingerprint(corpus: list[dict]) -> str:
return hashlib.sha256(
json.dumps(corpus, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
).hexdigest()
def retrieve(db: Session, payload: RetrievalRequest) -> RetrievalResponse:
started = perf_counter()
terms = list(dict.fromkeys(tokenize(payload.query)))
if not terms:
raise HTTPException(422, "问题中没有可检索词，请补充具体关键词")
corpus = load_corpus(db, payload.category)
fingerprint = hashlib.sha256(
json.dumps(corpus, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
).hexdigest()
fingerprint = corpus_fingerprint(corpus)
counters = [
Counter(tokenize(item["title"]) * CONFIG.title_weight + tokenize(item["content"]))
for item in corpus