Merge pull request 'release: synchronize retrieval export to main' (#35) from release/retrieval-export-main into main

This commit was merged in pull request #35.
This commit is contained in:
2026-10-16 23:15:19 +08:00
12 changed files with 801 additions and 3 deletions
+1
View File
@@ -57,6 +57,7 @@ Vite 通常使用 http://localhost:5173。
- GET/POST `/api/documents`、GET/PATCH/DELETE `/api/documents/{id}`、GET `/api/documents/{id}/history`
- POST `/api/retrieval`:已核验知识的关键词检索
- GET `/api/retrieval/runs`、GET `/api/retrieval/runs/{run_id}`:检索历史和原始结果
- POST `/api/retrieval/exports`:管理员按编号导出 JSONL 历史与可选准备上下文,见 [导出契约](docs/RETRIEVAL_EXPORT_API.md)
- `/api/indexes`:管理员构建/查询切分快照、发布/查询当前版本、查询 chunks 和发布历史,详见 [接口契约](docs/INDEX_API.md)
### 阶段 2 接口约定
+24
View File
@@ -1,12 +1,14 @@
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import Response
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models import RetrievalRun
from app.schemas.catalog import Page
from app.schemas.exports import ExportHistory
from app.schemas.hybrid import (
ChunkKeywordRequest,
ChunkKeywordResponse,
@@ -17,6 +19,7 @@ from app.schemas.retrieval import RetrievalRequest, RetrievalResponse
from app.schemas.vectors import VectorRetrievalResponse
from app.services.auth import require_admin
from app.services.embedding_runtime import get_embedding_provider
from app.services.exports import export_history
from app.services.hybrid import retrieve_chunks, retrieve_hybrid
from app.services.retrieval import retrieve
@@ -25,6 +28,27 @@ router = APIRouter(prefix="/retrieval", tags=["retrieval"])
RunResponse = RetrievalResponse | VectorRetrievalResponse | ChunkKeywordResponse | HybridResponse
@router.post(
"/exports",
response_class=Response,
responses={200: {"content": {"application/x-ndjson": {}}, "description": "JSONL 历史快照"}},
dependencies=[Depends(require_admin)],
)
def download_history(payload: ExportHistory, db: Session = Depends(get_db)):
"""管理员按明确编号导出历史;沿用 Origin/CSRF,失败不返回部分文件。"""
content, export_id, digest = export_history(db, payload)
return Response(
content,
media_type="application/x-ndjson; charset=utf-8",
headers={
"Content-Disposition": f'attachment; filename="nalati-retrieval-{export_id}.jsonl"',
"Cache-Control": "no-store",
"X-Records-SHA256": digest,
"X-Content-Type-Options": "nosniff",
},
)
@router.post("", response_model=RetrievalResponse)
def keyword_retrieval(payload: RetrievalRequest, db: Session = Depends(get_db)):
"""只检索已核验知识;保存原始检索结果供复核,不调用大模型。"""
+19
View File
@@ -0,0 +1,19 @@
"""Explicit, bounded selection of immutable experiment history."""
from uuid import UUID
from pydantic import Field, field_validator
from app.schemas.catalog import Input
class ExportHistory(Input):
run_ids: list[UUID] = Field(min_length=1, max_length=100)
include_contexts: bool = Field(False, strict=True)
@field_validator("run_ids")
@classmethod
def unique_ids(cls, value):
if len(value) != len(set(value)):
raise ValueError("检索记录编号不得重复")
return value
+201
View File
@@ -0,0 +1,201 @@
"""Read stored snapshots in one SQL statement; finish validation before any download."""
import hashlib
from typing import Annotated
from uuid import uuid4
from fastapi import HTTPException
from pydantic import Field, TypeAdapter, ValidationError
from sqlalchemy import String, literal, select, union_all
from sqlalchemy.orm import Session
from app.models.catalog import utc_now
from app.models.rag import RagContext
from app.models.retrieval import RetrievalRun
from app.schemas.exports import ExportHistory
from app.schemas.hybrid import ChunkKeywordResponse, HybridResponse
from app.schemas.rag import ContextRead
from app.schemas.retrieval import RetrievalResponse
from app.schemas.vectors import VectorRetrievalResponse
from app.services.completions import canonical_json, request_body
MAX_CONTEXTS = 100
MAX_BYTES = 8 * 1024 * 1024
SCHEMA_VERSION = "nalati-retrieval-export-v1"
RUN = TypeAdapter(
Annotated[
RetrievalResponse | ChunkKeywordResponse | VectorRetrievalResponse | HybridResponse,
Field(discriminator="mode"),
]
)
def invalid_history():
return HTTPException(409, "历史记录格式或关联不一致,未生成导出文件")
def same_fields(raw, parsed):
"""Reject unknown/missing fields instead of silently dropping snapshot metadata."""
if isinstance(parsed, dict):
return (
isinstance(raw, dict)
and raw.keys() == parsed.keys()
and all(same_fields(raw[key], value) for key, value in parsed.items())
)
if isinstance(parsed, list):
return (
isinstance(raw, list)
and len(raw) == len(parsed)
and all(same_fields(left, right) for left, right in zip(raw, parsed))
)
return True
def history_rows(db: Session, ids: list[str], include_contexts: bool):
runs = select(
literal(0).label("record_order"),
RetrievalRun.id.label("id"),
RetrievalRun.id.label("run_id"),
RetrievalRun.created_at.label("created_at"),
RetrievalRun.query.label("query"),
RetrievalRun.mode.label("mode"),
RetrievalRun.corpus_fingerprint.label("corpus_fingerprint"),
RetrievalRun.response.label("payload"),
).where(RetrievalRun.id.in_(ids))
if include_contexts:
contexts = select(
literal(1).label("record_order"),
RagContext.id.label("id"),
RagContext.retrieval_run_id.label("run_id"),
RagContext.created_at.label("created_at"),
literal(None, String).label("query"),
literal(None, String).label("mode"),
literal(None, String).label("corpus_fingerprint"),
RagContext.response.label("payload"),
).where(RagContext.retrieval_run_id.in_(ids))
statement = union_all(runs, contexts)
else:
statement = runs
# One PostgreSQL statement snapshot, including optional preparations. No offset
# pagination, ORM auth rows, current corpus lookups, or model dependencies.
return list(
db.execute(
statement.order_by("record_order", "created_at", "id").limit(
len(ids) + MAX_CONTEXTS + 1
)
).mappings()
)
def checked_run(row):
raw = row["payload"]
try:
value = RUN.validate_json(canonical_json(raw), strict=True)
if (
not same_fields(raw, value.model_dump(mode="json"))
or str(value.run_id) != row["id"]
or value.mode != row["mode"]
or value.query != row["query"]
or value.corpus_fingerprint != row["corpus_fingerprint"]
or value.created_at != row["created_at"]
or (value.status == "ok") != bool(value.results)
):
raise ValueError("run mismatch")
except (ValidationError, ValueError, TypeError) as exc:
raise invalid_history() from exc
return value
def checked_context(row, runs):
raw = row["payload"]
try:
value = ContextRead.model_validate_json(canonical_json(raw), strict=True)
run = runs[row["run_id"]]
if (
not same_fields(raw, value.model_dump(mode="json"))
or str(value.id) != row["id"]
or str(value.retrieval_run_id) != row["run_id"]
or value.created_at != row["created_at"]
or value.query != run.query
or value.retrieval_mode != run.mode
or value.corpus_fingerprint != run.corpus_fingerprint
or value.text_index_version_id != run.text_index_version_id
or value.text_generation != run.text_generation
or value.catalog_revision != getattr(run, "catalog_revision", value.catalog_revision)
or value.vector_version_id != getattr(run, "vector_version_id", None)
or value.vector_generation != getattr(run, "generation", None)
):
raise ValueError("context mismatch")
if value.status == "prepared":
expected = request_body(
value.query,
[evidence.model_dump(mode="json") for evidence in value.evidence],
value.completion,
)
if (
not value.evidence
or canonical_json(value.request_body) != canonical_json(expected)
or value.request_sha256
!= hashlib.sha256(canonical_json(expected).encode()).hexdigest()
):
raise ValueError("request mismatch")
elif value.request_body is not None or value.request_sha256 is not None or value.evidence:
raise ValueError("unexpected request")
except (ValidationError, ValueError, TypeError, KeyError, AttributeError) as exc:
raise invalid_history() from exc
def export_history(db: Session, payload: ExportHistory):
ids = sorted(str(value) for value in payload.run_ids)
rows = history_rows(db, ids, payload.include_contexts)
context_count = sum(row["record_order"] == 1 for row in rows)
if context_count > MAX_CONTEXTS:
raise HTTPException(413, "关联上下文超过100条,请缩小检索记录选择范围")
run_rows = [row for row in rows if row["record_order"] == 0]
if {row["id"] for row in run_rows} != set(ids):
raise HTTPException(404, "部分检索记录不存在,未生成导出文件")
runs = {row["id"]: checked_run(row) for row in run_rows}
lines = []
size = 0
for row in rows:
if row["record_order"]:
checked_context(row, runs)
record = {"record_type": "rag_context", "id": row["id"], "payload": row["payload"]}
else:
record = {
"record_type": "retrieval_run",
"id": row["id"],
"ranking_unit": "knowledge_item" if row["mode"] == "keyword" else "chunk",
"payload": row["payload"],
}
try:
line = (canonical_json(record) + "\n").encode("utf-8")
except (TypeError, ValueError, UnicodeError) as exc:
raise invalid_history() from exc
size += len(line)
if size > MAX_BYTES:
raise HTTPException(413, "导出超过8 MiB,请缩小检索记录选择范围")
lines.append(line)
records = b"".join(lines)
digest = hashlib.sha256(records).hexdigest()
export_id = str(uuid4())
manifest = {
"record_type": "manifest",
"schema_version": SCHEMA_VERSION,
"export_id": export_id,
"exported_at": utc_now().isoformat(),
"selection": {"run_ids": ids, "include_contexts": payload.include_contexts},
"counts": {"retrieval_runs": len(run_rows), "rag_contexts": context_count},
"records_sha256": digest,
"snapshot_semantics": "stored_history_not_current_corpus",
"generation_measurements": {
"status": "not_called",
"usage": None,
"cost": None,
"currency": None,
},
}
content = (canonical_json(manifest) + "\n").encode("utf-8") + records
if len(content) > MAX_BYTES:
raise HTTPException(413, "导出超过8 MiB,请缩小检索记录选择范围")
return content, export_id, digest
+342
View File
@@ -0,0 +1,342 @@
"""Export historical protocol fixtures, not domain scores or actual model usage."""
import copy
import hashlib
import json
from datetime import timedelta
from uuid import uuid4
import pytest
from sqlalchemy import event, func, select
from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import flag_modified
from test_rag import prepare
from test_retrieval import add_item, search
from test_runtime import server
from test_vectors import ToyProvider, text_version, vector_build, vector_publish
from app.models.auth import User
from app.models.catalog import utc_now
from app.models.rag import RagContext
from app.models.retrieval import RetrievalRun
from app.services import exports
from app.services.embedding_runtime import get_embedding_provider
PATH = "/api/retrieval/exports"
def download(client, runs, **fields):
return client.post(PATH, json={"run_ids": [r["run_id"] for r in runs], **fields})
def lines(response):
assert response.status_code == 200, response.text
assert response.content.endswith(b"\n")
return [json.loads(line) for line in response.content.rstrip(b"\n").split(b"\n")]
def chunk_run(client):
add_item(client, content="alpha query 合成占位正文")
text_version(client)
response = client.post("/api/retrieval/chunks", json={"query": "alpha"})
assert response.status_code == 200, response.text
return response.json()
def test_mixed_modes_exact_snapshots_manifest_digest_and_optional_contexts(
client, migrated_database
):
add_item(client, content="alpha query\n合成测试正文")
text_ = text_version(client)
client.app.dependency_overrides[get_embedding_provider] = ToyProvider
vector_publish(client, vector_build(client, text_))
runs = []
for path in ["", "/chunks", "/vector", "/hybrid"]:
response = client.post("/api/retrieval" + path, json={"query": "alpha\nquery"})
assert response.status_code == 200, response.text
runs.append(response.json())
context = prepare(client, runs[-1]["run_id"]).json()
client.app.dependency_overrides.pop(get_embedding_provider)
response = download(client, list(reversed(runs)))
records = lines(response)
manifest = records[0]
assert manifest["schema_version"] == "nalati-retrieval-export-v1"
assert manifest["counts"] == {"retrieval_runs": 4, "rag_contexts": 0}
assert manifest["selection"] == {
"run_ids": sorted(r["run_id"] for r in runs),
"include_contexts": False,
}
assert manifest["snapshot_semantics"] == "stored_history_not_current_corpus"
assert manifest["generation_measurements"] == {
"status": "not_called",
"usage": None,
"cost": None,
"currency": None,
}
data = response.content.split(b"\n", 1)[1]
assert hashlib.sha256(data).hexdigest() == manifest["records_sha256"]
assert response.headers["X-Records-SHA256"] == manifest["records_sha256"]
assert response.headers["content-type"] == "application/x-ndjson; charset=utf-8"
assert response.headers["content-disposition"] == (
f'attachment; filename="nalati-retrieval-{manifest["export_id"]}.jsonl"'
)
assert response.headers["cache-control"] == "no-store"
assert response.headers["x-content-type-options"] == "nosniff"
assert [r["payload"] for r in records[1:]] == runs
assert [r["ranking_unit"] for r in records[1:]] == ["knowledge_item", "chunk", "chunk", "chunk"]
repeated = download(client, runs)
assert repeated.content.split(b"\n", 1)[1] == data
assert lines(repeated)[0]["export_id"] != manifest["export_id"]
included = download(client, runs, include_contexts=True)
all_records = lines(included)
assert all_records[0]["counts"] == {"retrieval_runs": 4, "rag_contexts": 1}
assert all_records[-1] == {
"record_type": "rag_context",
"id": context["id"],
"payload": context,
}
for forbidden in [
b"created_by",
b"password_hash",
b"test_admin",
b"csrf_token",
b"nalati_session",
]:
assert forbidden not in included.content
with Session(migrated_database) as db:
assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 4
assert db.scalar(select(func.count()).select_from(RagContext)) == 1
def test_missing_selection_is_atomic_and_no_evidence_is_exportable(client):
run = search(client, query="没有语料的合成问题")
assert run["status"] == "no_evidence"
missing = download(client, [run, {"run_id": str(uuid4())}])
assert missing.status_code == 404 and "Content-Disposition" not in missing.headers
assert "manifest" not in missing.text
assert lines(download(client, [run]))[1]["payload"] == run
@pytest.mark.parametrize(
"payload",
[
{},
{"run_ids": []},
{"run_ids": ["not-a-uuid"]},
{"run_ids": [str(uuid4()) for _ in range(101)]},
{"run_ids": [str(uuid4())], "include_contexts": "true"},
{"run_ids": [str(uuid4())], "api_key": "test-secret-not-echoed"},
],
)
def test_selection_validation_rejects_unsafe_or_unbounded_input(client, payload):
response = client.post(PATH, json=payload)
assert response.status_code == 422
assert "test-secret-not-echoed" not in response.text
def test_duplicate_uuids_are_rejected_after_normalization(client):
run = search(client)
response = client.post(PATH, json={"run_ids": [run["run_id"], run["run_id"].upper()]})
assert response.status_code == 422
def test_admin_origin_csrf_and_no_auth_data_in_errors(anonymous_client, migrated_database):
payload = {"run_ids": [str(uuid4())]}
assert anonymous_client.post(PATH, json=payload).status_code == 401
credentials = {"username": "export_user", "password": "test-password-123"}
assert anonymous_client.post("/api/auth/register", json=credentials).status_code == 201
login = anonymous_client.post("/api/auth/login", json=credentials).json()
anonymous_client.headers["X-CSRF-Token"] = login["csrf_token"]
assert anonymous_client.post(PATH, json=payload).status_code == 403
with Session(migrated_database) as db:
user = db.scalar(select(User).where(User.username == credentials["username"]))
user.role = "admin"
db.commit()
token = anonymous_client.headers.pop("X-CSRF-Token")
assert anonymous_client.post(PATH, json=payload).status_code == 403
anonymous_client.headers["X-CSRF-Token"] = token
assert (
anonymous_client.post(
PATH, json=payload, headers={"Origin": "https://example.com"}
).status_code
== 403
)
assert anonymous_client.post(PATH, json=payload).status_code == 404
def test_export_retains_history_after_catalog_deletion_and_context_is_no_generation(client):
add_item(client, content="alpha query 合成正文")
text_version(client)
run = client.post("/api/retrieval/chunks", json={"query": "不存在的词"}).json()
context = prepare(client, run["run_id"]).json()
assert context["status"] == "insufficient_evidence"
item = client.get("/api/knowledge").json()[0]
assert (
client.delete(f"/api/knowledge/{item['id']}?expected_version={item['version']}").status_code
== 204
)
result = lines(download(client, [run], include_contexts=True))
assert result[1]["payload"] == run and result[2]["payload"] == context
assert result[2]["payload"]["model_called"] is False
@pytest.mark.parametrize(
"change", ["id", "mode", "query", "fingerprint", "created_at", "unknown", "score_type"]
)
def test_corrupt_run_fails_without_partial_download_or_private_error(
client, migrated_database, change
):
add_item(client, content="alpha 合成正文")
run = search(client, query="alpha")
with Session(migrated_database) as db:
row = db.get(RetrievalRun, run["run_id"])
raw = copy.deepcopy(row.response)
if change == "id":
raw["run_id"] = str(uuid4())
elif change == "mode":
row.mode = "vector"
elif change == "query":
row.query = "private-test-data"
elif change == "fingerprint":
row.corpus_fingerprint = "f" * 64
elif change == "created_at":
row.created_at += timedelta(seconds=1)
elif change == "unknown":
raw["api_key"] = "private-test-data"
else:
raw["results"][0]["score"] = "NaN"
row.response = raw
db.commit()
response = download(client, [run])
assert response.status_code == 409
assert (
"Content-Disposition" not in response.headers and "private-test-data" not in response.text
)
@pytest.mark.parametrize(
"change",
[
"request",
"request_numeric",
"hash",
"link",
"catalog_revision",
"model_called",
"creator_extra",
],
)
def test_corrupt_context_fails_atomically(client, migrated_database, change):
run = chunk_run(client)
context = prepare(client, run["run_id"]).json()
with Session(migrated_database) as db:
row = db.get(RagContext, context["id"])
raw = copy.deepcopy(row.response)
if change == "request":
raw["request_body"]["api_key"] = "private-test-data"
elif change == "request_numeric":
# Python equality alone treats True == 1; the actual wire body/hash must not.
raw["request_body"]["n"] = True
elif change == "hash":
raw["request_sha256"] = "f" * 64
elif change == "link":
raw["retrieval_run_id"] = str(uuid4())
elif change == "catalog_revision":
raw["catalog_revision"] += 1
elif change == "model_called":
raw["model_called"] = True
else:
raw["created_by"] = "private-test-data"
row.response = raw
# SQLAlchemy's Python comparison otherwise considers JSON true equal to 1.
flag_modified(row, "response")
db.commit()
if change == "request_numeric":
db.refresh(row)
assert row.response["request_body"]["n"] is True
response = download(client, [run], include_contexts=True)
assert response.status_code == 409 and "Content-Disposition" not in response.headers
assert "private-test-data" not in response.text
assert download(client, [run]).status_code == 200
@pytest.mark.parametrize("boundary", ["record", "manifest"])
def test_utf8_byte_cap_rejects_whole_file_including_manifest(client, monkeypatch, boundary):
run = search(client, query="多字节中文合成问题")
original = download(client, [run])
full_size = len(original.content)
record_size = len(original.content.split(b"\n", 1)[1])
monkeypatch.setattr(
exports, "MAX_BYTES", record_size - 1 if boundary == "record" else full_size - 1
)
response = download(client, [run])
assert response.status_code == 413 and "Content-Disposition" not in response.headers
assert client.get(f"/api/retrieval/runs/{run['run_id']}").json() == run
monkeypatch.setattr(exports, "MAX_BYTES", full_size)
assert len(download(client, [run]).content) == full_size
def copied_context(context, run_id, actor):
raw = copy.deepcopy(context)
raw.update(id=str(uuid4()), created_at=utc_now().isoformat())
return RagContext(
id=raw["id"],
retrieval_run_id=run_id,
created_by=actor,
response=raw,
created_at=raw["created_at"],
)
def test_context_capacity_rejects_instead_of_truncating(client, migrated_database):
run = chunk_run(client)
context = prepare(client, run["run_id"]).json()
with Session(migrated_database) as db:
actor = db.scalar(select(RagContext.created_by))
db.add_all(copied_context(context, run["run_id"], actor) for _ in range(100))
db.commit()
assert download(client, [run]).status_code == 200
response = download(client, [run], include_contexts=True)
assert response.status_code == 413 and "Content-Disposition" not in response.headers
with Session(migrated_database) as db:
assert db.scalar(select(func.count()).select_from(RagContext)) == 101
def test_one_statement_snapshot_excludes_concurrent_context_insert(client, migrated_database):
run = chunk_run(client)
context = prepare(client, run["run_id"]).json()
with Session(migrated_database) as db:
# Separate fixture actor avoids the authentication row lock held by the export request.
actor = User(username="snapshot_fixture", password_hash="not-a-login-hash", role="user")
db.add(actor)
db.commit()
actor_id = actor.id
inserted = False
def concurrent_insert(connection, cursor, statement, parameters, context_, many):
nonlocal inserted
if not inserted and "UNION ALL" in statement and "rag_contexts" in statement:
inserted = True
with Session(migrated_database) as db:
db.add(copied_context(context, run["run_id"], actor_id))
db.commit()
event.listen(migrated_database, "after_cursor_execute", concurrent_insert)
try:
result = lines(download(client, [run], include_contexts=True))
finally:
event.remove(migrated_database, "after_cursor_execute", concurrent_insert)
assert inserted and result[0]["counts"]["rag_contexts"] == 1
assert lines(download(client, [run], include_contexts=True))[0]["counts"]["rag_contexts"] == 2
def test_http_export_records_and_digest_survive_process_restart(migrated_database):
with server(migrated_database) as client:
run = search(client, query="HTTP导出合成问题")
first = download(client, [run])
assert first.status_code == 200
with server(migrated_database) as client:
second = download(client, [run])
assert second.status_code == 200
assert second.content.split(b"\n", 1)[1] == first.content.split(b"\n", 1)[1]
assert second.headers["X-Records-SHA256"] == first.headers["X-Records-SHA256"]
+2
View File
@@ -37,6 +37,8 @@ FastAPI + Pydantic
新增 `retrieval_runs` 保存查询、模式、语料指纹和完整结果快照,支持分页读取及按 UUID 复核。零命中如实记录;当前依据修改后失效不会改写历史,但后续检索必须重新检查审核与父记录状态。响应耗时是本次真实读取、分词、排序及片段生成耗时,不含写日志或模型调用。
管理员 `POST /api/retrieval/exports` 按明确编号导出四种模式的原保存响应,并可包含关联准备上下文;以一个 SQL 语句快照读取,分别标记条目/chunk 粒度。JSONL 首行记录版本、选择范围、条数和后续记录字节摘要;输出前校验、限制容量,不写新历史、不依赖当前发布或模型,不导出认证数据。生成用量和费用为未采集,完整规则见 [导出契约](RETRIEVAL_EXPORT_API.md)。
关键词基线仍按知识条目检索,显示片段最长 240 字符;独立的持久化 chunks 和文本索引版本见下方。该接口不读取 chunks;chunk 与向量/RRF 的独立接口见后文。全语料计算明确限制条目数及总字符数,超限报错,适用于首批小规模资料与后续对照实验准备,不能当作生产检索性能结论。
+97
View File
@@ -0,0 +1,97 @@
# 检索与上下文历史导出
本接口为后续实验保留当时的检索证据与配置,不重新检索、不加载 Embedding 或调用生成模型,也不生成质量分数。四种模式仍保持各自证据粒度,不能把条目与 chunk 混为同一评测样本。
## 请求与权限
`POST /api/retrieval/exports`,管理员会话、允许的 Origin 和 `X-CSRF-Token` 均为必需。普通用户 403,匿名/失效会话 401;错误 Origin 或 CSRF 为 403。继承 [认证契约](AUTH_API.md),不会返回用户、会话或凭据字段。
```json
{
"run_ids": ["00000000-0000-4000-8000-000000000001"],
"include_contexts": false
}
```
编号仅为格式示例,实际应先从管理员 `GET /api/retrieval/runs` 选择已存在的记录。`run_ids` 必须为 1–100 个不重复 UUID,大小写规范化后仍重复也拒绝;`include_contexts` 为严格布尔值,默认 false。额外字段、空列表和非法编号为 422。
明确选择避免导出无边界的全库记录。任一检索编号不存在即 404,不生成部分文件。可选上下文是这些检索记录关联的**全部已保存准备记录**,不含其他检索的上下文。
## 下载与 JSONL 契约
- 成功为 200,`Content-Type: application/x-ndjson; charset=utf-8`,`Content-Disposition: attachment; filename="nalati-retrieval-<export UUID>.jsonl"`。
- 响应固定 `Cache-Control: no-store` 和 `X-Content-Type-Options: nosniff`;`X-Records-SHA256` 等于文件首行的 `records_sha256`。
- UTF-8 编码,每行一个 JSON 对象,每行(包括末行)以 LF 结束;内容中的换行由 JSON 转义。不是 CSV,也不是一个 JSON 数组。
- 首行为 `record_type=manifest`。其余先为检索记录,再为上下文记录,各组按数据库 `created_at,id` 升序。请求 UUID 列表的顺序不影响记录区顺序。
首行字段:
| 字段 | 含义 |
|---|---|
| `schema_version` | 固定 `nalati-retrieval-export-v1` |
| `export_id`、`exported_at` | 本次导出 UUID 与带时区 UTC 时间 |
| `selection` | 排序后的检索 UUID 及是否包含上下文 |
| `counts` | 实际 `retrieval_runs` 与 `rag_contexts` 条数,不含首行 |
| `records_sha256` | **首行之后全部原始 UTF-8 字节**的 SHA-256,含每条末尾 LF |
| `snapshot_semantics` | 固定 `stored_history_not_current_corpus` |
| `generation_measurements` | `status=not_called`,`usage/cost/currency=null`;没有生成调用或费用测量,不能解释为零成本 |
检索行有 `record_type=retrieval_run`、`id`、`ranking_unit` 和 `payload`。`payload` 保留原已保存响应,不重新计算或修改分数、耗时、文本、来源及版本:
| `payload.mode` | `ranking_unit` | 保留的配置与证据 |
|---|---|---|
| `keyword` | `knowledge_item` | 条目 BM25、中文分词/标题权重、片段与来源版本 |
| `keyword_chunks` | `chunk` | 文本发布/切分/语料版本、正文 BM25、chunk 原文出处 |
| `vector` | `chunk` | 文本/向量发布、模型完整 revision、空间/维度与精确余弦配置 |
| `hybrid` | `chunk` | 同一 chunk 粒度、两路完整候选/排名/原始分数/贡献、RRF 与模型空间配置 |
上下文行有 `record_type=rag_context`、`id` 和原保存的 `payload`;保留准备参数、证据、排除原因、提示词版本、请求体及请求摘要。`model_called=false`,即使 `status=prepared` 也不是模型回答。原字段详见 [RAG_CONTEXT_API.md](RAG_CONTEXT_API.md)。不导出数据库 `created_by` 或认证表数据。
## 快照、容量与失败
- 检索与可选上下文通过**一个 PostgreSQL SQL 语句快照**读取,没有跨页读取造成的漂移;执行后新增的上下文留到下一次导出。
- 已修改、删除或重新发布的知识不会改写历史。导出成功只说明当时记录可读取,不说明它仍是当前有效发布或来源仍受支持。
- 每次最多 100 条检索、100 条关联上下文,完整 UTF-8 文件(含 manifest)最多 8 MiB。超量为 413,不静默截断;某个检索自身关联超过 100 条准备记录时,本版可先不含上下文导出,不能借此声称已导出全部准备记录。
- 输出前完成结构、编号/关联、数据库元数据和准备请求摘要校验;未知/缺失字段、非有限数值或关联异常为 409。它是历史格式一致性检查,不是来源真实性或事实正确性证明。
- 服务端先完整验证并组装文件,再返回下载响应;校验、容量或数据库失败不会先发出部分 JSONL。此接口不写入新的检索/上下文记录,不重建索引。
- 同一历史选择的记录区字节和摘要可重复;manifest 的导出 UUID/时间每次变化,因此不要比较整个文件的摘要。若新增上下文后再次启用 `include_contexts`,记录区也会变化。SHA-256 仅供字节完整性核对,不是审阅签名或质量指标。
`elapsed_ms` 沿用各模式原始计时范围,不是生成模型或完整浏览器交互耗时。模型、语料与参数不一致的记录不能只凭分数直接做实验对比。
## 调用与核对
已有管理员登录并按认证契约准备 Cookie 文件和当前 CSRF 后,可在仓库根目录调用。Cookie 文件、真实请求、JSONL 和实际日志均留在 Git 元数据目录或仓库外;下面的编号仍需替换成真实已保存的检索编号。
```bash
mkdir -p .git/exports
curl --fail-with-body --request POST http://127.0.0.1:8000/api/retrieval/exports \
--cookie "$COOKIE_JAR" \
--header 'Origin: http://localhost:5173' \
--header "X-CSRF-Token: $CSRF_TOKEN" \
--header 'Content-Type: application/json' \
--data '{"run_ids":["00000000-0000-4000-8000-000000000001"],"include_contexts":true}' \
--output .git/exports/history.jsonl
```
仅在 HTTP 200 后按 JSONL 读取文件。核对摘要时不应重新序列化记录:
```python
import hashlib
import json
from pathlib import Path
header, records = Path(".git/exports/history.jsonl").read_bytes().split(b"\n", 1)
manifest = json.loads(header)
assert manifest["schema_version"] == "nalati-retrieval-export-v1"
assert hashlib.sha256(records).hexdigest() == manifest["records_sha256"]
rows = [json.loads(line) for line in records.rstrip(b"\n").split(b"\n")]
assert len(rows) == sum(manifest["counts"].values())
```
前端调用层 `services/exports.ts` 的 `exportRetrievalHistory` 返回 Blob、受限文件名和记录摘要,沿用同一 Cookie/CSRF、401 撤销及错误处理,不放入 localStorage。此阶段没有新增下载页面,页面入口由后续前端集成;调用层测试是模拟网络的契约测试,不冒充浏览器下载验收。
## 验证与撤回
新增专项覆盖混合模式原始快照、可选准备记录、无命中、软删除后历史、摘要稳定性、权限/CSRF、未知字段/关联损坏拒绝、字节/条数上限、并发新增上下文以及真实 HTTP 后端重启后的同一记录区。坐标 fixture 仅用于导出协议,不构成真实领域模型实验。专项命令 `pytest -q backend/tests/test_exports.py`:**30 passed**;完整 `pytest -q backend/tests`:**254 passed, 1 skipped**,skip 为显式启用的真实 Embedding 用例,本阶段没有重跑模型。`ruff check backend`、`ruff format --check backend` 通过;前端在 `frontend/` 执行 `npm run test:session`:**3 passed**,`npm run build` 通过。
数据库不新增迁移,继续为 `0007_rag_contexts`。撤回导出路由、服务、调用层即可停止新下载;原检索/上下文历史与已保存文件不变,不为回滚删除数据库记录。
+8
View File
@@ -99,3 +99,11 @@
- 修正页面联调发现的采集时间缺失、审核/文档导航入口和过期功能说明;后端审核/授权不放宽,不猜测历史资料采集时间。具体路径、HTTP 与浏览器证据分工及合成截图见 [BROWSER_ACCEPTANCE.md](BROWSER_ACCEPTANCE.md)。
- 独立 HTTP 检查确认匿名401、普通用户八类管理资源403、缺失CSRF/错误Origin403、退出后401;目录写入使旧发布409,重新发布后恢复真实检索;准备上下文仍为 model_called=false。合成数据和本机运行环境、原始记录、凭据不入库。
- 该轮验收收尾了T02的真实浏览器API连接、T05知识录入/审核与文本/向量页面核心流程、Issue #5登录/Cookie/CSRF/角色的主要浏览器路径。Docker、HTTPS、多浏览器、浏览器改密及历史元数据修订页面仍待验收;生成模型和领域评测继续延期。本轮 `pytest -q backend/tests` **224 passed, 1 skipped**,`ruff check`、`ruff format --check` 和前端 `npm run build` 均通过;常规 skip 为须显式启用的真实模型专项,本轮另外通过浏览器与独立 HTTP 实际运行了固定 BGE small。
## 2026-10-16 T07/T10:可复现历史导出
- 浏览器联调与录入修正经 [PR #29](https://gitea.kronecker.cc/saint/border-red-post-nalati/pulls/29) 真正合入 dev,已核对 head/merge 都在最新 dev 祖先链;再从最新 dev 创建唯一阶段分支 `feat/backend-retrieval-export`。
- 新增管理员/Origin/CSRF 保护的 `POST /api/retrieval/exports`,按明确 UUID 导出 JSONL,保留四模式原保存的查询、命中、原始分数、两路候选、模型/语料/切分参数、来源和耗时;可选择关联的准备上下文。条目与 chunk 粒度分开,生成模型使用量和费用为未采集,不伪造生成调用或实验分数。
- 一个 SQL 语句读取历史快照,保留旧版本而不重新检索;导出前完整校验,最多100条检索/100条上下文/8MiB,不静默截断,不返回部分文件。不新增数据库迁移或环境配置。
- 前端仅增加下载类型与调用层,复用 Cookie/CSRF/错误处理,不新增页面;真实浏览器下载与实验展示留给后续集成。完整字段、摘要核对、复现与回滚见 [RETRIEVAL_EXPORT_API.md](RETRIEVAL_EXPORT_API.md)。
- 专项测试 **30 passed**,包括真实 PostgreSQL 并发快照和 HTTP 重启导出;完整回归 **254 passed, 1 skipped**(skip 为须显式启用的真实 Embedding 专项),`ruff check` 与 `ruff format --check` 通过。前端会话下载契约 **3 passed**,类型与生产构建通过。测试使用合成资料与坐标 fixture,未重跑模型推理或调用生成模型。
+2 -1
View File
@@ -6,7 +6,8 @@
"scripts": {
"dev": "vite",
"build": "vue-tsc -b && vite build",
"preview": "vite preview"
"preview": "vite preview",
"test:session": "node --test tests/session.test.mjs"
},
"dependencies": {
"lucide-vue-next": "^0.468.0",
+19
View File
@@ -0,0 +1,19 @@
import { apiDownload } from './session'
import type { RetrievalRunResult } from '../api'
import type { RagContext } from './rag'
export type ExportSelection = { run_ids: string[]; include_contexts?: boolean }
export type ExportManifest = {
record_type: 'manifest'; schema_version: 'nalati-retrieval-export-v1'; export_id: string; exported_at: string;
selection: Required<ExportSelection>; counts: { retrieval_runs: number; rag_contexts: number }; records_sha256: string;
snapshot_semantics: 'stored_history_not_current_corpus';
generation_measurements: { status: 'not_called'; usage: null; cost: null; currency: null }
}
export type ExportRecord = ExportManifest
| { record_type: 'retrieval_run'; id: string; ranking_unit: 'knowledge_item' | 'chunk'; payload: RetrievalRunResult }
| { record_type: 'rag_context'; id: string; payload: RagContext }
// Returns bytes/metadata; callers explicitly choose whether and where to save the file.
export const exportRetrievalHistory = (payload: ExportSelection) => apiDownload(
'/api/retrieval/exports', { method: 'POST', body: JSON.stringify(payload) }, true,
)
+16 -2
View File
@@ -6,10 +6,10 @@ export class ApiError extends Error {
export function clearSessionToken() { csrfToken = null }
export async function apiRequest<T>(path: string, init: RequestInit = {}, protect = false): Promise<T> {
async function request(path: string, init: RequestInit, protect: boolean, accept: string): Promise<Response> {
if (protect && !csrfToken) await apiRequest('/api/auth/me')
const headers = new Headers(init.headers)
headers.set('Accept', 'application/json')
headers.set('Accept', accept)
if (init.body) headers.set('Content-Type', 'application/json')
if (protect && csrfToken) headers.set('X-CSRF-Token', csrfToken)
const response = await fetch(path, { ...init, credentials: 'include', headers })
@@ -19,6 +19,11 @@ export async function apiRequest<T>(path: string, init: RequestInit = {}, protec
const detail = body && typeof body === 'object' && 'detail' in body ? body.detail : null
throw new ApiError(response.status, typeof detail === 'string' ? detail : `请求失败(${response.status}),请检查登录状态或填写内容。`)
}
return response
}
export async function apiRequest<T>(path: string, init: RequestInit = {}, protect = false): Promise<T> {
const response = await request(path, init, protect, 'application/json')
if (response.status === 204) return undefined as T
const data = await response.json()
if (path === '/api/auth/login' || path === '/api/auth/me') {
@@ -26,3 +31,12 @@ export async function apiRequest<T>(path: string, init: RequestInit = {}, protec
}
return data as T
}
export type ApiDownload = { blob: Blob; filename: string; recordsSha256: string | null }
export async function apiDownload(path: string, init: RequestInit = {}, protect = false): Promise<ApiDownload> {
const response = await request(path, init, protect, 'application/x-ndjson')
const disposition = response.headers.get('Content-Disposition') ?? ''
const filename = /^attachment; filename="([A-Za-z0-9._-]+)"$/.exec(disposition)?.[1] ?? 'download.jsonl'
return { blob: await response.blob(), filename, recordsSha256: response.headers.get('X-Records-SHA256') }
}
+70
View File
@@ -0,0 +1,70 @@
import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test'
import ts from 'typescript'
async function session(t, fetch_) {
const previous = globalThis.fetch
globalThis.fetch = fetch_
t.after(() => { globalThis.fetch = previous })
const source = await readFile(new URL('../src/services/session.ts', import.meta.url), 'utf8')
const { outputText } = ts.transpileModule(source, { compilerOptions: { target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.ESNext } })
return import(`data:text/javascript;base64,${Buffer.from(outputText).toString('base64')}#${t.name}`)
}
test('JSON login and binary export share Cookie/CSRF transport without parsing JSONL as JSON', async t => {
const content = '{"record_type":"manifest"}\n{"query":"合成测试"}\n'
const calls = []
const api = await session(t, async (path, init) => {
calls.push([path, init])
if (path === '/api/auth/login') return Response.json({ csrf_token: 'fixture-csrf' })
return new Response(content, { headers: { 'Content-Disposition': 'attachment; filename="fixture.jsonl"', 'X-Records-SHA256': 'a'.repeat(64) } })
})
await api.apiRequest('/api/auth/login', { method: 'POST', body: '{}' })
const file = await api.apiDownload('/api/retrieval/exports', { method: 'POST', body: '{"run_ids":[]}' }, true)
assert.equal(await file.blob.text(), content)
assert.equal(file.filename, 'fixture.jsonl')
assert.equal(file.recordsSha256, 'a'.repeat(64))
assert.equal(calls[1][1].credentials, 'include')
assert.equal(calls[1][1].headers.get('X-CSRF-Token'), 'fixture-csrf')
assert.equal(calls[1][1].headers.get('Content-Type'), 'application/json')
assert.equal(calls[1][1].headers.get('Accept'), 'application/x-ndjson')
assert.equal(calls[0][1].headers.get('Accept'), 'application/json')
})
test('download refreshes missing session token and preserves server authorization errors', async t => {
const calls = []
const api = await session(t, async (path, init) => {
calls.push([path, init])
if (path === '/api/auth/me') return Response.json({ csrf_token: 'refreshed-csrf' })
return Response.json({ detail: '需要管理员权限' }, { status: 403 })
})
await assert.rejects(api.apiDownload('/api/retrieval/exports', { method: 'POST', body: '{}' }, true), error => {
assert.ok(error instanceof api.ApiError)
assert.equal(error.status, 403)
assert.equal(error.message, '需要管理员权限')
return true
})
assert.equal(calls.length, 2)
assert.equal(calls[0][0], '/api/auth/me')
assert.equal(calls[1][1].headers.get('X-CSRF-Token'), 'refreshed-csrf')
})
test('401 download clears stale token and unsafe filename is not propagated', async t => {
let current = 'old-csrf'
let failed = false
const api = await session(t, async (path, init) => {
if (path === '/api/auth/login' || path === '/api/auth/me') return Response.json({ csrf_token: current })
assert.equal(init.headers.get('X-CSRF-Token'), current)
if (!failed) {
failed = true; current = 'new-csrf'
return Response.json({ detail: '请先登录' }, { status: 401 })
}
return new Response('fixture', { headers: { 'Content-Disposition': 'attachment; filename="../../fixture.jsonl"' } })
})
await api.apiRequest('/api/auth/login', { method: 'POST', body: '{}' })
await assert.rejects(api.apiDownload('/api/retrieval/exports', { method: 'POST' }, true), error => error.status === 401)
const file = await api.apiDownload('/api/retrieval/exports', { method: 'POST' }, true)
assert.equal(file.filename, 'download.jsonl')
assert.equal(file.recordsSha256, null)
})