从零开发可用插件

本教程会做出一个真正可运行的领域插件：输入阻尼振子的角频率 ω 和阻尼率 γ，计算品质因数：

Q = ω / (2γ)

插件会返回 quality_factor 指标，写出一份 JSON 分析产物，并附带一个无需 API key 的确定性 Reasoner，使你可以立即跑通端到端流程。

完成品已放在 examples/plugins/oscillator_quality，对应运行配置是 examples/plugin_tutorial.yaml。建议先自己跟着创建，再与完成品比较。

1. 准备开发环境

需要 Python 3.11 或更高版本。在仓库根目录执行：

python -m venv .venv
. .venv/bin/activate
python -m pip install -e '.[dev]'

创建目录：

mkdir -p examples/plugins/oscillator_quality/oscillator_quality
touch examples/plugins/oscillator_quality/oscillator_quality/__init__.py

最终结构是：

examples/plugins/oscillator_quality/
├── plugin.yaml
└── oscillator_quality/
    ├── __init__.py
    └── plugin.py

ZIP 根目录可以多一层文件夹，Registry 会查找清单；但整个包内必须只有一个明确的受支持清单（plugin.yaml、plugin.yml 或 plugin.json）。本教程统一使用 plugin.yaml。

2. 写插件清单

创建 examples/plugins/oscillator_quality/plugin.yaml：

name: tutorial.oscillator
display_name: Oscillator Quality Tutorial
version: 0.1.0
type: domain
description: A zero-dependency damped-oscillator quality-factor example.
category: domain
domain: classical_mechanics
sdk_version: "1"
entrypoint: oscillator_quality.plugin:OscillatorQualityPlugin
produced_artifacts:
  - OscillatorAssessment
capabilities:
  - name: tutorial.oscillator.evaluate
    description: Calculate the quality factor of a damped oscillator.
    input_schema:
      angular_frequency: number
      damping_rate: number
    required_inputs:
      - angular_frequency
      - damping_rate
    output_schema:
      metrics: object
      observations: object
      artifacts: array
    output_summary: A quality-factor metric and a JSON assessment artifact.
    metric_outputs:
      - quality_factor
    example:
      angular_frequency: 4.0
      damping_rate: 0.5
    provides:
      - scientific_evaluation
    produces_artifact_kinds:
      - oscillator_assessment
    scientific_roles:
      - numerical_simulation

这里有三个稳定标识：

  • 插件 ID：tutorial.oscillator；
  • Python 入口：oscillator_quality.plugin:OscillatorQualityPlugin；
  • 能力 ID：tutorial.oscillator.evaluate。

metric_outputs 很重要。创建研究合同时，Runtime 用它验证 quality_factor 确实有可能被所选能力产出。example 会帮助 Reasoner 使用正确字段名。

3. 写插件实现

保持 __init__.py 为空。创建 oscillator_quality/plugin.py，内容如下：

from __future__ import annotations

import json

from scientific_agent.context import ResearchContext
from scientific_agent.models import ActionType, ArtifactType, JsonObject, ResearchAction
from scientific_agent.plugins.sdk import (
    CapabilitySpec,
    PluginArtifact,
    PluginExecutionContext,
    PluginManifest,
    PluginResult,
    ScientificPlugin,
)
from scientific_agent.reasoning import Reasoner

CAPABILITY = "tutorial.oscillator.evaluate"


class OscillatorTutorialReasoner(Reasoner):
    """Make the tutorial runnable without an API key."""

    async def propose_action(self, context: ResearchContext) -> ResearchAction:
        completed = [
            experiment
            for experiment in context.recent_experiments
            if experiment.status == "COMPLETED"
        ]
        if not completed:
            return ResearchAction(
                action_type=ActionType.RUN_EXPERIMENT,
                purpose="Measure the quality factor for the declared oscillator",
                reasoning_summary=(
                    "No oscillator assessment has been recorded, so evaluate the reference case."
                ),
                capability=CAPABILITY,
                inputs={"angular_frequency": 4.0, "damping_rate": 0.5},
                expected_observation="A positive quality factor with a reusable JSON record.",
                success_criteria={"quality_factor_at_least": 3.0},
                estimated_cost={"experiment": 1},
            )
        return ResearchAction(
            action_type=ActionType.DONE,
            purpose="Conclude the oscillator assessment",
            reasoning_summary="The requested oscillator assessment has been recorded.",
        )


