import yaml from core import codex_informer from core.codex_informer import CodexInformer from core.eil_processor import EILProcessor from core.esil_inference import ESILInference from core.hei_inference import HEIInference from core.eris_reasoner import ERISReasoner from core.fec_controller import FECController # ─── GLOBAL INITIALIZATION ──────────────────────────────────────────────────────── codex = CodexInformer( emotion_families_path="config/emotion_families.yaml", arc_mapping_path="config/arc_mapping.yaml", resonance_mapping_path="config/resonance_mapping.yaml" ) # EIL & ESIL use codex for lookups eil = EILProcessor(codex) esil = ESILInference(codex) # HEIInference now takes codex so it can resolve families hei = HEIInference(codex) # ERISReasoner also takes codex to resolve arc/resonance eris = ERISReasoner(codex) # FECController remains parameterless fec = FECController() # Load Response Strategies YAML with open('config/response_strategies.yaml', 'r', encoding='utf-8') as f: response_strategies = yaml.safe_load(f)['response_strategies'] def run_mec_pipeline(user_input_text: str, force_hei: bool = False): """ Full MEC pipeline: 1) EILProcessor.infer_emotion 2) ESILInference.infer_esil 3) Routing: ERIS vs. HEI fallback based on confidence or force_hei 4) FECController.generate_prompt Returns: fusion_prompt (str), final_uesp (dict), empathic_response (str), EmID (str) """ # 1️⃣ EIL eil_packet = eil.infer_emotion(user_input_text) # 2️⃣ ESIL esil_packet = esil.infer_esil(eil_packet) # 3️⃣ Routing / HEI fallback if force_hei or esil_packet.get("confidence_score", 1.0) < 0.65: pseudo_esp = hei.detect_low_signal(esil_packet) final_uesp = eris.reason_emotion_state(pseudo_esp) else: final_uesp = eris.reason_emotion_state(esil_packet) # 4️⃣ FEC fusion_prompt = fec.generate_prompt(final_uesp) # 5️⃣ Simulated Empathic Response fam_code = final_uesp.get('primary_emotion_code') or final_uesp.get('Primary Emotion Code') # Load arc_lookup from codex or arc_mapping.yaml arc_lookup = codex.arc_mapping if hasattr(codex, 'arc_mapping') else {} emotion_arc = arc_lookup.get(fam_code, 'TBD') # Get the arc based on the family rsm_code = response_strategies.get(fam_code, {}).get('rsm_code', 'RSM-UNKNOWN') strategy_name = response_strategies.get(fam_code, {}).get('strategy', 'Strategy not defined') sample_response = response_strategies.get(fam_code, {}).get('sample_response', 'No response available') # Get Emotion ID (EmID) emid = final_uesp.get('emid', 'No EmID available') empathic_response = ( f"Response Strategy Code: {rsm_code}\n" f"Response Strategy: {strategy_name}\n\n" f"{sample_response}" ) return fusion_prompt, final_uesp, empathic_response, emid def get_demo_scenarios(): """ Load demo scenarios from the YAML file. Returns a dictionary of label-text pairs. """ with open('config/demo_scenarios.yaml', 'r', encoding='utf-8') as f: test_scenarios = yaml.safe_load(f).get('test_scenarios', []) return {item['label']: item['text'] for item in test_scenarios} # Example usage if __name__ == "__main__": prompt, uesp, empathic_response = run_mec_pipeline("I’m feeling anxious about tomorrow.", force_hei=False) print("Fusion Prompt:\n", prompt) print("Final UESP:\n", uesp) print("Empathic Response:\n", empathic_response)