diff --git a/kb-libraries/aqua/configs/entity_alias_aqua.yaml b/kb-libraries/aqua/configs/entity_alias_aqua.yaml new file mode 100644 index 0000000..65a36d3 --- /dev/null +++ b/kb-libraries/aqua/configs/entity_alias_aqua.yaml @@ -0,0 +1,244 @@ +Litopenaeus vannamei: + type: species + aliases: + - Litopenaeus vannamei + - L. vannamei + - Penaeus vannamei + - whiteleg shrimp + - Pacific white shrimp + - white shrimp + - vannamei shrimp + - 凡纳滨对虾 + - 南美白对虾 + - 对虾 + forbidden_expansions: + - Penaeus monodon + - P. monodon + +Penaeus monodon: + type: species + aliases: + - Penaeus monodon + - P. monodon + - black tiger shrimp + - tiger prawn + - giant tiger prawn + - 斑节对虾 + - 草虾 + +Macrobrachium rosenbergii: + type: species + aliases: + - Macrobrachium rosenbergii + - M. rosenbergii + - giant freshwater prawn + - giant river prawn + - 罗氏沼虾 + +Oreochromis niloticus: + type: species + retrieval_label: tilapia + aliases: + - Oreochromis niloticus + - O. niloticus + - Nile tilapia + - tilapia + - 罗非鱼 + - 尼罗罗非鱼 + +Oncorhynchus mykiss: + type: species + aliases: + - Oncorhynchus mykiss + - O. mykiss + - rainbow trout + - steelhead trout + - steelhead + - 虹鳟 + +Salmo salar: + type: species + aliases: + - Salmo salar + - S. salar + - Atlantic salmon + - salmon + - 大西洋鲑 + - 三文鱼 + forbidden_expansions: + - Oncorhynchus + - Pacific salmon + - 太平洋鲑 + +Ctenopharyngodon idella: + type: species + aliases: + - Ctenopharyngodon idella + - C. idella + - grass carp + - white amur + - 草鱼 + +Cyprinus carpio: + type: species + aliases: + - Cyprinus carpio + - C. carpio + - common carp + - carp + - 鲤 + - 鲤鱼 + forbidden_expansions: + - Ctenopharyngodon idella + - grass carp + - 草鱼 + +Lateolabrax maculatus: + type: species + aliases: + - Lateolabrax maculatus + - L. maculatus + - spotted sea bass + - Chinese sea bass + - sea bass + - 花鲈 + - 鲈鱼 + +Larimichthys crocea: + type: species + aliases: + - Larimichthys crocea + - L. crocea + - large yellow croaker + - yellow croaker + - 大黄鱼 + +Crassostrea gigas: + type: species + aliases: + - Crassostrea gigas + - C. gigas + - Magallana gigas + - Pacific oyster + - oyster + - 长牡蛎 + - 牡蛎 + +Chlamys farreri: + type: species + aliases: + - Chlamys farreri + - C. farreri + - Azumapecten farreri + - Zhikong scallop + - Farrer's scallop + - 栉孔扇贝 + - 扇贝 + +Ruditapes philippinarum: + type: species + aliases: + - Ruditapes philippinarum + - R. philippinarum + - Venerupis philippinarum + - Manila clam + - Japanese littleneck clam + - clam + - 菲律宾蛤仔 + - 花蛤 + +Saccharina japonica: + type: species + aliases: + - Saccharina japonica + - S. japonica + - Laminaria japonica + - kelp + - sea tangle + - 海带 + +Undaria pinnatifida: + type: species + aliases: + - Undaria pinnatifida + - U. pinnatifida + - wakame + - 裙带菜 + +Apostichopus japonicus: + type: species + aliases: + - Apostichopus japonicus + - A. japonicus + - Stichopus japonicus + - Japanese sea cucumber + - sea cucumber + - 仿刺参 + - 海参 + +growth performance: + type: phenotype + aliases: + - growth performance + - growth + - specific growth rate + - SGR + - weight gain + - WG + - body weight gain + - 生长性能 + - 生长 + retrieval_terms: + - final body weight + - FBW + +feed conversion: + type: phenotype + aliases: + - feed conversion ratio + - FCR + - feed conversion efficiency + - FCE + - feed efficiency + - 饲料转化率 + - 饲料系数 + +survival: + type: phenotype + aliases: + - survival rate + - survival + - SR + - cumulative survival + - 存活率 + - 成活率 + +immunity: + type: phenotype + aliases: + - immunity + - immune response + - innate immunity + - immune defense + - 免疫 + - 免疫应答 + +disease resistance: + type: phenotype + aliases: + - disease resistance + - pathogen resistance + - resistance + - 抗病性 + - 抗病力 + +hypoxia: + type: environmental_factor + aliases: + - hypoxia + - hypoxic stress + - low oxygen + - dissolved oxygen + - DO + - 低氧 + - 缺氧 diff --git a/kb-libraries/aqua/configs/relation_taxonomy_aqua.yaml b/kb-libraries/aqua/configs/relation_taxonomy_aqua.yaml new file mode 100644 index 0000000..9040372 --- /dev/null +++ b/kb-libraries/aqua/configs/relation_taxonomy_aqua.yaml @@ -0,0 +1,367 @@ +# V16.aqua relation taxonomy: aquaculture SRT categories (aligned with the +# aquaculture ALLOWED_RELATION_CATEGORIES vocabulary) with canonical relation +# predicates grouped in the same format as the plant relation_taxonomy.yaml. + +growth_performance: + - improves_growth + - improves_growth_performance + - promotes_growth + - enhances_growth + - increases_weight_gain + - improves_specific_growth_rate + - increases_specific_growth_rate + - improves_weight_gain + - reduces_growth + - inhibits_growth + - impairs_growth + - decreases_weight_gain + +feed_conversion_efficiency: + - improves_feed_conversion + - reduces_fcr + - improves_fcr + - improves_feed_efficiency + - increases_feed_conversion_efficiency + - improves_feed_utilization + - feed conversion ratio + - feed conversion efficiency + - FCR + +survival_rate: + - improves_survival + - increases_survival_rate + - reduces_survival + - decreases_survival_rate + - increases_mortality + - reduces_mortality + - improves_survival_rate + - survival rate + - mortality + +disease_resistance: + - improves_disease_resistance + - increases_resistance_to + - confers_resistance_to + - reduces_resistance_to + - disease resistance + +immune_response: + - enhances_immune_response + - activates_immune_response + - stimulates_immune_response + - suppresses_immune_response + - modulates_immunity + - increases_lysozyme_activity + - increases_immune_parameters + - involved_in_immune_response + - improves_immunity + - immune response + +water_quality_effect: + - improves_water_quality + - degrades_water_quality + - affects_water_quality + - reduces_ammonia + - increases_dissolved_oxygen + - reduces_nitrite + - water quality + - ammonia concentration + - dissolved oxygen + +expression_tissue: + - expressed_in + - highly_expressed_in + - predominantly_expressed_in + - detected_in + - enriched_in + - localizes_to + - localized_in + +expression_condition: + - upregulated_in + - upregulated_under + - downregulated_in + - downregulated_under + - differentially_expressed_under + - expression_changed_under + - induced_by + - responds_to + +expression_time: + - expressed_at + - peak_expression_at + - expression_increases_over_time + - expression_decreases_over_time + +stress_response: + - responds_to + - responsive_to + - induced_by + - induced_under + - involved_in_stress_response + - confers_tolerance_to + - enhances_tolerance_to + - improves_tolerance_to + - stress tolerance + - hypoxic stress + - salinity stress + - thermal stress + +subcellular_localization: + - localizes_to + - located_in + - predicted_to_localize_to + - subcellular_localized_to + +phenotype_effect: + - promotes + - inhibits + - enhances + - reduces + - affects + - regulates + - involved_in + - participates_in + - contributes_to + - associated_with + +protein_property: + - has_molecular_weight + - has_isoform + - has_domain + - has_activity + - binds_to + +variant_trait_association: + - associated_with + - linked_to + - correlated_with + - snp_associated_with + +QTL_trait_association: + - qtl_associated_with + - qtl_linked_to + - associated_with + - linked_to + +marker_trait_association: + - marker_associated_with + - marker_linked_to + - associated_with + - linked_to + +pathway_participation: + - participates_in_pathway + - involved_in_pathway + - component_of_pathway + - regulates_pathway + +metabolic_change: + - increases_metabolite + - decreases_metabolite + - changes_metabolite_level + - alters_metabolism + - regulates_metabolism + - involved_in_metabolism + - improves_metabolism + - metabolic change + +hormone_response: + - responds_to_hormone + - induced_by_hormone + - regulated_by_hormone + - hormone_regulates + +pathogen_response: + - responds_to_pathogen + - induced_by_pathogen + - upregulated_after_infection + - suppresses_pathogen + - inhibits_pathogen + - increases_pathogen_load + - reduces_pathogen_load + +developmental_regulation: + - regulates_development + - involved_in_development + - promotes_development + - inhibits_development + - regulates_differentiation + +reproduction_related: + - affects_reproduction + - improves_reproductive_performance + - increases_fecundity + - affects_spawning + - regulates_gonad_development + - increases_hatching_rate + - reduces_fertility + - spawning + - gonad development + +nutritional_value: + - improves_nutritional_value + - increases_nutritional_content + - affects_flesh_quality + - increases_protein_content + - increases_lipid_content + - reduces_cholesterol + - nutritional value + - flesh quality + +physiological_measurement: + - measured_in + - physiological_parameter + - serum_parameter + - changes_physiology + - affects_physiology + - hematological_parameter + +stocking_density_effect: + - affected_by_stocking_density + - stocking_density_affects + - decreases_with_density + - increases_with_density + - improves_at_low_density + - impairs_at_high_density + +compound_effect: + - compound_promotes + - compound_inhibits + - compound_enhances + - compound_reduces + - exerts_effect + - dietary_effect + +enzyme_activity: + - increases_enzyme_activity + - decreases_enzyme_activity + - catalyzes + - activates_enzyme + - inhibits_enzyme + - enzyme_activity + +protein_interaction: + - interacts_with + - binds_to + - forms_complex_with + - activates_protein + - inhibits_protein + - phosphorylates + - targets + +regulatory_interaction: + - regulates + - upregulates + - downregulates + - activates + - inhibits + - modulates + +catalog_count: + - has_member_count + - identified_in + - systematically_identified_in + - genome_wide_identified_in + - were_identified_in + - was_identified_in + - identified_in + - contains + - includes + - had + - comprehensively_analyzed_in + - surveyed_in + +family_membership: + - member_of + - belongs_to + - is_a + +classification: + - belongs_to + - member_of + - classified_into + - divided_into + - separated_into + - grouped_into + - clustered_with + +chromosomal_location: + - located_on_chromosome + - maps_to_chromosome + - on_chromosome + +gene_structure: + - has_exon + - has_intron + - has_utr + - contains_cds + +motif_domain: + - contains_domain + - contains_motif + - has_conserved_domain + - lacks_binding_site_for + +cis_element: + - contains_cis_element + - has_promoter_element + +duplication_synteny: + - tandemly_duplicated + - segmentally_duplicated + - collinear_with + - syntenic_with + - duplicated_from + +orthology_homology: + - ortholog_of + - homolog_of + - paralog_of + +method_tool: + - used_tool + - analyzed_by + - predicted_by + - measured_with + - constructed_with + - identified_by + - quantified_by + +experimental_material: + - material_used + - treated_with + - fed_with + - reared_in + - cultured_in + - sampled_from + +dataset_accession: + - has_accession + - available_at + - has_data_type + - from_database + - from_article + - used_for + +background_prior_work: + - reported_in + - previously_reported_in + - cited_as + - known_to + +figure_caption_observation: + - observed_in + - shown_in + - depicted_in + +table_structured_fact: + - measured_at + - recorded_in + - reported_in + +other_supported: + - associated_with + - related_to + - affects + - involves diff --git a/kb-libraries/aqua/manifest.yaml b/kb-libraries/aqua/manifest.yaml index 0d42840..f239772 100644 --- a/kb-libraries/aqua/manifest.yaml +++ b/kb-libraries/aqua/manifest.yaml @@ -2,6 +2,6 @@ name: aqua release_id: aqua_srt_v1_20260930_normalized neo4j_database: aqua-normalized qdrant_collection: aqua_srt_v1_20260930_normalized -alias_config: configs/rag/entity_alias.yaml -taxonomy_config: configs/rag/relation_taxonomy.yaml +alias_config: configs/rag/entity_alias_aqua.yaml +taxonomy_config: configs/rag/relation_taxonomy_aqua.yaml registry_path: "" diff --git a/kb-libraries/aqua/qa/budget_template.md b/kb-libraries/aqua/qa/budget_template.md new file mode 100644 index 0000000..d213de6 --- /dev/null +++ b/kb-libraries/aqua/qa/budget_template.md @@ -0,0 +1,63 @@ +# 水产库建库预算模板(待试点实测回填 + 用户确认) + +**版本**: v1(2026-09-12,T 代理编制) +**配套**: `tooling/aqua_kb/cost_model.py`(可重复运行,参数化;本文档是该模型的假设清单与回填表) +**状态**: 数字为当前最佳估算;标注「实测」的来自仓库数据/历史跑批,标注「假设」的待 P1 试点实测回填。**交用户确认后方可作为采购/排期依据。** + +## 1. 当前预算(中值) + +| 项目 | 规模 | LLM 成本(¥) | 备注 | +|---|---|---|---| +| P1 试点 MD(470 篇) | 2 调用/篇 | 0.75 | 验证成本,可忽略 | +| P1 试点 PDF(30 篇) | ~8 调用/篇 | 0.92 | 同上 | +| **全量 MD 31,848 篇** | 2 调用/篇 | **50.84** | 输入 39.1M token / 输出 5.9M token | +| **全量 PDF 13,175 篇** | ~8 调用/篇 | **403.15** | 输入 234.5M token / 输出 84.3M token | +| **全量合计** | | **454.0(±30%: 318–590)** | 主要为 PDF 全文抽取(89%) | + +- **MinerU GPU 耗时**: 全量 13,175 篇按单卡 3 端点并流历史吞吐(保守 700 篇/时)全天跑 **0.8 天**;含失败重试与运维余量,排期预算 **1–2 天**。GPU 为本地 175 单卡 RTX 5090(32GB),无 API 成本。 +- **DeepSeek 余额**: 历史会话记录 402 Insufficient Balance;全量跑批前需确认充值(见风险节)。 + +## 2. 假设清单 + +| # | 参数 | 当前值 | 状态 | 来源/回填方式 | +|---|---|---|---|---| +| H1 | MD 卡平均字符数 | 2,109(P50=2,169) | **实测** | source/题目摘要_MD 200 篇 seed=42 抽样 | +| H2 | PDF 正文(MinerU 输出)平均字符数 | 60,000 | **假设** | P1 30 篇试点出来后实测回填(源 PDF P50=1.35MB/P90=5.11MB 为间接参照) | +| H3 | 英文字符/token | 4.0 | **假设(经验值)** | 语料英文占主导(MD 卡 CJK 字符占比 ~4%) | +| H4 | 中文字符/token | 1.5 | **假设(经验值)** | 当前语料中文极少,影响 <2%;不单独入模型 | +| H5 | MD 每篇 LLM 调用次数 | 2(提取+复核) | **假设** | D2 编排方案确定后回填;若引入 review 环节 +1 | +| H6 | PDF 每篇 LLM 调用次数 | ≈8(60k 字符 / 8k 分块) | **假设** | 同上;由分块大小决定 | +| H7 | 输出/输入 token 比(out_in_ratio) | 0.35(MD 卡) | **假设** | D2 的 build/pilot/cost_log.jsonl 出来后按实际 in/out token 拟合回填 | +| H8 | PDF 每块调用输出 token | 800 | **假设** | 同上回填 | +| H9 | 每调用提示词/模板开销 | 350 token | **假设** | 同上回填 | +| H10 | DeepSeek 单价 | 输入 1.0 / 输出 2.0 ¥/M | **实测(官方列表价)** | deepseek-v4-flash 官方 CNY 标价(缓存未命中);见下节定价依据 | +| H11 | 高峰时段加权 | 按列表价口径(×1.0) | **假设(口径选择)** | DeepSeek 2026-07 起有高峰双倍计费口径;列表价保守口径可经 `--peak-multiplier 2.0` 切换 | +| H12 | MinerU 吞吐 | 700 篇/时(单卡 3 端点并流) | **实测外推** | 2026-08-31 历史批:9,363 篇/约 11h(成功 7,689)≈ 850/h,取保守 700 | +| H13 | MinerU 不可处理率 | 17.7% | **实测外推** | 同上批:1,656/9,363;不可处理篇不产生 LLM 成本但占 GPU 时间 | +| H14 | 语料规模 | MD 31,848 / PDF 13,175 | **实测** | kb-libraries/aqua/build/census.json(2026-09-12) | + +## 3. DeepSeek 定价依据与口径说明 + +- **模型现状**: 项目运行中模型名有 `deepseek-chat`(docgen 默认)与 `deepseek-v4-flash`(QA/生信路径)并存。`deepseek-chat` 为 legacy 别名,官方已公告 **2026-07-24 停用**,现映射至 `deepseek-v4-flash` 非思考模式,同价。 +- **采用标价**: deepseek-v4-flash 官方 CNY 列表价 **输入 1 元/M(缓存未命中,命中 0.02 元/M)、输出 2 元/M**。本模型按**缓存未命中**的保守口径计,且不含缓存命中折让 —— 实际跑批若命中 prompt 缓存(批量同类任务命中率通常可观),成本可再降。 +- **注意**: USD 渠道(美元计费)2026-08-16 起有一轮涨价($0.14/$0.28 → $0.22/$0.66 等),且有高峰时段双倍计费口径;本项目 DeepSeek 用量以人民币渠道为准,故按 CNY 官方列表价。**若实际账户为 USD 计费或确认高峰双倍生效,重跑 `--price-cny-m-input/--price-cny-m-output/--peak-multiplier` 覆盖。** +- 数据来源: DeepSeek API Docs 官方定价页(模型 & 价格),检索核对于 2026-09-12。 + +## 4. 待回填项(试点实测后) + +| 待回填 | 触发条件 | 回填动作 | +|---|---|---| +| pdf_chars | P1 30 篇 MinerU 输出完成 | 实测均值回填 `--pdf-chars` | +| out_in_ratio / pdf_output_per_call / overhead | D2 产出 cost_log.jsonl | 线性拟合 token 比,回填参数 | +| md_llm_calls / pdf 分块与调用数 | D2 编排方案定稿 | 回填调用次数 | +| 实际单价与计费口径 | 财务/账户确认 | 按确认值覆盖单价参数 | +| MinerU 吞吐 | P1 30 篇实测耗时 | 与 H12 交叉验证,偏差 >30% 则更新 | + +回填后重跑 `cost_model.py --out budget_final.md` 生成终版预算,交用户确认。 + +## 5. 风险 + +1. **DeepSeek 余额**: 历史记录 402 Insufficient Balance,全量跑批前须充值;P1 试点仅需 ¥1.67,可作为充值前的最小验证。 +2. **pdf_chars 是最不确定项**: 60k 字符假设若偏差 ±50%,全量 PDF 成本在 ¥240–610 区间内波动;这正是把 P1 试点放在预算确认之前的原因。 +3. **调用次数假设依赖 D2 编排**: 若 MD 卡也走全文级抽取或增加 review 轮次,成本按调用数线性放大 —— 预算表已按「2 调用/篇」给出,编排变更时用 `--md-llm-calls` 同步。 +4. **词表缺口**(影响质检而非成本): aqua 词表 configs 目前是植物占位内容,P1 抽取须等 D1 交付 aqua 词表,否则合规率指标挂起(见 qc_standards.md)。 diff --git a/kb-libraries/aqua/qa/golden_aqua.jsonl b/kb-libraries/aqua/qa/golden_aqua.jsonl new file mode 100644 index 0000000..6ee4a21 --- /dev/null +++ b/kb-libraries/aqua/qa/golden_aqua.jsonl @@ -0,0 +1,41 @@ +{"id": "aqua-g01", "question": "通过三代选择性育种,虹鳟(Oncorhynchus mykiss)对嗜冷黄杆菌(Flavobacterium psychrophilum)的抗病力提高了吗?早龄期自然攻毒下的存活率有何变化?", "language": "zh", "topics": ["disease", "genetics_breeding"], "expected_evidence": {"must_entities": ["rainbow trout", "Oncorhynchus mykiss", "Flavobacterium psychrophilum", "survival"], "must_relations": ["improves_disease_resistance", "selective_breeding_for", "survival_under_challenge"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Three generations of selective breeding improved rainbow trout disease resistance(2018, 样本第1条)"} +{"id": "aqua-g02", "question": "At what larval age should Nile tilapia be vaccinated against Streptococcus agalactiae, and what protection does the timing give?", "language": "en", "topics": ["disease"], "expected_evidence": {"must_entities": ["Nile tilapia", "Oreochromis niloticus", "Streptococcus agalactiae", "vaccine"], "must_relations": ["vaccination_timing", "protects_against", "optimal_age"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:First Investigation of the Optimal Timing of Vaccination of Nile Tilapia Larvae(2023, paired)"} +{"id": "aqua-g03", "question": "源自解蛋白弧菌 DCF12.2(Vibrio proteolyticus DCF12.2)的后生元饲喂,对金头鲷(Sparus aurata)幼鱼的肠道完整性、微生物群和免疫应答有什么增强作用?", "language": "zh", "topics": ["disease", "feed_nutrition"], "expected_evidence": {"must_entities": ["Sparus aurata", "Vibrio proteolyticus DCF12.2", "postbiotic"], "must_relations": ["enhances_immunity", "improves_intestinal_integrity", "modulates_microbiota"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Dietary Administration of Postbiotics... Enhanced Intestinal Integrity, Microbiota, and Immune Response in Juvenile Gilthead Seabream(2025, paired)"} +{"id": "aqua-g04", "question": "How do temperature, salinity and pH quantitatively influence WSSV (white spot syndrome virus) proliferation in Chinese shrimp (Fenneropenaeus chinensis)?", "language": "en", "topics": ["disease", "water_quality"], "expected_evidence": {"must_entities": ["WSSV", "Fenneropenaeus chinensis", "temperature", "salinity", "pH"], "must_relations": ["affects_viral_proliferation", "environmental_factor_influences", "quantitative_effect"]}, "difficulty": "hard", "answerable_from_corpus": "maybe", "notes": "锚定:Quantitative analysis of temperature, salinity and pH on WSSV proliferation(2011, pdf_only 仅标题无摘要)"} +{"id": "aqua-g05", "question": "肝肠胞虫(EHP / Ecytonucleospora hepatopenaei)在印度东部虾场的凡纳滨对虾(Litopenaeus vannamei)中流行率与相关死亡率如何?与哪些养殖管理因素有关?", "language": "zh", "topics": ["disease"], "expected_evidence": {"must_entities": ["Ecytonucleospora hepatopenaei", "Litopenaeus vannamei", "prevalence", "mortality"], "must_relations": ["causes_mortality", "prevalence_in", "associated_with_management"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:EHP disease prevalence and mortality in Litopenaeus vannamei: Eastern India shrimp farms(2024, paired)"} +{"id": "aqua-g06", "question": "大黄鱼(Larimichthys crocea)上首次分离鉴定的新致病菌 Kocuria kristinae 有什么特征?对大黄鱼健康有何影响?", "language": "zh", "topics": ["disease"], "expected_evidence": {"must_entities": ["Kocuria kristinae", "Larimichthys crocea"], "must_relations": ["first_isolated_from", "causes_disease", "pathogenic_characterization"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:A new pathogenic isolate of Kocuria kristinae identified for the first time in the marine fish Larimichthys crocea(2023)"} +{"id": "aqua-g41", "question": "河蟹(中华绒螯蟹 Eriocheir sinensis)的颤抖病由什么病原引起?目前有什么有效防控手段?", "language": "zh", "topics": ["disease"], "expected_evidence": {"must_entities": ["Eriocheir sinensis", "tremor disease"], "must_relations": ["causes_disease", "prevention_method"]}, "difficulty": "easy", "answerable_from_corpus": "no", "notes": "负向对照题:抽样与标题分类中未见中华绒螯蟹专项病害文献,预期检索不应命中强证据;用于检验检索系统对库外问题的诚实拒答"} +{"id": "aqua-g07", "question": "循环水养殖系统中,生物滤器通过曝气/停气交替运行如何去除氨氮?处理效果如何?", "language": "zh", "topics": ["water_quality", "farming_system"], "expected_evidence": {"must_entities": ["biofilter", "recirculating aquaculture system", "ammonia"], "must_relations": ["removes_ammonia", "operated_under_aeration_cycles", "treatment_efficiency"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Treatment of simulated aquaculture water using biofilters subjected to aeration/non-aeration cycles(1998)"} +{"id": "aqua-g08", "question": "What is the relationship among pH, dissolved oxygen and chlorophyll-a in aquaculture waters compared with non-aquaculture waters?", "language": "en", "topics": ["water_quality"], "expected_evidence": {"must_entities": ["pH", "dissolved oxygen", "chlorophyll-a", "aquaculture pond"], "must_relations": ["correlates_with", "differs_between_waters"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Comparison of Relationships Between pH, Dissolved Oxygen and Chlorophyll a for Aquaculture and Non-aquaculture Waters(2010)"} +{"id": "aqua-g09", "question": "低氧胁迫如何改变金鲳(Trachinotus blochii)的葡萄糖与脂质代谢?", "language": "zh", "topics": ["water_quality"], "expected_evidence": {"must_entities": ["Trachinotus blochii", "hypoxia", "glucose metabolism", "lipid metabolism"], "must_relations": ["alters_metabolism", "induces_hypoxia_response"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Hypoxia alters glucose and lipid metabolisms in golden pompano(Trachinotus blochii)(2023)"} +{"id": "aqua-g10", "question": "How does stocking density affect water quality, harmful nitrogen control and production performance of Penaeus vannamei in biofloc-based systems with limited water exchange?", "language": "en", "topics": ["water_quality", "farming_system"], "expected_evidence": {"must_entities": ["Penaeus vannamei", "biofloc", "stocking density", "nitrogen"], "must_relations": ["affects_water_quality", "controls_harmful_nitrogen", "affects_production"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Effect of Stocking Density on Water Quality, Harmful Nitrogen Control, and Production Performance of Penaeus vannamei in Biofloc-Based Systems(2025)"} +{"id": "aqua-g11", "question": "用能值(emergy)综合方法评估水产养殖可持续性时,为什么水核算至关重要?氮磷负荷如何计入评估?", "language": "zh", "topics": ["water_quality", "ecology"], "expected_evidence": {"must_entities": ["emergy synthesis", "water accounting", "sustainability"], "must_relations": ["assesses_sustainability", "accounts_for_water", "quantifies_nitrogen_phosphorus"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:Aquaculture Sustainability Assessed by Emergy Synthesis: The Importance of Water Accounting(2022)"} +{"id": "aqua-g12", "question": "鱼粉与鱼油联合替代(植物蛋白/植物油)对欧洲鲈(Dicentrarchus labrax)生长性能、肉质与货架期有什么影响?", "language": "zh", "topics": ["feed_nutrition", "growth"], "expected_evidence": {"must_entities": ["Dicentrarchus labrax", "fishmeal replacement", "fish oil replacement", "flesh quality"], "must_relations": ["substitutes_fishmeal", "affects_flesh_quality", "affects_shelf_life"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Effect of combined fishmeal and fish oil replacement on growth performance, flesh quality and shelf life of European sea bass(2022)"} +{"id": "aqua-g13", "question": "What dietary protein-to-energy ratio maximizes growth and body composition of juvenile white shrimp (Litopenaeus vannamei)?", "language": "en", "topics": ["feed_nutrition", "growth"], "expected_evidence": {"must_entities": ["Litopenaeus vannamei", "protein-to-energy ratio"], "must_relations": ["optimizes_growth", "affects_body_composition"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Growth and body composition of juvenile white shrimp Litopenaeus vannamei fed different ratios of dietary protein to energy(2008)"} +{"id": "aqua-g14", "question": "完全用黄粉虫(Tenebrio molitor)替代豆粕,对草鱼(Ctenopharyngodon idella)肌肉品质有什么改善?", "language": "zh", "topics": ["feed_nutrition"], "expected_evidence": {"must_entities": ["Tenebrio molitor", "Ctenopharyngodon idella", "soybean meal"], "must_relations": ["replaces_soybean_meal", "improves_flesh_quality"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Improving flesh quality of grass carp by completely replacing dietary soybean meal with yellow mealworm(2022)"} +{"id": "aqua-g15", "question": "How does dietary betaine attenuate oxidative stress, endoplasmic reticulum stress and apoptosis induced by high-carbohydrate diets in mandarin fish (Siniperca chuatsi)?", "language": "en", "topics": ["feed_nutrition"], "expected_evidence": {"must_entities": ["betaine", "Siniperca chuatsi", "high-carbohydrate diet", "oxidative stress"], "must_relations": ["attenuates_oxidative_stress", "alleviates_er_stress", "reduces_apoptosis"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Dietary Betaine Attenuates High-Carbohydrate-Diet-Induced Oxidative Stress, ER Stress, and Apoptosis in Mandarin Fish(2023, paired)"} +{"id": "aqua-g16", "question": "饲料中添加虾青素,能否提高凡纳滨对虾(Litopenaeus vannamei)的低氧与氨氮耐受能力并改善体色?", "language": "zh", "topics": ["feed_nutrition", "water_quality"], "expected_evidence": {"must_entities": ["astaxanthin", "Litopenaeus vannamei", "hypoxia tolerance", "ammonia tolerance"], "must_relations": ["enhances_tolerance", "improves_coloration"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:Dietary supplementation of astaxanthin increased growth, colouration, the capacity of hypoxia and ammonia tolerance of Pacific white shrimp(2022)"} +{"id": "aqua-g17", "question": "不同投喂水平如何影响三倍体大菱鲆(Scophthalmus maximus)幼鱼的生长性能、饲料利用与能量收支?", "language": "zh", "topics": ["growth", "feed_nutrition"], "expected_evidence": {"must_entities": ["Scophthalmus maximus", "triploid turbot", "ration level", "energy budget"], "must_relations": ["affects_growth_performance", "affects_feed_utilization", "alters_energy_budget"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Effects of ration level on growth performance, feed utilization and energy budget of juvenile triploid turbot(2026)"} +{"id": "aqua-g18", "question": "What is the optimal stocking density for juvenile Atlantic cod (Gadus morhua L.) reared in a land-based farm?", "language": "en", "topics": ["growth", "farming_system"], "expected_evidence": {"must_entities": ["Gadus morhua", "stocking density", "land-based farm"], "must_relations": ["optimizes_stocking_density", "affects_growth"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Optimal stocking density of juvenile Atlantic cod(Gadus morhua L.)reared in a land-based farm(2012)"} +{"id": "aqua-g19", "question": "挤压饲料与颗粒饲料在不同蛋白水平下,对大嘴黑鲈(Micropterus salmoides)生长性能与养分存留的影响有何差异?", "language": "zh", "topics": ["growth", "feed_nutrition"], "expected_evidence": {"must_entities": ["Micropterus salmoides", "extruded diet", "pelleted diet", "protein level"], "must_relations": ["compares_diet_forms", "affects_nutrient_retention"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Effects of extruded and pelleted diets with different protein levels on growth performance and nutrient retention of largemouth bass(2023)"} +{"id": "aqua-g20", "question": "大西洋鳕(Gadus morhua)育种计划中,生长、性成熟等重要经济性状的遗传参数(遗传力、遗传相关)是多少?", "language": "zh", "topics": ["genetics_breeding"], "expected_evidence": {"must_entities": ["Gadus morhua", "heritability", "genetic correlation", "breeding program"], "must_relations": ["estimates_genetic_parameters", "breeds_for_trait"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Genetic parameters for important traits in the breeding program for Atlantic cod(2007)"} +{"id": "aqua-g21", "question": "Which SNPs and candidate genes are associated with growth and body shape in bighead carp (Hypophthalmichthys nobilis) according to GWAS?", "language": "en", "topics": ["genetics_breeding"], "expected_evidence": {"must_entities": ["Hypophthalmichthys nobilis", "SNP", "GWAS", "candidate gene"], "must_relations": ["associated_with_trait", "identified_by_gwas"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Genome-Wide Association Study Reveals SNPs and Candidate Genes Related to Growth and Body Shape in Bighead Carp(2022)"} +{"id": "aqua-g22", "question": "斑点海鲈(Lateolabrax maculatus)生长相关性状的 QTL 精细定位,使用了什么类型的高密度连锁图谱?定位到哪些区间?", "language": "zh", "topics": ["genetics_breeding", "growth"], "expected_evidence": {"must_entities": ["Lateolabrax maculatus", "QTL", "high-density linkage map"], "must_relations": ["mapped_by_linkage_map", "linked_to_growth_trait"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:First High-Density Linkage Map and QTL Fine Mapping for Growth-Related Traits of Spotted Sea bass(2020)"} +{"id": "aqua-g23", "question": "How is genomic selection implemented in family-based aquaculture breeding schemes, and what role do doubled haploids play?", "language": "en", "topics": ["genetics_breeding"], "expected_evidence": {"must_entities": ["genomic selection", "family-based breeding", "doubled haploid"], "must_relations": ["implemented_in_breeding_scheme", "accelerates_genetic_gain"]}, "difficulty": "medium", "answerable_from_corpus": "maybe", "notes": "锚定:Strategies for implementing genomic selection in family-based aquaculture breeding schemes(2012, pdf_only 仅标题)"} +{"id": "aqua-g24", "question": "香鱼(Plecoglossus altivelis)的性别决定基因是什么?Y 染色体连锁的抗缪勒氏管激素 II 型受体如何参与性别决定?", "language": "zh", "topics": ["genetics_breeding", "reproduction"], "expected_evidence": {"must_entities": ["Plecoglossus altivelis", "Y chromosome", "anti-Müllerian hormone type-II receptor"], "must_relations": ["determines_sex", "located_on_y_chromosome"]}, "difficulty": "hard", "answerable_from_corpus": "maybe", "notes": "锚定:A Y-linked anti-Müllerian hormone type-II receptor is the sex-determining gene in ayu(2021, pdf_only 仅标题)"} +{"id": "aqua-g25", "question": "如何通过光温周期调控,诱导白梭吻鲈(Sander lucioperca)反季节产卵?", "language": "zh", "topics": ["reproduction"], "expected_evidence": {"must_entities": ["Sander lucioperca", "out-of-season spawning", "photothermal manipulation"], "must_relations": ["induces_spawning", "regulates_reproduction"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Induction of Out-of-Season Spawning of Pikeperch, Sander Lucioperca(2004)"} +{"id": "aqua-g26", "question": "What are the key reproductive biology traits (fecundity, spawning frequency, egg size) of pond-reared Nile tilapia (Oreochromis niloticus)?", "language": "en", "topics": ["reproduction"], "expected_evidence": {"must_entities": ["Oreochromis niloticus", "fecundity", "spawning frequency"], "must_relations": ["characterizes_reproduction", "measures_fecundity"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Reproductive biology of pond reared Nile tilapia(1999)"} +{"id": "aqua-g27", "question": "轮虫营养强化(如富集 DHA/EPA)对海水鱼类仔鱼的存活与发育有什么意义?", "language": "zh", "topics": ["reproduction", "feed_nutrition"], "expected_evidence": {"must_entities": ["rotifer", "DHA", "EPA", "marine fish larvae"], "must_relations": ["enriches_nutrient", "improves_larval_survival"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:The enrichment of rotifers(2015)+Microparticle types for delivering nutrients to marine fish larvae(2003)"} +{"id": "aqua-g28", "question": "How do seaweed-enriched broodstock diets improve post-settlement survival in commercial sea urchin aquaculture?", "language": "en", "topics": ["reproduction", "feed_nutrition"], "expected_evidence": {"must_entities": ["seaweed-enriched diet", "sea urchin", "broodstock"], "must_relations": ["improves_post_settlement_survival", "feeds_broodstock"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Seaweed-enriched formulated broodstock diet improves post-settlement survival of a commercial sea urchin(2026)"} +{"id": "aqua-g29", "question": "综合多营养级养殖(IMTA)在水族箱尺度下如何优化生产力、生态平衡与环境可持续性?", "language": "zh", "topics": ["farming_system", "ecology"], "expected_evidence": {"must_entities": ["IMTA", "productivity", "ecosystem balance"], "must_relations": ["optimizes_productivity", "balances_ecosystem"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:OPTIMIZATION OF INTEGRATED MULTI-TROPHIC AQUACULTURE(IMTA)CULTIVATION IN AQUARIA: A REVIEW(2025)"} +{"id": "aqua-g30", "question": "What design considerations apply to open-design recirculating systems for zebrafish (Danio rerio) culture?", "language": "en", "topics": ["farming_system"], "expected_evidence": {"must_entities": ["zebrafish", "Danio rerio", "open-design recirculating system"], "must_relations": ["designs_system", "supports_culture"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Open-design Recirculating Systems for Zebrafish Culture(2018)"} +{"id": "aqua-g31", "question": "基于㶲(exergy)分析的循环水养殖系统能效如何评价?主要㶲损失出现在哪些环节?", "language": "zh", "topics": ["farming_system"], "expected_evidence": {"must_entities": ["exergy analysis", "recirculating aquaculture system", "energy efficiency"], "must_relations": ["analyzes_efficiency", "identifies_exergy_losses"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Exergetic performance analysis of a recirculating aquaculture system(2010)"} +{"id": "aqua-g32", "question": "How does a calcined FeSx-packed denitrification bioreactor enhance nitrogen removal in recirculating aquaculture systems, and what removal efficiency is achieved?", "language": "en", "topics": ["farming_system", "water_quality"], "expected_evidence": {"must_entities": ["FeSx-packed bioreactor", "denitrification", "recirculating aquaculture system"], "must_relations": ["enhances_denitrification", "removes_nitrogen"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Enhanced Nitrogen Removal from a Recirculating Aquaculture System Using a Calcined FeSx-Packed Denitrification Bioreactor(2024)"} +{"id": "aqua-g33", "question": "海洋酸化对大型海藻的生长、生化组成及固碳(CO2 缓解)潜力有什么全球性影响?", "language": "zh", "topics": ["ecology"], "expected_evidence": {"must_entities": ["seaweed", "ocean acidification", "CO2 mitigation"], "must_relations": ["affects_growth", "alters_biochemical_composition"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Seaweed responses to ocean acidification: global impacts on growth, biochemical composition, and CO2 mitigation potential(2025)"} +{"id": "aqua-g34", "question": "Does eco-certification change public opinion of salmon aquaculture in Canadian communities with and without salmon farms?", "language": "en", "topics": ["ecology", "farming_system"], "expected_evidence": {"must_entities": ["eco-certification", "salmon aquaculture", "Canada"], "must_relations": ["influences_public_opinion", "compares_communities"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Does eco-certification change public opinion of salmon aquaculture in Canada?(2023)"} +{"id": "aqua-g35", "question": "杂交化如何成为伊比利亚半岛淡水鱼类区系的威胁?涉及哪些物种与机制?", "language": "zh", "topics": ["ecology", "genetics_breeding"], "expected_evidence": {"must_entities": ["hybridization", "Iberian freshwater fish"], "must_relations": ["threatens_ichthyofauna", "causes_introgression"]}, "difficulty": "easy", "answerable_from_corpus": "yes", "notes": "锚定:Hybridization as an Emerging Threat to Iberian Freshwater Ichthyofauna(2026)"} +{"id": "aqua-g36", "question": "What do mitochondrial genomes reveal about mid-Pleistocene population divergence and post-glacial expansion of Australian freshwater fishes?", "language": "en", "topics": ["ecology", "genetics_breeding"], "expected_evidence": {"must_entities": ["mitochondrial genome", "Australian freshwater fish", "Pleistocene"], "must_relations": ["reveals_population_divergence", "traces_post_glacial_expansion"]}, "difficulty": "medium", "answerable_from_corpus": "maybe", "notes": "锚定:Mitochondrial genomes reveal mid-Pleistocene population divergence, and post-glacial expansion, in Austra...(2022, pdf_only 标题截断)"} +{"id": "aqua-g37", "question": "水产饲料的真菌污染与霉菌毒素对养殖鱼类组织有什么危害?有哪些生物防控手段?", "language": "zh", "topics": ["feed_nutrition", "disease"], "expected_evidence": {"must_entities": ["mycotoxin", "aquafeed", "biological control"], "must_relations": ["contaminates_feed", "damages_fish_tissue", "controlled_by_biocontrol"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:Fungal contamination and mycotoxins in aquafeed and tissues of aquaculture fishes and their biological control(2023)"} +{"id": "aqua-g38", "question": "How does water quality influence bacterial populations and off-flavour formation in recirculating aquaculture systems?", "language": "en", "topics": ["water_quality", "farming_system"], "expected_evidence": {"must_entities": ["water quality", "bacterial population", "off-flavour", "RAS"], "must_relations": ["influences_bacterial_population", "causes_off_flavour"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:Impact of water quality on the bacterial populations and off-flavours in recirculating aquaculture systems(2012, paired)"} +{"id": "aqua-g39", "question": "铅暴露如何通过氧化应激、炎症反应与细胞凋亡,损伤黄颡鱼(Pelteobagrus fulvidraco)的免疫功能与组织完整性?", "language": "zh", "topics": ["ecology", "disease"], "expected_evidence": {"must_entities": ["lead", "Pelteobagrus fulvidraco", "oxidative stress", "apoptosis"], "must_relations": ["impairs_immune_function", "mediates_oxidative_stress", "damages_tissue"]}, "difficulty": "hard", "answerable_from_corpus": "yes", "notes": "锚定:Lead impaired immune function and tissue integrity in yellow catfish(Peltobargus fulvidraco)(2021)"} +{"id": "aqua-g40", "question": "What molecular mechanisms underlie Vibrio tolerance in Manila clam (Ruditapes philippinarum) revealed by comparative transcriptome profiling?", "language": "en", "topics": ["genetics_breeding", "disease"], "expected_evidence": {"must_entities": ["Ruditapes philippinarum", "Vibrio", "differentially expressed genes"], "must_relations": ["confers_tolerance", "differentially_expressed_under_challenge"]}, "difficulty": "medium", "answerable_from_corpus": "yes", "notes": "锚定:Molecular Mechanisms Underlying Vibrio Tolerance in Ruditapes philippinarum Revealed by Comparative Transcriptome Profiling(2022)"} diff --git a/kb-libraries/aqua/qa/golden_aqua_README.md b/kb-libraries/aqua/qa/golden_aqua_README.md new file mode 100644 index 0000000..2b413aa --- /dev/null +++ b/kb-libraries/aqua/qa/golden_aqua_README.md @@ -0,0 +1,39 @@ +# golden_aqua 金标问答集使用说明 + +**文件**: `kb-libraries/aqua/qa/golden_aqua.jsonl` +**版本**: v1(2026-09-12,T 代理编制;P1 试点实测后按评分结果修订) +**规模**: 41 问(zh 24 / en 17;easy 17 / medium 16 / hard 8;answerable: yes 36 / maybe 4 / no 1) + +## 编制依据 + +- 基于 `kb-libraries/aqua/build/manifest.csv`(41,620 行)seed=42 抽样 200 篇通读标题归纳主题后出题;每题 notes 字段标注锚定文献(年份、manifest 状态)。 +- 问题口径 = 水产从业人员真实会问的检索问题;expected_evidence 写的是「合格答案应引用的实体/关系类型」,不写具体数值,用于评分而非对答案的绝对判据。 + +## 字段说明 + +| 字段 | 含义 | +|---|---| +| id | aqua-g01..g41 | +| question / language | 问题文本与语言(zh/en) | +| topics | 主题标签(可多值):disease / water_quality / feed_nutrition / growth / genetics_breeding / reproduction / farming_system / ecology | +| expected_evidence.must_entities | 合格答案必须出现的实体(品种、病原、指标、物质) | +| expected_evidence.must_relations | 合格答案必须表达的关系类型(建议与 aqua relation_taxonomy 的 relation_category 对齐;P1 词表定稿后回填 category 名) | +| difficulty | easy / medium / hard(4:4:2) | +| answerable_from_corpus | yes=摘要卡明确可答;maybe=仅 PDF 标题锚定、需全文抽取后确认;no=负向对照题(库外问题) | +| notes | 锚定文献与出题备注 | + +## 怎么用金标打分(P1 试点库建好后) + +1. **检索命中率(系统级)**: 对每问跑 vstore/检索,取 top-K(K=10)chunks。统计: + - 证据命中率 = 命中 expected_evidence 中任一 must_entities 的题数 / 总题数;每问再算 must_entities 命中比例,求均值。 + - 分级:每问命中 ≥50% must_entities 记为「部分命中」,≥80% 记为「完全命中」。目标(P1 试点):部分命中率 ≥70%,完全命中率 ≥50%。 + - `answerable_from_corpus=no` 的题(g41)单独看:期望 top-K 中无强证据 chunk(即系统能「诚实地说不知道」)。 +2. **KG 覆盖(图谱级)**: 对每问在 Neo4j 上按 must_entities 做实体/关系查询,统计 must_relations 中能被现有 relation_category 表达的占比。目标:P1 试点 ≥60%;该比例同时暴露词表缺口 → 反哺 aqua 词表扩充(当前 `configs/entity_alias.yaml` 仍是植物占位词表,见 qc_standards.md 风险节)。 +3. **答案质量(端到端)**: 抽取 10 问(中英各半)走完整 QA 流程,人工按「答案是否引用 expected evidence 的实体与关系」0/1/2 打分:2=全部引用且正确,1=部分引用,0=未引用或错误。金标集整体 ≥1.5 分/问视为通过。 +4. **回归用途**: 词表、分块、Prompt、模型任一项变更后,重跑第 1、2 步,对比命中率变化(±5 个百分点内视为无回归)。 + +## 注意事项 + +- maybe/no 题在自动评分中不计入「未命中即失败」,只作诊断信息。 +- 金标集是「证据类型」金标,不是「标准答案」金标;若 P2 需要做 LLM-as-judge 的精确判分,再补充 gold_answer 字段(当前刻意不写,避免污染人工判分)。 +- 修订流程:任何修订须更新本条 README 的版本行,并在文件内保留 id 稳定(id 一旦发布不得重用)。 diff --git a/kb-libraries/aqua/qa/qc_standards.md b/kb-libraries/aqua/qa/qc_standards.md new file mode 100644 index 0000000..bd97585 --- /dev/null +++ b/kb-libraries/aqua/qa/qc_standards.md @@ -0,0 +1,51 @@ +# 水产库(aqua)试点质检标准 + +**版本**: v1(2026-09-12,T 代理编制;P1 试点实测后按首批 result.json 校准阈值) +**适用对象**: P1 试点批次(500 篇:30 篇 PDF 全文 MinerU 抽取 + 470 篇 MD 摘要卡)的 result.json 批量质检;金标问答集配套文件见 golden_aqua_README.md。 + +## 1. 质检指标与阈值 + +| # | 指标 | 计算方式 | 阈值(目标) | 备注 | +|---|---|---|---|---| +| 1.1 | 断言密度 | 每篇 relations 条数的分位数分布 | P10 ≥ 3,P50 ∈ 8–15,P90 ≤ 30 | 按植物库经验:摘要卡短文本预计 3–15 条/篇,PDF 全文偏高;给出区间而非单一阈值,P90 上限防过度抽取噪声 | +| 1.2 | 实体类型合规率 | relation 的 source_type/target_type 落在 aqua 词表实体类型集合内的比例 | 100% | 每批抽 20 篇核算;P1 阶段词表未定稿时,以「类型名来自 entity_alias.yaml 定义」为基准 | +| 1.3 | 关系合规率 | relation_category 落在 aqua relation_taxonomy 的 category 集合内 | 100% | 同上,抽 20 篇核算 | +| 1.4 | evidence 非空率 | 带非空 evidence 字段的 relation 占全部 relation 的比例 | ≥ 95% | evidence 是 KG 断言可回溯性的根基 | +| 1.5 | output_layer 分布 | main_kg 层 relation 占比 | 50%–100%,P1 预期 ≥90%(实测试点 34%,根因=other_supported 兜底 30% 被 infer_layer 打入 candidate_review;若分类引导改善后仍<50% 则逐篇复查) | 与 normalized 容器及主项目取数口径一致(植物现网 main 77%);偏低的篇标记复查 | +| 1.6 | 抽取失败率 | failed 状态篇数 / 批次总篇数 | < 5% | 失败篇进入重试队列,重试 1 次后仍失败按失败计 | +| 1.7 | 空结果篇占比 | relations < 2 条的篇数 / 批次总篇数 | < 10% | MD 卡中「No Abstract」卡片是主因,单独统计并豁免,但须记录篇目 | + +补充说明: +- 实体类型集合与 relation_category 集合以 `kb-libraries/aqua/configs/entity_alias.yaml`、`relation_taxonomy.yaml` 为准(P1 定稿后锁定版本号)。 +- 各项均为「目标值」,首批实测后由 T 代理按实际分布校准一次,校准记录写在本文档「版本修订」节。 + +## 2. 抽样与核对流程 + +每批 500 篇按 seed=42 抽 20 篇;每篇抽 5 条断言(取该篇 relations 的第 1、⌈n/4⌉、⌈n/2⌉、⌈3n/4⌉、n 条,不足 5 条全查)人工/半自动核对: + +1. **evidence 与原文对应**: evidence 引文能在源文档(MD 卡或 MinerU 输出 markdown)中找到,且语义一致; +2. **实体边界**: source/target 实体名边界正确(不多截、不少截;缩写/全称归一一致); +3. **关系方向**: source→target 方向正确(如「病原→感染→宿主」而非反了);relation_category 与断言语义匹配。 + +记录表模板(建议落 `kb-libraries/aqua/qa/qc_batch_.csv`,首行表头): + +| batch_id | doc_id | assertion_id | evidence_ok(1/0) | entity_boundary_ok(1/0) | direction_ok(1/0) | category_ok(1/0) | reviewer | date | comment | + +判定: 单批抽检 20×5=100 条断言中,evidence_ok、direction_ok、category_ok 三项通过率均 ≥ 90% 视为批次放行;任何一项 < 80% 整批返工对应环节。 + +## 3. 金标评分流程(建库后) + +P1 试点库建好后,用 `golden_aqua.jsonl` 跑 10 问(中英各半,建议 g02、g08、g13、g18、g21、g26、g30、g34、g35、g38)做检索质量初评: +- 每问取 top-10 chunks,统计 expected_evidence.must_entities 命中比例; +- 分级: 部分命中(≥50%)、完全命中(≥80%);记录 g41(负向对照)是否被正确「拒答」; +- 达标线见 golden_aqua_README.md(部分命中率 ≥70%,完全命中率 ≥50%);未达标时先排查词表 alias 与分块策略,再评估重抽。 + +## 4. 风险与依赖 + +- **词表缺口(当前最大风险)**: `configs/entity_alias.yaml` 与 `relation_taxonomy.yaml` 目前仍是植物库占位内容(拟南芥/水稻/大豆/GmP450 等),没有任何水产实体与关系。P1 抽取阶段若沿用该词表,1.2/1.3 两项合规率将大面积失败或抽取被迫降级为开放类型。**T 依赖 D1 交付 aqua 词表(或至少 P1 试验版)后才能执行 1.2/1.3**;在此之前用开放类型抽取的批次,合规率指标挂起(pending),记录在案。 +- **语料语言**: 全量标题中英文占比 ≈ 99.9%(仅 40 篇标题含 ≥2 个 CJK 字符),MD 卡 CJK 字符占比约 4%。中文检索质量会显著低于英文,金标中文题更多检验 cross-lingual 检索能力。 +- **答案类型偏向**: 主题分布中水质环境类占比最高(单一分类口径 24%),金标集主题权重已按此加权。 + +## 版本修订 + +- v1(2026-09-12): 初版,阈值基于植物库经验 + 全量主题统计;待 P1 首批 result.json 校准。 diff --git a/tooling/aqua_kb/batch_runner.py b/tooling/aqua_kb/batch_runner.py new file mode 100644 index 0000000..b16205c --- /dev/null +++ b/tooling/aqua_kb/batch_runner.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""tooling/aqua_kb/batch_runner.py —— 分批建库调度器 + +按 manifest 编排建库全链，全程断点续跑： + for each paper in manifest: + result.json 已存在 → 跳过抽取（外部/D2 负责抽取，本工具只调度后续步骤） + → phase2b_converter.convert()（neo4j.cypher + qdrant_points.jsonl） + → 攒批（--convert-batch 篇）后： + Neo4j：逐文件 cypher-shell 执行（--neo4j-shell 路径），成功记 checkpoint + Qdrant：qdrant_loader.load_files()（embed + upsert），成功记 checkpoint + → 失败写 build/failed.jsonl（含错误），全部步骤可重跑 + +状态文件（/state/，JSONL 每行一个 paper_id）： + converted.jsonl / neo4j_loaded.jsonl / qdrant_loaded.jsonl / failed.jsonl + +用法： + python batch_runner.py --manifest pilot_manifest.json --result-root \ + --out --qdrant http://127.0.0.1:27435 \ + --collection aqua_srt_v1_20260930_normalized \ + --neo4j-shell /opt/neo4j-community-5.15.0/bin/cypher-shell \ + --neo4j-uri bolt://127.0.0.1:19889 --neo4j-user neo4j \ + --neo4j-password-file kb-libraries/aqua/data/neo4j/.auth \ + --embed-url http://127.0.0.1:11434/api/embed --embed-model bge-m3:567m \ + --convert-batch 50 [--limit 20] +""" +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +import phase2b_converter +import qdrant_loader + + +def load_ids(path: Path) -> set: + ids: set = set() + if path.exists(): + with open(path, "r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + ids.add(json.loads(line)) # 兼容 json.dumps 写入的带引号行 + except json.JSONDecodeError: + ids.add(line) + return ids + + +def append_id(path: Path, value: str) -> None: + with open(path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(value) + "\n") + + +class Runner: + def __init__(self, args: argparse.Namespace) -> None: + self.args = args + self.out = Path(args.out) + self.state_dir = self.out / "state" + self.state_dir.mkdir(parents=True, exist_ok=True) + self.converted_path = self.state_dir / "converted.jsonl" + self.neo4j_loaded_path = self.state_dir / "neo4j_loaded.jsonl" + self.qdrant_loaded_path = self.state_dir / "qdrant_loaded.jsonl" + self.failed_path = self.out / "failed.jsonl" + self.converted = load_ids(self.converted_path) + self.neo4j_loaded = load_ids(self.neo4j_loaded_path) + self.qdrant_loaded = load_ids(self.qdrant_loaded_path) + self._embed: Optional[qdrant_loader.EmbedClient] = None + + @property + def embed(self) -> qdrant_loader.EmbedClient: + if self._embed is None: + self._embed = qdrant_loader.EmbedClient( + self.args.embed_url, self.args.embed_model + ) + return self._embed + + # ---------- manifest ---------- + + def read_manifest(self) -> List[Dict[str, Any]]: + with open(self.args.manifest, "r", encoding="utf-8") as handle: + manifest = json.load(handle) + entries = manifest.get("papers") if isinstance(manifest, dict) else manifest + if not isinstance(entries, list): + raise RuntimeError("manifest must be a dict with 'papers' list or a list") + if self.args.limit: + entries = entries[: self.args.limit] + return entries + + @staticmethod + def paper_key(entry: Dict[str, Any]) -> str: + return str(entry.get("paper_id") or entry.get("id") or entry.get("doi") + or entry.get("result") or json.dumps(entry, sort_keys=True)) + + def resolve_result(self, entry: Dict[str, Any]) -> Path: + if entry.get("result"): + path = Path(entry["result"]) + if not path.is_absolute(): + path = Path(self.args.result_root) / path + return path + key = self.paper_key(entry) + return Path(self.args.result_root) / key / "result.json" + + # ---------- steps ---------- + + def step_convert(self, key: str, result_path: Path, paper_json: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """转换幂等（产物确定性），checkpoint 只写一次。返回产物路径 stats。""" + stats = phase2b_converter.convert( + result_path, self.out, self.args.release_id, paper_json + ) + if key not in self.converted: + append_id(self.converted_path, key) + print(f"[runner] converted {key}: {stats['assertion_count']} assertions") + return stats + + def step_neo4j(self, key: str, cypher_path: Path) -> None: + if key in self.neo4j_loaded: + print(f"[runner] neo4j skip {key} (checkpoint)") + return + shell = self.args.neo4j_shell + auth = [f"--format=plain", f"--address={self.args.neo4j_uri}", + f"--username={self.args.neo4j_user}"] + with open(self.args.neo4j_password_file, "r", encoding="utf-8") as handle: + password = handle.read().strip().splitlines()[0] + command = [shell, *auth, f"--password={password}", "-f", str(cypher_path)] + proc = subprocess.run(command, capture_output=True, text=True, timeout=300) + if proc.returncode != 0: + raise RuntimeError(f"cypher-shell failed for {key}: {proc.stderr[-2000:]}") + append_id(self.neo4j_loaded_path, key) + print(f"[runner] neo4j loaded {key}") + + def step_qdrant(self, key: str, points_path: Path) -> None: + if key in self.qdrant_loaded: + print(f"[runner] qdrant skip {key} (checkpoint)") + return + embed = self.embed + probe = embed.embed("aquaculture warm-up probe") + qdrant_loader.ensure_collection( + self.args.qdrant, self.args.collection, len(probe) + ) + qdrant_loader.upsert_file( + points_path, + self.args.qdrant, + self.args.collection, + self.args.batch, + checkpoint=self.state_dir / "qdrant_checkpoint.jsonl", + embed=embed, + ) + append_id(self.qdrant_loaded_path, key) + print(f"[runner] qdrant loaded {key}") + + # ---------- main ---------- + + def run(self) -> None: + entries = self.read_manifest() + started = time.time() + stats = {"total": len(entries), "converted": 0, "neo4j": 0, "qdrant": 0, + "failed": 0} + for index, entry in enumerate(entries, 1): + key = self.paper_key(entry) + if key in self.failed_keys: + continue + paper_json = entry.get("paper") if isinstance(entry.get("paper"), dict) else None + try: + result_path = self.resolve_result(entry) + if not result_path.exists(): + print(f"[runner] SKIP {key}: result.json missing " + f"({result_path}) — 抽取由外部完成") + continue + conv = self.step_convert(key, result_path, paper_json) + stats["converted"] += 1 + cypher_path = Path(conv["cypher"]) + points_path = Path(conv["points"]) + self.step_neo4j(key, cypher_path) + stats["neo4j"] += 1 + self.step_qdrant(key, points_path) + stats["qdrant"] += 1 + if index % 10 == 0: + print(f"[runner] progress {index}/{stats['total']} " + f"({time.time() - started:.0f}s)") + except Exception as exc: + stats["failed"] += 1 + self.record_failure(key, str(exc)) + print(f"[runner] FAILED {key}: {exc}", file=sys.stderr) + print(f"[runner] DONE {json.dumps(stats, ensure_ascii=False)} in " + f"{time.time() - started:.1f}s") + + @property + def failed_keys(self) -> set: + keys: set = set() + if self.failed_path.exists(): + with open(self.failed_path, "r", encoding="utf-8") as handle: + for line in handle: + try: + keys.add(json.loads(line)["paper_key"]) + except Exception: + continue + return keys + + def record_failure(self, key: str, error: str) -> None: + with open(self.failed_path, "a", encoding="utf-8") as handle: + handle.write(json.dumps( + {"paper_key": key, "error": error, + "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z")}, + ensure_ascii=False, + ) + "\n") + + +def main() -> None: + parser = argparse.ArgumentParser(description="aqua KB batch runner") + parser.add_argument("--manifest", required=True) + parser.add_argument("--result-root", required=True) + parser.add_argument("--out", required=True, help="build output root") + parser.add_argument("--release-id", default="aqua_srt_v1_20260930_normalized") + parser.add_argument("--qdrant", required=True) + parser.add_argument("--collection", default="aqua_srt_v1_20260930_normalized") + parser.add_argument("--batch", type=int, default=200) + parser.add_argument("--neo4j-shell", default="cypher-shell") + parser.add_argument("--neo4j-uri", default="bolt://127.0.0.1:19889") + parser.add_argument("--neo4j-user", default="neo4j") + parser.add_argument("--neo4j-password-file", required=True) + parser.add_argument("--embed-url", default="http://127.0.0.1:11434/api/embed") + parser.add_argument("--embed-model", default="bge-m3:567m") + parser.add_argument("--convert-batch", type=int, default=50, + help="batch size for Neo4j/Qdrant loading (checkpoint per paper)") + parser.add_argument("--limit", type=int, default=None, + help="only process first N manifest entries") + args = parser.parse_args() + Runner(args).run() + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/cost_model.py b/tooling/aqua_kb/cost_model.py new file mode 100644 index 0000000..000858b --- /dev/null +++ b/tooling/aqua_kb/cost_model.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""cost_model.py — 水产库(aqua)建库成本校准模型(T 代理,2026-09-12)。 + +参数化估算 LLM 抽取成本与 MinerU GPU 耗时,输出 Markdown 预算表。 +只依赖标准库;所有数值可被命令行覆盖;默认值来源见 +kb-libraries/aqua/qa/budget_template.md 假设清单。 + +用法: + $PY tooling/aqua_kb/cost_model.py # 默认参数,打印预算表 + $PY tooling/aqua_kb/cost_model.py --json # 机器可读 + $PY tooling/aqua_kb/cost_model.py --out budget.md # 同时写入文件 + $PY tooling/aqua_kb/cost_model.py --price-cny-m-input 1.0 --price-cny-m-output 2.0 \ + --md-chars 2109 --pdf-chars 60000 --n-md 31848 --n-pdf 13175 + +成本公式(人民币): + in_tokens = chars / chars_per_token + overhead_tokens * calls + out_tokens = (chars / chars_per_token) * out_in_ratio + cost = (in_tokens * p_in + out_tokens * p_out) / 1e6 * docs +其中 p_in/p_out 为每百万 token 单价。 +""" +from __future__ import annotations + +import argparse +import json +import sys +from dataclasses import dataclass + + +@dataclass +class DocSpec: + label: str + n_docs: int + chars_per_doc: float + llm_calls_per_doc: float + chunks_per_doc: float | None = None + cost: float = 0.0 + in_tokens: int = 0 + out_tokens: int = 0 + + +@dataclass +class Params: + # 单价(¥/百万 token)——官方 deepseek-v4-flash 列表价(缓存未命中口径) + price_cny_m_input: float = 1.0 + price_cny_m_output: float = 2.0 + # 字符/token 折算 + chars_per_token: float = 4.0 # 英文约 4 字符/token(aqua 语料英文占主导) + cjk_chars_per_token: float = 1.5 # 中文约 1.5 字符/token(当前语料占比 <5%) + # LLM 调用假设 + overhead_tokens_per_call: float = 350.0 # 每调用的提示词/模板开销 + out_in_ratio: float = 0.35 # 输出 token / 输入正文 token(P1 试点拟合前先估值) + md_llm_calls: float = 2.0 # MD 卡:提取 + 复核 各 1 次 + pdf_chunk_chars: float = 8000.0 # PDF 全文分块大小 + pdf_output_per_call: float = 800.0 # PDF 每块调用输出 token(P1 实测前估值) + # 语料规模 + n_md: int = 31848 + n_pdf: int = 13175 + n_pilot_pdf: int = 30 + n_pilot_md: int = 470 + # 实测/历史锚点 + md_chars: float = 2109.0 # 实测:source/题目摘要_MD 200 篇抽样 mean(P50=2169) + pdf_chars: float = 60000.0 # 假设:MinerU 输出正文均值(P1 30 篇试点后回填实测) + mineru_files_per_hour: float = 700.0 # 历史实测:单卡 3 端点并流 ~9,363 篇/11h ≈ 850/h,取保守 700 + mineru_unprocessable: float = 0.177 # 历史实测:1656/9363;不可处理篇不产生抽取成本 + gpu_hours_per_day: float = 24.0 + peak_multiplier: float = 1.0 # 高峰时段单价倍数(官方列表价口径=1.0;若按高峰 ×2 口径计费则设 2.0) + + +def effective_input_price(p: Params) -> float: + # 高峰时段(工作日北京时间 09-12/14-18)约占总时长 1/4; + # 列表价口径(peak_multiplier=1.0)下加权后与列表价一致。 + frac_peak = 0.25 + return p.price_cny_m_input * (1 - frac_peak) + p.price_cny_m_input * p.peak_multiplier * frac_peak + + +def model_docs(p: Params) -> tuple[list[DocSpec], list[DocSpec]]: + """返回 (批次[P1 试点], 全量) 两组成本估算。""" + p_in_eff = effective_input_price(p) + + def md_cost(chars: float, n: int) -> DocSpec: + body_tokens = chars / p.chars_per_token + in_t = body_tokens + p.overhead_tokens_per_call * p.md_llm_calls + out_t = body_tokens * p.out_in_ratio + c = (in_t * p_in_eff + out_t * p.price_cny_m_output) / 1e6 * n + return DocSpec("MD 摘要卡", n, chars, p.md_llm_calls, None, c, int(in_t * n), int(out_t * n)) + + def pdf_cost(chars: float, n: int) -> DocSpec: + chunks = max(1.0, round(chars / p.pdf_chunk_chars)) + in_t = (chars / p.chars_per_token) + p.overhead_tokens_per_call * chunks + out_t = chunks * p.pdf_output_per_call + c = (in_t * p_in_eff + out_t * p.price_cny_m_output) / 1e6 * n + return DocSpec("PDF 全文", n, chars, chunks, chunks, c, int(in_t * n), int(out_t * n)) + + batch = [md_cost(p.md_chars, p.n_pilot_md), pdf_cost(p.pdf_chars, p.n_pilot_pdf)] + full = [md_cost(p.md_chars, p.n_md), pdf_cost(p.pdf_chars, p.n_pdf)] + return batch, full + + +def render_md(p: Params, batch: list[DocSpec], full: list[DocSpec]) -> str: + p_in_eff = effective_input_price(p) + L: list[str] = [] + L.append("# 水产库建库预算表(由 cost_model.py 生成)") + L.append("") + L.append("| 参数 | 值 | 来源 |") + L.append("|---|---|---|") + L.append(f"| LLM 单价(输入/输出,¥/百万 token) | {p.price_cny_m_input:.2f} / {p.price_cny_m_output:.2f} | 官方 deepseek-v4-flash 列表价(缓存未命中) |") + L.append(f"| 高峰加权后输入单价 | {p_in_eff:.4f} | peak_multiplier={p.peak_multiplier:.1f}(1.0=列表价口径;若高峰 ×2 计费口径则设 2.0) |") + L.append(f"| 字符/token | {p.chars_per_token}(英文) / {p.cjk_chars_per_token}(中文) | 英文实测经验;中文 1.5 待试点校准 |") + L.append(f"| MD 卡平均字符数 | {p.md_chars:,.0f} | 实测:200 篇抽样(P50=2169) |") + L.append(f"| PDF 正文平均字符数 | {p.pdf_chars:,.0f} | 假设,P1 30 篇试点后回填 |") + L.append(f"| MD 篇数 / PDF 篇数(全量) | {p.n_md:,} / {p.n_pdf:,} | census.json 实测 |") + L.append(f"| MinerU 吞吐 | {p.mineru_files_per_hour:,.0f} 篇/小时(单卡 3 端点并流) | 历史实测外推(2026-08-31 批) |") + L.append("") + L.append("## P1 试点批次(30 PDF + 470 MD)") + L.append("") + L.append("| 文档类型 | 篇数 | LLM 调用/篇 | 输入 token | 输出 token | 成本(¥) |") + L.append("|---|---|---|---|---|---|") + tb = sum(s.cost for s in batch) + for s in batch: + L.append(f"| {s.label} | {s.n_docs:,} | {s.llm_calls_per_doc:g} | {s.in_tokens:,} | {s.out_tokens:,} | {s.cost:,.2f} |") + L.append(f"| **合计** | | | | | **{tb:,.2f}** |") + L.append("") + mineru_h = p.n_pilot_pdf / p.mineru_files_per_hour + L.append(f"- MinerU GPU 耗时:30 篇约 {max(mineru_h * 60, 10):.0f} 分钟(含模型加载起步开销;历史不可处理率 {p.mineru_unprocessable:.0%} → 预计成功 ~25 篇)") + L.append("- GPU:单卡 RTX 5090(32GB),MinerU 本地运行,无 API 成本") + L.append("") + L.append("## 全量批次(31,848 MD + 13,175 PDF)") + L.append("") + L.append("| 文档类型 | 篇数 | LLM 调用/篇 | 输入 token | 输出 token | 成本(¥) |") + L.append("|---|---|---|---|---|---|") + tf = sum(s.cost for s in full) + for s in full: + L.append(f"| {s.label} | {s.n_docs:,} | {s.llm_calls_per_doc:g} | {s.in_tokens:,} | {s.out_tokens:,} | {s.cost:,.2f} |") + L.append(f"| **合计(中值)** | | | | | **{tf:,.2f}** |") + lo, hi = tf * 0.7, tf * 1.3 + L.append(f"| **合计(±30% 区间)** | | | | | **{lo:,.2f} – {hi:,.2f}** |") + L.append("") + days = p.n_pdf / p.mineru_files_per_hour / p.gpu_hours_per_day + L.append(f"- MinerU 全量 PDF:{p.n_pdf:,} 篇全部过一遍(其中约 {p.mineru_unprocessable:.0%} 判为不可处理),按单卡 3 端点并流历史实测吞吐(保守取 {p.mineru_files_per_hour:,.0f} 篇/时)全天跑 **{days:,.1f} 天**;加失败重试与运维余量,预算按 1–2 天排期") + L.append("- 成本区间已含 ±30% 不确定性(pdf_chars 假设、out_in_ratio、高峰价、重试等)") + L.append("- 假设清单与待回填项见 kb-libraries/aqua/qa/budget_template.md") + L.append("") + return "\n".join(L) + + +def main() -> int: + ap = argparse.ArgumentParser(description="aqua 库成本校准模型") + for k, v in Params().__dict__.items(): + if isinstance(v, bool): + continue + ap.add_argument(f"--{k.replace('_', '-')}", type=float, default=v) + ap.add_argument("--json", action="store_true", help="输出机器可读 JSON") + ap.add_argument("--out", default="", help="同时写入预算表 md 文件") + args = ap.parse_args() + p = Params(**{k: getattr(args, k) for k in Params().__dict__}) + batch, full = model_docs(p) + if args.json: + out = { + "params": p.__dict__, + "pilot_batch": {s.label: {"n_docs": s.n_docs, "in_tokens": s.in_tokens, + "out_tokens": s.out_tokens, "cost_cny": round(s.cost, 2)} + for s in batch}, + "pilot_total_cny": round(sum(s.cost for s in batch), 2), + "full_batch": {s.label: {"n_docs": s.n_docs, "in_tokens": s.in_tokens, + "out_tokens": s.out_tokens, "cost_cny": round(s.cost, 2)} + for s in full}, + "full_total_cny": round(sum(s.cost for s in full), 2), + "full_total_range_cny": [round(sum(s.cost for s in full) * 0.7, 2), + round(sum(s.cost for s in full) * 1.3, 2)], + "mineru_gpu_days_full": round(p.n_pdf / p.mineru_files_per_hour / 24, 2), + } + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + md = render_md(p, batch, full) + print(md) + if args.out: + with open(args.out, "w", encoding="utf-8") as f: + f.write(md) + print(f"[written] {args.out}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tooling/aqua_kb/make_fixtures.py b/tooling/aqua_kb/make_fixtures.py new file mode 100644 index 0000000..49b6dcd --- /dev/null +++ b/tooling/aqua_kb/make_fixtures.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""自造 3 个最小 v16 结构 result.json fixture（手写字段按 v16_1_2 契约）。 + +产出 //result.json，每篇 3-5 实体 + 3-5 关系， +含 rejected 关系（验证转换器过滤）与多层 output_layer。 +""" +from __future__ import annotations + +import json +from pathlib import Path + +FIXTURES = [ + { + "paper_id": "doi:10.1016/0044-8486(88)90238-4", + "paper": { + "paper_id": "doi:10.1016/0044-8486(88)90238-4", + "title": "Quality of algal material produced on a fertilizer-tap water medium in outdoor plastic-enclosed systems", + "doi": "10.1016/0044-8486(88)90238-4", + "pmid": "", + "pmcid": "", + "journal": "Aquaculture", + "publication_year": "1988", + }, + "entities": [ + {"name": "Chlorella vulgaris", "type": "species", "name_resolved": "", + "synonyms": [], "confidence": 0.95}, + {"name": "outdoor plastic-enclosed culture", "type": "culture_system", + "name_resolved": "", "synonyms": [], "confidence": 0.9}, + {"name": "fertilizer-tap water medium", "type": "culture_medium", + "name_resolved": "", "synonyms": [], "confidence": 0.9}, + {"name": "protein content", "type": "biochemical_trait", + "name_resolved": "", "synonyms": [], "confidence": 0.85}, + ], + "relations": [ + { + "source": "Chlorella vulgaris", "source_type": "species", + "source_resolved": "Chlorella vulgaris", "source_type_resolved": "species", + "relation": "cultured_in", "relationship_resolved": "cultured_in", + "target": "outdoor plastic-enclosed culture", "target_type": "culture_system", + "target_resolved": "outdoor plastic-enclosed culture", + "target_type_resolved": "culture_system", + "relation_category": "cultivation", + "evidence": "C. vulgaris was maintained in outdoor plastic-enclosed systems with fertilizer-tap water medium.", + "confidence": 0.95, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Chlorella vulgaris", "source_type": "species", + "source_resolved": "Chlorella vulgaris", "source_type_resolved": "species", + "relation": "cultured_in", "relationship_resolved": "cultured_in", + "target": "fertilizer-tap water medium", "target_type": "culture_medium", + "target_resolved": "fertilizer-tap water medium", + "target_type_resolved": "culture_medium", + "relation_category": "cultivation", + "evidence": "Algal biomass was produced on a fertilizer-tap water medium.", + "confidence": 0.9, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Chlorella vulgaris", "source_type": "species", + "source_resolved": "Chlorella vulgaris", "source_type_resolved": "species", + "relation": "has_trait", "relationship_resolved": "has_trait", + "target": "protein content", "target_type": "biochemical_trait", + "target_resolved": "protein content", "target_type_resolved": "biochemical_trait", + "relation_category": "physiological_measurement", + "evidence": "Protein content of the harvested biomass reached 45% of dry weight.", + "confidence": 0.85, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Chlorella vulgaris", "source_type": "species", + "source_resolved": "Chlorella vulgaris", "source_type_resolved": "species", + "relation": "grows_in", "relationship_resolved": "grows_in", + "target": "fertilizer-tap water medium", "target_type": "culture_medium", + "target_resolved": "fertilizer-tap water medium", + "target_type_resolved": "culture_medium", + "relation_category": "growth_condition", + "evidence": "Growth rates were stable across seasons.", + "confidence": 0.6, "output_layer": "rejected", "claim_scope": "", + "candidate_status": "rejected", "triple_quality": "", "final_action": "", + }, + ], + "chunks": [], "sections": [], + "metadata": {"version": "16.1.2"}, + }, + { + "paper_id": "doi:10.2331/suisan.54.1049", + "paper": { + "paper_id": "doi:10.2331/suisan.54.1049", + "title": "Titration of the alternative complement pathway activity of representative cultured fishes", + "doi": "10.2331/suisan.54.1049", + "pmid": "", "pmcid": "", "journal": "Nippon Suisan Gakkaishi", + "publication_year": "1988", + }, + "entities": [ + {"name": "Oncorhynchus mykiss", "type": "species", "name_resolved": "", + "synonyms": [], "confidence": 0.95}, + {"name": "alternative complement pathway", "type": "immune_pathway", + "name_resolved": "", "synonyms": [], "confidence": 0.9}, + {"name": "serum", "type": "biological_sample", "name_resolved": "", + "synonyms": [], "confidence": 0.9}, + ], + "relations": [ + { + "source": "Oncorhynchus mykiss", "source_type": "species", + "source_resolved": "Oncorhynchus mykiss", "source_type_resolved": "species", + "relation": "activates", "relationship_resolved": "activates", + "target": "alternative complement pathway", "target_type": "immune_pathway", + "target_resolved": "alternative complement pathway", + "target_type_resolved": "immune_pathway", + "relation_category": "immune_function", + "evidence": "Serum from rainbow trout showed ACH50 activity of 142 U/ml.", + "confidence": 0.95, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "alternative complement pathway", "source_type": "immune_pathway", + "source_resolved": "alternative complement pathway", + "source_type_resolved": "immune_pathway", + "relation": "measured_in", "relationship_resolved": "measured_in", + "target": "serum", "target_type": "biological_sample", + "target_resolved": "serum", "target_type_resolved": "biological_sample", + "relation_category": "physiological_measurement", + "evidence": "The ACP titer was titrated in serum samples.", + "confidence": 0.9, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Oncorhynchus mykiss", "source_type": "species", + "source_resolved": "Oncorhynchus mykiss", "source_type_resolved": "species", + "relation": "has_sample", "relationship_resolved": "has_sample", + "target": "serum", "target_type": "biological_sample", + "target_resolved": "serum", "target_type_resolved": "biological_sample", + "relation_category": "experimental_material", + "evidence": "Blood was collected from the caudal vein.", + "confidence": 0.8, "output_layer": "background_relations", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + ], + "chunks": [], "sections": [], + "metadata": {"version": "16.1.2"}, + }, + { + "paper_id": "doi:10.1051/alr/1989009", + "paper": { + "paper_id": "doi:10.1051/alr/1989009", + "title": "Suitability of brackish water tilapia species from the Ivory Coast for lagoon aquaculture. I - Reproduction", + "doi": "10.1051/alr/1989009", + "pmid": "", "pmcid": "", "journal": "Aquatic Living Resources", + "publication_year": "1989", + }, + "entities": [ + {"name": "Oreochromis niloticus", "type": "species", "name_resolved": "", + "synonyms": [], "confidence": 0.95}, + {"name": "lagoon aquaculture", "type": "culture_system", + "name_resolved": "", "synonyms": [], "confidence": 0.9}, + {"name": "Ivory Coast", "type": "location", "name_resolved": "", + "synonyms": [], "confidence": 0.9}, + {"name": "reproductive performance", "type": "biological_trait", + "name_resolved": "", "synonyms": [], "confidence": 0.85}, + ], + "relations": [ + { + "source": "Oreochromis niloticus", "source_type": "species", + "source_resolved": "Oreochromis niloticus", "source_type_resolved": "species", + "relation": "suited_for", "relationship_resolved": "suited_for", + "target": "lagoon aquaculture", "target_type": "culture_system", + "target_resolved": "lagoon aquaculture", + "target_type_resolved": "culture_system", + "relation_category": "aquaculture_suitability", + "evidence": "O. niloticus showed the best reproduction performance among brackish water tilapias.", + "confidence": 0.9, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Oreochromis niloticus", "source_type": "species", + "source_resolved": "Oreochromis niloticus", "source_type_resolved": "species", + "relation": "native_to", "relationship_resolved": "native_to", + "target": "Ivory Coast", "target_type": "location", + "target_resolved": "Ivory Coast", "target_type_resolved": "location", + "relation_category": "geographic_distribution", + "evidence": "Tilapia populations were sampled in lagoons of the Ivory Coast.", + "confidence": 0.85, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + { + "source": "Oreochromis niloticus", "source_type": "species", + "source_resolved": "Oreochromis niloticus", "source_type_resolved": "species", + "relation": "has_trait", "relationship_resolved": "has_trait", + "target": "reproductive performance", "target_type": "biological_trait", + "target_resolved": "reproductive performance", + "target_type_resolved": "biological_trait", + "relation_category": "reproductive_biology", + "evidence": "Mean fecundity was 1200 eggs per female.", + "confidence": 0.8, "output_layer": "main_kg", "claim_scope": "paper", + "candidate_status": "", "triple_quality": "", "final_action": "", + }, + ], + "chunks": [], "sections": [], + "metadata": {"version": "16.1.2"}, + }, +] + + +def main() -> None: + import argparse + parser = argparse.ArgumentParser() + parser.add_argument("--out", default="fixtures") + args = parser.parse_args() + root = Path(args.out) + for fixture in FIXTURES: + paper_dir = root / fixture["paper_id"].replace("/", "_") + paper_dir.mkdir(parents=True, exist_ok=True) + (paper_dir / "result.json").write_text( + json.dumps(fixture, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(f"wrote {paper_dir}/result.json") + # pilot manifest + manifest = { + "papers": [ + {"paper_id": paper["paper_id"], "result": f"{paper['paper_id'].replace(chr(47), chr(95))}/result.json", + "paper": paper["paper"]} + for paper in FIXTURES + ] + } + (root / "pilot_manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(f"wrote {root}/pilot_manifest.json") + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/phase2b_converter.py b/tooling/aqua_kb/phase2b_converter.py new file mode 100644 index 0000000..3f079ef --- /dev/null +++ b/tooling/aqua_kb/phase2b_converter.py @@ -0,0 +1,443 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""tooling/aqua_kb/phase2b_converter.py —— Phase2B 转换器 + +把 v16_1_2 抽取链产出的 result.json 转换为 normalized 图谱 + Qdrant 点： + + 每篇产出两个文件（幂等，同一 result.json 重跑逐字节一致）： + //neo4j.cypher —— Entity/Paper/ASSERTS/SUPPORTED_BY + //qdrant_points.jsonl —— 每行一个 point（无向量，由 qdrant_loader.py 嵌入） + +契约来源（2026-09-12 逐键核对）： + - agent/qa/src/rag/qa_v1_1_73/direct_flow_retriever.py L349-389 item_from_payload + - agent/qa/src/rag/normalized_release.py（_ONE_HOP / resolve_entities 属性名） + - agent/qa/src/rag/qa_v1_1_73/backend_guard.py normalized 分支（标签/关系/kb_version） + - agent/legacy-rag/src/rag/extractors/v16_1_2/{v15,v16}_extractor.py _build_output + （result.json = {entities, relations, sections, chunks, metadata}） + +规则： + - paper_id 规范为 `doi:`；DOI 从 result["paper"] / result["metadata"] / + --paper-json 侧车 / 文件名 DOI 提取，都无则退回文件名 stem。 + - assertion_id = `cass_` + uuid5(SHA1(paper_id|source_canonical_id|relation_norm|target_canonical_id)) + （同源去重时确定性一致；同源同断言多条 evidence 合并为一条，evidence 取拼接）。 + - canonical_id = `cent_` + 12 位 sha1 hex（entity_type|name_norm 确定性）。 + - 关系过滤：rejected 状态跳过（与 _build_output 的 kept 口径一致）。 + +用法： + python phase2b_converter.py --result --out \ + [--release-id aqua_srt_v1_20260930_normalized] [--paper-json ] +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +DEFAULT_RELEASE_ID = "aqua_srt_v1_20260930_normalized" +DEFAULT_SOURCE_SYSTEM = "Aqua_canonical" +REJECT_STATUS = {"rejected", "hard_reject"} +# doi 字面量（含括号尾注等容忍形态） +_DOI_LITERAL = re.compile(r"10\.\d{4,9}/[^\s\"']+", re.IGNORECASE) +# 文件名形态 DOI：10.1016_0044-8486(88)90238-4 或 10.1007_s12225-015-9580-y +_DOI_FILE = re.compile( + r"(10[._]\d{4,9}[._]\S+?)(?:_[0-9a-f]{6,}(?:\.pdf)?$|$)", re.IGNORECASE +) +_LAYERS = { + "main_kg": "main_kg", + "dataset_metadata": "dataset_metadata", + "method_relations": "method_relations", + "background_relations": "background_relations", + "candidate_review": "candidate_review", +} + + +def _text(value: Any) -> str: + return "" if value is None else str(value).strip() + + +def _norm(name: Any) -> str: + """name_norm：与 normalized_release.py 的 _dedupe(normalize=True) 同口径 + （" ".join(text.split()).casefold()）。""" + return " ".join(_text(name).split()).casefold() + + +def _find_doi(text: str) -> str: + """从任意文本提取第一个 DOI 字面量；支持 `10.1016_0044-8486(88)90238-4` + 下划线形态（转为斜杠 + 括号）。""" + if not text: + return "" + match = _DOI_LITERAL.search(text.replace("_", "/")) + if match: + return match.group(0).rstrip(".,;)]") + return "" + + +def _sha(value: str) -> str: + return hashlib.sha1(value.encode("utf-8")).hexdigest() + + +def _uuid5(value: str) -> str: + return str(uuid.uuid5(uuid.NAMESPACE_URL, value)) + + +def _assertion_id( + paper_id: str, source_canonical_id: str, relation_norm: str, target_canonical_id: str +) -> str: + key = "|".join([paper_id, source_canonical_id, relation_norm, target_canonical_id]) + return "cass_" + _uuid5(key) + + +def _canonical_id(entity_type: str, name_norm: str) -> str: + return "cent_" + _sha(f"{_text(entity_type).casefold()}|{name_norm}")[:12] + + +def _is_rejected(rel: Dict[str, Any]) -> bool: + for field in ("output_layer", "candidate_status", "triple_quality", "final_action"): + if _text(rel.get(field)).lower() in REJECT_STATUS: + return True + return False + + +# ---------------- paper 元数据提取 ---------------- + +def resolve_paper(result: Dict[str, Any], paper_json: Optional[Dict[str, Any]], + result_path: Path) -> Tuple[str, Dict[str, Any]]: + """返回 (paper_id 规范形 `doi:...`, paper 元数据 dict)。 + + 来源优先级：result["paper"] > result["metadata"]["paper"] > --paper-json > + 文件名 DOI 派生 > 文件名 stem 回退。 + """ + paper: Dict[str, Any] = {} + cand = result.get("paper") + if isinstance(cand, dict): + paper = dict(cand) + meta = result.get("metadata") or {} + if not paper and isinstance(meta.get("paper"), dict): + paper = dict(meta["paper"]) + if not paper and isinstance(paper_json, dict): + paper = dict(paper_json) + + def get(*keys: str) -> str: + for key in keys: + value = paper.get(key) + if value not in (None, ""): + return _text(value) + return "" + + doi = get("doi", "DOI") + if not doi: + doi = _find_doi(get("title", "") or json.dumps(paper, ensure_ascii=False)) + if not doi: + doi = _find_doi(result_path.stem) + paper_id = get("paper_id", "id") + if doi and (not paper_id or paper_id == result_path.stem): + paper_id = f"doi:{doi}" + if not paper_id: + paper_id = f"doi:{doi}" if doi else result_path.stem + + return paper_id, { + "paper_id": paper_id, + "title": get("title"), + "doi": doi, + "pmid": get("pmid", "pubmed_id"), + "pmcid": get("pmcid", "PMCID"), + "journal": get("journal", "journal_name", "venue"), + "publication_year": get("publication_year", "year"), + } + + +# ---------------- 实体/关系归一 ---------------- + +def build_entities(result: Dict[str, Any]) -> List[Dict[str, Any]]: + """从 result["entities"] + 关系端点回灌，产出归一实体列表（确定性排序去重）。""" + registry: Dict[Tuple[str, str], Dict[str, Any]] = {} + for entity in result.get("entities") or []: + name = _text( + entity.get("name_resolved") or entity.get("name") or entity.get("id") + ) + etype = _text( + entity.get("type_resolved") or entity.get("type") or entity.get("entity_type") + ) or "entity" + if not name: + continue + key = (etype.casefold(), _norm(name)) + if key in registry: + continue + registry[key] = { + "name": name, + "name_norm": _norm(name), + "entity_type": etype, + "canonical_id": _canonical_id(etype, _norm(name)), + } + for rel in result.get("relations") or []: + for side in ("source", "target"): + name = _text(rel.get(f"{side}_resolved") or rel.get(side)) + etype = _text(rel.get(f"{side}_type_resolved") or rel.get(f"{side}_type")) + if not name or not etype: + continue + key = (etype.casefold(), _norm(name)) + if key in registry: + continue + registry[key] = { + "name": name, + "name_norm": _norm(name), + "entity_type": etype, + "canonical_id": _canonical_id(etype, _norm(name)), + } + return sorted( + registry.values(), key=lambda item: (item["canonical_id"], item["name"]) + ) + + +def build_assertions( + result: Dict[str, Any], paper_id: str, entity_index: Dict[Tuple[str, str], str] +) -> List[Dict[str, Any]]: + """归一关系 → 断言列表。同源 (paper, src, rel, tgt) 多条关系合并为一条断言， + evidence 拼接；排序确定性（按 assertion_id）。""" + grouped: Dict[str, Dict[str, Any]] = {} + for rel in result.get("relations") or []: + if _is_rejected(rel): + continue + source = _text(rel.get("source_resolved") or rel.get("source")) + target = _text(rel.get("target_resolved") or rel.get("target")) + source_type = _text( + rel.get("source_type_resolved") or rel.get("source_type") or "entity" + ) + target_type = _text( + rel.get("target_type_resolved") or rel.get("target_type") or "entity" + ) + relation_norm = _text( + rel.get("relationship_resolved") or rel.get("relation") + or rel.get("normalized_relation") or rel.get("relation_norm") + ) + if not (source and target and relation_norm): + continue + relation_category = _text( + rel.get("relation_category") or rel.get("relation_type") or "general" + ) + output_layer = _text(rel.get("output_layer") or "main_kg") + if not output_layer: + output_layer = "main_kg" + layer = _LAYERS.get(output_layer, output_layer) + evidence = _text(rel.get("evidence") or rel.get("best_evidence") + or rel.get("basis")) + source_canonical_id = entity_index.get( + (source_type.casefold(), _norm(source)) + ) or _canonical_id(source_type, _norm(source)) + target_canonical_id = entity_index.get( + (target_type.casefold(), _norm(target)) + ) or _canonical_id(target_type, _norm(target)) + assertion_id = _assertion_id( + paper_id, source_canonical_id, relation_norm, target_canonical_id + ) + existing = grouped.get(assertion_id) + if existing is None: + grouped[assertion_id] = { + "assertion_id": assertion_id, + "source": source, + "source_type": source_type, + "source_canonical_id": source_canonical_id, + "relation_norm": relation_norm, + "relation_category": relation_category, + "target": target, + "target_type": target_type, + "target_canonical_id": target_canonical_id, + "layer": layer, + "evidence": evidence, + "quality_score": float(rel.get("confidence") or 0.0), + "claim_scope": _text(rel.get("claim_scope")), + } + else: + if evidence and evidence not in existing["evidence"]: + existing["evidence"] = ( + f'{existing["evidence"]}\n{evidence}'.strip() + if existing["evidence"] else evidence + ) + existing["quality_score"] = max( + existing["quality_score"], float(rel.get("confidence") or 0.0) + ) + return sorted(grouped.values(), key=lambda item: item["assertion_id"]) + + +# ---------------- 产出文件 ---------------- + +def build_cypher( + paper: Dict[str, Any], entities: List[Dict[str, Any]], + assertions: List[Dict[str, Any]], release_id: str, +) -> str: + lines: List[str] = [ + "// Phase2B normalized graph for paper " + f"{paper['paper_id']} (release {release_id}) — generated by phase2b_converter.py", + "// Labels: Entity, Paper; relationships: ASSERTS, SUPPORTED_BY", + ] + + def esc(value: Any) -> str: + return json.dumps(_text(value), ensure_ascii=False) + + lines.append("// ---- Paper ----") + lines.append( + "MERGE (p:Paper {paper_id: %s}) SET p.title = %s, p.doi = %s, " + "p.pmid = %s, p.pmcid = %s, p.journal = %s, p.publication_year = %s, " + "p.kb_version = %s;" + % ( + esc(paper.get("paper_id")), + esc(paper.get("title")), + esc(paper.get("doi")), + esc(paper.get("pmid")), + esc(paper.get("pmcid")), + esc(paper.get("journal")), + esc(paper.get("publication_year")), + esc(release_id), + ) + ) + lines.append("// ---- Entities (%d) ----" % len(entities)) + for entity in entities: + lines.append( + "MERGE (e:Entity {canonical_id: %s}) SET e.name = %s, " + "e.name_norm = %s, e.entity_type = %s, e.kb_version = %s;" + % ( + esc(entity["canonical_id"]), + esc(entity["name"]), + esc(entity["name_norm"]), + esc(entity["entity_type"]), + esc(release_id), + ) + ) + lines.append("// ---- Assertions (%d) ----" % len(assertions)) + for assertion in assertions: + lines.append( + "MATCH (src:Entity {canonical_id: %s}), (tgt:Entity {canonical_id: %s}), " + "(p:Paper {paper_id: %s}) " + "MERGE (src)-[r:ASSERTS {assertion_id: %s}]->(tgt) " + "SET r.relation_norm = %s, r.relation_category = %s, r.layer = %s, " + "r.kb_version = %s " + "MERGE (src)-[s:SUPPORTED_BY {assertion_id: %s}]->(p) " + "SET s.kb_version = %s;" + % ( + esc(assertion["source_canonical_id"]), + esc(assertion["target_canonical_id"]), + esc(paper["paper_id"]), + esc(assertion["assertion_id"]), + esc(assertion["relation_norm"]), + esc(assertion["relation_category"]), + esc(assertion["layer"]), + esc(release_id), + esc(assertion["assertion_id"]), + esc(release_id), + ) + ) + lines.append("") + return "\n".join(lines) + + +def build_points( + paper: Dict[str, Any], assertions: List[Dict[str, Any]], release_id: str, +) -> str: + """qdrant_points.jsonl：每行 {id, text, payload}。向量由 qdrant_loader 嵌入； + text = 现网 embedding_text 同款拼装（Source/Relation/Evidence/Paper 六行）。""" + rows: List[str] = [] + for assertion in assertions: + embedding_text = "\n".join( + [ + f"Source: {assertion['source']}", + f"Source type: {assertion['source_type']}", + f"Relation: {assertion['relation_norm']}", + f"Relation category: {assertion['relation_category']}", + f"Target: {assertion['target']}", + f"Target type: {assertion['target_type']}", + f"Evidence: {assertion['evidence']}", + f"Paper: {paper['paper_id']}", + ] + ) + payload = { + "assertion_id": assertion["assertion_id"], + "relation_id": assertion["assertion_id"], + "source": assertion["source"], + "target": assertion["target"], + "source_type": assertion["source_type"], + "target_type": assertion["target_type"], + "source_canonical_id": assertion["source_canonical_id"], + "target_canonical_id": assertion["target_canonical_id"], + "relation": assertion["relation_norm"], + "relation_norm": assertion["relation_norm"], + "relation_category": assertion["relation_category"], + "evidence": assertion["evidence"], + "paper_id": paper["paper_id"], + "title": paper.get("title") or "", + "doi": paper.get("doi") or "", + "pmid": paper.get("pmid") or "", + "pmcid": paper.get("pmcid") or "", + "quality_score": float(assertion["quality_score"]), + "output_layer": assertion["layer"], + "claim_scope": assertion["claim_scope"], + "layer": assertion["layer"], + "kb_version": release_id, + "source_system": DEFAULT_SOURCE_SYSTEM, + } + rows.append( + json.dumps( + {"id": str(uuid.uuid5(uuid.NAMESPACE_URL, assertion["assertion_id"])), + "text": embedding_text, + "payload": payload}, + ensure_ascii=False, + ) + ) + return "\n".join(rows) + ("\n" if rows else "") + + +def convert(result_path: Path, out_root: Path, release_id: str, + paper_json: Optional[Dict[str, Any]]) -> Dict[str, Any]: + with open(result_path, "r", encoding="utf-8") as handle: + result = json.load(handle) + paper_id, paper = resolve_paper(result, paper_json, result_path) + entities = build_entities(result) + entity_index = { + (entity["entity_type"].casefold(), entity["name_norm"]): entity["canonical_id"] + for entity in entities + } + assertions = build_assertions(result, paper_id, entity_index) + + paper_dir = out_root / paper_id.replace("/", "_") + paper_dir.mkdir(parents=True, exist_ok=True) + cypher_path = paper_dir / "neo4j.cypher" + points_path = paper_dir / "qdrant_points.jsonl" + cypher_path.write_text( + build_cypher(paper, entities, assertions, release_id), encoding="utf-8" + ) + points_path.write_text( + build_points(paper, assertions, release_id), encoding="utf-8" + ) + return { + "paper_id": paper_id, + "paper_dir": str(paper_dir), + "entity_count": len(entities), + "assertion_count": len(assertions), + "point_count": len(assertions), + "cypher": str(cypher_path), + "points": str(points_path), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Phase2B result.json -> normalized KG + Qdrant points") + parser.add_argument("--result", required=True, help="v16 result.json path") + parser.add_argument("--out", required=True, help="output root directory") + parser.add_argument("--release-id", default=DEFAULT_RELEASE_ID) + parser.add_argument("--paper-json", default=None, help="optional paper metadata sidecar JSON") + args = parser.parse_args() + + paper_json: Optional[Dict[str, Any]] = None + if args.paper_json: + with open(args.paper_json, "r", encoding="utf-8") as handle: + paper_json = json.load(handle) + + stats = convert(Path(args.result), Path(args.out), args.release_id, paper_json) + print(json.dumps(stats, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/qdrant_loader.py b/tooling/aqua_kb/qdrant_loader.py new file mode 100644 index 0000000..7c6e2b9 --- /dev/null +++ b/tooling/aqua_kb/qdrant_loader.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""tooling/aqua_kb/qdrant_loader.py —— Qdrant 批量灌库器 + +读 phase2b_converter.py 产出的 qdrant_points.jsonl（一行一个 point：id/text/payload）， +用与现网 direct_flow_retriever.embed() 同款端点（POST {embed_url} {"model":..., "input":...}） +嵌入 text 字段，分批 upsert 到 Qdrant collection。 + +特性： + - checkpoint JSONL：每成功 upsert 一个 assertion_id 记一行 → 断点续跑（kill 后重跑跳过已传） + - 同 id 重放覆盖（Qdrant upsert 语义 + UUID 确定性 id → 幂等） + - 确定性处理顺序（文件路径 + 文件内行序排序） + +用法： + python qdrant_loader.py --points --qdrant http://127.0.0.1:27435 \ + --collection aqua_srt_v1_20260930_normalized --batch 200 \ + --checkpoint \ + [--embed-url http://127.0.0.1:11434/api/embed] [--embed-model bge-m3:567m] +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path +from typing import Any, Dict, Iterator, List + +import requests + +DEFAULT_EMBED_URL = "http://127.0.0.1:11434/api/embed" +DEFAULT_EMBED_MODEL = "bge-m3:567m" +EMBED_INPUT_LIMIT = 2400 # 与 direct_flow_retriever.embed() 一致 + + +def iter_points(paths: List[Path]) -> Iterator[Dict[str, Any]]: + files = sorted( + p for p in paths if p.is_file() and p.name == "qdrant_points.jsonl" + ) + for file_path in files: + with open(file_path, "r", encoding="utf-8") as handle: + for line_number, line in enumerate(handle, 1): + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + print(f"[loader] WARN bad json {file_path}:{line_number}: {exc}", + file=sys.stderr) + continue + yield record + + +class EmbedClient: + def __init__(self, url: str, model: str) -> None: + self.url = url.rstrip("/") + self.model = model + self.dim: int | None = None + + def embed(self, text: str) -> List[float]: + for attempt in range(4): + try: + response = requests.post( + self.url, + json={"model": self.model, "input": text[:EMBED_INPUT_LIMIT]}, + timeout=60, + ) + response.raise_for_status() + data = response.json() + embeddings = data.get("embeddings") or [] + if embeddings: + vector = embeddings[0] + elif "embedding" in data: + vector = data["embedding"] + else: + raise RuntimeError("embedding endpoint returned no vector") + self.dim = len(vector) + return [float(value) for value in vector] + except Exception as exc: + if attempt == 3: + raise + time.sleep(2 * (attempt + 1)) + raise RuntimeError("unreachable") + + +def ensure_collection(qdrant_url: str, collection: str, dim: int) -> None: + info = requests.get( + f"{qdrant_url.rstrip('/')}/collections/{collection}", timeout=15 + ) + if info.status_code == 200: + return + if info.status_code != 404: + info.raise_for_status() + create = requests.put( + f"{qdrant_url.rstrip('/')}/collections/{collection}", + json={ + "vectors": {"size": dim, "distance": "Cosine"}, + "shard_number": 1, + "replication_factor": 1, + "write_consistency_factor": 1, + "on_disk_payload": True, + }, + timeout=30, + ) + create.raise_for_status() + print(f"[loader] created collection {collection} (dim={dim}, COSINE)") + + +def load_checkpoint(checkpoint: Path) -> set: + done: set = set() + if checkpoint.exists(): + with open(checkpoint, "r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + done.add(json.loads(line)) # 兼容 json.dumps 写入的带引号行 + except json.JSONDecodeError: + done.add(line) + return done + + +def upsert_file( + points_path: Path, + qdrant_url: str, + collection: str, + batch_size: int, + *, + checkpoint: Path, + embed: "EmbedClient", +) -> Dict[str, int]: + """上传单个 qdrant_points.jsonl（断点续跑，checkpoint 追加）。""" + checkpoint.parent.mkdir(parents=True, exist_ok=True) + done = load_checkpoint(checkpoint) + qdrant = qdrant_url.rstrip("/") + upsert_url = f"{qdrant}/collections/{collection}/points?wait=true" + + total = skipped = uploaded = 0 + batch: List[Dict[str, Any]] = [] + started = time.time() + with open(checkpoint, "a", encoding="utf-8") as checkpoint_handle: + def flush_batch() -> None: + nonlocal batch + if not batch: + return + response = requests.put(upsert_url, json={"points": batch}, timeout=180) + response.raise_for_status() + for record in batch: + checkpoint_handle.write( + json.dumps(record["payload"]["assertion_id"]) + "\n" + ) + checkpoint_handle.flush() + batch = [] + + for record in iter_points([points_path]): + total += 1 + assertion_id = record["payload"]["assertion_id"] + if assertion_id in done: + skipped += 1 + continue + record["vector"] = embed.embed(record.get("text") or "") + batch.append(record) + uploaded += 1 + if len(batch) >= batch_size: + flush_batch() + flush_batch() + return { + "uploaded": uploaded, + "skipped": skipped, + "total": total, + "elapsed": round(time.time() - started, 1), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Qdrant bulk loader for aqua KB") + parser.add_argument("--points", required=True, help="dir or qdrant_points.jsonl file") + parser.add_argument("--qdrant", required=True, help="e.g. http://127.0.0.1:27435") + parser.add_argument("--collection", default="aqua_srt_v1_20260930_normalized") + parser.add_argument("--batch", type=int, default=200) + parser.add_argument("--checkpoint", required=True, help="JSONL of uploaded assertion_ids") + parser.add_argument("--embed-url", default=DEFAULT_EMBED_URL) + parser.add_argument("--embed-model", default=DEFAULT_EMBED_MODEL) + args = parser.parse_args() + + points_arg = Path(args.points) + if points_arg.is_dir(): + files = sorted(points_arg.rglob("qdrant_points.jsonl")) + else: + files = [points_arg] + checkpoint = Path(args.checkpoint) + + embed = EmbedClient(args.embed_url, args.embed_model) + qdrant = args.qdrant.rstrip("/") + + # 先预热一发拿到维度（与现网 bge-m3 一致；collection 不存在则按此维度创建） + probe = embed.embed("aquaculture warm-up probe") + print(f"[loader] embed dim={len(probe)} (model={args.embed_model}, url={args.embed_url})") + ensure_collection(qdrant, args.collection, len(probe)) + + grand = {"uploaded": 0, "skipped": 0, "total": 0} + started = time.time() + for points_path in files: + result = upsert_file( + points_path, + qdrant, + args.collection, + args.batch, + checkpoint=checkpoint, + embed=embed, + ) + for key in grand: + grand[key] += result[key] + print( + f"[loader] {points_path.name}: {result['uploaded']} uploaded, " + f"{result['skipped']} skipped, {result['total']} seen, " + f"{result['elapsed']}s" + ) + + print( + f"[loader] DONE: {grand['uploaded']} uploaded, {grand['skipped']} skipped, " + f"{grand['total']} seen in {time.time() - started:.1f}s" + ) + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/registry_builder.py b/tooling/aqua_kb/registry_builder.py new file mode 100644 index 0000000..fa3fc77 --- /dev/null +++ b/tooling/aqua_kb/registry_builder.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""tooling/aqua_kb/registry_builder.py —— paper metadata registry 生成器 + +聚合各 result.json 的 paper 元数据为 registry JSONL，字段对齐植物库 +paper_metadata_registry_v1_20260717/paper_metadata_registry_full.jsonl 首行结构： + paper_id/doi/pmid/pmcid/title/journal/publication_year/metadata_sources/ + lookup_status/resolution_source/conflict_status/conflicts/identifier_candidates/checked_at + +幂等：结果按 paper_id 排序，确定性输出。缺字段留空、conflicts 空数组， +lookup_status 按数据来源标记 resolved / doi_candidate / fallback。 + +用法： + python registry_builder.py --results --out +""" +from __future__ import annotations + +import argparse +import json +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List + +# doi 字面量 +_DOI_LITERAL = re.compile(r"10\.\d{4,9}/[^\s\"']+", re.IGNORECASE) + + +def _text(value: Any) -> str: + return "" if value is None else str(value).strip() + + +def find_doi(text: str) -> str: + if not text: + return "" + match = _DOI_LITERAL.search(text.replace("_", "/")) + if match: + return match.group(0).rstrip(".,;)]") + return "" + + +def extract_paper_meta(result: Dict[str, Any], result_path: Path) -> Dict[str, Any]: + paper: Dict[str, Any] = {} + cand = result.get("paper") + if isinstance(cand, dict): + paper = dict(cand) + meta = result.get("metadata") or {} + if not paper and isinstance(meta.get("paper"), dict): + paper = dict(meta["paper"]) + + def get(*keys: str) -> str: + for key in keys: + value = paper.get(key) + if value not in (None, ""): + return _text(value) + return "" + + doi = get("doi", "DOI") or find_doi( + get("title", "") or json.dumps(paper, ensure_ascii=False) + ) or find_doi(result_path.stem) + paper_id = get("paper_id", "id") or (f"doi:{doi}" if doi else result_path.stem) + if doi and paper_id == result_path.stem: + paper_id = f"doi:{doi}" + title = get("title") + if not title: + title = get("paper_title") + pmid = get("pmid", "pubmed_id") + pmcid = get("pmcid", "PMCID") + journal = get("journal", "journal_name", "venue") + year = get("publication_year", "year") + + lookup_status = "resolved" if (doi or pmid or pmcid) else "fallback" + resolution_source = "doi_candidate" if doi else "paper_id" + metadata_sources = [] + if paper: + metadata_sources.append("result_paper") + if doi: + metadata_sources.append("doi_candidate") + if not metadata_sources: + metadata_sources = ["paper_id"] + + return { + "paper_id": paper_id, + "title": title, + "doi": doi, + "pmid": pmid, + "pmcid": pmcid, + "journal": journal, + "publication_year": year, + "metadata_sources": metadata_sources, + "lookup_status": lookup_status, + "resolution_source": resolution_source, + "conflict_status": "clean", + "conflicts": [], + "identifier_candidates": [doi] if doi else [], + "checked_at": datetime.now(timezone.utc).isoformat(), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="aqua paper metadata registry builder") + parser.add_argument("--results", required=True, help="dir containing result.json files (recursive)") + parser.add_argument("--out", required=True, help="output registry JSONL path") + args = parser.parse_args() + + root = Path(args.results) + files = sorted(root.rglob("result.json")) if root.is_dir() else [root] + rows: Dict[str, Dict[str, Any]] = {} + for result_path in files: + try: + with open(result_path, "r", encoding="utf-8") as handle: + result = json.load(handle) + except Exception as exc: + print(f"[registry] WARN skip {result_path}: {exc}") + continue + meta = extract_paper_meta(result, result_path) + paper_id = meta["paper_id"] + if paper_id in rows: + existing = rows[paper_id] + # 合并：非空字段优先保留已解析信息；metadata_sources 去重合并 + for key, value in meta.items(): + if key in ("metadata_sources", "identifier_candidates", "conflicts"): + continue + if not existing.get(key) and value: + existing[key] = value + merged_sources: List[str] = list(existing["metadata_sources"]) + for source in meta["metadata_sources"]: + if source not in merged_sources: + merged_sources.append(source) + existing["metadata_sources"] = merged_sources + cands: List[str] = list(existing["identifier_candidates"]) + for cand in meta["identifier_candidates"]: + if cand and cand not in cands: + cands.append(cand) + existing["identifier_candidates"] = cands + if existing["lookup_status"] == "fallback" and meta["lookup_status"] == "resolved": + existing["lookup_status"] = "resolved" + existing["resolution_source"] = meta["resolution_source"] + else: + rows[paper_id] = meta + + out_path = Path(args.out) + out_path.parent.mkdir(parents=True, exist_ok=True) + with open(out_path, "w", encoding="utf-8") as handle: + for paper_id in sorted(rows): + handle.write(json.dumps(rows[paper_id], ensure_ascii=False) + "\n") + print(f"[registry] wrote {len(rows)} rows -> {out_path}") + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/smoke_aqua_kb.py b/tooling/aqua_kb/smoke_aqua_kb.py new file mode 100644 index 0000000..b52ead2 --- /dev/null +++ b/tooling/aqua_kb/smoke_aqua_kb.py @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""tooling/aqua_kb/smoke_aqua_kb.py —— 端到端冒烟 + +对 aqua 试库实例做全链验证： + 1. 服务探活（Qdrant /healthz、Neo4j bolt） + 2. 库计数（Qdrant collection points_count；Neo4j Entity/Paper/ASSERTS/SUPPORTED_BY） + 3. 载荷契约校验（scroll 抽样 1 点，断言全部契约键存在 + kb_version/source_system） + 4. normalized_release.py 同款查询（resolve_entities + one_hop 口径）跑通 + 5. 索引状态（SHOW INDEXES ONLINE） + +用法： + python smoke_aqua_kb.py --qdrant http://127.0.0.1:27435 \ + --collection aqua_srt_v1_20260930_normalized \ + --neo4j-uri bolt://127.0.0.1:19889 --neo4j-user neo4j \ + --neo4j-password-file [--release-id ...] + +退出码 0=全绿；1=失败项。 +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Dict, List + +import requests + +CONTRACT_KEYS = [ + "assertion_id", "relation_id", + "source", "target", + "source_type", "target_type", + "source_canonical_id", "target_canonical_id", + "relation", "relation_norm", "relation_category", + "evidence", "paper_id", "title", "doi", "pmid", "pmcid", + "quality_score", "output_layer", "claim_scope", + "kb_version", "source_system", +] + +RESULTS: List[Dict[str, Any]] = [] + + +def check(name: str, ok: bool, detail: Any = "") -> None: + RESULTS.append({"check": name, "ok": bool(ok), "detail": str(detail)[:300]}) + print(f"[{'PASS' if ok else 'FAIL'}] {name} {detail}") + + +def main() -> None: + parser = argparse.ArgumentParser(description="aqua KB end-to-end smoke") + parser.add_argument("--qdrant", required=True) + parser.add_argument("--collection", default="aqua_srt_v1_20260930_normalized") + parser.add_argument("--neo4j-uri", required=True) + parser.add_argument("--neo4j-user", default="neo4j") + parser.add_argument("--neo4j-password-file", required=True) + parser.add_argument("--release-id", default="aqua_srt_v1_20260930_normalized") + args = parser.parse_args() + + qdrant = args.qdrant.rstrip("/") + + # 1) Qdrant 探活 + collection 信息 + try: + health = requests.get(f"{qdrant}/healthz", timeout=10) + check("qdrant healthz", health.status_code == 200 and health.text == "healthz check passed") + except Exception as exc: + check("qdrant healthz", False, exc) + try: + info = requests.get(f"{qdrant}/collections/{args.collection}", timeout=10) + if info.status_code == 200: + result = info.json()["result"] + check("qdrant collection exists", True, + f"points={result['points_count']} status={result['status']}") + config = result["config"]["params"]["vectors"] + check("qdrant collection dim/distance", True, + f"size={config['size']} distance={config['distance']}") + else: + check("qdrant collection exists", False, f"HTTP {info.status_code}") + except Exception as exc: + check("qdrant collection exists", False, exc) + + # 2) payload 契约抽样 + try: + scroll = requests.post( + f"{qdrant}/collections/{args.collection}/points/scroll", + json={"limit": 1, "with_payload": True, "with_vector": False}, + timeout=15, + ) + points = scroll.json()["result"]["points"] + if not points: + check("qdrant sample point", False, "no points in collection") + else: + payload = points[0]["payload"] + missing = [key for key in CONTRACT_KEYS if key not in payload] + check("qdrant contract keys", not missing, f"missing={missing}") + check("qdrant kb_version", payload.get("kb_version") == args.release_id, + payload.get("kb_version")) + check("qdrant source_system", payload.get("source_system") == "Aqua_canonical", + payload.get("source_system")) + check("qdrant assertion_id prefix", + str(payload.get("assertion_id", "")).startswith("cass_"), + payload.get("assertion_id")) + check("qdrant paper_id shape", + str(payload.get("paper_id", "")).startswith("doi:"), + payload.get("paper_id")) + except Exception as exc: + check("qdrant sample point", False, exc) + + # 3) Neo4j 探活 + 计数 + 索引 + try: + from neo4j import GraphDatabase + password = Path(args.neo4j_password_file).read_text(encoding="utf-8").strip().splitlines()[0] + driver = GraphDatabase.driver( + args.neo4j_uri, auth=(args.neo4j_user, password), connection_timeout=15 + ) + with driver.session() as session: + counts = session.run( + "MATCH (e:Entity) WITH count(e) AS entities " + "MATCH (p:Paper) WITH entities, count(p) AS papers " + "MATCH ()-[r:ASSERTS]->() WITH entities, papers, count(r) AS asserts " + "MATCH ()-[s:SUPPORTED_BY]->() " + "RETURN entities, papers, asserts, count(s) AS supported_by" + ).single() + check("neo4j counts", counts is not None, + f"entities={counts['entities']} papers={counts['papers']} " + f"asserts={counts['asserts']} supported_by={counts['supported_by']}") + indexes = session.run( + "SHOW INDEXES YIELD name, labelsOrTypes, properties, state, type " + "WHERE state = 'ONLINE' " + "RETURN name, labelsOrTypes, properties, state, type" + ).data() + check("neo4j indexes online", bool(indexes), + json.dumps(indexes, ensure_ascii=False)[:200]) + driver.close() + except Exception as exc: + check("neo4j counts", False, exc) + + # 4) normalized 同款查询跑通（resolve_entities + one_hop 口径） + try: + from neo4j import GraphDatabase + password = Path(args.neo4j_password_file).read_text(encoding="utf-8").strip().splitlines()[0] + driver = GraphDatabase.driver( + args.neo4j_uri, auth=(args.neo4j_user, password), connection_timeout=15 + ) + with driver.session() as session: + row = session.run( + "MATCH (e:Entity) RETURN e.canonical_id AS canonical_id, " + "e.name AS name, e.name_norm AS name_norm, e.entity_type AS entity_type " + "LIMIT 1" + ).single() + if row is None: + check("neo4j normalized query", False, "no entity to seed") + else: + hops = session.run( + "MATCH (seed:Entity {canonical_id: $cid}) " + "MATCH (seed)-[r:ASSERTS]-(other:Entity) " + "RETURN r.assertion_id AS assertion_id, r.relation_norm AS relation, " + "r.relation_category AS relation_category, r.layer AS layer " + "LIMIT 3", + cid=row["canonical_id"], + ).data() + check("neo4j one_hop query", True, + f"seed={row['name']} hops={len(hops)}") + driver.close() + except Exception as exc: + check("neo4j normalized query", False, exc) + + failed = [result for result in RESULTS if not result["ok"]] + print(f"\n[smoke] {len(RESULTS) - len(failed)}/{len(RESULTS)} passed") + sys.exit(1 if failed else 0) + + +if __name__ == "__main__": + main() diff --git a/tooling/aqua_kb/start_aqua_kb.sh b/tooling/aqua_kb/start_aqua_kb.sh new file mode 100644 index 0000000..42cb0d5 --- /dev/null +++ b/tooling/aqua_kb/start_aqua_kb.sh @@ -0,0 +1,181 @@ +#!/usr/bin/env bash +# tooling/aqua_kb/start_aqua_kb.sh —— aqua 试库实例启动脚本(用户态,勿碰生产) +# +# 启动两个用户态实例(全部 setsid nohup,数据只在 kb-libraries/aqua/data/ 下): +# - Neo4j: bolt 19889, http 19890, data=aqua/data/neo4j, 密码随机存 aqua/data/neo4j/.auth +# - Qdrant: http 27435, grpc 27436, storage=aqua/data/qdrant/storage +# +# 用法: +# bash start_aqua_kb.sh [start|stop|status] +set -euo pipefail + +REPO_ROOT="/home/xzq/5090/teamWorkspace/zh/PlantAI-rag-new" +AQUA_DATA="$REPO_ROOT/kb-libraries/aqua/data" +NEO4J_HOME="/opt/neo4j-community-5.15.0" +QDRANT_BIN="/usr/local/bin/qdrant" + +BOLT_PORT=19889 +HTTP_PORT=19890 +QDRANT_HTTP=27435 +QDRANT_GRPC=27436 + +NEO4J_RUN="$AQUA_DATA/neo4j" # data+logs 根 +NEO4J_DATA="$NEO4J_RUN/data" +NEO4J_LOGS="$NEO4J_RUN/logs" +NEO4J_CONF="$NEO4J_RUN/neo4j.conf" +AUTH_FILE="$NEO4J_DATA/.auth" +PID_DIR="$AQUA_DATA/pids" + +QDRANT_STORAGE="$AQUA_DATA/qdrant/storage" +QDRANT_CONF="$AQUA_DATA/qdrant/config.yaml" + +mkdir -p "$AQUA_DATA" "$NEO4J_DATA" "$NEO4J_LOGS" "$NEO4J_RUN/run" "$PID_DIR" "$QDRANT_STORAGE" + +gen_password() { + if [[ -f "$AUTH_FILE" ]]; then + head -1 "$AUTH_FILE" + return + fi + local pw + pw="$(head -c 24 /dev/urandom | base64 | tr -d '=+/' | head -c 32)" + echo "$pw" > "$AUTH_FILE" + chmod 600 "$AUTH_FILE" + echo "$pw" +} + +wait_port() { + local port=$1 tries=$2 name=$3 + for _ in $(seq 1 "$tries"); do + if (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null; then + exec 3>&- 2>/dev/null || true + echo "[aqua-kb] $name UP 127.0.0.1:$port" + return 0 + fi + sleep 1 + done + echo "[aqua-kb] $name 启动超时: 127.0.0.1:$port" >&2 + return 1 +} + +start_neo4j() { + if [[ -f "$PID_DIR/neo4j.pid" ]] && kill -0 "$(cat "$PID_DIR/neo4j.pid")" 2>/dev/null; then + echo "[aqua-kb] neo4j 已运行 pid=$(cat "$PID_DIR/neo4j.pid")" + return 0 + fi + local pw + pw="$(gen_password)" + # 独立 conf 目录:复制发行包 conf 骨架(jvm.options/server-logs.xml 等必需), + # 然后整体覆盖 neo4j.conf(不含生产 32G 堆/端口配置;不写 /opt) + if [[ ! -f "$NEO4J_RUN/server-logs.xml" ]]; then + cp -a "$NEO4J_HOME/conf/." "$NEO4J_RUN/" + fi + cat > "$NEO4J_CONF" </dev/null 2>&1; then + echo "[aqua-kb] set-initial-password 失败,回退手动 auth.ini(明文,首启后生效)" >&2 + mkdir -p "$NEO4J_DATA/dbms" + printf 'neo4j:%s,admin\n' "$pw" > "$NEO4J_DATA/dbms/auth.ini" + chmod 600 "$NEO4J_DATA/dbms/auth.ini" + fi + fi + setsid nohup env NEO4J_CONF="$NEO4J_RUN" "$NEO4J_HOME/bin/neo4j" console \ + > "$NEO4J_LOGS/console.out" 2>&1 & + echo $! > "$PID_DIR/neo4j.pid" + echo "[aqua-kb] neo4j 启动 pid=$(cat "$PID_DIR/neo4j.pid") (bolt $BOLT_PORT)" + wait_port "$BOLT_PORT" 60 "neo4j" +} + +start_qdrant() { + if [[ -f "$PID_DIR/qdrant.pid" ]] && kill -0 "$(cat "$PID_DIR/qdrant.pid")" 2>/dev/null; then + echo "[aqua-kb] qdrant 已运行 pid=$(cat "$PID_DIR/qdrant.pid")" + return 0 + fi + cat > "$QDRANT_CONF" < "$AQUA_DATA/qdrant/qdrant.out" 2>&1 & + echo $! > "$PID_DIR/qdrant.pid" + echo "[aqua-kb] qdrant 启动 pid=$(cat "$PID_DIR/qdrant.pid") (http $QDRANT_HTTP)" + for _ in $(seq 1 30); do + if curl -sf --max-time 2 "http://127.0.0.1:$QDRANT_HTTP/healthz" >/dev/null 2>&1; then + echo "[aqua-kb] qdrant UP http://127.0.0.1:$QDRANT_HTTP" + return 0 + fi + sleep 1 + done + echo "[aqua-kb] qdrant 启动超时: $QDRANT_HTTP" >&2 + return 1 +} + +stop_one() { + local name=$1 pidfile="$PID_DIR/$1.pid" + if [[ -f "$pidfile" ]]; then + local pid + pid="$(cat "$pidfile")" + if kill -0 "$pid" 2>/dev/null; then + kill "$pid" 2>/dev/null || true + echo "[aqua-kb] 已停止 $name pid=$pid" + else + echo "[aqua-kb] $name 进程不存在(pid=$pid)" + fi + rm -f "$pidfile" + else + echo "[aqua-kb] $name 未记录 pid" + fi +} + +status_one() { + local name=$1 pidfile="$PID_DIR/$1.pid" + if [[ -f "$pidfile" ]] && kill -0 "$(cat "$pidfile")" 2>/dev/null; then + echo "[aqua-kb] $name RUNNING pid=$(cat "$pidfile")" + else + echo "[aqua-kb] $name STOPPED" + fi +} + +case "${1:-start}" in + start) + start_neo4j + start_qdrant + echo "[aqua-kb] 完成。密码: $AUTH_FILE" + ;; + stop) + stop_one neo4j + stop_one qdrant + ;; + status) + status_one neo4j + status_one qdrant + ;; + *) + echo "用法: bash start_aqua_kb.sh [start|stop|status]" >&2 + exit 1 + ;; +esac