admin
0282cfeeeb
- 生信进度:后端新增 stage_progress 事件帧+进度解析白名单(rnaseq_qc/sample_qc), kernel 日志流式落盘(20行/1s flush+fsync),logs 分页端点(offset/limit/eof), 事件 seq 跨重启扫描续增(2026-09-17 断流修复);前端 FlowProgressCard 确定性/ 不确定两态进度条+点击定位到生信任务 TAB,BioJobsPane DAG 行展开滚动日志区 (3s 轮询),log_tail 4000 截断,重跑 stage_started 清零进度 - 图片注入:figures 三源重构(标准DOI图库主路径/遗留体系兜底/PMC兜底), [[FIG:]] 宽松归一化识别 PMC/PMID/DOI 三种形态,引文兜底扩展至 [PMCID:]/[DOI:] 且标记未满配额继续注入,注入预算 90s/10篇(175 实测校准), PMC graphic 无 content-type 宽松正则,空结果不缓存(176 冷连接三连败修复) - QA 启动自愈:start.sh 启动前等 Qdrant readyz(30min)+Neo4j TCP(10min); backend_guard 标签校验改子集判定 - 样本配对:同 inode 去重优先真实文件,R1/R2 归一配对一对一认领 - gitignore 新增 *.bak-*/*.before-*/bioflow_service/bioflow.db 测试:bioflow_service pytest 358/358,前端 bioflow vitest 205/205
444 lines
19 KiB
Python
444 lines
19 KiB
Python
"""normalize.py —— 上游响应 → 契约形态的归一化器（API.md §7/§8/§9）。
|
||
|
||
真实上游键以 DISCOVERY.md §4.1 为准；别名规则按 API.md 逐条实现。
|
||
本模块是纯函数集合（不发起 IO），供 qa 路由（A2）与测试共用。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
|
||
def pick(obj: dict, *names: str, default=None):
|
||
"""取第一个存在且非 None 的键（归一化核心工具）。保留 PM 脚手架语义。"""
|
||
for n in names:
|
||
if isinstance(obj, dict) and n in obj and obj[n] is not None:
|
||
return obj[n]
|
||
return default
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# §8 Source 归一化（缺失字段键必须存在，值为 null）
|
||
# ---------------------------------------------------------------------------
|
||
# title 回填优先级（缺陷 #27，2026-09-11）：title/paper_title 缺失时，
|
||
# 依次从以下字段取可读文本（论文标题优先于实体名），全部缺失才保持 null。
|
||
# 排除 content/evidence/text/snippet 正文字段——正文片段不是标题。
|
||
_TITLE_FALLBACK_TOPLEVEL = (
|
||
"publication_title", "article_title", "journal_title", "name",
|
||
)
|
||
_TITLE_FALLBACK_METADATA = (
|
||
"paper_title", "publication_title", "article_title", "journal_title",
|
||
"source_entity", "name", "source",
|
||
)
|
||
|
||
def normalize_source(item: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""单个 source item → API.md §8 形态（所有键存在，缺失为 null）。
|
||
|
||
别名映射（按 API.md §8 精确实现）：
|
||
- content|evidence|text|snippet → snippet
|
||
- doc_id|paper_id|pubmed_id|id|chunk_id → paper_id（chunk_id 单独保留）
|
||
- final_score|score|confidence|rerank_score → score（真实键 final_score）
|
||
- title|paper_title|source → title（真实 item 的 source 是断言主体实体名；
|
||
title 存在时优先取 title，两者并存时 source 落入 metadata.source_entity）
|
||
- url|doi_url → url；只有 doi 时派生 url = https://doi.org/{doi}
|
||
- evidence_id|id → id；relation_id|assertion_id → metadata.assertion_id
|
||
- output_layer / claim_scope / cluster_id / support_count / section /
|
||
species / vector_score / rrf_score / rerank_score / retrieval_methods /
|
||
retrieval_queries / cluster_member_* 原样透传（或进 metadata）
|
||
"""
|
||
if not isinstance(item, dict):
|
||
item = {}
|
||
|
||
source = pick(item, "source") # 断言主体实体名（上游语义）
|
||
title = pick(item, "title", "paper_title")
|
||
# 缺陷 #27：title 为空时按固定优先级回填——
|
||
# 顶层标题类字段 → metadata 标题类字段 → 顶层 source（断言主体实体名）。
|
||
# 正文字段绝不参与（正文片段不是标题）。
|
||
if not title:
|
||
for _key in _TITLE_FALLBACK_TOPLEVEL:
|
||
_candidate = pick(item, _key)
|
||
if _candidate:
|
||
title = _candidate
|
||
break
|
||
if not title and isinstance(pick(item, "metadata"), dict):
|
||
for _key in _TITLE_FALLBACK_METADATA:
|
||
_candidate = pick(item["metadata"], _key)
|
||
if _candidate:
|
||
title = _candidate
|
||
break
|
||
if not title:
|
||
title = source
|
||
|
||
doi = pick(item, "doi")
|
||
url = pick(item, "url", "doi_url")
|
||
if url is None and doi:
|
||
url = f"https://doi.org/{doi}"
|
||
|
||
metadata: dict[str, Any] = {}
|
||
if isinstance(pick(item, "metadata"), dict):
|
||
metadata.update(item["metadata"])
|
||
if source is not None and title is not None and source != title:
|
||
metadata.setdefault("source_entity", source)
|
||
for key in (
|
||
"vector_score", "rrf_score", "rerank_score", "kb",
|
||
"retrieval_methods", "retrieval_queries",
|
||
"cluster_member_count", "cluster_member_assertion_ids",
|
||
"cluster_member_paper_ids", "cluster_member_pmids",
|
||
"output_layer", "claim_scope", "species",
|
||
):
|
||
if key in item and item[key] is not None:
|
||
metadata.setdefault(key, item[key])
|
||
|
||
assertion_id = pick(item, "relation_id", "assertion_id")
|
||
if assertion_id is not None:
|
||
metadata.setdefault("assertion_id", assertion_id)
|
||
|
||
return {
|
||
"id": pick(item, "evidence_id", "id"),
|
||
"title": title,
|
||
"paper_id": pick(item, "doc_id", "paper_id", "pubmed_id", "id", "chunk_id"),
|
||
"chunk_id": pick(item, "chunk_id"),
|
||
"pmid": pick(item, "pmid"),
|
||
"pmcid": pick(item, "pmcid"),
|
||
"doi": doi,
|
||
"url": url,
|
||
"snippet": pick(item, "content", "evidence", "text", "snippet"),
|
||
"score": pick(item, "final_score", "score", "confidence", "rerank_score"),
|
||
"locator": pick(item, "locator"),
|
||
"section": pick(item, "section"),
|
||
"support_count": pick(item, "support_count"),
|
||
"type": pick(item, "type"),
|
||
"metadata": metadata,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# §9 图谱：hop / chain / {nodes,edges,chains}
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def normalize_hop(hop: dict[str, Any]) -> dict[str, Any]:
|
||
"""hop 归一化：
|
||
from_name|source → source、to_name|target → target、
|
||
from_id|source_canonical_id → source 节点 id、to_id|target_canonical_id → target 节点 id、
|
||
relation 原样、source_type|target_type → source/target 节点 type、
|
||
paper_id / pmid / pmcid / doi 透传。
|
||
"""
|
||
if not isinstance(hop, dict):
|
||
hop = {}
|
||
return {
|
||
"source": pick(hop, "from_name", "source"),
|
||
"target": pick(hop, "to_name", "target"),
|
||
"source_id": pick(hop, "from_id", "source_canonical_id"),
|
||
"target_id": pick(hop, "to_id", "target_canonical_id"),
|
||
"source_type": pick(hop, "source_type"),
|
||
"target_type": pick(hop, "target_type"),
|
||
"relation": pick(hop, "relation"),
|
||
"assertion_id": pick(hop, "assertion_id"),
|
||
"paper_id": pick(hop, "paper_id"),
|
||
"pmid": pick(hop, "pmid"),
|
||
"pmcid": pick(hop, "pmcid"),
|
||
"doi": pick(hop, "doi"),
|
||
}
|
||
|
||
|
||
def normalize_chain_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""单个 chain item（真实形态：chain_id / hops[] 等）→ {chain_id, hops}。"""
|
||
if not isinstance(item, dict):
|
||
item = {}
|
||
hops = [normalize_hop(h) for h in item.get("hops", []) if isinstance(h, dict)]
|
||
return {"chain_id": pick(item, "chain_id"), "hops": hops}
|
||
|
||
|
||
def extract_chains(raw: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
"""从可能的两类源提取 chain 列表：
|
||
- `native_graph_chains`：列表（真实形态，305 条）
|
||
- `graph_knowledge_chains`：dict 形态 `{mode, chains[], chain_count, …}`，取 `.chains`
|
||
两种形态都要能提取出 hop 列表。
|
||
"""
|
||
if not isinstance(raw, dict):
|
||
return []
|
||
chains = pick(raw, "native_graph_chains", "graph_knowledge_chains")
|
||
if isinstance(chains, dict):
|
||
chains = chains.get("chains")
|
||
if not isinstance(chains, list):
|
||
return []
|
||
return [normalize_chain_item(c) for c in chains]
|
||
|
||
|
||
def normalize_graph(raw: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""从 `knowledge_graph` / `graph_edges` / `native_graph_chains`（列表形态）/
|
||
`graph_knowledge_chains`（dict 形态）提取并归一化为 {nodes, edges, chains}。
|
||
|
||
- 节点去重（按 id；id 缺失的 hop 跳过）；label 取 from_name/to_name、type 取
|
||
source_type/target_type。
|
||
- 边携带 chain_id 与 hop 序号（hop_index 缺失时按列表序补）。
|
||
- 链保持 chain_id + hops 列表。
|
||
"""
|
||
raw = raw if isinstance(raw, dict) else {}
|
||
chains = extract_chains(raw)
|
||
|
||
nodes: dict[str, dict[str, Any]] = {}
|
||
edges: list[dict[str, Any]] = []
|
||
|
||
def add_node(node_id: Any, name: Any, typ: Any) -> None:
|
||
if not node_id:
|
||
return
|
||
key = str(node_id)
|
||
if key not in nodes:
|
||
nodes[key] = {"id": node_id, "label": name, "type": typ}
|
||
|
||
for chain in chains:
|
||
for idx, hop in enumerate(chain["hops"]):
|
||
hop_index = hop.get("hop_index")
|
||
if hop_index is None:
|
||
hop_index = idx
|
||
# 节点 id 优先 canonical id；label 优先名字，退化到 id
|
||
src_id = pick(hop, "source_id", "source") or hop.get("source")
|
||
tgt_id = pick(hop, "target_id", "target") or hop.get("target")
|
||
src_label = hop.get("source") or src_id
|
||
tgt_label = hop.get("target") or tgt_id
|
||
add_node(src_id, src_label, hop.get("source_type"))
|
||
add_node(tgt_id, tgt_label, hop.get("target_type"))
|
||
if src_id and tgt_id:
|
||
edges.append({
|
||
"source": src_id,
|
||
"target": tgt_id,
|
||
"relation": hop.get("relation"),
|
||
"label": hop.get("relation"),
|
||
"chain_id": chain.get("chain_id"),
|
||
"hop": hop_index,
|
||
"paper_id": hop.get("paper_id"),
|
||
"pmid": hop.get("pmid"),
|
||
"pmcid": hop.get("pmcid"),
|
||
"doi": hop.get("doi"),
|
||
})
|
||
|
||
# 兼容显式 {nodes, edges} 形态（knowledge_graph / graph_edges 直接给边列表时）
|
||
explicit_nodes = raw.get("nodes")
|
||
explicit_edges = raw.get("edges")
|
||
if isinstance(explicit_nodes, list):
|
||
for n in explicit_nodes:
|
||
if isinstance(n, dict) and n.get("id"):
|
||
key = str(n["id"])
|
||
if key not in nodes:
|
||
nodes[key] = {
|
||
"id": n["id"],
|
||
"label": pick(n, "label"),
|
||
"type": pick(n, "type"),
|
||
}
|
||
if isinstance(explicit_edges, list):
|
||
edges.extend(
|
||
{
|
||
"source": pick(e, "source"),
|
||
"target": pick(e, "target"),
|
||
"relation": pick(e, "relation"),
|
||
"label": pick(e, "label", "relation"),
|
||
"chain_id": pick(e, "chain_id"),
|
||
"hop": pick(e, "hop"),
|
||
"paper_id": pick(e, "paper_id"),
|
||
"pmid": pick(e, "pmid"),
|
||
"pmcid": pick(e, "pmcid"),
|
||
"doi": pick(e, "doi"),
|
||
}
|
||
for e in explicit_edges
|
||
if isinstance(e, dict)
|
||
)
|
||
|
||
return {
|
||
"nodes": list(nodes.values()),
|
||
"edges": edges,
|
||
"chains": chains,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# §7 归一化 QA 响应（/chat、/chat/stream complete、/search 共用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def normalize_qa_response(raw: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""上游 31 键响应 → API.md §7 形态。
|
||
|
||
别名（真实键优先）：
|
||
- answer|response|message → answer
|
||
- sources|citations → sources；used_sources|references → used_sources
|
||
- query_plan|agentic_plan|retrieval_plan → retrieval_plan（真实键 query_plan）
|
||
- retrieval_notes|notes → retrieval_notes
|
||
- timings|timing → timing（真实键 timings）
|
||
- citation_validation|citation_map → citation_map（真实键 citation_validation）
|
||
- 图谱四源 → graph + chains（见 normalize_graph）
|
||
- success 透传；raw 保存原始完整响应
|
||
"""
|
||
raw = raw if isinstance(raw, dict) else {}
|
||
|
||
sources = [normalize_source(s) for s in (pick(raw, "sources", "citations") or [])]
|
||
used_sources = [
|
||
normalize_source(s) for s in (pick(raw, "used_sources", "references") or [])
|
||
]
|
||
graph = normalize_graph(raw)
|
||
retrieval_plan = pick(raw, "query_plan", "agentic_plan", "retrieval_plan")
|
||
|
||
return {
|
||
"query_id": pick(raw, "query_id"),
|
||
"version": pick(raw, "version"),
|
||
"mode": pick(raw, "mode", default="chat"),
|
||
"answer": pick(raw, "answer", "response", "message", default=""),
|
||
"source_count": pick(raw, "source_count", default=len(sources)),
|
||
"sources": sources,
|
||
"used_sources": used_sources,
|
||
"chains": graph["chains"],
|
||
"graph": {"nodes": graph["nodes"], "edges": graph["edges"], "chains": graph["chains"]},
|
||
"retrieval_plan": retrieval_plan,
|
||
"retrieval_notes": pick(raw, "retrieval_notes", "notes"),
|
||
"timing": pick(raw, "timings", "timing"),
|
||
"citation_map": pick(raw, "citation_validation", "citation_map"),
|
||
"success": pick(raw, "success", default=True),
|
||
"raw": raw,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 图片注入（2026-08-28 图文并茂方案）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# 标记形态(2026-09-22 三修):LLM 输出不稳定(PMC/DOI/PMID、带不带冒号空格都出现过),
|
||
# 改为宽松捕获 [[FIG:任意内容]] + _marker_figures 归一化识别,不再逐形态猜正则。
|
||
_FIG_MARKER = re.compile(r"\[\[FIG:([^\]]+)\]\]")
|
||
_MAX_FIGURES = 24
|
||
|
||
|
||
def _normalize_marker_token(token: str) -> str:
|
||
"""把 [[FIG:...]] 内容归一化为裸标识:去空白、去 PMCID:/PMID:/DOI: 前缀。"""
|
||
value = (token or "").strip()
|
||
value = re.sub(r"^(?:PMCID|PMID|DOI)\s*[:：]\s*", "", value, flags=re.I)
|
||
return value.strip()
|
||
|
||
|
||
def _marker_figures(token: str, figures_map: dict[str, dict[str, Any]]) -> tuple[str, list[dict]]:
|
||
"""标记内容 → (map 键, 图清单)。三种形态:PMCxxxx / 纯数字 PMID / DOI。"""
|
||
norm = _normalize_marker_token(token)
|
||
upper = norm.upper()
|
||
if re.match(r"^PMC\d+$", upper):
|
||
return upper, ((figures_map.get(upper) or {}).get("figures") or [])
|
||
if re.match(r"^\d+$", norm):
|
||
for k, v in figures_map.items():
|
||
if str((v or {}).get("pmid") or "") == norm:
|
||
return k, ((v or {}).get("figures") or [])
|
||
key = f"PMID:{norm}"
|
||
return key, ((figures_map.get(key) or {}).get("figures") or [])
|
||
if norm.startswith("10.") or upper.startswith("DOI:"):
|
||
doi_val = norm[4:].strip() if upper.startswith("DOI:") else norm
|
||
for cand in (f"DOI:{doi_val}", f"DOI:{doi_val.replace('/', '_')}"):
|
||
if cand in figures_map:
|
||
return cand, ((figures_map.get(cand) or {}).get("figures") or [])
|
||
key = f"DOI:{doi_val}"
|
||
return key, ((figures_map.get(key) or {}).get("figures") or [])
|
||
return norm, []
|
||
|
||
|
||
def inject_figure_markdown(
|
||
answer: str,
|
||
figures_map: dict[str, dict[str, Any]],
|
||
base_url: str = "",
|
||
) -> str:
|
||
"""把 LLM 的 [[FIG:PMCID]] / [[FIG:DOI:...]] 标记替换为 markdown 图注块（纯同步函数）。
|
||
|
||
参数 figures_map: 键为 PMCID（大写）或 DOI 原串 -> {"license": str, "figures": [...]}
|
||
由调用方（sse_relay）预先异步解析后传入，本函数不做 IO，便于单测。
|
||
|
||
规则：
|
||
- 标记 → 该键对应前 2 张图；无图删标记
|
||
- 兜底：无任何标记时，按引用 [PMID: xxx] / [PMCID: xxx] / [DOI: xxx] 首次
|
||
出现位置插入对应论文前 2 张图（2026-09-22 起三种引文形态都注入）
|
||
- 总图数上限 8，超出删除标记/跳过
|
||
"""
|
||
answer = answer or ""
|
||
if not answer:
|
||
return answer
|
||
|
||
# 1. 解析标记(宽松捕获 + _marker_figures 归一化识别)
|
||
markers = _FIG_MARKER.findall(answer)
|
||
inserted = 0
|
||
if markers:
|
||
used_pmcids: dict[str, int] = {}
|
||
for token in markers:
|
||
if inserted >= _MAX_FIGURES:
|
||
break
|
||
key, figures = _marker_figures(token, figures_map)
|
||
take = min(2, len(figures), _MAX_FIGURES - inserted)
|
||
if not take:
|
||
continue
|
||
used_pmcids[key] = used_pmcids.get(key, 0) + 1
|
||
markdown_lines = _figure_lines(figures[:take], base_url)
|
||
replacement = "\n\n" + "\n\n".join(markdown_lines) + "\n\n"
|
||
# 只替换第一个出现的该标记（每论文一个标记由 prompt 保证）
|
||
answer = answer.replace(f"[[FIG:{token}]]", replacement, 1)
|
||
inserted += take
|
||
# 清理残余标记（无图/超限的）
|
||
answer = _FIG_MARKER.sub("", answer)
|
||
if inserted >= _MAX_FIGURES:
|
||
return answer
|
||
# 2026-09-22 修复:标记未用满配额(含一张未注入)时继续引文兜底。
|
||
# 此前标记存在即 return——「有标记但无图」时 PMID/PMCID/DOI 引文永不注入。
|
||
|
||
# 2. 兜底：标记未用满配额时,按引用位置注入(PMID/PMCID/DOI 三种引文形态)
|
||
citation_entries: list[tuple[str, str]] = [] # (citation 文本, figures_map 键)
|
||
pmcid_by_pmid = _pmid_to_pmcid_map(figures_map)
|
||
for pmid, pmcid in pmcid_by_pmid.items():
|
||
citation_entries.append((f"[PMID: {pmid}]", pmcid))
|
||
for key, data in figures_map.items():
|
||
upper = str(key).upper()
|
||
if upper.startswith("PMC"):
|
||
citation_entries.append((f"[PMCID: {upper}]", key))
|
||
elif str(key).startswith("DOI:"):
|
||
citation_entries.append((f"[DOI: {str(key)[4:].strip()}]", key))
|
||
alias = (data or {}).get("doi_alias")
|
||
if alias and str(alias).startswith("DOI:"):
|
||
citation_entries.append((f"[DOI: {str(alias)[4:].strip()}]", key))
|
||
seen_keys: set[str] = set()
|
||
for citation, key in citation_entries:
|
||
if inserted >= _MAX_FIGURES:
|
||
break
|
||
if key in seen_keys:
|
||
continue
|
||
data = figures_map.get(key) or {}
|
||
figures = (data.get("figures") or [])[:2]
|
||
if not figures:
|
||
continue
|
||
idx = answer.find(citation)
|
||
if idx < 0:
|
||
continue
|
||
seen_keys.add(key)
|
||
end = idx + len(citation)
|
||
markdown_lines = _figure_lines(figures, base_url)
|
||
block = "\n\n" + "\n\n".join(markdown_lines) + "\n\n"
|
||
answer = answer[:end] + block + answer[end:]
|
||
inserted += len(figures)
|
||
return answer
|
||
|
||
|
||
def _figure_lines(figures: list[dict], base_url: str) -> list[str]:
|
||
lines = []
|
||
for f in figures:
|
||
url = f.get("url") or ""
|
||
caption = f.get("caption") or ""
|
||
license_ = f.get("license") or ""
|
||
if base_url and not url.startswith("http"):
|
||
url = f"{base_url.rstrip('/')}{url}"
|
||
if not url:
|
||
continue
|
||
cap_text = caption
|
||
if license_:
|
||
cap_text = f"{caption}（许可: {license_}）" if caption else f"许可: {license_}"
|
||
lines.append(f"")
|
||
return lines
|
||
|
||
|
||
def _pmid_to_pmcid_map(figures_map: dict[str, dict[str, Any]]) -> dict[str, str]:
|
||
"""从 figures_map 的 metadata 里推导 PMID→PMCID（调用方在数据里带上 pmid）。"""
|
||
out = {}
|
||
for pmcid, data in figures_map.items():
|
||
pmid = (data or {}).get("pmid")
|
||
if pmid:
|
||
out[str(pmid)] = pmcid
|
||
return out
|