""" Template Engine for Intent-Driven Script Flow. [AC-IDS-06] Template mode script generation with variable filling. """ import asyncio import logging import re from typing import Any logger = logging.getLogger(__name__) class TemplateEngine: """ [AC-IDS-06] Template script engine. Fills template variables using context or LLM generation. """ VARIABLE_PATTERN = re.compile(r'\{(\w+)\}') DEFAULT_TIMEOUT = 5.0 def __init__(self, llm_client: Any = None): """ Initialize TemplateEngine. Args: llm_client: LLM client for variable generation (optional) """ self._llm_client = llm_client async def fill_template( self, template: str, context: dict[str, Any] | None, history: list[dict[str, str]] | None, ) -> str: """ [AC-IDS-06] Fill template variables with context or LLM-generated values. Args: template: Script template with {variable} placeholders context: Session context with collected inputs history: Conversation history for context Returns: Filled template string """ try: variables = self.VARIABLE_PATTERN.findall(template) if not variables: return template variable_values = {} for var in variables: value = await self._generate_variable_value( variable_name=var, context=context, history=history, ) variable_values[var] = value result = template for var, value in variable_values.items(): result = result.replace(f"{{{var}}}", value) logger.info( f"[AC-IDS-06] Filled template: " f"variables={list(variable_values.keys())}" ) return result except Exception as e: logger.error(f"[AC-IDS-06] Template fill failed: {e}, return original") return template async def _generate_variable_value( self, variable_name: str, context: dict[str, Any] | None, history: list[dict[str, str]] | None, ) -> str: """ Generate value for a single template variable. Args: variable_name: Variable name to generate value for context: Session context history: Conversation history Returns: Generated variable value """ if context and variable_name in context: return str(context[variable_name]) if context and context.get("inputs"): for inp in context["inputs"]: if isinstance(inp, dict): if inp.get("variable") == variable_name: return str(inp.get("input", f"[{variable_name}]")) if self._llm_client: prompt = self._build_variable_prompt( variable_name=variable_name, history=history, ) try: messages = [{"role": "user", "content": prompt}] response = await asyncio.wait_for( self._llm_client.generate(messages), timeout=self.DEFAULT_TIMEOUT, ) value = response.content.strip() if hasattr(response, 'content') else str(response).strip() return value except asyncio.TimeoutError: logger.warning( f"[AC-IDS-06] Variable generation timeout for {variable_name}" ) except Exception as e: logger.warning( f"[AC-IDS-06] Variable generation failed for {variable_name}: {e}" ) logger.warning( f"[AC-IDS-06] Failed to generate value for {variable_name}, " f"use placeholder" ) return f"[{variable_name}]" def _build_variable_prompt( self, variable_name: str, history: list[dict[str, str]] | None, ) -> str: """ Build prompt for variable value generation. """ prompt_parts = [ f'根据对话历史,为变量 "{variable_name}" 生成合适的值。', "", ] if history: prompt_parts.append("对话历史:") for msg in history[-3:]: role = "用户" if msg.get("role") == "user" else "客服" content = msg.get("content", "") prompt_parts.append(f"{role}: {content}") prompt_parts.append("") prompt_parts.extend([ "只返回变量值,不要解释。", ]) return "\n".join(prompt_parts) def extract_variables(self, template: str) -> list[str]: """ Extract variable names from template. Args: template: Template string with {variable} placeholders Returns: List of variable names """ return self.VARIABLE_PATTERN.findall(template)