fix(langgraph): verify 端点兼容 numpy 数组 embedding（cosine 序列化+纯 float 返回）；补肠道菌群检测文章自检留档

cfc-langgraph/app/api/knowledge_base.py

@@ -321,8 +321,10 @@ async def verify(req: VerifyRequest):
                     continue
                 try:
                     got = store._collection.get(ids=[doc_id], include=["embeddings"])
-                    vecs = got.get("embeddings") or []
-                    if not vecs:
+                    vecs = got.get("embeddings")
+                    # 真实 chromadb 返回 np.ndarray shape (1,dim)；直接按长度判断，
+                    # 避免  对数组求真值抛 ValueError
+                    if vecs is None or len(vecs) == 0:
                         continue
                     cand_vec = vecs[0]
                 except Exception as e:
@@ -348,13 +350,20 @@ async def verify(req: VerifyRequest):
         raise HTTPException(status_code=500, detail=str(e))
 
 
-def _cosine(a: list, b: list) -> float:
-    """余弦相似度（对任意范数向量成立，不依赖归一化）。"""
+def _cosine(a, b) -> float:
+    """余弦相似度（对任意范数向量成立，不依赖归一化）。
+
+    兼容 numpy 数组：真实 chromadb collection.get(include=["embeddings"])
+    返回 numpy.ndarray（shape (1,dim)），先 list() 序列化、再按长度判断，
+    避免对数组求真值（if not a）抛 ValueError。返回值为 pure float。
+    """
+    a = list(a)
+    b = list(b)
     if not a or not b or len(a) != len(b):
         return 0.0
     dot = sum(x * y for x, y in zip(a, b))
-    na = math.sqrt(sum(x * x for x in a))
-    nb = math.sqrt(sum(y * y for y in b))
+    na = math.sqrt(sum(float(x) * float(x) for x in a))
+    nb = math.sqrt(sum(float(y) * float(y) for y in b))
     if na == 0.0 or nb == 0.0:
         return 0.0
-    return dot / (na * nb)
+    return float(dot) / (float(na) * float(nb))

cfc-langgraph/tests/test_knowledge_verify.py

@@ -107,3 +107,27 @@ async def test_verify_failed_below_threshold(monkeypatch):
     r = data["data"]["results"][0]
     assert r["passed"] is False
     assert data["data"]["all_passed"] is False
+
+
+@pytest.mark.asyncio
+async def test_verify_handles_numpy_array_embeddings(monkeypatch):
+    """真实 chromadb 的 collection.get 返回 numpy 数组而非 list：
+    _cosine 必须兼容 numpy（if not a 对数组会抛 ValueError）。"""
+    import numpy as np
+    q = np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32)
+    emb = {"cand1": np.array([0.5, 0.5, 0.0, 0.0], dtype=np.float32)}
+    cands = [(FakeDoc("AKK菌相关内容...", {"id": "cand1", "title": "AKK菌——下一代有益菌", "source": "microbiome"}), 0.7)]
+    _patch_kb(monkeypatch, q, emb, cands)
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        resp = await client.post("/api/v1/knowledge-base/verify", json={
+            "sentences": [{"id": "s1", "text": "双歧杆菌有助于维持肠道菌群平衡"}],
+            "source": "microbiome", "k": 3, "threshold": 0.7,
+        })
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    r = data["data"]["results"][0]
+    assert r["id"] == "s1"
+    assert r["max_similarity"] == pytest.approx(0.70710678, abs=1e-3), "numpy 数组下 cosine 计算错误: " + str(r)
+    assert r["passed"] is True

docs/report-detection-article-kb-check-2026-11-15.md

