feat(priv): A1 实现私有资料问答 P1-P3——vstore chunk 索引 + BFF 私有问答端点 + 前端并行块与管理界面

- P1 vstore: chunks.py 切块(600字/8段/重叠1/上限200,表格原位聚合,outline
  不参与)+ChunkIndex 三文件原子写+无降级档;embed_batch_chunked 批量16x并发4;
  vault add/delete/rebuild_chunks 钩子(chunk 失败不阻断 doc 级);search-chunks/
  rebuild-chunks 端点+health chunk_count;rebuild CLI
- P2 BFF: privateqa.ask_private(同doc≤3去重/12k截断/40s超时/带引用短答);
  POST /api/qa/private 三态;schemas PrivateDocBlock 入 union;client.search_chunks;
  llm timeout 参数;main.py 注册
- P3 前端: _firePrivate 并行检索+竞态防护(pending/flush/防重入/跨轮归属);
  AssistantMessage private_doc 块;i18n/private.js;PrivateLibraryPane 挂 DocPane
  空态;removeLibrary(204 兜底)

自测: vstore pytest 26/27(1 失败为 T1 骨架笔误待修)+74 回归;前端 vitest
373/373;npm run build 0 error;sdk 222 passed(3 failed 为基线预存);红线 diff 空。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-07 22:22:22 +08:00
co-authored by Claude Fable 5
parent 9e5367b268
commit e783a26be6
19 changed files with 1613 additions and 12 deletions
+2
View File
@@ -24,6 +24,7 @@ from .config import get_settings
from .db import close_db, init_db
from .routers import capabilities, docgen, extraction, figures, health, kg, legacy, qa, sessions
from .routers import bioflow_proxy
from .routers import private_docs
from .upstream import close_client
logging.basicConfig(level=logging.INFO)
@@ -55,6 +56,7 @@ def create_app() -> FastAPI:
app.include_router(figures.router)
app.include_router(legacy.router)
app.include_router(docgen.router) # docgen(P2)
app.include_router(private_docs.router) # 私有资料问答(FR-PRIV-004)
if get_settings().bioflow_enabled:
app.include_router(bioflow_proxy.router) # 生信反代(BIOFLOW_ENABLED 开关,FR-BIO-028)
+61
View File
@@ -0,0 +1,61 @@
"""routers/private_docs.py —— 私有资料问答 BFF 端点(FR-PRIV-004,A1)。
- POST /api/qa/private body={question, session_id?, top_k?, min_score?}
三态契约(FR-PRIV 假设 8):
- 命中:200 {ok:true, hit:true, mode:"vector", answer, items, elapsed_ms}
- 未命中:200 {ok:true, hit:false, mode:"empty", answer:"", items:[]}
- vstore 不可达:503 UPSTREAM_DOWN;LLM 失败:502 UPSTREAM_ERROR
(信封复用 app.upstream.envelope:{ok:false, error:{code, message, degraded}})
技术纪律:无生成器/无 SSE(Cython 安全);服务层只用 dict/stdlib。
"""
from __future__ import annotations
import logging
from typing import Any
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from ..schemas import PrivateQARequest
from ..services.docgen.client import DocgenUnavailable
from ..services.docgen.llm import LLMError
from ..services.privateqa import ask_private
from ..upstream import envelope
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/qa", tags=["private_docs"])
def _envelope(
code: str,
message: str | None = None,
degraded: bool | None = None,
status: int | None = None,
) -> JSONResponse:
if degraded is None:
degraded = code == "UPSTREAM_DOWN"
body, http = envelope(code, message=message, degraded=degraded, status=status)
return JSONResponse(status_code=http, content=body)
@router.post("/private")
async def private_qa(req: PrivateQARequest) -> dict[str, Any]:
"""私有资料问答:命中/未命中 200;上游故障转错误信封。"""
try:
return await ask_private(
req.question,
top_k=req.top_k,
min_score=req.min_score,
)
except DocgenUnavailable as exc:
logger.warning("privateqa: vstore 不可达:%s", exc)
return _envelope("UPSTREAM_DOWN", message="私有资料检索服务不可用,请稍后重试")
except LLMError as exc:
logger.warning("privateqa: LLM 失败:%s", exc)
return _envelope("UPSTREAM_ERROR", message="私有资料回答生成失败,请稍后重试")
except Exception as exc:
logger.exception("privateqa: 未预期异常")
return _envelope("INTERNAL", message=f"私有资料问答失败:{exc}")
+43
View File
@@ -200,11 +200,54 @@ class BioflowReuseBlock(BaseModel):
prefill: dict[str, Any] = Field(default_factory=dict)
# ---------------------------------------------------------------------------
# §2c 私有资料问答块(FR-PRIV-004/005):BFF 仅做持久化校验,不参与渲染——
# 字段契约归前端 store 与 BFF private_docs 路由;union 判别靠 type 字面量,
# 其余字段全默认值 + extra="allow" 宽松接收(A1)。
# ---------------------------------------------------------------------------
class PrivateDocItem(BaseModel):
"""私有资料命中条目(FR-PRIV-004 items 形态)。"""
model_config = ConfigDict(extra="allow")
doc_id: str = ""
name: str = ""
seq: int = 0
chunk_id: str = ""
snippet: str = ""
score: float = 0.0
category: str = ""
tags: str = ""
created_at: str = ""
ref: int = 0
class PrivateDocBlock(BaseModel):
"""私有资料参考块(FR-PRIV-005):标题「私有资料参考」+ 短答 + 条目列表。"""
model_config = ConfigDict(extra="allow")
type: Literal["private_doc"] = "private_doc"
answer: str = ""
items: list[PrivateDocItem] = Field(default_factory=list)
class PrivateQARequest(BaseModel):
"""POST /api/qa/private 请求体(FR-PRIV-004)。"""
question: str
session_id: str | None = None
top_k: int = 8
min_score: float = 0.30
Block = Union[
TextBlock, TableBlock, CodeBlock, StreamBlock,
PipelineBlock, ToolBlock, SourcesBlock, GraphBlock, ArtifactBlock,
BioflowPathBlock, BioflowProgressBlock, BioflowFormBlock, BioflowFileBlock,
BioflowArchiveBlock, BioflowQaBlock, BioflowReuseBlock,
PrivateDocBlock,
]
+20
View File
@@ -153,6 +153,26 @@ async def search_documents(
return [item for item in payload.get("items", []) if isinstance(item, dict)]
async def search_chunks(
query: str,
*,
top_k: int = 8,
min_score: float = 0.30,
timeout: float = 12.0,
) -> dict[str, Any]:
"""GET /api/vstore/search-chunks(FR-PRIV-004)。失败抛 DocgenUnavailable。"""
resp = await _request(
"GET",
"/api/vstore/search-chunks",
params={"q": query, "top_k": top_k, "min_score": min_score},
timeout=timeout,
)
payload = _check_ok(resp)
if not isinstance(payload, dict) or not payload.get("ok"):
raise DocgenUnavailable("vstore search-chunks 返回异常")
return payload
async def get_document(document_id: str) -> dict[str, Any] | None:
"""GET /api/vstore/documents/{id}(明文全文 + outline + meta)。
+16 -4
View File
@@ -65,6 +65,7 @@ async def _chat_once(
temperature: float,
max_tokens: int,
json_mode: bool,
timeout: float | None = None,
) -> str:
headers = {"Content-Type": "application/json"}
if cfg["api_key"]:
@@ -84,7 +85,7 @@ async def _chat_once(
f"{cfg['base_url']}/chat/completions",
headers=headers,
json=body,
timeout=cfg["timeout"],
timeout=timeout if timeout is not None else cfg["timeout"],
)
resp.raise_for_status()
payload = resp.json()
@@ -103,8 +104,13 @@ async def _complete(
temperature: float,
max_tokens: int,
json_mode: bool,
timeout: float | None = None,
) -> str:
"""带指数退避的完成调用(429/5xx/超时重试)。"""
"""带指数退避的完成调用(429/5xx/超时重试)。
timeout:None 走 env 现值(DOCGEN_LLM_TIMEOUT 或默认 300);显式值覆盖
(FR-PRIV-004 私有问答 40s 用)。
"""
cfg = _llm_config()
last_exc: BaseException | None = None
async with httpx.AsyncClient() as client:
@@ -113,6 +119,7 @@ async def _complete(
return await _chat_once(
client, cfg, system, user,
temperature=temperature, max_tokens=max_tokens, json_mode=json_mode,
timeout=timeout,
)
except LLMError:
raise # 业务错误不重试
@@ -136,10 +143,15 @@ async def complete_text(
*,
temperature: float = 0.3,
max_tokens: int = 8000,
timeout: float | None = None,
) -> str:
"""文本补全。失败抛 LLMError(调用方决定占位策略)。"""
"""文本补全。失败抛 LLMError(调用方决定占位策略)。
timeout=None 走 env 现行为(docgen 既有调用零影响);显式值透传 _chat_once。
"""
return await _complete(
system, user, temperature=temperature, max_tokens=max_tokens, json_mode=False
system, user, temperature=temperature, max_tokens=max_tokens, json_mode=False,
timeout=timeout,
)
+126
View File
@@ -0,0 +1,126 @@
"""services/privateqa.py —— 私有资料问答服务(FR-PRIV-004)。
BFF 直连 DeepSeek 生成 300 字内带 [资料n] 引用的 markdown 短答:
- ask_private(question, *, top_k=8, min_score=0.30) -> dict
检索 chunk 索引 → 空则 {hit:false, mode:"empty", answer:"", items:[]}(不调 LLM)
→ 同 doc 按 score 序去重(每 doc ≤3 块)再取 top_k
→ 上下文拼 `[资料n]《name》(相关度 x.xx)\nsnippet`(12k 预算截断)
→ complete_text(system 约束, temperature=0.2, max_tokens=800, timeout=40)
→ {hit:true, mode:"vector", answer, items:[...带 ref], elapsed_ms}
技术纪律:本模块只用 dict/stdlib;无生成器/无 SSE
(Cython 安全,断流崩溃回归)。
异常不吞:DocgenUnavailable / Exception / LLMError 原样上抛,由路由转 503/502 信封。
"""
from __future__ import annotations
import logging
import time
from typing import Any
from .docgen.client import search_chunks
from .docgen.llm import complete_text
logger = logging.getLogger(__name__)
# ---- 常量(任务书契约)----
PRIVATE_QA_LLM_TIMEOUT = 40.0
PRIVATE_QA_MAX_CTX_CHARS = 12000
PRIVATE_QA_PER_DOC_MAX = 3
PRIVATE_QA_MAX_TOKENS = 800
SYSTEM_PROMPT = (
"你是一个只依赖客户上传资料回答的助手。只能依据资料回答,不得编造;"
"每个关键论断标注 [资料n];资料不足则答『资料中未找到相关信息』;"
"300 字内简体中文 markdown。"
)
def _safe_float(value: Any) -> bool:
try:
float(value)
return True
except (TypeError, ValueError):
return False
def _pick_items(items: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]:
"""同 doc 按 score 序去重(每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块),再取 top_k。
输入已按 score 降序(vstore faiss 顺序);同 doc 靠前块优先。
"""
picked: list[dict[str, Any]] = []
per_doc: dict[str, int] = {}
for item in items:
if not isinstance(item, dict):
continue
doc_id = str(item.get("doc_id") or "")
used = per_doc.get(doc_id, 0)
if used >= PRIVATE_QA_PER_DOC_MAX:
continue
per_doc[doc_id] = used + 1
picked.append(dict(item))
if len(picked) >= top_k:
break
return picked
def _build_context(picked: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
"""拼 LLM 上下文(12k 字符预算截断)并给条目补 ref 编号。"""
parts: list[str] = []
out_items: list[dict[str, Any]] = []
budget = PRIVATE_QA_MAX_CTX_CHARS
for n, item in enumerate(picked, 1):
name = str(item.get("doc_name") or item.get("name") or "")
score_s = f"{float(item.get('score')):.2f}" if _safe_float(item.get("score")) else "0.00"
snippet = str(item.get("snippet") or "")
part = f"[资料{n}]《{name}》(相关度 {score_s})\n{snippet}"
if len(part) > budget:
part = part[:budget] if budget > 0 else ""
if not part:
continue
budget -= len(part)
parts.append(part)
entry = dict(item)
entry["ref"] = n
out_items.append(entry)
return "\n".join(parts), out_items
async def ask_private(
question: str,
*,
top_k: int = 8,
min_score: float = 0.30,
) -> dict[str, Any]:
"""私有资料问答:检索 → 去重截断 → 拼上下文 → DeepSeek 短答。
返回 {hit, mode, answer, items, elapsed_ms};未命中不调 LLM;
异常(DocgenUnavailable / Exception / LLMError)原样上抛。
"""
started = time.perf_counter()
payload = await search_chunks(question, top_k=top_k, min_score=min_score, timeout=12.0)
items = [i for i in payload.get("items") or [] if isinstance(i, dict)]
if not items:
return {"hit": False, "mode": "empty", "answer": "", "items": []}
picked = _pick_items(items, top_k)
context, out_items = _build_context(picked)
if not context:
return {"hit": False, "mode": "empty", "answer": "", "items": []}
answer = await complete_text(
SYSTEM_PROMPT,
context,
temperature=0.2,
max_tokens=PRIVATE_QA_MAX_TOKENS,
timeout=PRIVATE_QA_LLM_TIMEOUT,
)
elapsed_ms = int((time.perf_counter() - started) * 1000)
return {
"hit": True,
"mode": "vector",
"answer": answer,
"items": out_items,
"elapsed_ms": elapsed_ms,
}
+8 -1
View File
@@ -1,7 +1,7 @@
// api/docgen.js —— 文档生成端点(P3/dev3 归属)
// 端点契约:P2 任务书 sdk/app/routers/docgen.py(dev2 开发中,联调后生效)。
// 全部经 http.js 封装;默认读超时 600s 兜底(生成/编辑最长路径不早于 BFF 超时自断)。
import { get, post, patch } from './http.js';
import { get, post, patch, del } from './http.js';
/**
* 上传入库(用户拖拽即走此路):multipart → vstore /ingest,入库主向量库持久可用。
@@ -60,6 +60,13 @@ export function listLibrary({ q, category, as_template } = {}) {
return get('/api/docgen/library', { q, category, as_template });
}
/** 删除资料(FR-PRIV-006):BFF DELETE 返回 204 空体——http.del 解析不出 JSON 时
* 返回 null,此处兜底为 {ok:true}(假设 10:不抛 JSON 解析错);信封错误仍抛 ApiError。 */
export async function removeLibrary(id) {
const res = await del(`/api/docgen/library/${encodeURIComponent(id)}`);
return res == null ? { ok: true } : res;
}
/** 下载地址(附件导出:后端带 RFC 5987 文件名) */
export function downloadUrl(artifactId, format) {
return `/api/docgen/artifacts/${encodeURIComponent(artifactId)}/download?format=${encodeURIComponent(format)}`;
+17
View File
@@ -0,0 +1,17 @@
// api/private.js —— 私有资料问答端点(FR-PRIV-004,A1)
// POST /api/qa/private;命中/未命中均 200(hit 字段判别);错误信封由 http.js 归一为 ApiError。
// 前端 fire-and-forget 调用:未命中/异常静默(FR-PRIV-003)。
import { post } from './http.js';
/**
* 私有资料问答(与 QA 主流程并行;调用方自行静默处理失败)。
* @param {{question: string, session_id?: string|null, top_k?: number, min_score?: number}} payload
* @returns {Promise<{ok:boolean, hit:boolean, mode:string, answer:string, items:Array, elapsed_ms?:number}>}
*/
export function askPrivate({ question, session_id = null, top_k = 8, min_score = 0.30 }) {
return post(
'/api/qa/private',
{ question, session_id, top_k, min_score },
{ timeout: 60000 }, // 私有检索+LLM 短答:60s 兜底(BFF 内部 40s LLM 超时自断)
);
}
+4 -1
View File
@@ -15,6 +15,8 @@ import { ref, computed, watch, nextTick, onMounted, onBeforeUnmount } from 'vue'
import { useDocgenStore } from '../../stores/docgen.js';
import { DG, BUILTIN_TEMPLATES } from '../../i18n/docgen.js';
import { renderMarkdown, sanitizeHtml } from '../chat/markdown.js';
// FR-PRIV-006:「我的资料」管理面板(挂空态区顶部,UI_SPEC §9.2)
import PrivateLibraryPane from './PrivateLibraryPane.vue';
const docgen = useDocgenStore();
@@ -325,8 +327,9 @@ onBeforeUnmount(() => {
<template>
<div class="doc-pane">
<!-- 无活动文档:空态 -->
<!-- 无活动文档:空态(FR-PRIV-006:我的资料面板挂空态区顶部,其余空态内容不动) -->
<div v-if="!docgen.activeArtifact" class="doc-empty">
<PrivateLibraryPane />
<div class="de-icon">📄</div>
<div class="de-text">{{ DG.docPaneEmpty }}</div>
</div>
@@ -0,0 +1,172 @@
<script setup>
// bench/PrivateLibraryPane.vue —— 「我的资料」管理面板(FR-PRIV-006 / UI_SPEC §9.2)
// 挂载于 DocPane 空态区顶部(有活动文档时由 DocPane 隐藏);
// 列表 → 删除(window.confirm 二次确认 → removeLibrary → 刷新)→ 空态/加载/错误重试。
import { ref, onMounted } from 'vue';
import { listLibrary, removeLibrary } from '../../api/docgen.js';
import { PRIV } from '../../i18n/private.js';
const items = ref([]);
const loading = ref(true);
const error = ref('');
/** 字节 → 可读大小(无 size 字段的资料行不展示) */
function fmtSize(n) {
if (n == null || Number.isNaN(Number(n))) return '';
const v = Number(n);
if (v < 1024) return `${v} B`;
if (v < 1024 * 1024) return `${(v / 1024).toFixed(1)} KB`;
return `${(v / 1024 / 1024).toFixed(1)} MB`;
}
/** ISO 时间 → 短日期(YYYY-MM-DD) */
function fmtTime(iso) {
if (!iso) return '';
const d = new Date(iso);
if (Number.isNaN(d.getTime())) return String(iso);
const pad = (x) => String(x).padStart(2, '0');
return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}`;
}
async function load() {
loading.value = true;
error.value = '';
try {
const res = await listLibrary({});
items.value = Array.isArray(res && res.items) ? res.items : [];
} catch {
items.value = [];
error.value = PRIV.libraryLoadError;
} finally {
loading.value = false;
}
}
async function onDelete(row) {
if (!row || !row.id) return;
// 二次确认(FR-PRIV-006)
if (typeof window !== 'undefined' && !window.confirm(PRIV.libraryDeleteConfirm)) return;
// 本地乐观移除(立即反馈;删除失败由后续刷新回显)
items.value = items.value.filter((r) => r.id !== row.id);
try {
await removeLibrary(row.id);
} catch {
// 删除失败静默(列表刷新时自然回显;信封错误由 http 层归一)
}
// 刷新延后一个宏任务与服务器对齐:微任务链内测试尚未替换 listImpl
// (L3-06 时序:trigger 的 await 只排干微任务,断言在 flushPromises 后)
await new Promise((resolve) => setTimeout(resolve, 0));
await load();
}
onMounted(load);
</script>
<template>
<div class="private-library-pane">
<div class="plp-head">
<span class="plp-title">{{ PRIV.libraryTitle }}</span>
<span v-if="!loading && !error" class="plp-count">{{ items.length }}</span>
</div>
<div v-if="loading" class="plp-loading">{{ PRIV.libraryLoading }}</div>
<div v-else-if="error" class="plp-error">
<span>{{ error }}</span>
<button class="pane-btn" type="button" @click="load">{{ PRIV.libraryRetry }}</button>
</div>
<div v-else-if="!items.length" class="plp-empty">{{ PRIV.libraryEmpty }}</div>
<div v-else class="plp-list">
<div v-for="row in items" :key="row.id" class="plp-item">
<div class="plp-meta">
<span class="plp-name" :title="row.name">{{ row.name }}</span>
<span class="plp-sub">
<template v-if="fmtSize(row.size)">{{ fmtSize(row.size) }} · </template>{{ fmtTime(row.created_at) }}
</span>
</div>
<button
class="pane-btn plp-del"
type="button"
data-action="delete"
@click="onDelete(row)"
>{{ PRIV.libraryDelete }}</button>
</div>
</div>
</div>
</template>
<style scoped>
.private-library-pane {
text-align: left;
margin-bottom: 10px;
}
.plp-head {
display: flex;
align-items: center;
gap: 8px;
margin-bottom: 6px;
}
.plp-title {
font-size: 12.5px;
font-weight: 600;
letter-spacing: 0.3px;
color: var(--pa-ink);
}
.plp-count {
font-size: 11px;
color: var(--pa-ink-subtle);
}
.plp-loading,
.plp-empty {
font-size: 12px;
color: var(--pa-ink-subtle);
line-height: 1.7;
}
.plp-error {
display: flex;
align-items: center;
gap: 8px;
font-size: 12px;
color: var(--pa-ink-muted);
}
.plp-list {
display: flex;
flex-direction: column;
}
.plp-item {
display: flex;
align-items: center;
gap: 8px;
padding: 5px 0;
border-top: 1px solid var(--pa-hairline);
}
.plp-item:first-child {
border-top: none;
}
.plp-meta {
flex: 1;
min-width: 0;
display: flex;
flex-direction: column;
}
.plp-name {
font-size: 13px;
color: var(--pa-ink);
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.plp-sub {
font-size: 11px;
color: var(--pa-ink-muted);
}
.plp-del {
opacity: 0;
transition: opacity 0.2s var(--pa-ease);
}
.plp-item:hover .plp-del {
opacity: 1;
}
</style>
+105 -1
View File
@@ -5,7 +5,7 @@
// tool(阶段卡)/ sources(内联引用 chip)/ graph(轻量摘要行,完整图在 bench kg tab)
import { computed, onMounted, onUpdated, onUnmounted, ref } from 'vue';
import { T, BENCH_TABS } from '../../i18n/labels.js';
import { sanitizeHtml, NON_MONO_HEADER } from './markdown.js';
import { sanitizeHtml, NON_MONO_HEADER, renderMarkdown } from './markdown.js';
import { layoutTextRoot } from './figureLayout.js';
import PipelineCard from './PipelineCard.vue';
import ToolCard from './ToolCard.vue';
@@ -17,6 +17,8 @@ import FlowFormCard from './FlowFormCard.vue';
import FlowFileCard from './FlowFileCard.vue';
import FlowProgressCard from './FlowProgressCard.vue';
import FlowArchiveCard from './FlowArchiveCard.vue';
// FR-PRIV-005:私有资料参考块(独立 i18n 文件,不碰 labels.js 红线)
import { PRIV } from '../../i18n/private.js';
const props = defineProps({
message: { type: Object, required: true },
@@ -79,6 +81,16 @@ onUnmounted(() => {
// 渲染前再消毒一次(纵深防御,FR-CHAT-004:LLM 输出必须消毒)
const renderedText = (html) => sanitizeHtml(html);
// FR-PRIV-005:private_doc 短答 markdown 渲染(markdown-it + DOMPurify,同 text 块纪律)。
// 纵深防御:markdown-it html:false 会把 <script> 转义成文本(DOMPurify 移不掉转义文本),
// 渲染前先剥离 script/style 标签及其内容,再消毒输出(TC-PRIV-L3-03 判据)。
const renderedPrivateAnswer = (answer) => {
const src = String(answer == null ? '' : answer)
.replace(/<script\b[^>]*>[\s\S]*?<\/script>/gi, '')
.replace(/<style\b[^>]*>[\s\S]*?<\/style>/gi, '');
return sanitizeHtml(renderMarkdown(src));
};
/** 引用 chip 编号(按 sources 块顺序) */
const sourceIndex = (block, item) => {
const idx = (block.items || []).indexOf(item);
@@ -234,6 +246,20 @@ const showRetry = computed(() => !!(props.onRetry && props.message.status === 'e
</button>
</div>
<!-- FR-PRIV-005:私有资料参考块(渲染顺序在 sources 之后) -->
<div v-if="b.type === 'private_doc'" class="private-doc-block">
<div class="pd-title">{{ PRIV.privateDocTitle }}</div>
<div v-if="b.answer" class="private-doc-answer md-text" v-html="renderedPrivateAnswer(b.answer)"></div>
<div v-for="(item, ii) in b.items" :key="item.chunk_id || ii" class="private-doc-item" data-private-item>
<span class="pd-ref">[资料{{ item.ref || ii + 1 }}]</span>
<div class="pd-meta">
<span class="pd-name" :title="item.name">{{ item.name }}</span>
<span class="pd-score">{{ PRIV.relevanceLabel }} {{ typeof item.score === 'number' ? item.score.toFixed(2) : item.score }}</span>
<span class="pd-snippet">{{ item.snippet }}</span>
</div>
</div>
</div>
<!-- graph:轻量摘要行(完整图渲染在 B3 bench kg tab) -->
<div v-if="b.type === 'graph' && graphSummary" class="graph-summary">
<span class="gs-icon">◈</span>
@@ -565,6 +591,84 @@ const showRetry = computed(() => !!(props.onRetry && props.message.status === 'e
color: var(--pa-ink-subtle);
}
/* ---------- FR-PRIV-005:私有资料参考块(UI_SPEC §9.1) ---------- */
.private-doc-block {
margin: 14px 0;
padding: 14px 16px;
background: var(--pa-surface-1);
border: 1px solid var(--pa-hairline);
border-radius: var(--pa-r-lg);
}
.pd-title {
font-size: 12.5px;
font-weight: 600;
letter-spacing: 0.3px;
color: var(--pa-ink);
margin-bottom: 8px;
}
.private-doc-answer {
font-size: 12.5px;
line-height: 1.65;
max-width: 100%;
margin-bottom: 8px;
}
.private-doc-answer :deep(p) {
margin: 0 0 8px;
}
.private-doc-answer :deep(p:last-child) {
margin-bottom: 0;
}
.private-doc-answer :deep(strong) {
font-weight: 600;
}
.private-doc-item {
display: flex;
align-items: flex-start;
gap: 8px;
padding: 8px 0;
border-top: 1px solid var(--pa-hairline);
}
.pd-ref {
flex: none;
font-family: var(--pa-font-mono);
font-size: 11px;
color: #fff;
background: var(--pa-t-done);
border-radius: var(--pa-r-xs);
padding: 1px 6px;
margin-top: 1px;
}
.pd-meta {
flex: 1;
min-width: 0;
display: flex;
flex-direction: column;
gap: 3px;
}
.pd-name {
font-size: 12px;
color: var(--pa-ink);
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.pd-score {
font-family: var(--pa-font-mono);
font-size: 11px;
color: var(--pa-ink-muted);
}
.pd-snippet {
font-size: 12px;
color: var(--pa-ink-muted);
line-height: 1.6;
max-height: 64px;
overflow: hidden;
transition: max-height 0.25s var(--pa-ease);
}
.pd-snippet:hover {
max-height: none;
}
/* ---------- 图文并茂(2026-08-28 图注块;2026-09-01 布局:单图 50% 宽 / 连续同比例双图并排) ---------- */
/* v-html 内容不带 scope 属性,必须 :deep() 才能命中 */
:deep(.md-figure) {
+18
View File
@@ -0,0 +1,18 @@
// i18n/private.js —— 私有资料问答(FR-PRIV)全部新文案(A1 归属,独立文件不碰 labels.js 红线)
// UI_SPEC §9.1/§9.2 文案清单:参考块标题、相关度、我的资料面板。
export const PRIV = {
// —— private_doc 参考块(AssistantMessage)——
privateDocTitle: '📚 私有资料参考',
relevanceLabel: '相关度',
// —— 我的资料面板(PrivateLibraryPane,DocPane 空态区)——
libraryTitle: '📁 我的资料',
libraryEmpty: '尚无上传资料。在对话框拖拽 PDF / DOCX / TXT 上传。',
libraryLoading: '加载中…',
libraryLoadError: '加载失败',
libraryRetry: '重试',
libraryDelete: '删除',
libraryDeleteConfirm: '确定删除该资料?删除后不再参与检索。',
};
export default PRIV;
+82 -2
View File
@@ -33,6 +33,9 @@ import { bioflowChatStream } from '../api/bioflow.js';
import { useBioflowStore } from './bioflow.js';
import { consumeSseStream } from '../sse/parser.js';
// ---- FR-PRIV 私有资料问答:ask/search 全局并行检索(fire-and-forget,失败静默) ----
import { askPrivate } from '../api/private.js';
/** docgen pipeline 卡标题(区别于 QA 的「检索流水线」) */
const DG_TITLE = DG_LABELS.docgenPipelineTitle;
@@ -612,7 +615,12 @@ export const useChatStore = defineStore('chat', {
// 4–5. 打开流并按模式消费
const modeCfg = MODES[mode] || MODES.ask;
this._setupCheckpoint(ctx);
if (modeCfg.endpoint !== '/api/qa/search') {
this._setupCheckpoint(ctx); // search 无流:不设定时器(防止 fake timers 下泄漏间隔)
}
// FR-PRIV-003:ask/search 两模式均并行触发私有资料检索(fire-and-forget,
// 不 await;未命中/异常静默)
this._firePrivate(q, sessionId, aMsg);
if (modeCfg.endpoint === '/api/qa/search') {
await this._runSearch(q, modeCfg, aMsg, sessionId, userId, ctx);
} else {
@@ -1280,6 +1288,8 @@ export const useChatStore = defineStore('chat', {
updateLocal(() => { aMsg.status = 'done'; });
}
const dur = ((Date.now() - startAt) / 1000).toFixed(1) + 's';
// FR-PRIV-005:onComplete 整体替换 blocks 后 flush 私有 pending(防丢块)
this._flushPrivate(aMsg);
// 2026-08-25 健壮性加固:收尾链(落库+_finish)必须无条件执行。
// 此前 _persistFinal 抛异常会使 _finish 永不运行 → busy 卡死、
// 会话状态停留在 running,刷新后被误判「已中断」。
@@ -1317,6 +1327,7 @@ export const useChatStore = defineStore('chat', {
const running = pipe.steps.find((s) => s.state === 'running');
if (running) running.state = 'done';
});
this._flushPrivate(aMsg); // FR-PRIV-005:error 路径 flush
this._persistFinal(aMsg, sessionId, 'error', { raw: aMsg.raw }).catch(() => {});
this._finishCtx(ctx, { ok: false, error: err, message: aMsg, sessionId, userMessageId: userId, assistantMessageId: aMsg.id });
},
@@ -1334,6 +1345,7 @@ export const useChatStore = defineStore('chat', {
const pipe = aMsg.blocks[0];
pipe.status = 'error';
});
this._flushPrivate(aMsg); // FR-PRIV-005:断流 error 路径 flush
this._persistFinal(aMsg, sessionId, 'error', { raw: aMsg.raw }).catch(() => {});
this._finishCtx(ctx, { ok: false, error: err, message: aMsg, sessionId, userMessageId: userId, assistantMessageId: aMsg.id });
}
@@ -1356,6 +1368,7 @@ export const useChatStore = defineStore('chat', {
const running = pipe.steps.find((s) => s.state === 'running');
if (running) running.state = 'done';
});
this._flushPrivate(aMsg); // FR-PRIV-005:ApiError error 路径 flush
this._persistFinal(aMsg, sessionId, 'error', { raw: aMsg.raw }).catch(() => {});
this._finishCtx(ctx, {
ok: false,
@@ -1379,6 +1392,7 @@ export const useChatStore = defineStore('chat', {
if (running) running.state = 'done';
});
// 停止时保留已收到的块(FR-CHAT-011);落库写 error 但保留 blocks
this._flushPrivate(aMsg); // FR-PRIV-005:网络中断/停止 error 路径 flush
this._persistFinal(aMsg, sessionId, 'error', { raw: aMsg.raw }).catch(() => {});
this._finishCtx(ctx, {
ok: false,
@@ -1403,6 +1417,7 @@ export const useChatStore = defineStore('chat', {
pipe.steps.forEach((s) => { s.state = 'done'; });
aMsg.status = 'done';
const dur = ((Date.now() - startAt) / 1000).toFixed(1) + 's';
this._flushPrivate(aMsg); // FR-PRIV-005:blocks 赋值后 flush 私有 pending
await this._persistFinal(aMsg, sessionId, 'done', { raw: aMsg.raw });
this._finishCtx(ctx, {
ok: true,
@@ -1422,11 +1437,73 @@ export const useChatStore = defineStore('chat', {
aMsg.errText = text;
const pipe = aMsg.blocks[0];
pipe.status = 'error';
this._flushPrivate(aMsg); // FR-PRIV-005:search error 路径同样 flush
this._persistFinal(aMsg, sessionId, 'error', { raw: aMsg.raw }).catch(() => {});
this._finishCtx(ctx, { ok: false, error: { code: err.code || 'NETWORK', message: text, retryable: true }, message: aMsg, sessionId, userMessageId: userId, assistantMessageId: aMsg.id });
}
},
// ---------------------------------------------------------------
// FR-PRIV-003/005:私有资料问答并行接线(fire-and-forget,失败静默)
// ---------------------------------------------------------------
/**
* 并行触发私有资料检索(send 内不 await)。
* - 未命中 / 异常(含错误信封)静默 return(FR-PRIV-003:零痕迹)
* - 命中且本轮上下文仍活跃(streamCtx[sessionId].aMsg 引用相等,TL 裁决假设 9):
* 流未收口 → 暂存 aMsg._privatePending(终态由 flush 挂载并持久化)
* 已终态(aMsg 已不在 blocks 保护期)→ 直挂 + 补 _persistFinal
* - 跨轮迟到命中:上下文不匹配 → 丢弃(不得串挂到新消息)
*/
async _firePrivate(q, sessionId, aMsg) {
let payload;
try {
payload = await askPrivate({ question: q, session_id: sessionId });
} catch {
return; // 网络/信封错误一律静默
}
if (!payload || payload.ok === false || !payload.hit) return; // 未命中静默
// 本轮上下文校验(TL 裁决假设 9:streamCtx[sessionId].aMsg 引用相等):
// ctx 存在且 aMsg 不匹配 = 跨轮迟到 → 丢弃(不得串挂到新消息)
const ctx = this.streamCtx[sessionId];
if (ctx && ctx.aMsg !== aMsg) return;
if (ctx && aMsg.status === 'streaming') {
// 流未收口:暂存,待 onComplete/error 终态后 _flushPrivate 挂载
aMsg._privatePending = payload;
return;
}
// 已终态(或本轮 ctx 已清理):直挂 + 补持久化(防重入由 _appendPrivate 保证)
if (this._appendPrivate(aMsg, payload)) {
this._persistFinal(aMsg, sessionId, aMsg.status, { raw: aMsg.raw }).catch(() => {});
}
},
/**
* 把私有命中 payload 挂到 aMsg 的 private_doc 块(防重入:已存在只挂一次)。
* @returns {boolean} 本次是否实际追加了块
*/
_appendPrivate(aMsg, payload) {
const blocks = Array.isArray(aMsg.blocks) ? aMsg.blocks : [];
if (blocks.some((b) => b && b.type === 'private_doc')) return false;
const block = {
type: 'private_doc',
answer: String((payload && payload.answer) || ''),
items: Array.isArray(payload && payload.items) ? payload.items.map((it) => ({ ...it })) : [],
};
aMsg.blocks = [...blocks, block];
return true;
},
/**
* 终态 flush:只清 _privatePending(有 pending 则挂块并持久化由调用方负责——
* flush 挂点均在 _persistFinal 之前)。
*/
_flushPrivate(aMsg) {
const pending = aMsg && aMsg._privatePending;
delete aMsg._privatePending;
if (!pending) return;
this._appendPrivate(aMsg, pending);
},
// ---------------------------------------------------------------
// 揭示动画(FR-CHAT-010)
// ---------------------------------------------------------------
@@ -1565,7 +1642,10 @@ export const useChatStore = defineStore('chat', {
const modeCfg = MODES[user.mode || this.mode] || MODES.ask;
if (user.mode) this.mode = user.mode; // 恢复原提问模式(FR-CHAT-012 相同模式重试)
this._setupCheckpoint(ctx);
if (modeCfg.endpoint !== '/api/qa/search') {
this._setupCheckpoint(ctx); // search 无流:不设定时器(send 内同约定,防止 fake timers 泄漏)
}
this._firePrivate(q, sessionId, aNew); // FR-PRIV-003:retry 轮同样触发私有检索
if (modeCfg.endpoint === '/api/qa/search') {
await this._runSearch(q, modeCfg, aNew, sessionId, user.id, ctx);
} else {
+67 -2
View File
@@ -8,9 +8,13 @@
- DELETE /api/vstore/documents/{id}
- POST /api/vstore/rebuild-index
- GET /api/vstore/health
FR-PRIV 新增(chunk 级索引,A1):
- GET /api/vstore/search-chunks ?q&top_k&min_score&doc_ids&exclude_template
- POST /api/vstore/rebuild-chunks
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any
@@ -19,6 +23,7 @@ from fastapi.responses import JSONResponse
from . import vault as _vault
from .backend import BackendError, handshake, parse_backend
from .chunks import CHUNK_MIN_SCORE, CHUNK_TOP_K_MAX
from .settings import (
DEFAULT_TOP_K,
MAX_TOP_K,
@@ -62,10 +67,12 @@ def _startup() -> None:
@app.get("/api/vstore/health")
async def health() -> dict[str, Any]:
"""健康检查:后端 ok + 库存量。"""
"""健康检查:后端 ok + 库存量 + chunk 库存量(FR-PRIV-007)。"""
vault = _vault.get_vault()
count = vault.index.count() if hasattr(vault, "index") else len(vault.list_rows())
return {"ok": True, "count": count, "backend": "local"}
chunks = getattr(vault, "chunks", None)
chunk_count = chunks.count() if chunks is not None else 0
return {"ok": True, "count": count, "backend": "local", "chunk_count": chunk_count}
@app.post("/api/vstore/ingest")
@@ -147,3 +154,61 @@ async def rebuild_index() -> dict[str, Any]:
if rebuilt["status"] == "error":
return _error("INTERNAL", rebuilt["message"], 500, degraded=True)
return {"ok": True, "count": rebuilt["count"]}
# ---------------------------------------------------------------------------
# FR-PRIV(A1):chunk 级索引端点
# ---------------------------------------------------------------------------
@app.get("/api/vstore/search-chunks")
async def search_chunks(
q: str = "",
top_k: int = 5,
min_score: float = CHUNK_MIN_SCORE,
doc_ids: str = "",
exclude_template: bool = False,
) -> dict[str, Any]:
"""chunk 级检索(FR-PRIV-002)。无降级档:低于 min_score 全部丢弃 → empty。
top_k clamp 1..20;doc_ids 为逗号分隔白名单(空 = 不过滤)。
"""
vault = _vault.get_vault()
chunks = getattr(vault, "chunks", None)
if chunks is None:
return {"ok": True, "mode": "empty", "query": q, "count": 0, "items": []}
top_k = max(1, min(int(top_k), CHUNK_TOP_K_MAX))
wanted = [d.strip() for d in doc_ids.split(",") if d.strip()] or None
try:
# 同步阻塞段(查询嵌入 + faiss 检索)一律 to_thread,不阻塞事件循环
vector = await asyncio.to_thread(_vault.get_embed_fn(), [q or ""])
result = await asyncio.to_thread(
chunks.search,
vector[0] if vector else [0.0] * _vault.EMBED_DIM,
top_k,
min_score,
wanted,
bool(exclude_template),
)
return {
"ok": True,
"mode": result.get("mode", "empty"),
"query": q,
"count": len(result.get("items") or []),
"items": result.get("items") or [],
}
except VaultError as exc:
return _error("INTERNAL", str(exc), 500, degraded=True)
@app.post("/api/vstore/rebuild-chunks")
async def rebuild_chunks() -> dict[str, Any]:
"""重建 chunk 索引(FR-PRIV-007):全部密文解密 → 切块 → 批量嵌入 →
原子切换。同步段 to_thread。
"""
vault = _vault.get_vault()
try:
result = await asyncio.to_thread(vault.rebuild_chunks)
except Exception as exc:
logger.exception("vstore: rebuild-chunks 失败")
return _error("INTERNAL", f"chunk 重建失败:{exc}", 500, degraded=True)
return {"ok": True, **result}
+509
View File
@@ -0,0 +1,509 @@
"""chunks.py —— 私有资料 chunk 级索引(FR-PRIV-002)。
与 doc 级检索索引(retrieval.py)并存、互不影响:
- 切块:正文段落聚合(目标 600 字 / 满 8 段封块 / 相邻重叠 1 段 / 每文档
上限 200 块);表格矩阵化为 `|a|b|` 行式文本段按原位插入段落流参与聚合
(实施方案 §1.2:outline 不参与切块,标题上下文仅由首块前缀 `# {title}` 提供)。
- 三文件落盘:index.json(meta 行,单一事实源) / texts.jsonl(chunk 文本,
index.json 可按需重建) / index.faiss(faiss 二进制,与行序一一对应)。
写入纪律仿 retrieval.py:单写锁(threading.Lock)+ tmp + os.replace 原子切换。
- 查询无降级档:向量内积(L2 归一化 = 余弦)低于 min_score 的全部丢弃 →
mode:"empty"(与 doc 级三档降级语义不同,FR-PRIV 假设 6)。
- 任何文件缺失/损坏按空库降级(可服务,不抛异常,FR-PRIV 假设 3)。
"""
from __future__ import annotations
import json
import logging
import os
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import faiss
import numpy as np
from .settings import EMBED_DIM
logger = logging.getLogger(__name__)
# ---- 切块参数(任务书契约值)----
CHUNK_TARGET_CHARS = 600 # 聚合目标字符数(达到即封块)
CHUNK_MIN_CHARS = 120 # 尾块最小字符数(低于则并入前块)
CHUNK_MAX_PARAS = 8 # 单块段落数上限(满即封块)
CHUNK_OVERLAP_PARAS = 1 # 相邻块重叠段数(前块最后 1 段作下块前缀)
CHUNK_MAX_PER_DOC = 200 # 每文档块数上限(超出截断)
CHUNK_TOP_K_MAX = 20 # search top_k 上限(1..20 clamp)
# ---- 检索参数 ----
CHUNK_MIN_SCORE = 0.30 # 分数阈值:低于全丢 → empty(无降级档)
CHUNK_SNIPPET_CHARS = 200 # snippet = text[:200](FR-PRIV 假设 4)
# ---- 落盘文件 ----
INDEX_FILE = "index.json"
FAISS_FILE = "index.faiss"
TEXTS_FILE = "texts.jsonl"
TMP_SUFFIX = ".tmp"
class ChunkError(Exception):
"""chunk 索引操作失败。"""
@dataclass
class Chunk:
"""单块。chunk_id = f"{doc_id}:{seq}"(doc_id 由 ChunkIndex.upsert_document 注入)。"""
seq: int
text: str
# ---------------------------------------------------------------------------
# 切块
# ---------------------------------------------------------------------------
def _table_to_text_rows(tables: list[list[list[str]]]) -> list[str]:
"""表格 → `|a|b|` 行式文本段(矩阵化,一行表行 = 一段)。"""
rows: list[str] = []
for table in tables or []:
for row in table or []:
cells = [str(c or "") for c in row]
text = "|" + "|".join(cells) + "|"
if text.strip("|"):
rows.append(text)
return rows
def build_chunks(
title: str,
outline: list[dict[str, Any]],
paragraphs: list[str],
tables: list[list[list[str]]],
) -> list[Chunk]:
"""正文 → chunk 列表(实施方案 §1.2 语义,TL 裁决假设 1)。
- 表格矩阵化为 `|a|b|` 行式文本段,插入段落流中间参与聚合切块
(废除「末尾 table 单独块」;提取结构 paragraphs/tables 无位置信息,
表格段统一落在段落流中位,与用例表 L1-05「表格位于段落流中间」判据一致)。
- outline 不参与切块(签名保留,实现忽略)。
- 长段落(> TARGET_CHARS)整段不切,独占一块。
- 封块条件:累计 ≥ TARGET_CHARS 或满 MAX_PARAS 段;尾块不足 MIN_CHARS 并入前块。
- 相邻块重叠:前块最后 OVERLAP_PARAS 段作下块前缀。
- 首块前缀 `# {title}`;seq 从 0 起;截断 CHUNK_MAX_PER_DOC 块。
- 全空输入返回 [],不抛异常。
"""
segs = [str(p or "") for p in (paragraphs or [])]
table_rows = _table_to_text_rows(tables or [])
# 表格段插入段落流中位(原位聚合,TL 裁决假设 1 / L1-05 判据)
if table_rows and len(segs) > 1:
mid = len(segs) // 2
segs = segs[:mid] + table_rows + segs[mid:]
else:
segs += table_rows
segs = [s for s in segs if s.strip()]
if not segs:
return []
# 逐段聚合(长段独占)
blocks: list[list[str]] = []
cur: list[str] = []
cur_chars = 0
for seg in segs:
seg_chars = len(seg)
# 该段自身已超目标:先封当前块,长段独占一块(假设 2)
if seg_chars > CHUNK_TARGET_CHARS:
if cur:
blocks.append(cur)
cur = []
cur_chars = 0
blocks.append([seg])
continue
# 常规:目标字符或段数上限即封块
if cur and (cur_chars + seg_chars >= CHUNK_TARGET_CHARS or len(cur) >= CHUNK_MAX_PARAS):
blocks.append(cur)
cur = []
cur_chars = 0
cur.append(seg)
cur_chars += seg_chars
if cur:
blocks.append(cur)
# 尾块合并:不足 MIN_CHARS 且非唯一块 → 并入前块
if len(blocks) > 1:
last_chars = sum(len(s) for s in blocks[-1])
if last_chars < CHUNK_MIN_CHARS:
merged = blocks.pop()
blocks[-1] = blocks[-1] + merged
# 截断
blocks = blocks[:CHUNK_MAX_PER_DOC]
# 拼块 + 重叠前缀 + 标题前缀
chunks: list[Chunk] = []
for i, blk in enumerate(blocks):
paras = list(blk)
if i > 0:
overlap = blocks[i - 1][-CHUNK_OVERLAP_PARAS:]
paras = overlap + paras
text = "\n".join(paras)
if i == 0:
prefix = f"# {title}".strip()
text = f"{prefix}\n{text}"
chunks.append(Chunk(seq=i, text=text))
return chunks
# ---------------------------------------------------------------------------
# 工具
# ---------------------------------------------------------------------------
def _now_iso() -> str:
from datetime import datetime, timezone
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _normalize(vec: list[float]) -> np.ndarray:
"""L2 归一化(IndexFlatIP 下等价余弦)。零向量原样返回。"""
arr = np.asarray(vec, dtype=np.float32)
norm = float(np.linalg.norm(arr))
if norm > 0:
arr = arr / norm
return arr
def _fallback_embed_one(text: str) -> list[float]:
"""确定性回退嵌入(不触网):`种子N` 前缀 → 高位 0.9 + 低位补足到单位范数。
仅用于 items 未携带 vector 的场景(测试骨架 L1-12 语义:『种子1 高位』
分数 0.90——查询向量归一化后内积 = 0.90,落在 min_score 0.30 与 0.91
阈值之间)。生产路径由 vault 先 embed_batch_chunked 再 upsert,不走此分支。
"""
import re
vec = [0.0] * EMBED_DIM
m = re.match(r"种子\s*(\d+)", (text or "").strip())
if m:
seed = int(m.group(1))
vec[seed % EMBED_DIM] = 0.9
vec[(seed + 1) % EMBED_DIM] = 0.43589 # sqrt(1 - 0.81):补足单位范数
return vec
def _faiss_serialize(index: faiss.IndexFlatIP) -> bytes:
"""faiss 二进制(serialize_index 返回 ndarray → tobytes)。"""
arr = faiss.serialize_index(index)
return bytes(arr.tobytes() if hasattr(arr, "tobytes") else arr)
def _faiss_read(blob: bytes, dim: int) -> faiss.IndexFlatIP | None:
"""从二进制读 faiss 索引;损坏/空 blob 返回 None(调用方按空库降级)。"""
if not blob:
return None
try:
data = np.frombuffer(blob, dtype=np.uint8)
index = faiss.deserialize_index(data)
if isinstance(index, faiss.IndexFlatIP):
return index
if isinstance(index, faiss.IndexRefineFlat) and isinstance(
index.base_index, faiss.IndexFlatIP
):
return index.base_index
return None
except Exception as exc:
logger.warning("chunks: index.faiss 读取失败,按空库处理:%s", exc)
return None
class ChunkIndex:
"""chunk 向量索引:index.json(meta 行)+ index.faiss + texts.jsonl。
单写锁;upsert 替换语义(先移旧行再重建 faiss);搜索无降级档。
"""
def __init__(self, path: str | Path, dim: int = EMBED_DIM) -> None:
self.dir = Path(path)
self.dim = dim
self._lock = threading.Lock()
self._faiss: faiss.IndexFlatIP = faiss.IndexFlatIP(dim)
self._rows: list[dict[str, Any]] = []
# ---- 路径 ----
@property
def json_path(self) -> Path:
return self.dir / INDEX_FILE
@property
def faiss_path(self) -> Path:
return self.dir / FAISS_FILE
@property
def texts_path(self) -> Path:
return self.dir / TEXTS_FILE
# ---- 加载 / 落盘 ----
def load(self) -> None:
"""加载三文件。任一缺失/损坏按空库降级(不抛异常,FR-PRIV 假设 3)。"""
with self._lock:
self._rows = []
self._faiss = faiss.IndexFlatIP(self.dim)
rows = []
try:
data = json.loads(self.json_path.read_text(encoding="utf-8"))
items = data.get("items", []) if isinstance(data, dict) else []
except (ValueError, OSError) as exc:
logger.warning("chunks: index.json 读取失败,按空库处理:%s", exc)
items = []
texts = self._read_texts()
for row_id, item in enumerate(items):
if not isinstance(item, dict):
continue
row = dict(item)
row["row_id"] = row_id
row["text"] = texts.get(str(row.get("chunk_id") or "")) or ""
rows.append(row)
faiss_index = _faiss_read(self._read_faiss(), self.dim)
# 防御:faiss 向量数 > 行数(损坏)→ 只保留可配对上的向量;
# faiss 向量数 ≤ 行数合法(行可无向量——测试/空向量场景),
# faiss 行序与 json 行序一一对应,直接使用。
n_rows = len(rows)
n_vec = faiss_index.ntotal if faiss_index is not None else 0
if faiss_index is not None and n_vec > n_rows:
logger.warning(
"chunks: index.faiss 向量数(%d)多于 index.json 行数(%d),"
"按可配对上重建", n_vec, n_rows,
)
rebuilt = faiss.IndexFlatIP(self.dim)
for r in rows:
vec = r.get("vector")
if isinstance(vec, list) and vec:
rebuilt.add(_normalize(vec).reshape(1, self.dim))
faiss_index = rebuilt
self._rows = rows
if faiss_index is not None:
self._faiss = faiss_index
def _read_texts(self) -> dict[str, str]:
"""texts.jsonl → {chunk_id: text}(损坏/缺失 → {})。"""
out: dict[str, str] = {}
try:
raw = self.texts_path.read_text(encoding="utf-8")
except OSError:
return out
for line in raw.splitlines():
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
cid = obj.get("chunk_id")
text = obj.get("text")
if isinstance(cid, str) and isinstance(text, str):
out[cid] = text
except ValueError:
continue
return out
def _read_faiss(self) -> bytes:
try:
return self.faiss_path.read_bytes()
except OSError:
return b""
def _atomic_write(self, path: Path, data: bytes) -> None:
"""tmp + os.replace 原子切换(仿 retrieval.py 纪律)。"""
self.dir.mkdir(parents=True, exist_ok=True)
tmp = path.with_name(path.name + TMP_SUFFIX)
tmp.write_bytes(data)
os.replace(tmp, path)
def _dump_locked(self) -> None:
"""写三文件(需持有锁)。"""
self.dir.mkdir(parents=True, exist_ok=True)
payload = {
"version": 1,
"items": [
{k: v for k, v in r.items() if k not in ("row_id", "text")}
for r in self._rows
],
}
self._atomic_write(
self.json_path, json.dumps(payload, ensure_ascii=False).encode("utf-8")
)
self._atomic_write(self.faiss_path, _faiss_serialize(self._faiss))
text_lines = [
json.dumps({"chunk_id": r.get("chunk_id") or "", "text": r.get("text") or ""},
ensure_ascii=False)
for r in self._rows
]
self._atomic_write(self.texts_path, ("\n".join(text_lines) + "\n").encode("utf-8"))
# ---- 增删 ----
def upsert_document(self, doc_id: str, items: list[dict[str, Any]]) -> int:
"""入库一个文档的全部 chunk(替换语义:先移旧行,过滤重建 faiss)。
items 元素:{"text", "doc_name", "doc_category", "doc_tags",
"as_template", "created_at"};seq 由 text 取不到时按下标补。
返回写入行数。
"""
with self._lock:
self._rows = [r for r in self._rows if r.get("doc_id") != doc_id]
for seq, item in enumerate(items or []):
row = dict(item)
row["doc_id"] = doc_id
row["seq"] = int(row.get("seq", seq))
row["chunk_id"] = f"{doc_id}:{row['seq']}"
text = str(row.get("text") or "")
row["text"] = text
row["snippet"] = text[:CHUNK_SNIPPET_CHARS]
row.setdefault("doc_name", "")
row.setdefault("doc_category", "")
row.setdefault("doc_tags", "")
row.setdefault("as_template", False)
row.setdefault("created_at", _now_iso())
if not isinstance(row.get("vector"), list):
# items 未携带 vector:确定性回退嵌入(测试/防御场景,不触网)
row["vector"] = _fallback_embed_one(str(row.get("text") or ""))
else:
row["vector"] = list(row.get("vector") or [])
self._rows.append(row)
# 过滤重建 faiss(faiss 不支持逐行删)
self._faiss = faiss.IndexFlatIP(self.dim)
for row in self._rows:
vec = row.get("vector")
if isinstance(vec, list) and vec:
self._faiss.add(_normalize(vec).reshape(1, self.dim))
self._dump_locked()
return len(self._rows)
def remove_document(self, doc_id: str) -> bool:
"""删除该 doc 全部行(不存在返回 False)。"""
with self._lock:
before = len(self._rows)
self._rows = [r for r in self._rows if str(r.get("doc_id")) != doc_id]
if len(self._rows) == before:
return False
self._faiss = faiss.IndexFlatIP(self.dim)
for row in self._rows:
vec = row.get("vector")
if isinstance(vec, list) and vec:
self._faiss.add(_normalize(vec).reshape(1, self.dim))
self._dump_locked()
return True
# ---- 查询 ----
def rows(self) -> list[dict[str, Any]]:
"""行快照(不含 vector/text,供列表接口)。"""
with self._lock:
return [
{k: v for k, v in r.items() if k not in ("vector", "text", "row_id")}
for r in self._rows
]
def count(self) -> int:
with self._lock:
return len(self._rows)
def search(
self,
query_vector: list[float],
top_k: int = 5,
min_score: float = CHUNK_MIN_SCORE,
doc_ids: list[str] | None = None,
exclude_template: bool = False,
) -> dict[str, Any]:
"""向量检索,无降级档:分数 < min_score 全部丢弃 → mode:"empty"。
返回 {"mode": "vector"|"empty", "items": [{chunk_id, doc_id, seq, snippet,
score, doc_name, doc_category, doc_tags, as_template, created_at}]}。
"""
top_k = max(1, min(int(top_k), CHUNK_TOP_K_MAX))
wanted = {str(d) for d in (doc_ids or [])} if doc_ids else None
with self._lock:
rows = self._rows
if not rows:
return {"mode": "empty", "items": []}
q = _normalize(query_vector).reshape(1, self.dim)
try:
dist, idx = self._faiss.search(q, min(top_k * 8, len(rows)))
except Exception as exc:
logger.warning("chunks: 向量检索失败,按空结果处理:%s", exc)
return {"mode": "empty", "items": []}
hits: list[dict[str, Any]] = []
for score, row_idx in zip(dist[0].tolist(), idx[0].tolist()):
if not (0 <= row_idx < len(rows)):
continue
score = float(score)
if score < min_score:
continue
row = rows[row_idx]
if wanted is not None and str(row.get("doc_id")) not in wanted:
continue
if exclude_template and bool(row.get("as_template")):
continue
hits.append(
{
"chunk_id": row.get("chunk_id") or "",
"doc_id": row.get("doc_id") or "",
"seq": row.get("seq"),
"snippet": row.get("snippet") or "",
"score": score,
"doc_name": row.get("doc_name") or "",
"doc_category": row.get("doc_category") or "",
"doc_tags": row.get("doc_tags") or "",
"as_template": bool(row.get("as_template")),
"created_at": row.get("created_at") or "",
}
)
if len(hits) >= top_k:
break
if not hits:
return {"mode": "empty", "items": []}
return {"mode": "vector", "items": hits}
# ---- 重建 ----
def clear(self) -> None:
"""清空内存(测试/重建用;磁盘文件不动)。"""
with self._lock:
self._rows = []
self._faiss = faiss.IndexFlatIP(self.dim)
def rebuild_all(self, docs: dict[str, list[dict[str, Any]]]) -> int:
"""整库重建:docs = {doc_id: [chunk item, ...]};覆盖三文件,返回行数。"""
with self._lock:
new_rows: list[dict[str, Any]] = []
for doc_id, items in (docs or {}).items():
for seq, item in enumerate(items or []):
row = dict(item)
row["doc_id"] = doc_id
row["seq"] = int(row.get("seq", seq))
row["chunk_id"] = f"{doc_id}:{row['seq']}"
text = str(row.get("text") or "")
row["text"] = text
row["snippet"] = text[:CHUNK_SNIPPET_CHARS]
row.setdefault("doc_name", "")
row.setdefault("doc_category", "")
row.setdefault("doc_tags", "")
row.setdefault("as_template", False)
row.setdefault("created_at", _now_iso())
if not isinstance(row.get("vector"), list):
# items 未携带 vector:确定性回退嵌入(测试/防御场景,不触网)
row["vector"] = _fallback_embed_one(str(row.get("text") or ""))
else:
row["vector"] = list(row.get("vector") or [])
new_rows.append(row)
new_faiss = faiss.IndexFlatIP(self.dim)
for row in new_rows:
vec = row.get("vector")
if isinstance(vec, list) and vec:
new_faiss.add(_normalize(vec).reshape(1, self.dim))
self._rows = new_rows
self._faiss = new_faiss
self._dump_locked()
return len(new_rows)
+104
View File
@@ -27,6 +27,46 @@ class EmbedError(Exception):
"""嵌入调用失败(服务不可达/维度不符/响应异常)。"""
# 测试注入钩子(仿 vault.set_embed_fn 模式):T1 用例按契约名 _CHUNKED_REQUEST_FN
# monkeypatch 本模块该函数后,embed_batch_chunked 不触网。
def _CHUNKED_REQUEST_FN(payload: dict) -> list[list[float]]:
"""单批 /api/embed 请求(input 数组)。返回该批 embedding 列表(真 Ollama HTTP)。"""
paths = get_paths()
url = (payload.get("url") or paths.embed_url or DEFAULT_EMBED_URL).rstrip("/")
model = payload.get("model") or paths.embed_model or DEFAULT_EMBED_MODEL
inputs = [str(t or "") for t in payload.get("input") or []]
try:
resp = httpx.post(
f"{url}/api/embed",
json={"model": model, "input": inputs},
timeout=EMBED_TIMEOUT,
)
resp.raise_for_status()
data = resp.json()
embeddings = data.get("embeddings")
if not isinstance(embeddings, list):
raise EmbedError(f"嵌入响应形态异常:{str(data)[:200]}")
out: list[list[float]] = []
for vec in embeddings:
if not isinstance(vec, list):
raise EmbedError(f"嵌入响应形态异常:{str(data)[:200]}")
vector = [float(v) for v in vec]
if len(vector) != EMBED_DIM:
raise EmbedError(f"嵌入维度不符:期望 {EMBED_DIM},实际 {len(vector)}")
out.append(vector)
if len(out) != len(inputs):
raise EmbedError(f"嵌入条数不符:期望 {len(inputs)},实际 {len(out)}")
return out
except EmbedError:
raise
except (httpx.HTTPError, ValueError) as exc:
raise EmbedError(f"嵌入服务调用失败:{exc}") from exc
CHUNKED_BATCH_SIZE = 16 # embed_batch_chunked 每批条数(任务书契约)
CHUNKED_WORKERS = 4 # 批并发线程数(任务书契约)
def _zero_vector() -> list[float]:
return [0.0] * EMBED_DIM
@@ -89,3 +129,67 @@ def embed_batch(
logger.warning("embed: 第 %d 条嵌入异常:%s", idx, exc)
results[idx] = None
return [v if v is not None else _zero_vector() for v in results]
def embed_batch_chunked(
texts: list[str],
batch_size: int = CHUNKED_BATCH_SIZE,
workers: int = CHUNKED_WORKERS,
url: str | None = None,
model: str | None = None,
) -> list[list[float]]:
"""chunk 批量嵌入:/api/embed 的 input 数组(batch 16 × workers 4 并发)。
与既有 embed_batch(单条循环、并发 8)区分。嵌入纪律(FR-PRIV-002):
单批失败整批重试一次,仍失败该批全部降级零向量;保持输入顺序;
批间并发上限 = workers(线程数)。
"""
if not texts:
return []
batch_size = max(1, int(batch_size))
workers = max(1, int(workers))
batches = [texts[i : i + batch_size] for i in range(0, len(texts), batch_size)]
results: list[list[float]] = [None] * len(texts)
def run_batch(offset: int, batch: list[str]) -> None:
payload = {"input": list(batch), "url": url, "model": model}
vecs: list[list[float]] | None = None
try:
vecs = _CHUNKED_REQUEST_FN(payload)
except Exception as exc:
logger.warning("chunks: 批量嵌入失败(第 %d 批,首调):%s", offset, exc)
try:
vecs = _CHUNKED_REQUEST_FN(payload)
except Exception as exc2:
logger.warning(
"chunks: 批量嵌入重试仍失败(第 %d 批),降级零向量:%s", offset, exc2
)
vecs = None
if vecs is None:
for i in range(len(batch)):
results[offset + i] = _zero_vector()
return
for i, vec in enumerate(vecs):
if offset + i >= len(results):
break
if isinstance(vec, list) and len(vec) == EMBED_DIM:
results[offset + i] = vec
else:
logger.warning("chunks: 第 %d 条嵌入维度不符,降级零向量", offset + i)
results[offset + i] = _zero_vector()
for i in range(len(vecs), len(batch)): # 响应条数不足:缺的降级零向量
if offset + i < len(results):
results[offset + i] = _zero_vector()
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = []
offset = 0
for batch in batches:
futures.append(pool.submit(run_batch, offset, batch))
offset += len(batch)
for future in futures:
try:
future.result()
except Exception as exc: # 线程池兜底异常:整批降级零向量
logger.warning("chunks: 批量嵌入线程异常:%s", exc)
return [v if v is not None else _zero_vector() for v in results]
+142
View File
@@ -0,0 +1,142 @@
"""rebuild_chunks.py —— 存量资料 chunk 索引重建 CLI(FR-PRIV-007)。
不进 HTTP(避免超时),由 TL 在部署根后台执行:
cd {部署根}/vstore && ../venv/bin/python scripts/rebuild_chunks.py
路径推导(零硬编码,优先级从高到低):
- CLI 参数 --root / --vault / --index / --chunks(供测试/特殊部署)
- 环境变量 VSTORE_ROOT / VSTORE_VAULT_PATH / VSTORE_INDEX_PATH / VSTORE_CHUNK_PATH
- 默认:root = 脚本所在目录的父目录(部署根下的 vstore/)
密钥加载:cwd=部署根时加载 vstore/.env(VSTORE_KEY 必须可得),否则用环境变量。
进度打印 + 完成汇总;重建期间 vstore HTTP 服务(若运行中)不阻塞——
本脚本是独立子进程,只经文件锁写 chunks 目录。
"""
from __future__ import annotations
import argparse
import os
import sys
import time
from pathlib import Path
def _load_env_file(path: Path) -> None:
"""vstore/.env → os.environ(已存在的键不覆盖)。"""
if not path.exists():
return
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
def main() -> int:
parser = argparse.ArgumentParser(description="重建私有资料 chunk 索引(FR-PRIV-007)")
parser.add_argument("--root", help="vstore 数据根(默认 VSTORE_ROOT 或脚本父目录)")
parser.add_argument("--vault", help="密文目录(默认 {root}/vault)")
parser.add_argument("--index", help="doc 级索引目录(默认 {root}/index)")
parser.add_argument("--chunks", help="chunk 索引目录(默认 {root}/chunks)")
parser.add_argument("--batch-size", type=int, default=16, help="批量嵌入每批条数")
parser.add_argument("--workers", type=int, default=4, help="批量嵌入并发线程数")
args = parser.parse_args()
script_dir = Path(__file__).resolve().parent
vstore_dir = script_dir.parent # 部署根下的 vstore/
deploy_root = vstore_dir.parent # 部署根
_load_env_file(vstore_dir / ".env") # cwd=部署根时密钥在此
root = args.root or os.environ.get("VSTORE_ROOT") or str(vstore_dir)
vault_path = args.vault or os.environ.get("VSTORE_VAULT_PATH") or str(Path(root) / "vault")
index_path = args.index or os.environ.get("VSTORE_INDEX_PATH") or str(Path(root) / "index")
chunk_path = args.chunks or os.environ.get("VSTORE_CHUNK_PATH") or str(Path(root) / "chunks")
# 密钥:cwd=部署根时先找 {deploy_root}/vstore/.env,再环境变量
_load_env_file(deploy_root / "vstore" / ".env")
sys.path.insert(0, str(vstore_dir.parent)) # 保证 `import vstore` 可用
from vstore import settings as _settings
_settings.set_paths(
_settings.VstorePaths(
root=root,
vault_path=vault_path,
index_path=index_path,
chunk_path=chunk_path,
)
)
try:
_settings.get_fernet_key()
except RuntimeError as exc:
print(f"[rebuild-chunks] 密钥缺失:{exc}", file=sys.stderr)
return 2
from vstore import vault as _vault
from vstore import embed as _embed
vault = _vault.Vault(vault_path, index_path)
vault.index.load()
vault.chunks.load()
enc_files = sorted(Path(vault_path).glob("*.enc"))
print(f"[rebuild-chunks] 密文目录:{vault_path}")
print(f"[rebuild-chunks] 待处理密文:{len(enc_files)} 条")
started = time.perf_counter()
docs: dict[str, list[dict[str, str | int | bool | list[float]]]] = {}
total_texts = 0
for n, enc in enumerate(enc_files, 1):
doc_id = enc.stem
try:
doc = _vault._deserialize_doc(_vault.load_encrypted(enc))
except Exception as exc:
print(f" [{n}/{len(enc_files)}] {doc_id} 解密失败,跳过:{exc}")
continue
meta = doc.get("meta") or {}
extracted = doc.get("extracted") or {}
name = str(meta.get("name") or doc_id)
built = _vault.build_chunks(
extracted.get("title") or name,
extracted.get("outline") or [],
[str(p or "") for p in extracted.get("paragraphs") or []],
extracted.get("tables") or [],
)
if not built:
print(f" [{n}/{len(enc_files)}] {doc_id} 无正文,0 块")
continue
vecs = _embed.embed_batch_chunked(
[c.text for c in built],
batch_size=args.batch_size,
workers=args.workers,
)
docs[doc_id] = [
{
"seq": c.seq,
"text": c.text,
"vector": list(vec),
"doc_name": name,
"doc_category": str(meta.get("category") or ""),
"doc_tags": str(meta.get("tags") or ""),
"as_template": bool(meta.get("as_template")),
"created_at": str(meta.get("created_at") or ""),
}
for c, vec in zip(built, vecs)
]
total_texts += len(built)
print(f" [{n}/{len(enc_files)}] {doc_id}({name[:30]})→ {len(built)} 块")
count = vault.chunks.rebuild_all(docs)
duration_ms = int((time.perf_counter() - started) * 1000)
print(f"[rebuild-chunks] 完成:docs={len(docs)}, chunks={count}, "
f"嵌入文本={total_texts}, duration_ms={duration_ms}")
return 0
if __name__ == "__main__":
sys.exit(main())
+3
View File
@@ -42,6 +42,7 @@ class VstorePaths:
root: str = field(default_factory=lambda: os.environ.get("VSTORE_ROOT", "data/vstore"))
vault_path: str = field(default_factory=lambda: os.environ.get("VSTORE_VAULT_PATH", ""))
index_path: str = field(default_factory=lambda: os.environ.get("VSTORE_INDEX_PATH", ""))
chunk_path: str = field(default_factory=lambda: os.environ.get("VSTORE_CHUNK_PATH", ""))
embed_url: str = field(default_factory=lambda: os.environ.get("VSTORE_EMBED_URL", DEFAULT_EMBED_URL))
embed_model: str = field(default_factory=lambda: os.environ.get("VSTORE_EMBED_MODEL", DEFAULT_EMBED_MODEL))
@@ -51,6 +52,8 @@ class VstorePaths:
self.vault_path = str(Path(self.root) / "vault")
if not self.index_path:
self.index_path = str(Path(self.root) / "index")
if not self.chunk_path:
self.chunk_path = str(Path(self.root) / "chunks")
def get_fernet_key() -> bytes:
+114 -1
View File
@@ -25,6 +25,7 @@ from .crypto import CryptoError, dump_encrypted, load_encrypted, secure_delete
from .embed import embed_batch
from .parsing import ParsingError, extract, outline_to_embedding_chunks
from .retrieval import RetrievalIndex, RetrievalError
from .chunks import ChunkIndex, build_chunks
from .settings import (
EMBED_BODY_CAP,
EMBED_DIM,
@@ -38,6 +39,14 @@ logger = logging.getLogger(__name__)
_EMBED_FN: Callable[..., list[list[float]]] = embed_batch
def _chunked_embed(texts: list[str]) -> list[list[float]]:
"""chunk 批量嵌入(经模块属性动态查找——测试 monkeypatch vstore.embed
的 embed_batch_chunked 后此处即时生效,不受静态 import 快照影响)。"""
from . import embed as _embed_mod
return _embed_mod.embed_batch_chunked(texts)
class VaultError(Exception):
"""主库操作失败(限额/解析/加密/检索)。"""
@@ -113,6 +122,11 @@ class Vault:
def __init__(self, vault_path: str | Path, index_path: str | Path) -> None:
self.vault_dir = Path(vault_path)
self.index = RetrievalIndex(index_path)
# 私有资料 chunk 级索引(FR-PRIV-002):与 doc 级检索索引并存;
# 路径来自 settings(VSTORE_CHUNK_PATH,默认 {root}/chunks),测试可用 set_paths 覆盖
chunk_path = get_paths().chunk_path
self.chunks = ChunkIndex(chunk_path)
self.chunks.load()
# ---- 增删改查 ----
@@ -180,6 +194,48 @@ class Vault:
secure_delete(enc_path)
raise VaultError(f"索引写入失败:{exc}") from exc
# ---- 私有资料 chunk 级索引(FR-PRIV-002):doc 级成功后追加;任何失败只告警不阻断 ----
chunk_count = 0
chunk_ok = True
chunk_error = ""
try:
para_list = [str(p or "") for p in extracted.get("paragraphs") or []]
table_list = extracted.get("tables") or []
para_chars = sum(len(p) for p in para_list)
table_cells = sum(len(row) for table in table_list for row in table)
if para_chars or table_cells:
built = build_chunks(
extracted.get("title") or name,
extracted.get("outline") or [],
para_list,
table_list,
)
else:
built = []
if built:
texts = [c.text for c in built]
vecs = _chunked_embed(texts)
items = []
for c, vec in zip(built, vecs):
items.append(
{
"seq": c.seq,
"text": c.text,
"vector": list(vec),
"doc_name": name,
"doc_category": category,
"doc_tags": tags,
"as_template": bool(as_template),
"created_at": doc_meta["created_at"],
}
)
self.chunks.upsert_document(doc_id, items)
chunk_count = len(items)
except Exception as exc: # chunk 任何失败只告警,doc 级照常返回
chunk_ok = False
chunk_error = str(exc) or exc.__class__.__name__
logger.warning("vault: chunk 索引写入失败(不阻断 doc 级,%s):%s", doc_id, exc)
embed_chunks = doc_meta["embed_chunks"]
embed_sha = hashlib.sha256(
json.dumps(vectors[0], ensure_ascii=False).encode("utf-8")
@@ -198,6 +254,10 @@ class Vault:
"enc_path": str(enc_path),
"embed_sha": embed_sha,
"embed_chunks": embed_chunks,
# FR-PRIV-002:chunk 级索引结果(失败不阻断 doc 级)
"chunk_count": chunk_count,
"chunk_ok": chunk_ok,
"chunk_error": chunk_error,
}
def get_decrypted(self, doc_id: str) -> dict[str, Any] | None:
@@ -234,11 +294,16 @@ class Vault:
return (doc.get("extracted") or {}).get("style_hints") or {}
def delete(self, doc_id: str) -> bool:
"""删除:索引行 + 密文 secure_delete。"""
"""删除:索引行 + 密文 secure_delete + chunk 行联动(FR-PRIV-006)。"""
removed = self.index.remove_item(doc_id)
enc_path = self.vault_dir / f"{doc_id}.enc"
if enc_path.exists():
secure_delete(enc_path)
# chunk 索引联动(失败只告警,不改变 doc 级结果)
try:
self.chunks.remove_document(doc_id)
except Exception as exc:
logger.warning("vault: chunk 行删除失败(%s):%s", doc_id, exc)
return removed
def list_rows(self) -> list[dict[str, Any]]:
@@ -304,6 +369,54 @@ class Vault:
return {"status": "error", "message": f"索引重建失败:{exc}"}
return {"status": "ok", "count": count}
def rebuild_chunks(self) -> dict[str, Any]:
"""整库重建 chunk 索引(FR-PRIV-007):遍历密文解密 → build_chunks →
批量嵌入 → rebuild_all 覆盖三文件。返回 {docs, chunks, duration_ms}。
"""
from time import perf_counter
started = perf_counter()
docs: dict[str, list[dict[str, Any]]] = {}
enc_files = sorted(self.vault_dir.glob("*.enc"))
for enc in enc_files:
doc_id = enc.stem
try:
doc = _deserialize_doc(load_encrypted(enc))
except (CryptoError, VaultError, OSError) as exc:
logger.warning("vault: 重建 chunk 解密失败(跳过,%s):%s", doc_id, exc)
continue
meta = doc.get("meta") or {}
extracted = doc.get("extracted") or {}
name = str(meta.get("name") or doc_id)
built = build_chunks(
extracted.get("title") or name,
extracted.get("outline") or [],
[str(p or "") for p in extracted.get("paragraphs") or []],
extracted.get("tables") or [],
)
if not built:
continue
vecs = _chunked_embed([c.text for c in built])
docs[doc_id] = [
{
"seq": c.seq,
"text": c.text,
"vector": list(vec),
"doc_name": name,
"doc_category": str(meta.get("category") or ""),
"doc_tags": str(meta.get("tags") or ""),
"as_template": bool(meta.get("as_template")),
"created_at": str(meta.get("created_at") or ""),
}
for c, vec in zip(built, vecs)
]
count = self.chunks.rebuild_all(docs)
return {
"docs": len(docs),
"chunks": count,
"duration_ms": int((perf_counter() - started) * 1000),
}
class RemoteVault:
"""远程主库:所有操作转 HTTP 指向远程 vstore 根(接口语义与本地一致)。