feat(aqua): KB build tooling, aqua configs, QA artifacts (P1)

- phase2b_converter: result.json -> neo4j.cypher + qdrant_points.jsonl
  (payload keys replicate direct_flow_retriever contract; kb_version=
  aqua_srt_v1_20260930_normalized, source_system=Aqua_canonical)
- qdrant_loader: bge-m3 embed + batched upsert + checkpoint resume
- registry_builder: paper blocks -> registry jsonl (plant format)
- batch_runner: manifest -> convert -> neo4j -> qdrant, all checkpointed
- start_aqua_kb.sh: user-space trial instances (Neo4j 19889 / Qdrant 27435)
- smoke 11/11 green incl. resume drills
- aqua configs: entity_alias_aqua.yaml (22 entries), relation_taxonomy_aqua.yaml
  (45 groups); manifest points to aqua configs
- QA artifacts: golden_aqua.jsonl (41 Qs zh/en), qc_standards.md,
  budget_template.md + cost_model.py (measured pilot tokens)
- pilot: 500 MD + 30 PDF extracted, 0 failures; QC stats computed

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-12 22:19:40 +08:00
co-authored by Claude Fable 5
parent a9a0ece4b4
commit 730692a1f6
15 changed files with 2642 additions and 2 deletions
@@ -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
- 低氧
- 缺氧
@@ -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
+2 -2
View File
@@ -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: ""
+63
View File
@@ -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)。
+41
View File
@@ -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)"}
@@ -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 一旦发布不得重用)。
+51
View File
@@ -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_<batch_id>.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 校准。
+239
View File
@@ -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(含错误),全部步骤可重跑
状态文件(<out>/state/,JSONL 每行一个 paper_id):
converted.jsonl / neo4j_loaded.jsonl / qdrant_loaded.jsonl / failed.jsonl
用法:
python batch_runner.py --manifest pilot_manifest.json --result-root <dir> \
--out <build dir> --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()
+185
View File
@@ -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())
+234
View File
@@ -0,0 +1,234 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""自造 3 个最小 v16 结构 result.json fixture(手写字段按 v16_1_2 契约)。
产出 <dir>/<paper_id>/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()
+443
View File
@@ -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 重跑逐字节一致):
<out>/<paper_id>/neo4j.cypher —— Entity/Paper/ASSERTS/SUPPORTED_BY
<out>/<paper_id>/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>`;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 <result.json> --out <dir> \
[--release-id aqua_srt_v1_20260930_normalized] [--paper-json <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()
+229
View File
@@ -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 <dir|file> --qdrant http://127.0.0.1:27435 \
--collection aqua_srt_v1_20260930_normalized --batch 200 \
--checkpoint <file.jsonl> \
[--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()
+151
View File
@@ -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 <dir of result.json...> --out <registry.jsonl>
"""
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()
+173
View File
@@ -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 <path to .auth> [--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()
+181
View File
@@ -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" <<EOF
server.bolt.listen_address=127.0.0.1:$BOLT_PORT
server.http.listen_address=127.0.0.1:$HTTP_PORT
server.https.listen_address=127.0.0.1:19891
server.directories.data=$NEO4J_DATA
server.directories.logs=$NEO4J_LOGS
server.directories.run=$NEO4J_RUN/run
server.directories.plugins=$NEO4J_HOME/plugins
dbms.security.auth_enabled=true
server.memory.heap.initial_size=1G
server.memory.heap.max_size=4G
server.memory.pagecache.size=2G
EOF
# 首次初始化:auth.ini 不存在时用 set-initial-password 写入(与生产同法)
if [[ ! -f "$NEO4J_DATA/dbms/auth.ini" ]]; then
echo "[aqua-kb] 首次初始化 neo4j(写入初始密码)..."
if ! NEO4J_CONF="$NEO4J_RUN" "$NEO4J_HOME/bin/neo4j-admin" dbms set-initial-password "$pw" >/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" <<EOF
storage:
storage_path: $QDRANT_STORAGE
snapshots_path: $QDRANT_STORAGE/snapshots
service:
host: 127.0.0.1
http_port: $QDRANT_HTTP
grpc_port: $QDRANT_GRPC
enable_tls: false
max_workers: 8
telemetry_disabled: true
EOF
# 用户态 shell ulimit -n 默认 1024,255 worker 会 Too many open files(actix panic);
# hard limit 1048576,先提升再启动(与生产 systemd LimitNOFILE 同法)。
ulimit -n 1048576
setsid nohup "$QDRANT_BIN" --config-path "$QDRANT_CONF" \
< /dev/null > "$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