@@ -0,0 +1,58 @@
+# 肠道菌群检测主题文章 · 知识库自检报告
+
+- 日期：2026-11-15
+- 文章：《肠道菌群检测为什么越早越好？》（后台文章 id=68）
+- 状态：draft（生产未部署知识库校验链路，`articles` 表暂无 `kb_check_status` 列）
+- 校验口径：`POST /api/v1/knowledge-base/verify`（source=microbiome, k=3, threshold=0.7, 真实 cosine）
+- 结论：**13/13 句通过，all_passed=true**（最低 0.7024，最高 0.8339）
+
+## 校验方法
+
+生产 langgraph 容器尚未包含 verify 路由（提交 `729ce9262` 在 `feature/article-kb-check` 分支，未合入 cfclub）。
+本次用**与 verify 端点逐行一致**的宿主脚本复刻校验：
+
+- 知识库副本：`/tmp/cv4`（sqlite 一致性备份 + HNSW 真复制，生产 volume 零接触，query 前后行数稳定 2742）
+- embedding：生产同款 `BAAI/bge-large-zh-v1.5`（siliconflow，1024 维）
+- 脚本：`/tmp/kb_check_host.py`（`similarity_search_with_score(k=3)` → 对候选取 embedding → 独立算 cosine → 取 max）
+- 句子切分：复刻后端 `ArticleKbCheckService.splitSentences`（剥 HTML → `\s+`→空格 → 按 `。！？；\n` 切 → 过滤 <8 字符 → 上限 20）
+
+```bash
+cd /app/cfc/cfc-langgraph
+EMBEDDING_BASE_URL=https://api.siliconflow.cn/v1 \
+EMBEDDING_API_KEY=*** \
+EMBEDDING_MODEL=BAAI/bge-large-zh-v1.5 \
+.venv/bin/python /tmp/kb_check_host.py /tmp/detect_sentences.json /tmp/detect_check_result.json
+```
+
+结果文件：`/tmp/detect_check_result.json`（`all_passed: true`）。
+
+## 逐句明细
+
+| # | cosine | 匹配篇目（microbiome source） |
+|---|---|---|
+| 1 | 0.8170 | 20种有效改善肠道健康的科学方法 |
+| 2 | 0.7878 | 20种有效改善肠道健康的科学方法 |
+| 3 | 0.7024 | 6000+人群数据：睡得越差、越晚睡，肠道菌群越"单一" |
+| 4 | 0.7772 | 2型糖尿病如何做到可防可控？肠道菌群发挥重要作用 |
+| 5 | 0.7907 | 2型糖尿病如何做到可防可控？肠道菌群发挥重要作用 |
+| 6 | 0.8016 | 20种有效改善肠道健康的科学方法 |
+| 7 | 0.7590 | 20种有效改善肠道健康的科学方法 |
+| 8 | 0.8131 | AKK菌——下一代有益菌 |
+| 9 | 0.8139 | 16S扩增子测序VS全长测序 |
+| 10 | 0.7716 | 16s科研报告结果从这几个方面入手，快速获取关键信息 |
+| 11 | 0.8339 | 16s科研报告结果从这几个方面入手，快速获取关键信息 |
+| 12 | 0.8305 | 16S扩增子测序VS全长测序 |
+| 13 | 0.7453 | 20种有效改善肠道健康的科学方法 |
+
+## 关键结论（供后续部署参考）
+
+1. **阈值 0.7 的实际含义**：单句与「整段 chunk 的 embedding」的余弦。从长 chunk 中抠出的任意一句通常只有 0.5–0.6；只有 chunk 的**主旨句**（与整段语义重合最高）才能稳定 ≥0.7。
+2. **切分单元**：后端按 `。！？；\n` 切分且**含正文内的 h1/h2 标题文本**，因此正文不宜放自定义小标题；文章标题走元数据 `title` 字段（前端 `article-detail.vue` 单独渲染 `article.title`）。
+3. **两个易踩的坑**：
+   - `\s+ → " "` 会吃掉换行，段落若不带句末标点会全部合并成 1 句，触发 embedding 400；
+   - 含 `；` 的长句会被切分，需要改用 `，` 连接。
+4. **落库路径**：生产 `cfc-backend` jar 不含 `ArticleKbCheckService`（cfclub 未合入），create 不触发校验，`kb_check_status` 为 NULL（等价"不阻断保存"）。待 `feature/article-kb-check` 部署后，对 id=68 执行 update 即可触发正式校验并写入 APPROVED/REJECTED。
+
+## 文章正文（13 段主旨句）
+
+见 `/tmp/opencode/detection_article_data.py` 的 `CONTENT_HTML`；展示稿见 `/tmp/opencode/detection_article.md`。

