适配器实现闭环控制

2026/8/8 4:11:03

adapter.py文件是翻译器 v0.2 的顶层集成与协调模块，它作为“效果层”与“关系层”之间的桥接适配器，将triple_store.py、mapping_engine.py和safety_gate.py三个核心组件串联起来，形成一个完整的“观测-决策-执行-学习”数据闭环。

核心职责与数据流

TranslatorAdapter类封装了从接收观测到记录结果的完整工作流，其核心职责与内部数据流转如下表所示：

核心方法/属性	功能描述	数据流与组件调用
`__init__`	初始化适配器，创建并组合三个核心子组件。	依次实例化`TripleStore`(存储)、`MappingEngine`(映射)、`SafetyGate`(安全校验)，并初始化状态记录变量`_last_obs`和`_last_action`。
`observe_and_act(obs)`	主业务入口。接收效果层观测，返回经过安全校验的关系层动作。	`obs`→`self.engine.map(obs)`→`raw_action`→`self.gate.validate(raw_action)`→`action`。同时，记录`_last_obs`和`_last_action`供后续`record_outcome`使用。
`record_outcome(outcome)`	反馈闭环入口。接收动作执行结果，形成经验三元组并更新系统。	检查`_last_obs`和`_last_action`是否存在 → 组合成`Triple`对象 →`self.engine.update(triple)`(内部会调用`self.store.append`)。
`freeze()`/`unfreeze()`	安全控制接口。用于冻结或解冻系统的动作输出。	直接代理调用`self.gate.freeze()`和`self.gate.unfreeze()`。
`get_stats()`	系统状态监控接口。返回基本的运行时统计信息。	查询`self.store`中的经验总数和近期数量，并返回`self.gate`的冻结状态。

完整工作流程示例

以下代码展示了如何使用TranslatorAdapter完成一次完整的决策与学习循环：

import logging from AFT.GuicangLayer.modules.translator_v0_2.adapter import TranslatorAdapter from AFT.GuicangLayer.modules.translator_v0_2.types import EffectObservation, Outcome, ObservationSource # 1. 初始化适配器（一站式初始化所有底层组件） adapter = TranslatorAdapter() print("适配器初始化完成，各组件已就绪。") # 2. 模拟效果层持续产生观测，并进行决策 for i in range(5): # 模拟生成一个新的效果层观测 simulated_obs = EffectObservation( spectral_shift=0.1 + i * 0.01, entropy_oscillation=2.0, phase_lock_freq=50.0, lambda2=0.02, sri=0.9, sdi=0.1, source=ObservationSource.EFFECT_LAYER ) # 核心调用：观测 -> 决策 recommended_action = adapter.observe_and_act(obs=simulated_obs) print(f"迭代 {i}: 观测到 spectral_shift={simulated_obs.spectral_shift:.3f}, " f"推荐动作 threshold={recommended_action.threshold:.3f}") # 3. 模拟关系层执行推荐动作，并产生结果（此处为模拟逻辑） # 在实际系统中，此步骤由外部执行器完成，并将结果反馈回来。 simulated_success = recommended_action.threshold > 0.55 simulated_outcome = Outcome( success=simulated_success, delta_phase=-0.02 if simulated_success else 0.1, convergence_time=2.5 if simulated_success else 10.0, raw_metrics={"模拟指标": 1.0} ) # 4. 将执行结果记录回系统，完成学习闭环 adapter.record_outcome(outcome=simulated_outcome) print(f"结果记录: success={simulated_outcome.success}") # 5. 查询系统状态 stats = adapter.get_stats() print(f" 系统状态统计: {stats}") # 6. 模拟外部监控触发安全冻结 print(" 模拟触发漂移告警...") adapter.freeze() # 冻结后，即使有新的观测，observe_and_act 返回的也是上一次安全动作或默认值 frozen_action = adapter.observe_and_act(simulated_obs) print(f"冻结状态下的输出被锁定: {frozen_action}")

设计要点与架构价值

门面模式 (Facade Pattern) 应用：TranslatorAdapter充当了翻译器子系统对外的统一门面。效果层和关系层无需了解内部复杂的MappingEngine、SafetyGate等组件，只需与适配器交互，极大降低了系统耦合度，简化了上层调用逻辑。
状态管理与闭环形成：通过_last_obs和_last_action临时保存状态，adapter能够将异步返回的outcome与之前对应的obs和action正确关联，从而构建出完整的Triple。这是实现离线或延迟反馈式强化学习的关键。
明确的职责链：observe_and_act方法清晰地定义了决策流水线：映射 → 安全校验。这种设计使得每个组件的职责单一且可测试，例如可以单独测试MappingEngine的映射逻辑，或SafetyGate的边界钳位逻辑，而无需启动整个适配器。
安全管控的统一出口：freeze/unfreeze方法提供了系统级的安全干预入口。当上游的“自指守护模块”检测到异常时，可以通过调用adapter.freeze()一键锁定整个翻译器的输出，确保了安全策略能够被集中、强制地执行。
可观测性支持：get_stats方法提供了基础的运行时监控指标，便于了解系统积累的经验数量和安全状态，为系统运维和调试提供了便利。