"""Conversation-scoped dynamic variables for prompt pipeline assistants. The renderer is deliberately small: it only understands ``{{ name }}`` placeholders and never evaluates expressions. A value is substituted once, so user input cannot introduce a second template expansion. """ from __future__ import annotations import json import re from copy import deepcopy from datetime import datetime from typing import Any from uuid import uuid4 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from models import AssistantConfig Primitive = str | int | float | bool VARIABLE_NAME = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,63}$") PLACEHOLDER = re.compile(r"{{\s*([A-Za-z][A-Za-z0-9_]*)\s*}}") MAX_VARIABLES = 50 MAX_VALUE_LENGTH = 2048 MAX_HISTORY_ENTRIES = 50 MAX_HISTORY_JSON_LENGTH = 32_000 class DynamicVariableError(ValueError): pass def _primitive(value: Any, name: str) -> Primitive: if not isinstance(value, (str, int, float, bool)) or value is None: raise DynamicVariableError(f"动态变量 {name} 仅支持字符串、数字或布尔值") if len(str(value)) > MAX_VALUE_LENGTH: raise DynamicVariableError(f"动态变量 {name} 超过 {MAX_VALUE_LENGTH} 字符") return value def _validate_public(values: dict[str, Any] | None) -> dict[str, Primitive]: values = values or {} if len(values) > MAX_VARIABLES: raise DynamicVariableError(f"动态变量最多允许 {MAX_VARIABLES} 个") result: dict[str, Primitive] = {} for name, value in values.items(): if not VARIABLE_NAME.fullmatch(name): raise DynamicVariableError(f"动态变量名称不合法: {name}") if name.startswith(("system__", "secret__")): raise DynamicVariableError(f"客户端不能设置保留变量: {name}") result[name] = _primitive(value, name) return result def _type_matches(value: Primitive, expected: str) -> bool: if expected == "string": return isinstance(value, str) if expected == "boolean": return isinstance(value, bool) if expected == "number": return isinstance(value, (int, float)) and not isinstance(value, bool) return False def _system_values( *, assistant_id: str | None, conversation_id: str, timezone: str ) -> dict[str, Primitive]: try: zone = ZoneInfo(timezone) except ZoneInfoNotFoundError as exc: raise DynamicVariableError(f"无效时区: {timezone}") from exc now = datetime.now(zone) return { "system__agent_id": assistant_id or "", "system__conversation_id": conversation_id, "system__time": now.strftime("%A, %H:%M %d %B %Y"), "system__time_utc": now.astimezone(ZoneInfo("UTC")).isoformat(), "system__timezone": timezone, "system__agent_turns": 0, "system__conversation_history": json.dumps( {"entries": []}, ensure_ascii=False, separators=(",", ":") ), } class DynamicVariableStore: """Mutable state owned by exactly one conversation.""" def __init__( self, values: dict[str, Primitive], secrets: dict[str, str] | None = None, ): self.values = dict(values) self.secrets = dict(secrets or {}) self.history: list[dict[str, str]] = [] @classmethod def from_config(cls, cfg: AssistantConfig) -> "DynamicVariableStore": return cls(cfg.dynamic_variables, cfg.secret_dynamic_variables) def render(self, template: str, *, allow_secrets: bool = False) -> str: if not template: return template timezone = str(self.values.get("system__timezone") or "Asia/Shanghai") try: now = datetime.now(ZoneInfo(timezone)) self.values["system__time"] = now.strftime("%A, %H:%M %d %B %Y") self.values["system__time_utc"] = now.astimezone(ZoneInfo("UTC")).isoformat() except ZoneInfoNotFoundError: pass def replace(match: re.Match[str]) -> str: name = match.group(1) if name.startswith("secret__"): if not allow_secrets: raise DynamicVariableError(f"密钥变量 {name} 只能用于 HTTP Header") if name not in self.secrets: raise DynamicVariableError(f"缺少密钥变量: {name}") return self.secrets[name] if name not in self.values: raise DynamicVariableError(f"缺少动态变量: {name}") value = self.values[name] if isinstance(value, bool): return "true" if value else "false" return str(value) return PLACEHOLDER.sub(replace, template) def render_data(self, value: Any, *, allow_secrets: bool = False) -> Any: if isinstance(value, str): return self.render(value, allow_secrets=allow_secrets) if isinstance(value, list): return [self.render_data(item, allow_secrets=allow_secrets) for item in value] if isinstance(value, dict): return { key: self.render_data(item, allow_secrets=allow_secrets) for key, item in value.items() } return deepcopy(value) def record(self, role: str, content: str, *, completed_agent_turn: bool = False) -> None: if content: self.history.append({"role": role, "message": content}) self.history = self.history[-MAX_HISTORY_ENTRIES:] if completed_agent_turn: self.values["system__agent_turns"] = int( self.values.get("system__agent_turns", 0) ) + 1 serialized = json.dumps( {"entries": self.history}, ensure_ascii=False, separators=(",", ":") ) while len(serialized) > MAX_HISTORY_JSON_LENGTH and len(self.history) > 1: self.history.pop(0) serialized = json.dumps( {"entries": self.history}, ensure_ascii=False, separators=(",", ":"), ) self.values["system__conversation_history"] = serialized def assign(self, name: str, value: Any) -> None: if name.startswith(("system__", "secret__")) or not VARIABLE_NAME.fullmatch(name): raise DynamicVariableError(f"工具不能更新保留变量: {name}") self.values[name] = _primitive(value, name) def prepare_dynamic_config( cfg: AssistantConfig, client_values: dict[str, Any] | None, *, assistant_id: str | None, timezone: str = "Asia/Shanghai", trusted_values: dict[str, Any] | None = None, ) -> AssistantConfig: """Validate and merge one call's values without mutating stored config.""" if cfg.type != "prompt": if client_values: raise DynamicVariableError("动态变量目前仅支持 prompt 助手") return cfg supplied = _validate_public(client_values) # Server-side CRM/order/auth providers can pass trusted_values. They have # precedence over browser input but still cannot claim reserved prefixes. supplied.update(_validate_public(trusted_values)) definitions = cfg.dynamic_variable_definitions or {} merged: dict[str, Primitive] = {} for name, definition in definitions.items(): if name.startswith(("system__", "secret__")) or not VARIABLE_NAME.fullmatch(name): raise DynamicVariableError(f"动态变量定义名称不合法: {name}") if name in supplied: value = supplied.pop(name) elif "default" in definition and definition.get("default") is not None: value = _primitive(definition["default"], name) elif definition.get("required", False): raise DynamicVariableError(f"缺少必填动态变量: {name}") else: continue expected = str(definition.get("type") or "string") if not _type_matches(value, expected): raise DynamicVariableError(f"动态变量 {name} 类型应为 {expected}") merged[name] = value merged.update(supplied) conversation_id = f"conv_{uuid4().hex[:20]}" merged.update( _system_values( assistant_id=assistant_id, conversation_id=conversation_id, timezone=timezone, ) ) prepared = cfg.model_copy(deep=True) prepared.dynamic_variables = merged prepared.conversation_id = conversation_id # Validate prompt and greeting before media resources are allocated. store = DynamicVariableStore.from_config(prepared) store.render(prepared.prompt) store.render(prepared.greeting) return prepared def value_at_path(payload: Any, path: str) -> Any: current = payload for part in path.split("."): if isinstance(current, dict) and part in current: current = current[part] elif isinstance(current, list) and part.isdigit() and int(part) < len(current): current = current[int(part)] else: raise KeyError(path) return current