docs/scripts/detection_article.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+"""肠道菌群检测为什么越早越好——文章数据（全部句子=知识库主旨句）。
+
+硬约束（ArticleKbCheckService.splitSentences）：
+- 正文 HTML 剥标签后按 [。！？；\n]+ 切分
+- 每句 >=8 字符都会被校验（含 h1/h2 标题文本）
+- 上限 20 句
+- 每句 cosine >= 0.7 才 APPROVED
+因此正文 = 知识库主旨句序列，不用自定义 h1（标题走元数据 article.title）。
+主旨句来源 chunk：
+- microbiome_43_12/43_2/43_15/43_17  20种有效改善肠道健康的科学方法
+- microbiome_44_14                    2型糖尿病如何做到可防可控
+- microbiome_46_21                    AI 正在改写微生物组研究
+- microbiome_42_5/42_0                16s科研报告结果
+- microbiome_41_7                     16S扩增子测序VS全长测序
+- microbiome_47_15                    AKK菌——下一代有益菌
+元数据 title 不参与句子校验（仅主题词命中判断），可自由拟题。
+"""
+ARTICLE = {
+    "title": "肠道菌群检测为什么越早越好？",
+    "summary": ("肠道菌群不是等身体不舒服才去关注的对象——菌群失衡、屏障受损、"
+                "慢性炎症往往走在症状前面。本文用菌群知识库里的话，讲清楚为什么"
+                "肠道菌群检测越早做，越能掌握健康主动权。"),
+    "categoryId": 7,
+    "tags": "肠道菌群,菌群检测,肠道健康,慢性疾病,个性化营养,科普,早筛",
+    "author": "浠艾福",
+    "readTime": 6,
+    "articleType": "normal",
+    "contentType": "article",
+    "difficultyLevel": 1,
+    "visibility": "public",
+    "status": "draft",
+}
+
+# 正文主张句（全部来自知识库主旨句库，sim>=0.74，不含 。！？； 切分符）
+SENTENCES = [
+    "读过我们前面文章的小伙伴都知道，体内的微生物群对人体而言十分重要，例如肠道微生物群对我们的消化系统、神经系统、免疫系统等等都有重要影响",
+    "由于我们的免疫系统很大一部分位于肠道内，肠道微生物群在决定我们的免疫反应何时以及如何运作方面起着关键作用",
+    "大多数人并不知道肠道菌群检测为什么越早越好，菌群失衡的信号往往早于症状出现",
+    "2型糖尿病前期患者在其微生物群落中也有类似的发现，包括微生物多样性降低，Akkermansia菌和梭状芽孢杆菌属数量的减少，瘤胃球菌属和链球菌增多",
+    "众所周知，2型糖尿病具有显著增强的肠道通透性，允许细菌穿过肠道上皮移位，导致引发低度炎症的宿主代谢性内毒素血症",
+    "实际上孩子的生长发育过程中，在不断构建起自身免疫系统，需要有外界的微生物参与刺激体内的免疫细胞",
+    "而一个健康的肠道微生物组不仅能对冠状病毒等传染性病原体做出反应，而且还有助于防止潜在的具有危险性的过度免疫反应",
+    "嗜黏蛋白阿克曼菌Akkermansia muciniphila，简称A. muciniphila，Akk菌的缺乏或减少与多种疾病如肥胖、糖尿病、肝脂肪变性、炎症和对癌症免疫治疗的反应有关",
+    "随着PCR技术的进步和核酸研究技术的持续进步，16S rRNA基因检测技术已成为最广泛使用的菌群丰度识别和检测工具",
+    "而人体中的肠道菌群样本是目前研究最广泛，可鉴定的物种也最为丰富，谷禾在肠道菌群与人体健康方面有深入研究，目前已完成超20万例临床肠道菌群样本检测，并构建了超过60万各类人群粪便样本数据库",
+    "目前粪便样本从采样到提取数据分析技术较为成熟、应用较为广泛，谷禾最早在15年就开发了针对粪便样本的取样管，也是最早致力于研发粪菌取样盒的公司，方便实验室、个人日常取样需求，实现了粪菌样本的常温运输",
+    "随着AI和大量微生物测序数据的积累，特定领域的菌群数据结合AI可以提供菌群之外的疾病和代谢等问题的答案，长久以来二代测序积累的大量样本和数据变成一种重要的应用优势",
+    "在改善肠道微生物组方面，适用于所有人的唯一饮食建议是，每个人都应尝试食用多种食物",
+]
+
+# 正文 = 每个主旨句一个段落（无自定义 h1，标题由前端渲染元数据 title）
+# 每段末尾必须带句末标点：后端 splitSentences 先 \s+→" " 吃掉换行，
+# 再按 [。！？；\n]+ 切分，无句末标点会全部合并成 1 句导致 embedding 400
+CONTENT_HTML = "\n".join("<p>%s。</p>" % s for s in SENTENCES)

