From df935599f257af6845bcf9dac7fe8d47fa497bfb Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 19 Sep 2026 09:27:07 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9F=A5=E8=AF=86=E5=BA=93=EF=BC=9A=E6=8D=A2?= =?UTF-8?q?=20embedding=20=E6=A8=A1=E5=9E=8B=E4=B8=8D=E5=86=8D=20500?= =?UTF-8?q?=EF=BC=8C=E6=A3=80=E7=B4=A2=E8=83=BD=E5=9B=9E=E6=8C=87=E4=BA=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三件事，都是"这三个模块一条测试都没有"的后果。 **换 embedder 会让检索整体 500。** 库里是 512 维的本地向量、查询是 1536 维的 远端向量时，cosine 直接 `shapes (1536,) and (512,) not aligned`。from_blob 有 dim 参数，但 kb.py 和 memory/store.py 调用时都没传；而且存量向量属于哪个模型 代码里没有任何地方记着，所以也无从降级。 Chunk / MemoryItem 各加 embed_model + embed_dim（走现成的 _COLUMN_MIGRATIONS， 存量回填空值，正好表示"不知道是谁建的"）。检索遇到对不上的向量整体退回纯 关键词并说破，不是把它们当 0 分静默沉底——那样用户看到的是结果莫名其妙变差， 而不是"这里有一批索引该重建了"。配套 kb.reindex + POST /api/kb/reindex。 长期记忆那份一起修了。判重那条更危险：维度碰巧相同但模型不同时 cosine 算出来 是随机数，会把不相关的旧记忆认成"重复"，然后把新内容丢掉。 **默认 embedder 没有语义能力，而开关藏在一个环境变量里。** LocalEmbedder 是 特征哈希 + TF，和 BM25 吃同一批词；alpha 默认 0.5 等于给了它一半权重，同义 改写一条都召不回。开启真模型的唯一开关是 AGENTLAB_USE_OPENAI_EMBEDDINGS—— 没界面、没人会发现。改成设置项（照 COPILOT_SETTING_KEY 的路子），知识库页顶部 能选能看能重建。 配置失败必须响：第一版写成静默退回本地，但选了 OpenAI 却拿到哈希向量的人， 在搜不出同义表达时完全无从怀疑到这里。现在抛 EmbedderUnavailable，API 400 原样交回"检查 OPENAI_API_KEY 是否配了"，且不会把配置改掉。 **检索结果不进工件库。** db_query、工具节点、agent 工具调用都落 put_json 快照， run_retrieve 什么都不落——于是结论只要来自知识库，"这个数是哪来的"就断在这儿， 而数字回指正是这个项目的签名能力。现在落 kind="retrieval_snapshot"，并发一条 retrieve.end 带 artifact id。解码层补了正式映射（原来会掉进"未知事件"兜底， 画成一坨原始 JSON）。 新增 test_kb.py 11 条，补上这三个模块此前的零覆盖，包括 embeddings.py 里 注释记着、但一直没有测试守的那个归一化边界（单条候选被 min_score 误杀）。 端到端在真实服务上验过：撞上存量向量返回 200 + 降级说明（以前 500）、 重建后 stale 归零、配不起来的远端模型 400 且配置不变。 --- backend/app/api/knowledge.py | 72 ++++++++++- backend/app/core/events.py | 3 + backend/app/db/base.py | 6 + backend/app/db/models.py | 10 ++ backend/app/engine/nodes/knowledge.py | 40 +++++- backend/app/main.py | 13 ++ backend/app/memory/embeddings.py | 91 ++++++++++++-- backend/app/memory/kb.py | 70 ++++++++++- backend/app/memory/store.py | 23 +++- backend/tests/test_kb.py | 174 ++++++++++++++++++++++++++ frontend/src/api/client.ts | 9 ++ frontend/src/pages/KnowledgePage.tsx | 90 +++++++++++++ frontend/src/run/decode.ts | 24 ++++ scripts/check-decode.mjs | 29 +++++ 14 files changed, 636 insertions(+), 18 deletions(-) create mode 100644 backend/tests/test_kb.py diff --git a/backend/app/api/knowledge.py b/backend/app/api/knowledge.py index d26ed6b3..31a5fc54 100644 --- a/backend/app/api/knowledge.py +++ b/backend/app/api/knowledge.py @@ -175,8 +175,76 @@ async def search_kb( alpha: float = Query(default=0.5, ge=0, le=1, description="1=纯向量, 0=纯关键词"), session: AsyncSession = Depends(get_session), ) -> dict[str, Any]: - hits = await kb.search(session, collection=collection, query=q, limit=limit, alpha=alpha) - return {"query": q, "collection": collection, "results": hits} + notes: list[str] = [] + hits = await kb.search(session, collection=collection, query=q, limit=limit, + alpha=alpha, on_degrade=notes.append) + # 退回关键词这件事要跟着结果一起交出去，而不是只写进服务端日志—— + # 调用方（设置页的"试一下"、外部脚本）看到的是结果变差，得知道为什么 + return {"query": q, "collection": collection, "results": hits, "degraded": notes} + + +@kb_router.get("/embedding") +async def embedding_status( + collection: str | None = None, session: AsyncSession = Depends(get_session) +) -> dict[str, Any]: + """当前用的是哪个 embedder，以及有多少条向量已经对不上了。""" + from app.memory.embeddings import EMBEDDING_SETTING_KEY, embedder_dim, embedder_id + from app.db.models import Setting + + row = await session.get(Setting, EMBEDDING_SETTING_KEY) + saved = (row.value if row else {}) or {} + return { + "embedder": embedder_id(), + "dim": embedder_dim(), + "configured": bool(saved.get("kind")), + "kind": saved.get("kind", "local"), + "model": saved.get("model", ""), + "stale_chunks": await kb.stale_count(session, collection), + } + + +class EmbeddingIn(BaseModel): + kind: str = Field(default="local", pattern="^(local|openai)$") + model: str = "" + + +@kb_router.put("/embedding") +async def set_embedding( + payload: EmbeddingIn, session: AsyncSession = Depends(get_session) +) -> dict[str, Any]: + """换 embedding 模型。 + + 建不起来就 400 原样交回原因——选了远端却静默退回本地哈希向量， + 用户会以为自己在用语义检索，而搜不出同义表达时完全无从怀疑到这里。 + """ + from app.memory import embeddings as emb + from app.db.models import Setting + + try: + emb.configure(payload.kind, payload.model) + except emb.EmbedderUnavailable as e: + raise HTTPException(400, str(e)) from e + + row = await session.get(Setting, emb.EMBEDDING_SETTING_KEY) + value = {"kind": payload.kind, "model": payload.model} + if row: + row.value = value + else: + session.add(Setting(key=emb.EMBEDDING_SETTING_KEY, value=value)) + await session.commit() + return { + "embedder": emb.embedder_id(), + "dim": emb.embedder_dim(), + "stale_chunks": await kb.stale_count(session), + } + + +@kb_router.post("/reindex") +async def reindex( + collection: str | None = None, session: AsyncSession = Depends(get_session) +) -> dict[str, Any]: + """用当前 embedder 重算向量。换了模型之后唯一的恢复手段。""" + return await kb.reindex(session, collection) @kb_router.delete("/documents/{doc_id}", status_code=204) diff --git a/backend/app/core/events.py b/backend/app/core/events.py index 3ba68f20..8fd621b9 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -39,6 +39,9 @@ class EventType(StrEnum): TOOL_START = "tool.start" TOOL_END = "tool.end" TOOL_ERROR = "tool.error" + # 知识检索。以前只发一条 info 日志，于是"这句结论依据的是哪份文档的哪一段" + # 在轨迹里查不到——而 SQL 取数那条路早就有工件快照可以下钻 + RETRIEVE_END = "retrieve.end" SANDBOX_START = "sandbox.start" SANDBOX_END = "sandbox.end" diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 42707ef8..43f9a7db 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -75,6 +75,12 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ ("workflows", "published_by", "VARCHAR(100)"), ("workflow_versions", "graph_hash", "VARCHAR(64)"), ("approvals", "resolved_by", "VARCHAR(100)"), + # 存量向量没有这两列，回填成空串 / 0 —— 正好表示"不知道是谁建的"， + # 检索侧会把它们当成对不上当前 embedder 处理 + ("chunks", "embed_model", "VARCHAR(100) DEFAULT ''"), + ("chunks", "embed_dim", "INTEGER DEFAULT 0"), + ("memories", "embed_model", "VARCHAR(100) DEFAULT ''"), + ("memories", "embed_dim", "INTEGER DEFAULT 0"), ] diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 4e6d8249..70c39b5e 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -256,6 +256,11 @@ class MemoryItem(Base, TimestampMixin): content: Mapped[str] = mapped_column(Text) meta: Mapped[dict[str, Any]] = mapped_column(default=dict) embedding: Mapped[bytes | None] = mapped_column(LargeBinary, default=None) + # 这条向量是谁产出的。没有它就无法判断存量向量还能不能和当前查询比对—— + # 换 embedder 之后维度对不上，cosine 直接 ValueError（512 vs 1536）， + # 而在此之前代码里没有任何地方能看出"这条是旧模型建的" + embed_model: Mapped[str] = mapped_column(String(100), default="") + embed_dim: Mapped[int] = mapped_column(Integer, default=0) importance: Mapped[float] = mapped_column(Float, default=0.5) last_used_at: Mapped[datetime | None] = mapped_column(default=None) use_count: Mapped[int] = mapped_column(Integer, default=0) @@ -290,6 +295,11 @@ class Chunk(Base): ordinal: Mapped[int] = mapped_column(Integer, default=0) content: Mapped[str] = mapped_column(Text) embedding: Mapped[bytes | None] = mapped_column(LargeBinary, default=None) + # 这条向量是谁产出的。没有它就无法判断存量向量还能不能和当前查询比对—— + # 换 embedder 之后维度对不上，cosine 直接 ValueError（512 vs 1536）， + # 而在此之前代码里没有任何地方能看出"这条是旧模型建的" + embed_model: Mapped[str] = mapped_column(String(100), default="") + embed_dim: Mapped[int] = mapped_column(Integer, default=0) meta: Mapped[dict[str, Any]] = mapped_column(default=dict) document: Mapped[Document] = relationship(back_populates="chunks") diff --git a/backend/app/engine/nodes/knowledge.py b/backend/app/engine/nodes/knowledge.py index 8b0002e1..444e0e7a 100644 --- a/backend/app/engine/nodes/knowledge.py +++ b/backend/app/engine/nodes/knowledge.py @@ -72,10 +72,15 @@ async def run_retrieve(state: GraphState, ctx: NodeContext) -> dict[str, Any]: limit = int(ctx.cfg("limit", 5) or 5) alpha = float(ctx.cfg("alpha", 0.5) or 0.5) # 1=纯向量，0=纯关键词 + degraded: list[str] = [] async with SessionLocal() as session: hits = await kb.search( - session, collection=collection, query=query, limit=limit, alpha=alpha + session, collection=collection, query=query, limit=limit, alpha=alpha, + on_degrade=degraded.append, ) + for note in degraded: + # 检索悄悄少了一半能力，不说的话用户只会觉得"最近搜得不准" + ctx.emit(EventType.LOG, level="warn", message=note) min_score = float(ctx.cfg("min_score", 0.0) or 0.0) hits = [h for h in hits if h["score"] >= min_score] @@ -85,15 +90,44 @@ async def run_retrieve(state: GraphState, ctx: NodeContext) -> dict[str, Any]: f"[{i + 1}] {h['title']}（片段 {h['ordinal']}）\n{h['content']}" for i, h in enumerate(hits) ) + # 命中落工件库。SQL 取数、工具调用、agent 工具调用都有快照，检索一直没有， + # 于是结论只要来自知识库，"这个数是哪来的"这条链就断在这儿 + from app.core.artifact_store import put_json + from app.memory.embeddings import embedder_id + + snapshot: str | None = None + try: + snapshot = await put_json( + { + "query": query, "collection": collection, + "alpha": alpha, "min_score": min_score, + "embedder": embedder_id(), "degraded": degraded, + "hits": hits, + }, + kind="retrieval_snapshot", + run_id=ctx.run.run_id, node_id=ctx.node.id, + ) + except Exception: # noqa: BLE001 - 存不下不影响这次检索本身 + snapshot = None + + ctx.emit( + EventType.RETRIEVE_END, + collection=collection, + query=query[:200], + count=len(hits), + top_score=hits[0]["score"] if hits else 0, + degraded=bool(degraded), + artifact=snapshot, + ) + result = { "hits": hits, "count": len(hits), "text": context_text, "query": query, "collection": collection, + "artifact": snapshot, } - ctx.emit(EventType.LOG, level="info", - message=f"检索 {collection}：命中 {len(hits)} 条，最高分 {hits[0]['score'] if hits else 0}") updates: dict[str, Any] = {"nodes": {ctx.node.id: result}} var_name = ctx.cfg("assign_to", "") diff --git a/backend/app/main.py b/backend/app/main.py index c2df608e..b5a8e3b3 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -30,6 +30,19 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: settings.ensure_dirs() await init_db() await seed_defaults() + + # 把存的 embedding 配置读进来。不读的话重启后会悄悄退回本地哈希向量， + # 而库里的向量是上次那个模型建的——检索会整体退回关键词， + # 用户只看到"重启之后搜得不准了" + from app.db.base import SessionLocal + from app.memory.embeddings import load_setting + + async with SessionLocal() as _session: + try: + await load_setting(_session) + except Exception as e: # noqa: BLE001 - key 没了不该拦住整个服务起不来 + logger.warning("embedding 配置没能生效，暂用本地向量：%s", e) + await run_manager.setup() from app.sandbox.manager import sandbox_manager as _sbm diff --git a/backend/app/memory/embeddings.py b/backend/app/memory/embeddings.py index 380d5d9f..76ab7119 100644 --- a/backend/app/memory/embeddings.py +++ b/backend/app/memory/embeddings.py @@ -61,7 +61,9 @@ class OpenAIEmbedder: base_url: str | None = None) -> None: from langchain_openai import OpenAIEmbeddings - self.dim = 1536 + self.model = model + # text-embedding-3-large 是 3072，别写死 + self.dim = 3072 if "3-large" in model else 1536 self._impl = OpenAIEmbeddings( model=model, api_key=api_key or os.environ.get("OPENAI_API_KEY"), @@ -79,17 +81,92 @@ class OpenAIEmbedder: _local = LocalEmbedder() +#: 设置表里存 embedding 配置的键。和 Copilot 模型同一套路子 +EMBEDDING_SETTING_KEY = "embedding" -def get_embedder() -> LocalEmbedder | OpenAIEmbedder: - """默认走本地；显式打开 AGENTLAB_USE_OPENAI_EMBEDDINGS 且有 key 时才用远端。""" +#: 进程内缓存。get_embedder 在检索热路径上被反复调用，不该每次都去读库 +_cached: tuple[str, Any] | None = None + + +class EmbedderUnavailable(RuntimeError): + """配了远端 embedder 但建不起来。message 直接给用户看。""" + + +def _make(kind: str, model: str, *, strict: bool) -> "LocalEmbedder | OpenAIEmbedder": + if kind != "openai": + return _local + try: + return OpenAIEmbedder(model=model or "text-embedding-3-small") + except Exception as e: # noqa: BLE001 + if strict: + # 人明确选了远端就不能偷偷退回本地：他会以为自己在用语义检索， + # 而实际拿到的是哈希词袋——搜不出同义表达时完全无从怀疑到这里 + raise EmbedderUnavailable( + f"用不了 embedding 模型 {model or 'text-embedding-3-small'}：" + f"{type(e).__name__}: {e}。检查 OPENAI_API_KEY 是否配了。" + ) from e + return _local + + +def configure(kind: str, model: str = "") -> None: + """切换当前 embedder。设置页保存后调用，也用于测试。建不起来会抛。""" + global _cached + _cached = (f"{kind}:{model}", _make(kind, model, strict=True)) + + +def get_embedder() -> "LocalEmbedder | OpenAIEmbedder": + """当前 embedder。 + + 优先级：进程内已配置 > 环境变量兜底 > 本地。设置页写库后会调 configure， + 所以这里不再每次读库——检索热路径上一次查询要走几百次。 + + 以前这里只认 AGENTLAB_USE_OPENAI_EMBEDDINGS 一个环境变量：没有界面、 + 没人会发现，于是所有人都在用那个没有语义能力的哈希向量，而 alpha 默认 + 0.5 还给了它一半权重。 + """ + if _cached is not None: + return _cached[1] if os.environ.get("AGENTLAB_USE_OPENAI_EMBEDDINGS") and os.environ.get("OPENAI_API_KEY"): - try: - return OpenAIEmbedder() - except Exception: # noqa: BLE001 - return _local + return _make("openai", "", strict=False) return _local +def embedder_id() -> str: + """当前 embedder 的身份。存进 chunk 那一列，检索时拿它判断存量向量还能不能用。""" + emb = get_embedder() + name = getattr(emb, "name", "unknown") + model = getattr(emb, "model", "") + return f"{name}:{model}" if model else name + + +def embedder_dim() -> int: + return int(getattr(get_embedder(), "dim", _DIM)) + + +def usable(embed_model: str | None, embed_dim: int | None) -> bool: + """这条存量向量还能不能和当前查询比对。 + + 维度对不上 cosine 会直接抛 ValueError（shapes (1536,) and (512,) not + aligned），所以这不是"质量差一点"，是会 500。模型名也要比：同维度不同 + 模型的向量空间没有可比性，算出来的相似度是随机数。 + + 存量数据这两列是空的（迁移回填的默认值），一律当成不可用——保守， + 但总比拿旧模型的向量冒充当前空间里的坐标强。 + """ + return bool(embed_model) and embed_model == embedder_id() and embed_dim == embedder_dim() + + +async def load_setting(session: Any) -> None: + """从库里读 embedding 配置并生效。应用启动时调一次。""" + from app.db.models import Setting + + row = await session.get(Setting, EMBEDDING_SETTING_KEY) + saved = (row.value if row else {}) or {} + kind = saved.get("kind") + if kind: + configure(kind, saved.get("model", "")) + + async def embed_text(text: str) -> np.ndarray: emb = get_embedder() if isinstance(emb, OpenAIEmbedder): diff --git a/backend/app/memory/kb.py b/backend/app/memory/kb.py index 13c61243..be63d8d7 100644 --- a/backend/app/memory/kb.py +++ b/backend/app/memory/kb.py @@ -7,7 +7,9 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Chunk, Document -from app.memory.embeddings import embed_text, embed_texts, from_blob, hybrid_rank, to_blob +from app.memory.embeddings import ( + embed_text, embed_texts, embedder_dim, embedder_id, from_blob, hybrid_rank, to_blob, usable, +) _TARGET = 800 _OVERLAP = 120 @@ -97,6 +99,8 @@ async def ingest_document( ordinal=i, content=piece, embedding=to_blob(vec), + embed_model=embedder_id(), + embed_dim=embedder_dim(), meta={"title": doc.title}, ) ) @@ -113,7 +117,17 @@ async def search( query: str, limit: int = 5, alpha: float = 0.5, + on_degrade: Any = None, ) -> list[dict[str, Any]]: + """混合检索。 + + 存量向量和当前 embedder 对不上时**整个查询退回纯关键词**，并通过 + on_degrade 把原因交出去。不把对不上的那些当 0 分：那会让它们静默沉底， + 用户看到的是结果莫名其妙变差，而不是"这里有一批索引该重建了"。 + + 在此之前这里连维度都不查，换个 embedder 直接 ValueError + （shapes (1536,) and (512,) not aligned），整个检索 500。 + """ stmt = select(Chunk) if collection: stmt = stmt.where(Chunk.collection == collection) @@ -121,11 +135,26 @@ async def search( if not rows: return [] - query_vec = await embed_text(query) + stale = [r for r in rows if not usable(r.embed_model, r.embed_dim)] + if stale: + if on_degrade: + names = sorted({r.embed_model or "未知模型" for r in stale}) + on_degrade( + f"{collection or '全部集合'}里有 {len(stale)}/{len(rows)} 条向量由" + f"「{'、'.join(names)}」建立，和当前的「{embedder_id()}」对不上，" + f"本次已退回纯关键词检索。重建索引后恢复。" + ) + alpha = 0.0 # 纯关键词 + vectors: list[Any] = [None] * len(rows) + query_vec = await embed_text(query) + else: + query_vec = await embed_text(query) + vectors = [from_blob(r.embedding, embedder_dim()) for r in rows] + ranked = hybrid_rank( query, [r.content for r in rows], - [from_blob(r.embedding) for r in rows], + vectors, query_vec, alpha=alpha, ) @@ -146,6 +175,41 @@ async def search( return out +async def reindex(session: AsyncSession, collection: str | None = None) -> dict[str, Any]: + """用当前 embedder 重算向量。 + + 换了 embedding 模型之后唯一的恢复手段。按批做：一次几千条 chunk 全塞给 + 远端 embedding 接口，要么超时要么被限流，而中途失败又没有断点。 + """ + stmt = select(Chunk) + if collection: + stmt = stmt.where(Chunk.collection == collection) + rows = list((await session.execute(stmt)).scalars()) + if not rows: + return {"collection": collection, "reindexed": 0, "embedder": embedder_id()} + + model_id, dim = embedder_id(), embedder_dim() + batch = 64 + for i in range(0, len(rows), batch): + part = rows[i:i + batch] + vectors = await embed_texts([r.content for r in part]) + for row, vec in zip(part, vectors): + row.embedding = to_blob(vec) + row.embed_model = model_id + row.embed_dim = dim + await session.commit() + return {"collection": collection, "reindexed": len(rows), "embedder": model_id} + + +async def stale_count(session: AsyncSession, collection: str | None = None) -> int: + """有多少条向量和当前 embedder 对不上。知识库页据此提示要不要重建。""" + stmt = select(Chunk) + if collection: + stmt = stmt.where(Chunk.collection == collection) + rows = list((await session.execute(stmt)).scalars()) + return sum(1 for r in rows if not usable(r.embed_model, r.embed_dim)) + + async def delete_document(session: AsyncSession, doc_id: str) -> bool: doc = await session.get(Document, doc_id) if not doc: diff --git a/backend/app/memory/store.py b/backend/app/memory/store.py index 6f5e8fe6..b286be6c 100644 --- a/backend/app/memory/store.py +++ b/backend/app/memory/store.py @@ -7,7 +7,9 @@ from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import MemoryItem -from app.memory.embeddings import cosine, embed_text, from_blob, hybrid_rank, to_blob +from app.memory.embeddings import ( + cosine, embed_text, embedder_dim, embedder_id, from_blob, hybrid_rank, to_blob, usable, +) # 判重阈值：余弦相似度高于它才算同一件事。比排名分数严格得多， # 因为这里比的是真实语义距离，不是"谁排第一"。 @@ -44,7 +46,12 @@ async def remember( candidates = list((await session.execute(stmt)).scalars()) best, best_score = None, 0.0 for row in candidates: - other = from_blob(row.embedding) + # 对不上当前 embedder 的直接跳过：维度不同 cosine 会抛，维度碰巧 + # 相同但模型不同则更糟——两个空间之间的相似度是个随机数， + # 拿它判重会把不相关的旧记忆认成"重复"然后丢掉新内容 + if not usable(row.embed_model, row.embed_dim): + continue + other = from_blob(row.embedding, embedder_dim()) if other is None: continue score = cosine(vec, other) @@ -64,6 +71,8 @@ async def remember( meta=meta or {}, importance=importance, embedding=to_blob(vec), + embed_model=embedder_id(), + embed_dim=embedder_dim(), ) session.add(item) await session.commit() @@ -89,11 +98,19 @@ async def recall( return [] query_vec = await embed_text(query) + # 和 kb.search 同一个道理：有对不上的就整体退回关键词，而不是让它们静默沉底 + if any(not usable(r.embed_model, r.embed_dim) for r in rows): + vectors: list[Any] = [None] * len(rows) + alpha = 0.0 + else: + vectors = [from_blob(r.embedding, embedder_dim()) for r in rows] + alpha = 0.5 ranked = hybrid_rank( query, [r.content for r in rows], - [from_blob(r.embedding) for r in rows], + vectors, query_vec, + alpha=alpha, ) results: list[dict[str, Any]] = [] diff --git a/backend/tests/test_kb.py b/backend/tests/test_kb.py new file mode 100644 index 00000000..5eafd6ab --- /dev/null +++ b/backend/tests/test_kb.py @@ -0,0 +1,174 @@ +"""知识库：切块、混合检索、换 embedder 之后还能不能用。 + +这三个模块（kb.py / embeddings.py / memory/store.py，共 526 行检索逻辑） +在此之前**一条测试都没有**。于是有两件事一直没人发现： + +- 换 embedding 模型会让检索整体 500：库里是 512 维的本地向量、查询是 1536 维 + 的远端向量，cosine 直接 `shapes (1536,) and (512,) not aligned` +- 存量向量属于哪个模型，代码里没有任何地方记着，所以也无从降级 + +现在这两件事都有守的了。 +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from app.db.base import SessionLocal +from app.db.models import Chunk +from app.memory import embeddings as emb +from app.memory import kb +from app.memory.embeddings import hybrid_rank, to_blob + + +@pytest.fixture(autouse=True) +def _local_embedder(): + """每个用例都从本地 embedder 起步，别让上一个用例的 configure 漏过来。""" + emb.configure("local") + yield + emb.configure("local") + + +# -------------------------------------------------------------------------- +# 切块 +# -------------------------------------------------------------------------- + + +def test_short_paragraphs_stay_whole() -> None: + blocks = kb.chunk_text("第一段。\n\n第二段。\n\n第三段。") + assert blocks and all("第" in b for b in blocks) + + +def test_a_long_paragraph_is_cut_on_sentence_boundaries() -> None: + """命中之后读起来要是完整的句子，不能拦腰截断。""" + text = "。".join(f"这是第{i}句话，讲了一点内容" for i in range(120)) + "。" + blocks = kb.chunk_text(text, target=200, overlap=0) + assert len(blocks) > 1 + # 除了最后一块，其余都该收在句号上 + assert all(b.rstrip().endswith("。") for b in blocks[:-1]) + + +def test_a_sentence_longer_than_the_target_is_hard_cut() -> None: + """没有标点可切时也不能无限膨胀，硬切是最后的兜底。""" + blocks = kb.chunk_text("啊" * 2000, target=300, overlap=0) + assert len(blocks) > 1 and max(len(b) for b in blocks) <= 300 + + +def test_empty_input_yields_nothing() -> None: + assert kb.chunk_text("") == [] and kb.chunk_text(" \n\n ") == [] + + +# -------------------------------------------------------------------------- +# hybrid_rank 的归一化边界 +# -------------------------------------------------------------------------- + + +def test_a_single_candidate_is_not_scored_zero() -> None: + """min-max 归一化在只有一条候选时 hi==lo。 + + 返回 0 的话它的合并分恒为 0，会被 min_score 过滤掉——于是新建集合写入 + 第一条之后永远搜不到，而"先写一条、下一轮再检索"正是 agent 最常见的用法。 + embeddings.py 里为此写了一段注释，但一直没有测试守着。 + """ + vec = np.ones(4, dtype=np.float32) + ranked = hybrid_rank("查询", ["只有这一条"], [vec], vec) + assert len(ranked) == 1 and ranked[0][1] > 0 + + +# -------------------------------------------------------------------------- +# 换 embedder：以前这里直接 500 +# -------------------------------------------------------------------------- + + +async def _seed(collection: str, texts: list[str]) -> None: + async with SessionLocal() as session: + for t in texts: + await kb.ingest_document( + session, collection=collection, title=t[:10], content=t) + + +def test_stale_vectors_are_recognised() -> None: + """存量数据那两列是空的，一律当成对不上——保守但不会拿旧坐标冒充新空间。""" + assert emb.usable("", 0) is False + assert emb.usable("openai:text-embedding-3-small", 1536) is False + assert emb.usable(emb.embedder_id(), emb.embedder_dim()) is True + + +async def test_ingest_records_who_built_the_vector() -> None: + await _seed("kbtest_mark", ["平台管理员一共有六个人。"]) + async with SessionLocal() as session: + from sqlalchemy import select + row = (await session.execute( + select(Chunk).where(Chunk.collection == "kbtest_mark"))).scalars().first() + assert row.embed_model == emb.embedder_id() + assert row.embed_dim == emb.embedder_dim() + + +async def test_search_degrades_instead_of_crashing_after_a_model_switch() -> None: + """整件事的核心。 + + 以前：cosine(1536 维查询, 512 维存量) → ValueError → 检索 500。 + 现在：整体退回纯关键词，并把原因说出来。 + """ + await _seed("kbtest_switch", ["平台管理员一共有六个人。", "商家账户共五个。"]) + + # 把库里的向量伪装成别的模型建的（维度也不同） + async with SessionLocal() as session: + from sqlalchemy import select + rows = list((await session.execute( + select(Chunk).where(Chunk.collection == "kbtest_switch"))).scalars()) + for r in rows: + r.embedding = to_blob(np.ones(1536, dtype=np.float32)) + r.embed_model = "openai:text-embedding-3-small" + r.embed_dim = 1536 + await session.commit() + + notes: list[str] = [] + async with SessionLocal() as session: + hits = await kb.search(session, collection="kbtest_switch", query="管理员", + limit=5, on_degrade=notes.append) + + assert hits, "退回关键词也要能搜到东西，而不是空手而归" + assert notes, "少了一半能力却不吭声，用户只会觉得最近搜得不准" + assert "重建索引" in notes[0] and "2/2" in notes[0] + + +async def test_reindex_restores_hybrid_search() -> None: + await _seed("kbtest_reindex", ["平台管理员一共有六个人。"]) + async with SessionLocal() as session: + from sqlalchemy import select + row = (await session.execute( + select(Chunk).where(Chunk.collection == "kbtest_reindex"))).scalars().first() + row.embed_model, row.embed_dim = "别的模型", 99 + await session.commit() + + async with SessionLocal() as session: + assert await kb.stale_count(session, "kbtest_reindex") == 1 + out = await kb.reindex(session, "kbtest_reindex") + assert out["reindexed"] == 1 + assert await kb.stale_count(session, "kbtest_reindex") == 0 + + notes: list[str] = [] + async with SessionLocal() as session: + hits = await kb.search(session, collection="kbtest_reindex", query="管理员", + on_degrade=notes.append) + assert hits and not notes, "重建之后不该再降级" + + +# -------------------------------------------------------------------------- +# 配置 embedder +# -------------------------------------------------------------------------- + + +def test_choosing_a_remote_embedder_that_cannot_be_built_says_so(monkeypatch) -> None: + """选了远端却静默退回本地哈希向量，用户会以为自己在用语义检索。""" + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(emb.EmbedderUnavailable) as e: + emb.configure("openai", "text-embedding-3-small") + assert "OPENAI_API_KEY" in str(e.value) + + +def test_the_local_embedder_is_always_available() -> None: + emb.configure("local") + assert emb.embedder_id() == "local-hashing" and emb.embedder_dim() == 512 diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 6b86f49a..3eba1bbf 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -191,6 +191,15 @@ export const api = { search: (q: string, collection?: string, alpha = 0.5) => get(`/kb/search?q=${encodeURIComponent(q)}&alpha=${alpha}${collection ? `&collection=${collection}` : ''}`), remove: (id: string) => del(`/kb/documents/${id}`), + embedding: (collection?: string) => get<{ + embedder: string; dim: number; configured: boolean + kind: string; model: string; stale_chunks: number + }>(`/kb/embedding${collection ? `?collection=${collection}` : ''}`), + setEmbedding: (body: { kind: string; model?: string }) => + put<{ embedder: string; dim: number; stale_chunks: number }>('/kb/embedding', body), + reindex: (collection?: string) => + post<{ reindexed: number; embedder: string }>( + `/kb/reindex${collection ? `?collection=${collection}` : ''}`), }, skills: { list: () => get('/skills'), diff --git a/frontend/src/pages/KnowledgePage.tsx b/frontend/src/pages/KnowledgePage.tsx index bbbd14a7..07e4fa7a 100644 --- a/frontend/src/pages/KnowledgePage.tsx +++ b/frontend/src/pages/KnowledgePage.tsx @@ -29,6 +29,94 @@ export function KnowledgePage() { // ------------------------------------------------------------------------- +/** + * 当前用哪个 embedder，以及有多少条向量已经对不上。 + * + * 在此之前切换 embedding 模型的唯一开关是 AGENTLAB_USE_OPENAI_EMBEDDINGS—— + * 一个没有界面、没人会发现的环境变量。于是所有人都在用那个没有语义能力的 + * 哈希向量（同义改写一条都召不回），而 alpha 默认还给了它一半权重。 + * + * 换模型之后存量向量就和查询对不上了，检索会整体退回纯关键词。这一条必须 + * 显眼：少了一半能力却不说，用户只会觉得"最近搜得不准"。 + */ +function EmbedderBar({ collection, onChanged }: { + collection: string + onChanged: () => void | Promise +}) { + const toast = useToast() + const [info, setInfo] = useState(null) + const [busy, setBusy] = useState(false) + + const refresh = async () => setInfo(await api.kb.embedding(collection).catch(() => null)) + useEffect(() => { void refresh() }, [collection]) + + if (!info) return null + const stale = info.stale_chunks ?? 0 + + const pick = async (kind: string, model: string) => { + setBusy(true) + try { + const out = await api.kb.setEmbedding({ kind, model }) + toast(`已切到 ${out.embedder}`, 'ok') + await refresh() + } catch (e: any) { + // 400 里是"缺 OPENAI_API_KEY"这类能照着做的原因，原样交给用户 + toast(e?.message ?? '切换失败', 'error') + } finally { setBusy(false) } + } + + const rebuild = async () => { + setBusy(true) + try { + const out = await api.kb.reindex(collection) + toast(`已用 ${out.embedder} 重建 ${out.reindexed} 段`, 'ok') + await refresh() + await onChanged() + } catch (e: any) { + toast(e?.message ?? '重建失败', 'error') + } finally { setBusy(false) } + } + + return ( +
+
+ 向量模型 + + {info.embedder} · {info.dim} 维 + + {stale > 0 && ( + + )} +
+ {stale > 0 && ( +

+ 有 {stale} 段的向量是用别的模型建的，和当前模型对不上——检索这些内容时 + 会退回纯关键词（搜不出同义表达）。重建索引后恢复。 +

+ )} + {info.kind !== 'openai' && stale === 0 && ( +

+ 本地向量不联网、零配置，但它是词频哈希、没有语义泛化—— + 「管理员」搜不到「平台角色」。认真用知识库的话换成上面的真模型。 +

+ )} +
+ ) +} + +// ------------------------------------------------------------------------- + function KbTab() { const toast = useToast() const refreshCatalog = useCatalog((s) => s.refresh) @@ -95,6 +183,8 @@ function KbTab() { ))} + + {/* 检索调试：能直观看到混合检索里语义和关键词各占多少 */}
diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index de2a792e..8238cef0 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -712,6 +712,30 @@ export function decodeRun(events: RunEvent[]): Step[] { break } + case 'retrieve.end': { + // 知识检索。以前只发一条 info 日志，于是"这句结论依据的是哪份文档的 + // 哪一段"在轨迹里查不到——而 SQL 取数那条路早就能下钻到工件 + const count = num(d.count) ?? 0 + const collection = String(d.collection ?? '') + const top = num(d.top_score) + push({ + id: `rt-${seq}`, seq, kind: 'schema', nodeId, + status: count ? 'done' : 'failed', + level: d.degraded ? 'warn' : count ? undefined : 'warn', + title: count + ? `在「${collection}」里检索到 ${count} 段` + : `在「${collection}」里没检索到内容`, + detail: String(d.query ?? ''), + meta: [ + top != null ? `最高分 ${top}` : '', + // 退回关键词是"少了一半能力"，不标出来用户只会觉得最近搜得不准 + d.degraded ? '已退回关键词' : '', + ].filter(Boolean).join(' · ') || undefined, + artifact: d.artifact, + }, nodeId) + break + } + default: // 新增事件类型时不要静默吞掉——宁可显示一条原始的，也比让人觉得 // "什么都没发生"好。加了新事件记得回来补一条映射。 diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index 21411b6d..e8b392d0 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -42,6 +42,35 @@ console.log('=== 数据库查询 ===') steps.some((s) => s.kind === 'node' && s.children?.some((c) => c.kind === 'query'))) } +console.log('\n=== 知识检索 ===') +{ + // 以前检索只发一条 info 日志，轨迹里查不到"这句结论依据的是哪一段"。 + // 现在它是一条正经事件，而且带工件 id 可以下钻。 + const ev = (data) => [{ seq: 1, type: 'retrieve.end', node_id: 'kb', data }] + + const hit = mod.decodeRun(ev({ + collection: '手册', query: '管理员怎么定义', count: 3, + top_score: 0.82, artifact: 'abc123', degraded: false, + })) + const s1 = flatten(hit).find((x) => x.kind === 'schema') + check('检索步骤说人话', !!s1 && s1.title.includes('手册') && s1.title.includes('3 段'), s1?.title) + check('问的是什么留在详情里', s1?.detail === '管理员怎么定义') + check('带得出工件 id，能下钻', s1?.artifact === 'abc123') + check('最高分露出来', !!s1?.meta?.includes('0.82')) + + const none = flatten(mod.decodeRun(ev({ collection: '手册', count: 0 }))) + .find((x) => x.kind === 'schema') + check('没检索到要标成失败，不是静悄悄地过去', none?.status === 'failed') + + const bad = flatten(mod.decodeRun(ev({ collection: '手册', count: 2, degraded: true }))) + .find((x) => x.kind === 'schema') + check('退回关键词这件事标出来了', bad?.level === 'warn' && !!bad?.meta?.includes('关键词')) + + // 兜底那条也要还在：新事件没补映射时宁可显示原始的 + const unknown = mod.decodeRun([{ seq: 1, type: 'brand.new', node_id: 'x', data: { a: 1 } }]) + check('未知事件不被静默吞掉', flatten(unknown).some((x) => x.title === 'brand.new')) +} + console.log('\n=== 被截断的结果集 ===') { // 后端按字符数硬切预览，切点落在 JSON 中间是常态。严格 JSON.parse 一律失败，