HadTavern 0.01: Gemini/Claude fixes; UI _origId reuse; docs; .bat open

agentui/pipeline/templating.py

@@ -0,0 +1,308 @@
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Dict, List, Optional
+
+__all__ = [
+    "_OUT_MACRO_RE",
+    "_VAR_MACRO_RE",
+    "_PROMPT_MACRO_RE",
+    "_OUT_SHORT_RE",
+    "_BARE_MACRO_RE",
+    "_BRACES_RE",
+    "_split_path",
+    "_get_by_path",
+    "_stringify_for_template",
+    "_deep_find_text",
+    "_best_text_from_outputs",
+    "render_template_simple",
+]
+
+# Regex-макросы (общие для бэка)
+_OUT_MACRO_RE = re.compile(r"\[\[\s*OUT\s*[:\s]\s*([^\]]+?)\s*\]\]", re.IGNORECASE)
+_VAR_MACRO_RE = re.compile(r"\[\[\s*VAR\s*[:\s]\s*([^\]]+?)\s*\]\]", re.IGNORECASE)
+# Единый фрагмент PROMPT (провайдеро-специфичный JSON-фрагмент)
+_PROMPT_MACRO_RE = re.compile(r"\[\[\s*PROMPT\s*\]\]", re.IGNORECASE)
+# Короткая форма: [[OUT1]] — best-effort текст из ноды n1
+_OUT_SHORT_RE = re.compile(r"\[\[\s*OUT\s*(\d+)\s*\]\]", re.IGNORECASE)
+# Голые переменные: [[NAME]] или [[path.to.value]] — сначала ищем в vars, затем в контексте
+_BARE_MACRO_RE = re.compile(r"\[\[\s*([A-Za-z_][A-Za-z0-9_]*(?:\.[^\]]+?)?)\s*\]\]")
+# Подстановки {{ ... }} (включая простейший фильтр |default(...))
+_BRACES_RE = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")
+
+
+def _split_path(path: str) -> List[str]:
+    return [p.strip() for p in str(path).split(".") if str(p).strip()]
+
+
+def _get_by_path(obj: Any, path: Optional[str]) -> Any:
+    if path is None or path == "":
+        return obj
+    cur = obj
+    for seg in _split_path(path):
+        if isinstance(cur, dict):
+            if seg in cur:
+                cur = cur[seg]
+            else:
+                return None
+        elif isinstance(cur, list):
+            try:
+                idx = int(seg)
+            except Exception:  # noqa: BLE001
+                return None
+            if 0 <= idx < len(cur):
+                cur = cur[idx]
+            else:
+                return None
+        else:
+            return None
+    return cur
+
+
+def _stringify_for_template(val: Any) -> str:
+    if val is None:
+        return ""
+    if isinstance(val, bool):
+        # JSON-friendly booleans (useful when embedding into JSON-like templates)
+        return "true" if val else "false"
+    if isinstance(val, (dict, list)):
+        try:
+            return json.dumps(val, ensure_ascii=False)
+        except Exception:  # noqa: BLE001
+            return str(val)
+    return str(val)
+
+
+def _deep_find_text(obj: Any, max_nodes: int = 5000) -> Optional[str]:
+    """
+    Best-effort поиск первого текстового значения в глубине структуры JSON.
+    Сначала пытаемся по ключам content/text, затем общий обход.
+    """
+    try:
+        # Быстрые ветки
+        if isinstance(obj, str):
+            return obj
+        if isinstance(obj, dict):
+            c = obj.get("content")
+            if isinstance(c, str):
+                return c
+            t = obj.get("text")
+            if isinstance(t, str):
+                return t
+            parts = obj.get("parts")
+            if isinstance(parts, list) and parts:
+                for p in parts:
+                    if isinstance(p, dict) and isinstance(p.get("text"), str):
+                        return p.get("text")
+
+        # Общий нерекурсивный обход в ширину
+        queue: List[Any] = [obj]
+        seen = 0
+        while queue and seen < max_nodes:
+            cur = queue.pop(0)
+            seen += 1
+            if isinstance(cur, str):
+                return cur
+            if isinstance(cur, dict):
+                # часто встречающиеся поля
+                for k in ("text", "content"):
+                    v = cur.get(k)
+                    if isinstance(v, str):
+                        return v
+                # складываем все значения
+                for v in cur.values():
+                    queue.append(v)
+            elif isinstance(cur, list):
+                for it in cur:
+                    queue.append(it)
+    except Exception:
+        pass
+    return None
+
+
+def _best_text_from_outputs(node_out: Any) -> str:
+    """
+    Унифицированное извлечение "текста" из выхода ноды.
+    Поддерживает:
+    - PromptTemplate: {"text": ...}
+    - LLMInvoke: {"response_text": ...}
+    - ProviderCall/RawForward: {"result": <provider_json>}, извлекаем текст для openai/gemini/claude
+    - Общий глубокий поиск текста, если специфичные ветки не сработали
+    """
+    # Строка сразу
+    if isinstance(node_out, str):
+        return node_out
+
+    if not isinstance(node_out, dict):
+        return ""
+
+    # Явные короткие поля
+    if isinstance(node_out.get("response_text"), str) and node_out.get("response_text"):
+        return str(node_out["response_text"])
+    if isinstance(node_out.get("text"), str) and node_out.get("text"):
+        return str(node_out["text"])
+
+    res = node_out.get("result")
+    base = res if isinstance(res, (dict, list)) else node_out
+
+    # OpenAI
+    try:
+        if isinstance(base, dict):
+            ch0 = (base.get("choices") or [{}])[0]
+            msg = ch0.get("message") or {}
+            c = msg.get("content")
+            if isinstance(c, str):
+                return c
+    except Exception:
+        pass
+
+    # Gemini
+    try:
+        if isinstance(base, dict):
+            cand0 = (base.get("candidates") or [{}])[0]
+            content = cand0.get("content") or {}
+            parts0 = (content.get("parts") or [{}])[0]
+            t = parts0.get("text")
+            if isinstance(t, str):
+                return t
+    except Exception:
+        pass
+
+    # Claude
+    try:
+        if isinstance(base, dict):
+            blocks = base.get("content") or []
+            texts = [b.get("text") for b in blocks if isinstance(b, dict) and isinstance(b.get("text"), str)]
+            if texts:
+                return "\n".join(texts)
+    except Exception:
+        pass
+
+    # Общий глубокий поиск
+    txt = _deep_find_text(base)
+    return txt or ""
+
+
+def render_template_simple(template: str, context: Dict[str, Any], out_map: Dict[str, Any]) -> str:
+    """
+    Простая подстановка:
+    - {{ path }} — берёт из context (или {{ OUT.node.path }} для выходов)
+    - Поддержка фильтра по умолчанию: {{ path|default(value) }}
+      value может быть числом, строкой ('..'/".."), массивом/объектом в виде литерала.
+    - [[VAR:path]] — берёт из context
+    - [[OUT:nodeId(.path)*]] — берёт из out_map
+    Возвращает строку.
+    """
+    if template is None:
+        return ""
+    s = str(template)
+
+    # 1) Макросы [[VAR:...]] и [[OUT:...]]
+    def repl_var(m: re.Match) -> str:
+        path = m.group(1).strip()
+        val = _get_by_path(context, path)
+        return _stringify_for_template(val)
+
+    def repl_out(m: re.Match) -> str:
+        body = m.group(1).strip()
+        if "." in body:
+            node_id, rest = body.split(".", 1)
+            node_val = out_map.get(node_id)
+            val = _get_by_path(node_val, rest)
+        else:
+            val = out_map.get(body)
+        return _stringify_for_template(val)
+
+    s = _VAR_MACRO_RE.sub(repl_var, s)
+    s = _OUT_MACRO_RE.sub(repl_out, s)
+
+    # [[OUT1]] → текст из ноды n1 (best-effort)
+    def repl_out_short(m: re.Match) -> str:
+        try:
+            num = int(m.group(1))
+            node_id = f"n{num}"
+            node_out = out_map.get(node_id)
+            txt = _best_text_from_outputs(node_out)
+            return _stringify_for_template(txt)
+        except Exception:
+            return ""
+    s = _OUT_SHORT_RE.sub(repl_out_short, s)
+
+    # [[PROMPT]] — провайдеро-специфичный JSON-фрагмент, подготовленный в context["PROMPT"]
+    s = _PROMPT_MACRO_RE.sub(lambda _m: str(context.get("PROMPT") or ""), s)
+
+    # 1.5) Голые [[NAME]] / [[path.to.value]]
+    def repl_bare(m: re.Match) -> str:
+        name = m.group(1).strip()
+        # Зарезервированные формы уже обработаны выше; бережно пропускаем похожие
+        if name.upper() in {"OUT", "VAR", "PROMPT"} or re.fullmatch(r"OUT\d+", name.upper() or ""):
+            return m.group(0)
+        # Сначала пользовательские переменные
+        vmap = context.get("vars") or {}
+        if isinstance(vmap, dict) and name in vmap:
+            return _stringify_for_template(vmap.get(name))
+        # Затем путь из общего контекста
+        val = _get_by_path(context, name)
+        return _stringify_for_template(val)
+    s = _BARE_MACRO_RE.sub(repl_bare, s)
+
+    # 2) Подстановки {{ ... }} (+ simple default filter)
+    def repl_braces(m: re.Match) -> str:
+        expr = m.group(1).strip()
+
+        def eval_path(p: str) -> Any:
+            p = p.strip()
+            # Приоритет пользовательских переменных для простых идентификаторов {{ NAME }}
+            vmap = context.get("vars") or {}
+            if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", p) and isinstance(vmap, dict) and p in vmap:
+                return vmap.get(p)
+            if p.startswith("OUT."):
+                body = p[4:]
+                if "." in body:
+                    node_id, rest = body.split(".", 1)
+                    node_val = out_map.get(node_id)
+                    return _get_by_path(node_val, rest)
+                return out_map.get(body)
+            return _get_by_path(context, p)
+
+        default_match = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", expr)
+        if default_match:
+            base_path = default_match.group(1).strip()
+            fallback_raw = default_match.group(2).strip()
+
+            # Рекурсивная обработка вложенных default(...) и путей
+            def eval_default(raw: str) -> Any:
+                raw = raw.strip()
+                # Вложенный default: a|default(b)
+                dm = re.match(r"([^|]+)\|\s*default\((.*)\)\s*$", raw)
+                if dm:
+                    base2 = dm.group(1).strip()
+                    fb2 = dm.group(2).strip()
+                    v2 = eval_path(base2)
+                    if v2 not in (None, ""):
+                        return v2
+                    return eval_default(fb2)
+                # Пробуем как путь
+                v = eval_path(raw)
+                if v not in (None, ""):
+                    return v
+                # Явная строка в кавычках
+                if len(raw) >= 2 and ((raw[0] == '"' and raw[-1] == '"') or (raw[0] == "'" and raw[-1] == "'")):
+                    return raw[1:-1]
+                # Пробуем распарсить как JSON литерал (число/объект/массив/true/false/null)
+                try:
+                    return json.loads(raw)
+                except Exception:
+                    # Последний вариант: вернуть сырой текст. Для строк рекомендуется default('...') с кавычками.
+                    return raw
+
+            raw_val = eval_path(base_path)
+            val = raw_val if raw_val not in (None, "") else eval_default(fallback_raw)
+        else:
+            val = eval_path(expr)
+
+        return _stringify_for_template(val)
+
+    s = _BRACES_RE.sub(repl_braces, s)
+    return s