Spaces:
Paused
Paused
| """ | |
| Flare β Prompt Builder (v6 Β· date handling) | |
| ============================================================== | |
| """ | |
| from typing import List, Dict | |
| from datetime import datetime, timedelta | |
| from utils import log | |
| from config_provider import ConfigProvider | |
| # Date helper for Turkish date expressions | |
| def _get_date_context() -> Dict[str, str]: | |
| """Generate date context for Turkish date expressions""" | |
| now = datetime.now() | |
| # Weekday names in Turkish | |
| weekdays_tr = ["Pazartesi", "SalΔ±", "ΓarΕamba", "PerΕembe", "Cuma", "Cumartesi", "Pazar"] | |
| today_weekday = weekdays_tr[now.weekday()] | |
| # Calculate various dates | |
| dates = { | |
| "today": now.strftime("%Y-%m-%d"), | |
| "tomorrow": (now + timedelta(days=1)).strftime("%Y-%m-%d"), | |
| "day_after_tomorrow": (now + timedelta(days=2)).strftime("%Y-%m-%d"), | |
| "this_weekend_saturday": (now + timedelta(days=(5-now.weekday())%7)).strftime("%Y-%m-%d"), | |
| "this_weekend_sunday": (now + timedelta(days=(6-now.weekday())%7)).strftime("%Y-%m-%d"), | |
| "next_week_same_day": (now + timedelta(days=7)).strftime("%Y-%m-%d"), | |
| "two_weeks_later": (now + timedelta(days=14)).strftime("%Y-%m-%d"), | |
| "today_weekday": today_weekday, | |
| "today_day": now.day, | |
| "today_month": now.month, | |
| "today_year": now.year | |
| } | |
| return dates | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # INTENT PROMPT | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def build_intent_prompt(general_prompt: str, | |
| conversation: List[Dict[str, str]], | |
| user_input: str, | |
| intents: List, | |
| project_name: str = None) -> str: | |
| # Get config when needed | |
| cfg = ConfigProvider.get() | |
| # Get internal prompt from config | |
| internal_prompt = cfg.global_config.internal_prompt or "" | |
| # Extract intent names and captions | |
| intent_names = [it.name for it in intents] | |
| intent_captions = [it.caption or it.name for it in intents] | |
| # Get project language | |
| project_language = "Turkish" # Default | |
| if project_name: | |
| project = next((p for p in cfg.projects if p.name == project_name), None) | |
| if project: | |
| # Language code'u language name'e Γ§evir | |
| lang_map = { | |
| "tr": "Turkish", | |
| "en": "English", | |
| "de": "German", | |
| "fr": "French", | |
| "es": "Spanish" | |
| } | |
| project_language = lang_map.get(project.default_language, "Turkish") | |
| # Replace placeholders in internal prompt | |
| if internal_prompt: | |
| # Intent names - tΔ±rnak iΓ§inde ve virgΓΌlle ayrΔ±lmΔ±Ε | |
| intent_names_str = ', '.join([f'"{name}"' for name in intent_names]) | |
| internal_prompt = internal_prompt.replace("<intent names>", intent_names_str) | |
| # Intent captions - tΔ±rnak iΓ§inde ve virgΓΌlle ayrΔ±lmΔ±Ε | |
| intent_captions_str = ', '.join([f'"{caption}"' for caption in intent_captions]) | |
| internal_prompt = internal_prompt.replace("<intent captions>", intent_captions_str) | |
| # Project language | |
| internal_prompt = internal_prompt.replace("<project language>", project_language) | |
| # === INTENT INDEX === | |
| lines = ["### INTENT INDEX ###"] | |
| for it in intents: | |
| # IntentConfig object attribute access | |
| det = it.detection_prompt.strip() if it.detection_prompt else "" | |
| det_part = f' β’ detection_prompt β "{det}"' if det else "" | |
| exs = " | ".join(it.examples) if it.examples else "" | |
| ex_part = f" β’ examples β {exs}" if exs else "" | |
| newline_between = "\n" if det_part and ex_part else "" | |
| lines.append(f"{it.name}:{det_part}{newline_between}{ex_part}") | |
| intent_index = "\n".join(lines) | |
| # === HISTORY === | |
| history_block = "\n".join( | |
| f"{m['role'].upper()}: {m['content']}" for m in conversation[-10:] | |
| ) | |
| # Combine prompts | |
| combined_prompt = internal_prompt + "\n\n" + general_prompt if internal_prompt else general_prompt | |
| prompt = ( | |
| f"{combined_prompt}\n\n" | |
| f"{intent_index}\n\n" | |
| f"Conversation so far:\n{history_block}\n\n" | |
| f"USER: {user_input.strip()}" | |
| ) | |
| log("β Intent prompt built (with internal prompt)") | |
| return prompt | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # RESPONSE PROMPT | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def build_api_response_prompt(api_config, api_response: Dict) -> str: | |
| """Build prompt for API response with mappings""" | |
| response_prompt = api_config.response_prompt or "API yanΔ±tΔ±nΔ± kullanΔ±cΔ±ya aΓ§Δ±kla:" | |
| # Response mappings varsa, mapping bilgilerini ekle | |
| if api_config.response_mappings: | |
| mapping_info = [] | |
| for mapping in api_config.response_mappings: | |
| # JSON path'e gΓΆre deΔeri bul | |
| value = extract_value_from_json_path(api_response, mapping.json_path) | |
| if value is not None: | |
| # Type'a gΓΆre formatlama | |
| if mapping.type == "date": | |
| # ISO date'i TΓΌrkΓ§e formata Γ§evir | |
| try: | |
| dt = datetime.fromisoformat(value.replace('Z', '+00:00')) | |
| value = dt.strftime("%d %B %Y %H:%M") | |
| except: | |
| pass | |
| elif mapping.type == "float": | |
| try: | |
| value = f"{float(value):,.2f}" | |
| except: | |
| pass | |
| mapping_info.append(f"{mapping.caption}: {value}") | |
| if mapping_info: | |
| # Response prompt'a mapping bilgilerini ekle | |
| mapping_text = "\n\nΓnemli Bilgiler:\n" + "\n".join(f"β’ {info}" for info in mapping_info) | |
| response_prompt = response_prompt.replace("{{api_response}}", f"{{{{api_response}}}}{mapping_text}") | |
| # API response'u JSON string olarak ekle | |
| response_json = json.dumps(api_response, ensure_ascii=False, indent=2) | |
| final_prompt = response_prompt.replace("{{api_response}}", response_json) | |
| return final_prompt | |
| def extract_value_from_json_path(data: Dict, path: str): | |
| """Extract value from JSON using dot notation path""" | |
| try: | |
| parts = path.split('.') | |
| value = data | |
| for part in parts: | |
| if isinstance(value, dict): | |
| value = value.get(part) | |
| elif isinstance(value, list) and part.isdigit(): | |
| value = value[int(part)] | |
| else: | |
| return None | |
| return value | |
| except: | |
| return None | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # PARAMETER PROMPT | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| _FMT = """#PARAMETERS:{"extracted":[{"name":"<param>","value":"<val>"},...],"missing":["<param>",...]}""" | |
| def build_parameter_prompt(intent_cfg, | |
| missing_params: List[str], | |
| user_input: str, | |
| conversation: List[Dict[str, str]]) -> str: | |
| date_ctx = _get_date_context() | |
| parts: List[str] = [ | |
| "You are extracting parameters from user messages in TURKISH.", | |
| f"Today is {date_ctx['today']} ({date_ctx['today_weekday']}). Tomorrow is {date_ctx['tomorrow']}.", | |
| "Extract ONLY the parameters listed below from the conversation.", | |
| "Look at BOTH the current message AND previous messages to find parameter values.", | |
| "If a parameter cannot be found, is invalid, or wasn't provided, keep it in the \"missing\" list.", | |
| "Never guess or make up values. Only extract values explicitly given by the user.", | |
| "" | |
| ] | |
| # Add parameter descriptions | |
| parts.append("Parameters to extract:") | |
| for p in intent_cfg.parameters: | |
| if p.name in missing_params: | |
| # Special handling for date type parameters | |
| if p.type == "date": | |
| date_prompt = ( | |
| f"β’ {p.name}: {p.extraction_prompt}\n" | |
| f" IMPORTANT DATE RULES:\n" | |
| f" - 'bugΓΌn' = {date_ctx['today']}\n" | |
| f" - 'yarΔ±n' = {date_ctx['tomorrow']}\n" | |
| f" - 'ΓΆbΓΌr gΓΌn' = {date_ctx['day_after_tomorrow']}\n" | |
| f" - 'bu hafta sonu' = {date_ctx['this_weekend_saturday']} or {date_ctx['this_weekend_sunday']}\n" | |
| f" - 'bu cumartesi' = {date_ctx['this_weekend_saturday']}\n" | |
| f" - 'bu pazar' = {date_ctx['this_weekend_sunday']}\n" | |
| f" - 'haftaya' or 'gelecek hafta' = add 7 days to current date\n" | |
| f" - 'haftaya bugΓΌn' = {date_ctx['next_week_same_day']}\n" | |
| f" - 'iki hafta sonra' = {date_ctx['two_weeks_later']}\n" | |
| f" - '15 gΓΌn sonra' = add 15 days to today\n" | |
| f" - '10 Temmuz' = {date_ctx['today_year']}-07-10\n" | |
| f" - Turkish months: Ocak=01, Εubat=02, Mart=03, Nisan=04, MayΔ±s=05, Haziran=06, " | |
| f"Temmuz=07, AΔustos=08, EylΓΌl=09, Ekim=10, KasΔ±m=11, AralΔ±k=12" | |
| ) | |
| parts.append(date_prompt) | |
| else: | |
| parts.append(f"β’ {p.name}: {p.extraction_prompt}") | |
| # Add format instruction | |
| parts.append("") | |
| parts.append("IMPORTANT: Your response must start with '#PARAMETERS:' followed by the JSON.") | |
| parts.append("Return ONLY this format with no extra text before or after:") | |
| parts.append(_FMT) | |
| # Add conversation history | |
| history_block = "\n".join( | |
| f"{m['role'].upper()}: {m['content']}" for m in conversation[-10:] | |
| ) | |
| prompt = ( | |
| "\n".join(parts) + | |
| "\n\nConversation so far:\n" + history_block + | |
| "\n\nUSER: " + user_input.strip() | |
| ) | |
| log(f"π Parameter prompt built for missing: {missing_params}") | |
| return prompt |