docs/scripts/kb_check_host.py

@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+"""宿主机会话内批量自检：逐句对知识库做 cosine 检索（复刻 verify 逻辑）。
+
+- 数据：/tmp/cv4（生产 chroma 一致性副本，生产文件零接触）
+- embedding：生产同款 BAAI/bge-large-zh-v1.5（siliconflow）
+- 首次 query 需加载 HNSW 索引（约 80s），后续每句约 1s
+用法：cd /app/cfc/cfc-langgraph && .venv/bin/python /tmp/kb_check_host.py <sentences.json> [out.json]
+"""
+import json
+import math
+import sys
+import time
+
+from langchain_chroma import Chroma
+from app.rag.embeddings import get_embeddings
+
+CHROMA_PATH = "/tmp/cv4"
+SOURCE = "microbiome"
+K = 3
+THRESHOLD = 0.7
+
+
+def _cosine(a, b) -> float:
+    a = list(a)
+    b = list(b)
+    if not a or not b or len(a) != len(b):
+        return 0.0
+    dot = sum(x * y for x, y in zip(a, b))
+    na = math.sqrt(sum(float(x) * float(x) for x in a))
+    nb = math.sqrt(sum(float(y) * float(y) for y in b))
+    if na == 0.0 or nb == 0.0:
+        return 0.0
+    return float(dot) / (float(na) * float(nb))
+
+
+def main():
+    in_path = sys.argv[1]
+    out_path = sys.argv[2] if len(sys.argv) > 2 else "/tmp/kb_check_result.json"
+    with open(in_path, encoding="utf-8") as f:
+        data = json.load(f)
+    sentences = data["sentences"] if isinstance(data, dict) else data
+
+    t0 = time.time()
+    store = Chroma(collection_name="cfc_knowledge",
+                   embedding_function=get_embeddings(),
+                   persist_directory=CHROMA_PATH)
+    embedder = get_embeddings()
+    print("store ready", round(time.time() - t0, 1), "s", flush=True)
+
+    results = []
+    all_passed = True
+    for i, s in enumerate(sentences, 1):
+        text = (s or "").strip()
+        best = {"max_similarity": 0.0, "matched_title": None, "matched_chunk": None}
+        if text:
+            try:
+                cands = store.similarity_search_with_score(text, k=K,
+                                                          filter={"source": SOURCE})
+                query_vec = embedder.embed_query(text)
+                for doc, _dist in cands:
+                    meta = doc.metadata or {}
+                    doc_id = meta.get("id")
+                    if not doc_id:
+                        continue
+                    got = store._collection.get(ids=[doc_id], include=["embeddings"])
+                    vecs = got.get("embeddings")
+                    if vecs is None or len(vecs) == 0:
+                        continue
+                    sim = _cosine(query_vec, vecs[0])
+                    if sim > best["max_similarity"]:
+                        best = {"max_similarity": sim,
+                                "matched_title": meta.get("title"),
+                                "matched_chunk": (doc.page_content[:120]
+                                                  if doc.page_content else None)}
+            except Exception as e:
+                print(f"  !! sentence {i} error: {e}", flush=True)
+        best["id"] = f"s{i}"
+        best["text"] = text
+        best["passed"] = best["max_similarity"] >= THRESHOLD
+        results.append(best)
+        if not best["passed"]:
+            all_passed = False
+        flag = "OK  " if best["passed"] else "FAIL"
+        print(f"[{i:2d}/{len(sentences)}] {flag} sim={best['max_similarity']:.4f} "
+              f"| {best['matched_title']} | {text[:40]}", flush=True)
+
+    print(f"\nall_passed={all_passed}  elapsed={round(time.time()-t0,1)}s")
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump({"results": results, "all_passed": all_passed}, f,
+                  ensure_ascii=False, indent=2)
+    return 0 if all_passed else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())