triple_store.py文件实现了一个用于存储和管理Triple观测-执行-结果三元组经验数据的持久化引擎。它采用 JSON 文件作为 Phase-I 的简化存储方案并提供了数据追加、查询和相似度检索等核心功能是构建闭环学习系统经验回放机制的关键组件。核心功能与设计解析功能模块实现机制关键特性与说明初始化与加载__init__方法创建存储目录并调用_load_from_disk加载历史数据到内存缓存_cache。确保服务重启后经验不丢失实现数据持久化与内存快速访问的结合 。数据追加append和append_batch方法将Triple对象加入内存缓存并立即调用_persist写入磁盘。采用“写时持久化”策略每条数据独立存储为一个以时间戳命名的 JSON 文件避免单点故障导致数据全量丢失。数据查询query_recent(n)返回最近n条记录利用缓存列表的切片操作实现。提供高效的时间序列查询常用于获取最新样本用于模型训练或分析。相似度查询query_by_obs(obs, radius)基于欧氏距离查找与给定观测向量相似的过往经验。Phase-I 采用线性扫描计算适用于数据量不大的场景。该功能是基于案例推理或最近邻策略的核心能从历史中找到类似状态下的成功/失败经验 。序列化与反序列化_to_dict和_from_dict方法负责Triple对象与 JSON 可序列化字典之间的转换。确保复杂数据类 (EffectObservation,ActionParameters,Outcome) 能正确、完整地保存到文件并重新加载是数据契约落地的关键 。持久化策略每个Triple存储为独立的{timestamp}.json文件。优点简单、原子性写入、易于并行和增量备份。缺点大量小文件可能影响 I/O 效率此为 Phase-I 的简化设计。核心代码流程示例以下代码展示了TripleStore的典型使用流程import logging from pathlib import Path # 假设 types.py 中的类已定义 from AFT.GuicangLayer.modules.translator_v0_2.types import ( EffectObservation, ActionParameters, Outcome, Triple, ObservationSource ) from AFT.GuicangLayer.modules.translator_v0_2.triple_store import TripleStore # 初始化存储引擎指定数据存放目录 store TripleStore(base_path./experiment_data/triples_v1) logging.info(f存储初始化完成已加载历史样本数: {len(store._cache)}) # 1. 创建一条新的经验数据 new_observation EffectObservation( spectral_shift0.08, entropy_oscillation1.9, phase_lock_freq48.5, lambda20.015, sri0.92, sdi0.08, sourceObservationSource.EFFECT_LAYER ) new_action ActionParameters( threshold0.65, hold_duration4.0, tighten_coefficient1.1, relax_coefficient0.85 ) new_outcome Outcome( successTrue, delta_phase-0.03, convergence_time2.8, raw_metrics{overshoot: 0.05, settling_time: 2.5} ) new_triple Triple(obsnew_observation, actionnew_action, outcomenew_outcome) # 2. 存储经验store.append(new_triple) # 此时一条名为 {int(new_triple.timestamp)}.json 的文件已生成在指定目录 # 3. 查询最新经验例如用于模型训练 recent_experiences store.query_recent(n500) print(f获取到最近 {len(recent_experiences)} 条经验用于训练。) # 4. 基于当前状态查询相似历史经验例如用于决策参考 current_obs EffectObservation(...) # 假设是当前的观测状态 similar_past_triples store.query_by_obs(current_obs, radius0.15) print(f找到 {len(similar_past_triples)} 条相似历史经验。) if similar_past_triples: # 分析相似历史中的行动与结果为当前决策提供参考 successful_actions [t.action for t in similar_past_triples if t.outcome.success] print(f其中 {len(successful_actions)} 条经验取得了成功。)工程化设计要点松耦合与可测试性存储引擎与核心数据契约 (types.py) 分离通过明确的接口 (append,query) 交互便于单独测试和替换存储后端如未来升级到 Parquet 或数据库。可观测性集成内置logging模块记录加载数据量、文件读取错误等关键事件符合生产级 ML 应用对可观测性的要求 。渐进式架构注释明确标注当前为 “Phase-I 使用 JSON 简化实现”为后续性能优化如批量写入、索引加速相似查询和格式升级如 Parquet 列式存储预留了清晰的演进路径 。数据完整性保障_from_dict方法包含异常处理 (try-except)在加载损坏或格式不兼容的 JSON 文件时返回None并记录警告避免了因单个文件问题导致整个存储加载失败 。参考来源开源多智能体工作流驱动癌症药物发现开源大模型多智能体驱动的癌症药物发现工作流ML Enabled Application从模型部署到生产级AI应用的工程化落地从Notebook到生产环境机器学习模型交付实战指南Graph-RAG实战用知识图谱增强RAG提升技术文档问答准确率