""" Profile configuration mapping for the Qwen Pipeline. Converts ChromaDB profile data (HTML element IDs from the dashboard) into server config dicts that PipelineSession expects. """ import json # ── Profile → Config mapping ───────────────────────────────────── # Profile keys are HTML element IDs; config keys are what PipelineSession expects. PROFILE_TO_CONFIG_MAP = { 'agentAName': ('agent_a_name', str), # fixedGreeting, agentAReturnedGreeting, instructions handled specially below # (per-language: accept string, dict, or JSON-string of dict) 'agentData': ('agent_data', str), 'vadThreshold': ('vad_threshold', float), 'silenceDuration': ('silence_duration_ms', int), 'bargeInEnabled': ('barge_in_enabled', bool), 'bargeInMinDuration':('barge_in_min_duration_ms', int), 'llmModel': ('llm_model', str), 'temperature': ('temperature', float), 'maxTokens': ('max_tokens', lambda v: int(v) if v else None), 'topP': ('top_p', float), 'enableThinking': ('enable_thinking', bool), 'ttsMode': ('tts_mode', str), 'toolRouting': ('tool_routing', str), 'speechRate': ('speech_rate', float), 'pitchRate': ('pitch_rate', float), 'ttsVolume': ('tts_volume', int), 'enableLanguageAgent': ('enable_language_agent', bool), 'languageAgentGreeting': ('language_agent_greeting', str), 'languageAgentInstructions': ('language_agent_instructions', str), 'transferDelay': ('transfer_delay', float), 'sttProvider': ('stt_provider', str), 'geminiModel': ('gemini_model', str), 'geminiLangHint': ('gemini_lang_hint', str), 'enableOmni': ('enable_omni', bool), 'omniModel': ('omni_model', str), 'omniVoice': ('omni_voice', str), 'omniSpeechStyle': ('omni_speech_style', str), 'omniBargeInRms': ('omni_barge_in_rms', float), 'omniSmoothBargeIn': ('omni_smooth_barge_in', bool), 'noiseSuppression': ('noise_suppression', str), 'idleTimeoutEnabled': ('idle_timeout_enabled', bool), 'idleTimeoutSeconds': ('idle_timeout_s', int), 'omniAudioBufferDelay': ('omni_audio_buffer_delay', float), 'variableMappings': ('variable_mappings', lambda v: json.loads(v) if isinstance(v, str) else (v or [])), 'disabledBuiltinTools': ('disabled_builtin_tools', lambda v: json.loads(v) if isinstance(v, str) else (v or [])), 'placeholderFields': ('placeholder_fields', lambda v: json.loads(v) if isinstance(v, str) else (v or {})), } def profile_to_config(profile_data): """Convert a ChromaDB profile (HTML element IDs) to a server config dict.""" config = {'type': 'config'} for profile_key, (config_key, converter) in PROFILE_TO_CONFIG_MAP.items(): if profile_key not in profile_data: continue val = profile_data[profile_key] try: config[config_key] = converter(val) except (ValueError, TypeError): pass # Special: per-language text fields — accept string, dict, or JSON-string of dict. # The server's PipelineSession normalizes via _norm_lang_map on receipt. for profile_key, config_key in ( ('fixedGreeting', 'fixed_greeting'), ('agentAReturnedGreeting', 'agent_a_returned_greeting'), ('instructions', 'instructions'), ): if profile_key not in profile_data: continue val = profile_data[profile_key] if isinstance(val, str): # Could be a JSON-encoded dict (dashboard) or legacy plain text stripped = val.strip() if stripped.startswith('{'): try: parsed = json.loads(stripped) if isinstance(parsed, dict): val = parsed except (json.JSONDecodeError, ValueError, TypeError): pass config[config_key] = val # Special: voiceMap → voice_map (JSON string or dict) if 'voiceMap' in profile_data: try: vm = profile_data['voiceMap'] if isinstance(vm, str): vm = json.loads(vm) if isinstance(vm, dict): config['voice_map'] = vm except (json.JSONDecodeError, TypeError): pass # Special: omniVoiceMap → omni_voice_map (separate voice map for Omni mode) if 'omniVoiceMap' in profile_data: try: ovm = profile_data['omniVoiceMap'] if isinstance(ovm, str): ovm = json.loads(ovm) if isinstance(ovm, dict): config['omni_voice_map'] = ovm except (json.JSONDecodeError, TypeError): pass # Special: acrs config (JSON string or dict) if 'acrs' in profile_data: try: acrs = profile_data['acrs'] if isinstance(acrs, str): acrs = json.loads(acrs) if isinstance(acrs, dict): config['acrs'] = acrs except (json.JSONDecodeError, TypeError): pass # Special: conversationValidator (JSON string or dict) if 'conversationValidator' in profile_data: try: cv = profile_data['conversationValidator'] if isinstance(cv, str): cv = json.loads(cv) if isinstance(cv, dict): config['conversation_validator'] = cv except (json.JSONDecodeError, TypeError): pass # Special: emotionConfig (JSON string or dict) if 'emotionConfig' in profile_data: try: emo = profile_data['emotionConfig'] if isinstance(emo, str): emo = json.loads(emo) if isinstance(emo, dict): config['emotion'] = emo except (json.JSONDecodeError, TypeError): pass # Special: functionsTextarea → custom_tools (JSON array) if 'functionsTextarea' in profile_data: try: tools = json.loads(profile_data['functionsTextarea']) if isinstance(tools, list): config['custom_tools'] = tools except (json.JSONDecodeError, TypeError): config['custom_tools'] = [] # Special: transferAgents (JSON string or list) if 'transferAgents' in profile_data: try: ta = profile_data['transferAgents'] if isinstance(ta, str): ta = json.loads(ta) if isinstance(ta, list): config['transfer_agents'] = ta except (json.JSONDecodeError, TypeError): pass # Special: allowedLanguages (JSON string or list) if 'allowedLanguages' in profile_data: try: al = profile_data['allowedLanguages'] if isinstance(al, str): al = json.loads(al) if isinstance(al, list): config['allowed_languages'] = al except (json.JSONDecodeError, TypeError): pass # Backward compat: old enableAgentB / agentBName / agentBInstructions if 'enableAgentB' in profile_data and 'transfer_agents' not in config: enabled = profile_data.get('enableAgentB') if enabled in (True, 'true', '1', 1): config['transfer_agents'] = [{ 'name': profile_data.get('agentBName', 'Sales Department'), 'instructions': profile_data.get('agentBInstructions', 'You are a sales representative. Help the caller with product information, ' 'pricing, and purchases. Be friendly and professional. ' 'Keep responses brief (2-3 sentences). This is a live phone call.') }] return config def load_profile_config(profile_name, profiles_collection): """Load a profile from ChromaDB and return it as a config dict, or None. Args: profile_name: The profile ID to load. profiles_collection: The ChromaDB collection for profiles. """ if not profiles_collection: return None try: result = profiles_collection.get(ids=[profile_name]) if result['ids']: profile_data = json.loads(result['documents'][0]) config = profile_to_config(profile_data) # Debug: show what ChromaDB returned (instructions/greeting may be # per-language dicts now, so stringify before slicing). instr = config.get('instructions', '') instr_str = str(instr) print(f'[Profiles] Loaded "{profile_name}" — ' f'instructions: {len(instr_str)} chars = "{instr_str[:120]}..."') greeting = config.get('fixed_greeting', '') if greeting: print(f'[Profiles] Fixed greeting: "{str(greeting)[:80]}..."') return config except Exception as e: print(f'[Profiles] Load error: {e}') return None