跳转至

06 二次开发指南

面向"想扩展系统"的开发者:新增领域、换本体、扩展工具。核心思想:通用引擎 + 领域适配器分离——引擎不含任何领域词汇,领域语义全部外置在 domains/ 领域包,运行时约定式发现(domains/registry.py)。

6.1 架构速览

reasoner/engine.py       通用 OWL 2 RL 引擎(加载/推理/SPARQL/查询,无领域词汇)
reasoner/{derivation,proof,semantic}.py   推导链 / 结构化证明 / 语义配置
domains/registry.py      领域包约定式注册表(list_domains / load_domain)
domains/<name>/          一个领域一个包(插件式)
  ├─ NAME / DESCRIPTION / create_adapter(engine, config=None)   # 必须暴露
  ├─ config.json          语义配置(类名/属性名映射 + 推导链展开规则)
  ├─ derivation.json      推导链说明
  ├─ access_policy.json   语义白名单(受控语义接口)
  ├─ guard_config.json    调用护栏配置(预算/循环/缓存/审计,默认关闭)
  ├─ gates_smoke.json     pipeline_gate G3 物化冒烟断言集
  ├─ cq_tbox.py / cq_prompts.json   能力查询 TBox / 提示词
  ├─ cq_gen.py / cq_tests.json      CQ 出题器 / 题库
  ├─ adapter.py           领域查询实现(capability_match 等)
  ├─ mcp.py               MCP 工具注册(可选)
  └─ *.owl                该领域本体

6.2 新增一个领域(四步)

第 1 步:准备本体。编写或复用 OWL 文件,如 domains/my_domain/my_domain.owl。

第 2 步:建领域包。domains/my_domain/__init__.py 暴露三要素:

NAME = "my_domain"                                    # 领域标识
DESCRIPTION = "我的领域:……"                          # 展示给智能体/用户
from reasoner.engine import OntologyEngine

def create_adapter(engine: OntologyEngine, config=None):
    """把通用引擎包装为该领域的查询语义"""
    from .adapter import MyAdapter
    return MyAdapter(engine, config)

第 3 步:语义配置(config.json)。把领域查询依赖的类名/属性名外置,换同构本体时可整体覆盖:

{"semantic": {"agent_class": "Agent", "task_class": "Task",
  "has_capability": "has_capability", "requires_capability": "required_by"}}

第 4 步:注册并验证。约定式发现,无需改注册表代码:

from domains import list_domains, load_domain
list_domains()                                # 自动发现 my_domain
adapter = load_domain("my_domain", engine)    # 加载并调用领域查询

领域查询函数签名遵循默认领域(见 domains/agent_task/adapter.py:capability_match / orchestrate / assign / classify / consistency / explain 等),返回结构与推导链一致。

6.3 扩展 MCP 工具

领域包内 mcp.py 用 server.tool 注册领域工具(默认领域 14 个:能力匹配/编排/分配/分类/一致性/图谱/实体/状态/CQ 建模两件套/SHACL 体检/CQ 出题两件套);通用工具(SPARQL/个体/类/类型/属性/重推理/证明/策略/时序/护栏状态/质量门禁)在 mcp/mcp_server.py 注册,与领域无关。新增领域工具 = 在 domains/<name>/mcp.py 增加 @server.tool 函数并调用适配器方法。

护栏自动生效:通用工具经 mcp/tool_guard.py 的 wrap_server_tool 注册管道统一包装——guard_config.json 启用后自动获得预算熔断/循环拦截/缓存/审计;领域工具如需护栏,可在领域包注册时走同一包装器。

6.4 建模流水线:质量门禁(G1–G4)

核心工具 pipeline_gate(ttl, abox?, gates?)(MCP 通用工具,reasoner/pipeline_gate.py)把建模草稿按四道闸门 fail-fast 过闸,废品不出车间:

闸门 检查 打回方向
G1 DL 格式 防 OWL-Full(owlready2 is_owl_full() + 结构断言);数据属性 inverse functional 主键陷阱拦截 回 S2
G2 一致性 owlrl 物化 + Nothing/disjointWith 矛盾检查 + 空类提示(HermiT 可选增强,未启用 mode=degraded 显式标注) 回 S2
G3 物化冒烟 领域 gates_smoke.json 断言集(预期谓词 min_count,如 can_execute ≥1);断言集为空如实 skipped 检查骨架
G4 SHACL 复用 pyshacl + pipeline/shapes/<domain>.ttl,violations 短名化透传 修数据字典或 SHACL

实测(默认本体 TTL):

.venv/bin/python -c "
from reasoner.engine import ONTO_PATH, OntologyEngine
from reasoner import pipeline_gate
e = OntologyEngine(ONTO_PATH)
r = pipeline_gate.run_gates(e.raw.serialize(format='turtle'))
print(r['overall'], r['failed_at'], r['next'])   # pass None S4 查询暴露(四道闸门全部通过)
"
  • 遇 fail 即停(后续闸门标 not_run),failed_at 定位卡闸位置;每闸输出 {status, evidence, remediation};
  • gates 参数支持单闸调试(如 ['G3']),空列表全部 skipped(非假通过);只读(拷贝图物化,不写回主图);
  • 术语/结构层校验对应 validate_tbox_draft 的 L1–L5,与闸门互补。

SHACL 形状:pipeline/shapes/<domain>.ttl 供 G4 与 validate_shapes 工具使用。结合 docs/MODELING_PIPELINE.md、docs/DOMAIN_PACKAGE_GENERATION.md(从 OWL + 业务知识库自动生成领域包:能力分层表、规则.md、术语表.md、场景.md)可半自动化落地新领域,过闸后再进入查询暴露。

6.5 扩展通用查询 / 新规则

  • 通用查询(SPARQL/个体/类/类型/属性/重推理)在 reasoner/engine.py,与本体无关;
  • 业务规则 R2/R4 为代码规则(OWL 2 DL 表达不了的语义),位于 reasoner/engine.py 推理管线;R1/R3 分别用 OWL 属性链与传递属性表达;
  • 新增规则建议先在 reasoner/ 实现并加测试(tests/),再在适配器与文档中同步。

6.6 测试与验收

.venv/bin/python -m pytest tests/ -q        # 全量测试
.venv/bin/python scripts/query.py --local ...   # 领域查询 --local 回归

新增领域后建议补充:本体样例、适配器单测、MCP 工具冒烟、docs/ 领域说明。