class OscillatorQualityPlugin(ScientificPlugin):
    @property
    def manifest(self) -> PluginManifest:
        # plugin.yaml is authoritative when loaded from disk. This declaration
        # keeps direct SDK use and unit tests self-contained.
        return PluginManifest(
            name="tutorial.oscillator",
            version="0.1.0",
            plugin_type="domain",
            display_name="Oscillator Quality Tutorial",
            description="A zero-dependency damped-oscillator quality-factor example.",
            category_type="domain",
            domain="classical_mechanics",
            entrypoint="oscillator_quality.plugin:OscillatorQualityPlugin",
            produced_artifacts=("OscillatorAssessment",),
            capabilities=(
                CapabilitySpec(
                    name=CAPABILITY,
                    description="Calculate the quality factor of a damped oscillator.",
                    input_schema={
                        "angular_frequency": "number",
                        "damping_rate": "number",
                    },
                    required_inputs=("angular_frequency", "damping_rate"),
                    output_schema={
                        "metrics": "object",
                        "observations": "object",
                        "artifacts": "array",
                    },
                    output_summary=(
                        "A quality-factor metric and a JSON assessment artifact."
                    ),
                    metric_outputs=("quality_factor",),
                    example={"angular_frequency": 4.0, "damping_rate": 0.5},
                    provides=("scientific_evaluation",),
                    produces_artifact_kinds=("oscillator_assessment",),
                    scientific_roles=("numerical_simulation",),
                ),
            ),
        )

    def create_reasoner(self) -> Reasoner:
        return OscillatorTutorialReasoner()

    async def execute(
        self,
        capability: str,
        inputs: JsonObject,
        context: PluginExecutionContext,
    ) -> PluginResult:
        if capability != CAPABILITY:
            return PluginResult(
                success=False,
                error=f"unsupported capability: {capability}",
            )

        angular_frequency = float(inputs["angular_frequency"])
        damping_rate = float(inputs["damping_rate"])
        if angular_frequency <= 0:
            return PluginResult(
                success=False,
                error="angular_frequency must be positive",
            )
        if damping_rate <= 0:
            return PluginResult(
                success=False,
                error="damping_rate must be positive",
            )

        quality_factor = angular_frequency / (2.0 * damping_rate)
        assessment = {
            "angular_frequency": angular_frequency,
            "damping_rate": damping_rate,
            "quality_factor": quality_factor,
            "regime": "underdamped" if quality_factor > 0.5 else "strongly_damped",
            "run_id": context.run_id,
            "action_id": context.action_id,
        }
        return PluginResult(
            success=True,
            metrics={"quality_factor": quality_factor},
            observations={
                "quality_factor": quality_factor,
                "regime": assessment["regime"],
            },
            artifacts=(
                PluginArtifact.text(
                    name="oscillator-assessment.json",
                    artifact_type=ArtifactType.ANALYSIS_RESULT,
                    content=json.dumps(assessment, indent=2, sort_keys=True),
                    metadata={
                        "format": "json",
                        "method": "Q = angular_frequency / (2 * damping_rate)",
                    },
                ),
            ),
        )

这段代码做了什么

execute() 是插件的核心边界：Runtime 已经依据清单检查字段形状，然后把能力名、普通 JSON 输入和执行上下文交给插件。插件仍要验证领域规则，例如频率和阻尼率必须为正。

失败时返回 PluginResult(success=False, error="...")，不要抛出一个用来表达普通输入错误的异常。成功时：

  • metrics 只放可比较的有限数值；
  • observations 放 JSON 可序列化摘要；
  • artifacts 放需要持久化的实际证据；
  • context.run_id 和 action_id 可写入来源信息，但不要猜测存储路径。

这里同时在 Python 中声明 manifest，是为了直接实例化和单元测试。插件从目录加载时，plugin.yaml 是权威清单；两个声明必须保持一致。

确定性 Reasoner 只是为了让教程一键运行。真实领域插件通常不需要 create_reasoner()，而是让配置的 LLM Reasoner 根据能力说明与示例安排调用。

4. 写运行配置

创建 examples/plugin_tutorial.yaml：

title: "Oscillator plugin tutorial"
goal:
  description: "Confirm that the reference damped oscillator has quality factor >= 3.0"

acceptance:
  metric: quality_factor
  operator: ">="
  value: 3.0

budget:
  max_actions: 3
  max_experiments: 1
  max_failures: 1

plugins:
  - examples/plugins/oscillator_quality

capabilities:
  - tutorial.oscillator.evaluate

research_questions:
  - "What is the quality factor of the reference oscillator?"

插件路径相对于执行命令时的当前目录解析。本例应从仓库根目录运行。

5. 运行端到端测试

scientific-agent \
  --data-dir /tmp/scientific-agent-plugin-demo \
  run examples/plugin_tutorial.yaml

