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/ 领域说明。