chenxin
491136c770
- engine/bindings：nanobind 官方薄绑定包 hunjing-lingtai（LingtaiEngine RAII、HjStatus→LingtaiError 映射、HJ_ABI_VERSION 导出）；scikit-build-core 构建，直接编译引擎源码自包含 - backend 以 editable 路径依赖接入；typings/lingtai.pyi 类型存根（mypy strict 全绿） - app/core/lingtai.py：懒初始化单例 + LingtaiUnavailable fail-closed（禁止宿主第二实现降级） - 迁移 3：message.parser_version；post_message 的 markdown 输入经 hj_transform 规范化为 canonical blocks（展开 code/mermaid/math 并标注 parser_version，契约第 4 节） - 测试 29 项全绿：消息展开、Python/C++ golden 字节一致、引擎注入失败时消息路径显式 500（fail-closed 验证）
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""灵台引擎薄封装（Python 宿主接入，ADR-016）。
|
||
|
||
- 懒初始化进程级单例；初始化或调用失败一律抛 LingtaiUnavailable，
|
||
由调用方 fail closed（禁止在宿主内用第二实现降级）。
|
||
- 仅做类型转换与错误映射，不缓存权威状态、不复制业务语义。
|
||
"""
|
||
|
||
import json
|
||
import threading
|
||
from typing import Any
|
||
|
||
import lingtai
|
||
|
||
_lock = threading.Lock()
|
||
_engine: lingtai.LingtaiEngine | None = None
|
||
|
||
|
||
class LingtaiUnavailable(RuntimeError):
|
||
"""引擎缺失、初始化失败或调用失败：依赖方必须显式失败。"""
|
||
|
||
|
||
def get_engine() -> lingtai.LingtaiEngine:
|
||
global _engine
|
||
with _lock:
|
||
if _engine is None:
|
||
try:
|
||
_engine = lingtai.LingtaiEngine()
|
||
except Exception as exc: # nanobind LingtaiError 等
|
||
raise LingtaiUnavailable(f"灵台引擎初始化失败：{exc}") from exc
|
||
return _engine
|
||
|
||
|
||
def reset_engine() -> None:
|
||
"""测试辅助：丢弃单例，下次访问重新初始化。"""
|
||
global _engine
|
||
with _lock:
|
||
_engine = None
|
||
|
||
|
||
def transform_markdown(text: str) -> dict[str, Any]:
|
||
"""Markdown → canonical ContentBlock（dict：parser_version + blocks）。"""
|
||
try:
|
||
engine = get_engine()
|
||
except LingtaiUnavailable:
|
||
raise
|
||
try:
|
||
raw = engine.transform_markdown(text)
|
||
except Exception as exc:
|
||
raise LingtaiUnavailable(f"灵台引擎调用失败：{exc}") from exc
|
||
parsed: dict[str, Any] = json.loads(raw)
|
||
for block in parsed["blocks"]:
|
||
block.setdefault("id", None)
|
||
return parsed
|
||
|
||
|
||
def transform_version() -> str:
|
||
return str(lingtai.transform_version())
|