应该看到：

[Metric] quality_factor = 4
[CriteriaEngine] Acceptance criteria satisfied
DONE run_id=run_... status=SUCCESS

保存 Run ID 后检查完整事件：

scientific-agent \
  --data-dir /tmp/scientific-agent-plugin-demo \
  status run_... --events

在桌面端打开同一数据目录，进入该运行的 Lineage，应能看到实验到 oscillator-assessment.json 的关系。也可以启动 Control API，通过 /artifacts 下载 JSON。

如果创建合同时提示 acceptance metric has no producer，检查清单与 Python manifest 中的 metric_outputs。若提示 capability missing，检查三个地方的能力 ID 是否逐字符一致。

6. 添加快速单元测试

在插件目录旁创建 test_oscillator_plugin.py：

import asyncio

from oscillator_quality.plugin import OscillatorQualityPlugin
from scientific_agent.plugins.sdk import PluginExecutionContext


def test_quality_factor(tmp_path):
    plugin = OscillatorQualityPlugin()
    context = PluginExecutionContext(
        run_id="run_test",
        action_id="act_test",
        workspace=tmp_path,
    )

    result = asyncio.run(
        plugin.execute(
            "tutorial.oscillator.evaluate",
            {"angular_frequency": 4.0, "damping_rate": 0.5},
            context,
        )
    )

    assert result.success
    assert result.metrics == {"quality_factor": 4.0}
    assert result.artifacts[0].name == "oscillator-assessment.json"


def test_rejects_nonpositive_damping(tmp_path):
    plugin = OscillatorQualityPlugin()
    context = PluginExecutionContext("run_test", "act_test", tmp_path)

    result = asyncio.run(
        plugin.execute(
            "tutorial.oscillator.evaluate",
            {"angular_frequency": 4.0, "damping_rate": 0.0},
            context,
        )
    )

    assert not result.success
    assert result.error == "damping_rate must be positive"

运行时把插件包加入导入路径：

PYTHONPATH=src:examples/plugins/oscillator_quality \
  pytest -q test_oscillator_plugin.py

至少测试一条成功路径、一条领域输入失败路径和一个端到端运行。若插件读取产物，再测试错误类型、错误 hash 和跨运行引用会被拒绝。

7. 打包并在桌面端安装

从 examples/plugins 执行：

cd examples/plugins
zip -r oscillator-quality.zip oscillator_quality \
  -x '*__pycache__*' '*.pyc' '.DS_Store'

在 Desktop 中：

  1. 打开 Settings → Plugins。
  2. 选择 Install custom plugin。
  3. 选择 oscillator-quality.zip。
  4. 检查清单与依赖，然后选择 Enable。
  5. 新建运行并勾选 tutorial.oscillator.evaluate。

安装成功但启用失败通常表示 entrypoint 导入路径错误或依赖未装。解压 ZIP，确认 plugin.yaml 与 oscillator_quality/plugin.py 的相对位置和本教程一致。

8. 把教学插件改成真实插件

下一步可以按需要逐项扩展：

  • 在 input_schema 增加质量、刚度或测量数据字段，同时更新 required_inputs 和 example。
  • 用 artifact_inputs 接收 CSV/JSON 输入产物，通过 context.resolved_artifacts 或 read_artifact() 读取。
  • 通过 parent_names 建立一次调用内多个新产物的父子关系。
  • 在 estimate_cost() 中预估 solver、CPU、GPU 或训练样本成本。
  • 用 resource_usage 报告真实消耗，使预算准确扣减。
  • 在 start() 中初始化可复用资源，在 stop() 中释放。
  • 删除教学用确定性 Reasoner，让通用 Scientific Reasoner 编排能力。

不要仅为了让条件通过而报告虚构指标，也不要把普通插件标成 trusted evaluator。受信任评估需要独立数据边界、明确权限和额外测试。

发布前检查表

  • plugin.yaml 能独立说明插件用途、输入、输出和依赖。
  • 插件 ID、能力 ID 和 artifact kind 稳定且有命名空间。
  • 清单和 Python manifest 一致。
  • 所有输入都有类型与领域验证，错误信息可操作。
  • metric_outputs 与实际 PluginResult.metrics 一致。
  • 产物内容可复核，metadata 记录方法和来源。
  • 不写入 Runtime 数据目录，不读取其他运行或私有评估数据。
  • 依赖版本有边界，不在插件代码中自动安装包。
  • 单元测试、ZIP 安装测试和真实端到端运行都通过。

更完整的字段和生命周期说明见插件 SDK 参考。

找到 条结果：“”

    没有找到匹配结果：“”