|
|
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 |
|
|
|
|
|
|
|
|
codex = CodexInformer( |
|
|
emotion_families_path="config/emotion_families.yaml", |
|
|
arc_mapping_path="config/arc_mapping.yaml", |
|
|
resonance_mapping_path="config/resonance_mapping.yaml" |
|
|
) |
|
|
|
|
|
|
|
|
eil = EILProcessor(codex) |
|
|
esil = ESILInference(codex) |
|
|
|
|
|
|
|
|
hei = HEIInference(codex) |
|
|
|
|
|
|
|
|
eris = ERISReasoner(codex) |
|
|
|
|
|
|
|
|
fec = FECController() |
|
|
|
|
|
|
|
|
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) |
|
|
""" |
|
|
|
|
|
eil_packet = eil.infer_emotion(user_input_text) |
|
|
|
|
|
|
|
|
esil_packet = esil.infer_esil(eil_packet) |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
fusion_prompt = fec.generate_prompt(final_uesp) |
|
|
|
|
|
|
|
|
fam_code = final_uesp.get('primary_emotion_code') or final_uesp.get('Primary Emotion Code') |
|
|
|
|
|
arc_lookup = codex.arc_mapping if hasattr(codex, 'arc_mapping') else {} |
|
|
emotion_arc = arc_lookup.get(fam_code, 'TBD') |
|
|
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') |
|
|
|
|
|
|
|
|
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} |
|
|
|
|
|
|
|
|
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) |
|
|
|