插件开发指南¶
一个 ABI 插件对应一种生物学分析，并通过共享生命周期对外提供能力。新插件优先使用 下面的声明式接口；只有 manifest 无法表达所需行为时，才使用底层 Python 协议。
推荐的声明式接口¶
将 abi-plugin.yaml 放在插件模块旁边，并继承 DeclarativeABIPlugin。基类会从
manifest 读取插件身份、工具注册表和标准表路径，避免在多个位置重复维护：
from abi.plugin import DeclarativeABIPlugin
class MyPlugin(DeclarativeABIPlugin):
def load_config(self, config_path=None, **kwargs): ...
def build_plan(self, config, *, check_files=True): ...
def parse_outputs(self, tool_id, output_dir, sample_id): ...
def write_report(self, plan, result_dir): ...
如果 monorepo 将声明文件与 Python 模块分开存放，只需设置一个类属性，
例如 plugin_root = Path("plugins/my_analysis")。
基类会在导入时校验 manifest 及其声明的所有路径。插件发现还会强制
entry-point 名称、manifest 中的 plugin_id 与 entry_point 一致。运行时
注册表、工具合约和环境校验保持不变；发布前运行
abi contract-lint --strict。
底层 Python 接口¶
实现 abi.interfaces.ABIPlugin 协议：
plugin_iddisplay_namedescriptionreport_titleload_config()build_plan()registry()table_schemas()parse_outputs()write_report()
通过以下方式注册插件：
[project.entry-points."abi.plugins"]
my_analysis = "my_package.plugins:MyPlugin"
entry-point 的键必须与 abi-plugin.yaml 中的 plugin_id 完全一致。
插件目录¶
推荐布局：
plugins/my_analysis/
abi-plugin.yaml
config_default.yaml
sample_sheet_template.tsv
tool_registry.yaml
standard_tables.yaml
limitations.yaml ← 强制且非空
tool_contracts/
tool_a.yaml
skills/ ← 随包捆绑的 SKILL.md 文件
tool_a/SKILL.md
_engine/ ← 可选：复杂引擎代码（参见 metagenomic_plasmid）
每个插件必须提供非空的 limitations.yaml；否则 contract lint 以
missing_limitations、invalid_limitations 或 empty_limitations 失败。
生成的报告始终包含 limitations 章节（列表为空时使用明确的兜底文本）。
对于具有大量内部逻辑的复杂插件，使用带有私有 _engine/ 子目录的自包含包。参见 plugins/metagenomic_plasmid/ 获取规范示例。
技能与 Agent 集成¶
每个工具应在 skills/<tool_name>/SKILL.md 下有一个 SKILL.md 文件。
技能捆绑在包内 src/abi/skills/ 中，并通过以下方式安装到 Claude Code：
abi install-skills # → ~/.claude/skills/abi/
要添加新技能，在 src/abi/skills/<tool_name>/SKILL.md 下创建目录和 SKILL.md 文件。abi_agent/SKILL.md 技能教会 Claude Code 如何使用 abi CLI 本身；其他技能记录各个生物信息学工具。
工具 Skill 解释科学目的、参数、输入、输出和失败模式，但不会新增 MCP 函数，也不能绕过 ABI 生命周期。
让插件可被 Agent 调用¶
Agent 传输层暴露共享生命周期工具。新插件通过 entry-point 被发现后，Agent 使用 analysis_type 参数调用它；不要创建独立的 abi_my_plugin_plan 或 abi_my_plugin_run 工具。
提供可发现的生物学身份¶
manifest 身份是未经专门训练的 Agent 通过 abi_list_types 和 abi_export_agent_context 看到的信息：
abi_version: "0.1"
plugin_id: my_analysis
display_name: My Biological Analysis
description: "分析 <输入> 并生成 <主要生物学结果>。"
report_title: My Biological Analysis Report
entry_point: my_package.plugin:MyPlugin
description 用于插件选择，应说明生物学目标、输入类型和主要结果，避免只写实现细节或宣传性内容。
entry-point 键、manifest plugin_id 和插件类身份必须一致。提供 config_default.yaml 和 sample_sheet_template.tsv，让用户与 Agent 从明确 Schema 开始，而不是编造元数据。
让工作流可以被查询¶
abi_query 读取声明式 DAG 和工具注册表。使用稳定的节点与工具 ID，声明平台，并为每个节点提供有意义的输入、输出、依赖关系和说明。
Agent 应能在不读取源码的情况下回答：
abi query --type my_analysis --what stages
abi query --type my_analysis --what tools
abi query --type my_analysis --what platforms
abi query --type my_analysis --step qc_tool --what inputs
abi query --type my_analysis --step qc_tool --what outputs
资源属于特定节点时，应明确声明关联关系，使步骤级资源查询和预检诊断能够解释缺少的内容。
保持规划安全且确定¶
load_config()可以读取配置，但不得安装资源或运行工具。build_plan()必须根据显式输入生成确定的步骤和路径。plan和dry_run可以写入审查产物，但不得执行外部分析工具。真实工具只能在共享确认门之后通过
run执行。report必须读取已有发布结果，不能静默重新运行工作流。
Agent 权限 profile 由 ABI 核心层管理。插件不得在规划、解析、报告或诊断中创建隐藏执行路径。
发布 Agent 能解释的结果¶
插件的 table_schemas() 定义 abi_export_agent_context 返回的 standard_tables。表名、列、单位、标识符和缺失值约定必须稳定。
解析器应把工具输出标准化到这些表格。报告和图形应读取已发布表格，而不是重新解析原始日志或私有中间文件。
不适合标准表格的稳定最终产物通过 published_outputs(plan) 发布。工作流包含多个输出分支时，发布方法、溯源、局限性和带分支限定的 manifest。
可恢复失败应抛出 ABI 错误类型或提供结构化诊断。Agent 使用 error_code 和 diagnostic_hints；原始 traceback 不是操作契约。
检查 Agent 上下文¶
安装插件后，检查其机器可读契约：
abi list-types --output-json
abi export-agent-context --type my_analysis
abi doctor-agent --type my_analysis
abi export-tools --type my_analysis --format openai
确认插件说明足以支持选择，标准表格列表完整，且执行工具只有在明确请求时才被导出。
不要为插件手工编辑 OpenAI、Anthropic、Gemini 或 MCP Schema。ABI 从单一事实来源生成生命周期描述符，并在导出时注入插件范围。
添加端到端 Agent 契约测试¶
除插件单元测试外，还应通过 ABIAgentInterface 覆盖发现、上下文导出、规划、dry-run、执行确认和结果验证。
import json
from abi.agent import ABIAgentInterface
def call(tool, arguments):
return json.loads(ABIAgentInterface().dispatch(tool, arguments))
def test_agent_can_discover_and_plan_my_plugin(tmp_path, plugin_fixture):
listed = call("abi_list_types", {})
plugin_ids = {
row["analysis_type"]
for row in listed["result"]["analysis_types"]
}
assert "my_analysis" in plugin_ids
context = call(
"abi_export_agent_context",
{"analysis_type": "my_analysis"},
)
assert context["status"] == "success"
assert context["result"]["standard_tables"]
assert "abi_run" in context["result"]["unsafe_tools"]
common = {
"analysis_type": "my_analysis",
"config_path": str(plugin_fixture.config),
"sample_sheet": str(plugin_fixture.sample_sheet),
}
plan = call("abi_plan", {**common, "outdir": str(tmp_path / "plan")})
assert plan["status"] == "success"
dry = call("abi_dry_run", {**common, "outdir": str(tmp_path / "dry")})
assert dry["status"] == "success"
run = call("abi_run", {**common, "confirm_execution": False})
assert run["status"] == "confirmation_required"
在 golden_traces/<plugin_id>.jsonl 添加已知正确的生命周期，并在 tests/integration/test_golden_traces.py 回放。Trace 应包含错误恢复和需要确认的 run 尝试。
在插件测试套件中定义 plugin_fixture，让它生成有效的最小配置和样本表。Fixture 应使用小型合成数据，并且不依赖特定机器的资源路径。
Agent 可调用插件检查清单¶
插件出现在
abi list-types --output-json中。上下文导出包含正确的标准表格、产物、权限和错误码。
DAG 阶段、工具、平台、输入、输出和资源都可以查询。
默认配置和样本表模板没有含义模糊的字段。
plan、check 和 dry-run 不执行外部工具即可完成。
未授权时
abi_run返回confirmation_required。真实结果通过
abi_validate_result，报告使用标准表格。模型描述符和 MCP profile 不需要插件特有传输代码。
Golden trace 和回归测试覆盖完整 Agent 生命周期。
工具合约¶
合约是权威的、机器可读的工具声明：
tool_idcategoryexecution.executableexecution.command_template声明的输入/输出模板字段
标准化的标准表名称
tool_registry.yaml 是精简的运行策略索引。每个合约必须有且仅有一个
注册表条目，但条目只需要工具 ID 和运行策略：
tools:
- id: fastp
required: true
default_enabled: true
skill_path: skills/fastp/SKILL.md
不要在其中重复 name、category、executable 或 command_template。
旧注册表仍可包含这些字段，但如果重复声明冲突，ABI 会直接拒绝，
而不是静默选择其中一份。
环境名称不存储在单独的合约或注册表中。它们集中在 environments.yaml 中的
tool_assignments: 下（每个插件一个映射），由 ToolRegistry 在运行时注入正确的
env_name。参见：environments.yaml、scripts/emit_env_yamls.py。
在插件测试中使用 assert_plugin_contract(plugin)。
步骤输出合约¶
复杂插件可以在执行计划中嵌入逐步骤合约。对于 DAG 驱动的 metagenomic plasmid 插件，pipeline_dag.yaml 中的每个节点声明其 outputs 和可选的 assertions；规划器将这些字段复制到 PlanStep.params["_contract"] 中以供运行时执行。
契约覆盖率是强制的：每个调用外部工具的节点必须为输出声明 contract 检查；无产出的节点（纯聚合或标记步骤）必须显式设置 contract: {exempt: true, reason: ...}。该门槛由 abi contract-lint（missing_output_contract、exempt_missing_reason、exempt_mixed_with_checks）和插件校验强制执行；scripts/audit_contract_coverage.py 可输出各插件的覆盖率报告。
支持的输出检查包括：
min_size：文件或目录的最小字节大小，如"1KB"。extensions：允许的文件后缀，如[.fastq, .fastq.gz]。contains：输出目录中必须包含的文件。min_files：目录下最少的常规文件数，适用于生成的索引。min_contigs：最少的 FASTA contig 数量。required_keys：JSON 输出必需的顶层键。schema：带有简单类型/范围约束的点分 JSON 字段。
检查项必须嵌套在输出的 contract 键下；将 min_size 等检查与 type
并列是无效声明，运行时不会执行：
outputs:
clean_read1:
type: file
format: fastq.gz
path: "{outdir}/{sample_id}.clean.fastq.gz"
contract:
min_size: "1KB"
extensions: [".fastq.gz"]
断言在输出验证之后根据 output_files、output_json 和 return_code 进行评估。示例：
assertions:
- "output_json.summary.after_filtering.total_reads > 0"
- "output_files.clean_read1 exists"
当声明创建自身输出目录的工具的输出时，继续使用 output_dir。通用执行器有意仅创建父目录，因为某些工具在 output_dir 于执行前已存在时会失败。
如果规划器输出抽象输出路径而工具写入固定名称，请确保合约的 format 和文件名约定是明确的。执行器在检查合约之前按 output_dir、format、样本 ID 和 R1/R2 双端提示解析实际文件。
标准表¶
解析器必须仅写入插件声明的表。空表仍应以稳定的表头存在，以便 Agent 在不解析原始工具输出的情况下检查结果。
发布输出¶
插件可以实现 published_outputs(plan)，将插件特有的最终产物加入传输无关的
RuntimeResult.outputs 映射。只应返回稳定且已经存在的路径，并使用不会与 ABI
公共结果键冲突的标签。该钩子适合暴露带版本的产物清单等最终结果；中间文件仍
通过执行计划发现，不应逐个发布。
如果一个预设会生成多个最终报告，应发布带分支限定的标签；只有在恰好存在一套完整
报告时才添加通用别名。例如，EasyMetagenome 单分支运行发布 report_manifest，组合
预设则发布 taxonomy_report_manifest 和 functional_report_manifest。带版本的清单
应标明工作流，并关联其汇总的标准表和报告。
共享基础设施¶
插件应从公开 SDK 导入：
模块 |
用途 |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
DAG 驱动的计划构建¶
与其手写遍历样本、构造 PlanStep 对象的 build_plan()（约 200 行样板代码），插件必须在 pipeline_dag.yaml 文件中声明工作流，并使用通用 DAG 规划器：
# 在插件的 build_plan() 中：
def build_plan(self, config, *, check_files=True):
context = self.build_sample_context(config, check_files=check_files)
from abi.dag_planner import build_plan_from_dag
return build_plan_from_dag(
self.root / "pipeline_dag.yaml", config, context
)
pipeline_dag.yaml 结构¶
pipeline_id: my_analysis
platforms: [illumina]
# 类别 → 子目录映射
category_dirs:
qc: 01_qc
alignment: 02_alignment
nodes:
qc_fastp:
tool_id: fastp
category: qc
scope: per_sample # per_sample（默认）或 cross_sample
depends_on: []
inputs:
read1: {type: file, source: sample_sheet}
read2: {type: file, source: sample_sheet}
outputs:
clean_read1:
type: file
path: "{outdir}/{category_dir}/{sample_id}/{sample_id}_R1.clean.fastq.gz"
output_dir:
type: directory
path: "{outdir}/{category_dir}/{sample_id}"
aggregation_step:
tool_id: my_aggregator
scope: cross_sample # 运行一次，收集所有 per-sample 输出
depends_on: [qc_fastp]
inputs:
per_sample_data: {aggregate: per_sample_outputs}
声明式 TSV 解析¶
对于输出简单 TSV/JSON/日志的工具，在 parsers.yaml 中声明列映射，代替手写 Python 解析器函数。支持三种源类型：
源类型 |
用途 |
示例工具 |
|---|---|---|
|
CSV/TSV 列重映射 |
AMRFinderPlus、featureCounts |
|
嵌套 JSON 展平 |
fastp（summary before/after 块） |
|
分隔符日志解析 |
STAR（Log.final.out 管道分隔） |
tsv_mapping 示例：
parsers:
my_tool:
source:
type: tsv_mapping
pattern: "*.tsv"
delimiter: "\t"
target_table: my_standard_table
columns:
gene_name: {sources: [Gene, gene_name], default: ""}
coverage: {sources: [Coverage, cov_pct], default: "0"}
constants:
tool: my_tool
在 parse_outputs() 中接入：
def parse_outputs(self, tool_id, output_dir, sample_id):
if self._tsv_mapper.has_parser(tool_id):
rows = self._tsv_mapper.parse(tool_id, output_dir, sample_id=sample_id)
if rows:
return {self._tsv_mapper.get_target_table(tool_id): rows}
# 复杂解析器保留为 Python
...
测试插件¶
每个插件必须包含测试。新插件的最小测试套件覆盖三个方面：合约合规、注册表加载和计划生成。
最小测试文件 (tests/test_my_plugin.py)¶
import pytest
from abi.testing import assert_plugin_contract
from your_package.plugin import MyPlugin # 你的已安装插件入口点
def test_plugin_contract():
"""插件满足 ABIPlugin 协议。"""
plugin = MyPlugin()
assert_plugin_contract(plugin)
def test_registry_loads():
"""工具注册表 YAML 解析无错误。"""
plugin = MyPlugin()
registry = plugin.registry()
tools = registry.list_tools()
assert len(tools) > 0
# 验证预期工具已注册
tool_ids = [t["id"] for t in tools]
assert "fastp" in tool_ids
def test_build_plan(mock_sample_context, tmp_path):
"""build_plan() 对默认配置返回有效的 ExecutionPlan。"""
plugin = MyPlugin()
config = plugin.load_config()
config["outdir"] = str(tmp_path)
plan = plugin.build_plan(config)
assert len(plan.steps) > 0
# QC 步骤始终最先
assert plan.steps[0].step_id.startswith("qc_")
def test_parse_outputs_handles_missing_files(tmp_path):
"""解析器对缺失文件返回空结果（而非错误）。"""
plugin = MyPlugin()
result = plugin.parse_outputs("fastp", tmp_path, "S1")
assert isinstance(result, dict)
@pytest.mark.smoke
@pytest.mark.requires_tools
def test_real_execution_smoke(tmp_path):
"""使用合成数据执行完整管线。"""
# 生成最小测试数据，运行真实工具，验证输出……
pass
插件测试可用的 Fixtures¶
tests/conftest.py 中的所有 fixtures 无需导入即可使用：
Fixture |
类型 |
用途 |
|---|---|---|
|
|
带有 illumina 平台的单样本输入 |
|
|
包含两个分组的单样本上下文 |
|
|
用于脚手架的最小有效工具合约 |
|
|
包含 results/logs/provenance/tables/ 的临时目录 |
基准测试¶
对于数值级验证，使用 run_benchmark()：
from abi.testing.benchmark import run_benchmark
@pytest.mark.smoke
@pytest.mark.requires_tools
def test_my_plugin_benchmark(tmp_path):
result = run_benchmark(
plugin_id="my_analysis",
dataset_path=Path("data/benchmarks/my_analysis"),
outdir=tmp_path / "results",
)
assert result.total > 0
assert result.passed >= result.total * 0.7 # 开发阶段阈值
参见 docs/zh/testing.md 获取完整测试指南。
资源管理¶
ABI 为生物信息学数据库提供资源发现和自动安装系统。插件作者声明资源需求；ABI 负责检查、下载和安装后钩子。
声明资源¶
资源在 plugins/<name>/abi-plugin.yaml 中声明：
resources:
my_database:
name: "My Database"
description: "MyTool 的参考数据库"
url: "https://example.com/my_database.tar.gz"
size_gb: 2.5
required_by: [my_tool]
install_post: "makeblastdb -in {resource_dir}/sequences.fasta -dbtype nucl"
env_name: my_env
ResourceSpec 字段¶
字段 |
类型 |
描述 |
|---|---|---|
|
|
人类可读的名称 |
|
|
资源提供的内容 |
|
|
下载 URL（支持 http/https/S3） |
|
|
大约下载大小 |
|
|
依赖此资源的工具 ID |
|
|
下载后运行的 shell 命令（如 |
|
|
提供安装后工具的 conda 环境 |
CLI 命令¶
# 检查哪些资源可用/缺失
abi check-resources --type my_analysis
# 下载并安装缺失的资源（需要确认）
abi setup-resources --type my_analysis --confirm
环境解析¶
工具 → 环境分配在 environments.yaml 中，而非在单个工具合约中。注册工具时，将其环境分配添加到 environments.yaml：
tool_assignments:
my_analysis:
my_tool: my_env
ToolRegistry 在运行时注入正确的 env_name。运行 scripts/emit_env_yamls.py 重新生成每个环境的 envs/*.yml 文件。
断言表达式参考¶
步骤合约支持以简单表达式语言编写的断言。断言在工具执行后根据解析后的输出进行评估。
变量¶
变量 |
类型 |
示例 |
|---|---|---|
|
Any |
|
|
Path |
|
|
Path |
|
|
int |
|
运算符¶
运算符 |
示例 |
含义 |
|---|---|---|
|
|
大于 |
|
|
大于等于 |
|
|
小于 |
|
|
小于等于 |
|
|
等于 |
|
|
不等于 |
|
|
文件/目录存在 |
|
|
字符串包含 |
在 pipeline_dag.yaml 中编写断言¶
nodes:
qc_fastp:
tool_id: fastp
# ...
assertions:
- "output_json.summary.after_filtering.total_reads > 0"
- "output_json.summary.after_filtering.q30_rate >= 0.8"
- "output_files.clean_read1 exists"
- "output_files.clean_read2 exists"
- "return_code == 0"
断言评估¶
断言在输出验证之后进行评估。如果任何断言失败，步骤将被标记为失败，并抛出包含失败断言详情的 ContractViolationError。所有断言必须通过步骤才能成功。
执行安全¶
插件应使 plan 和 dry_run 对 Agent 安全。真实的外部工具执行只能在显式确认后通过 run 进行。