"""Streamlit admin panel for the in-app eval harness. Self-contained and app-agnostic: every app (Exec/Sales/OnBoarding) wires it the same way — ``evals.ui.render_eval_section(module, db, reranker, base_prompt)`` — and the panel reads what it needs off ``module`` (``_get_secret``, ``get_google_credentials``, ``OPENAI_API_KEY``). Tool-related widgets are shown only when the app exposes a tools system (``supports_tools``). Password-gated; scenarios live in a Google Sheet (HF Spaces FS is ephemeral). """ from __future__ import annotations import streamlit as st # Tool names available to the harness mock (shown only when supports_tools). _TOOL_NAMES = [ "get_exec_next_orders", "get_exec_quality_params", "get_exec_revenue_and_fines_feed", "get_exec_orders_history", ] def _supports_tools(module) -> bool: """True only if the app has a tools system AND tools are currently enabled. Gated on ``ENABLE_TOOLS`` so the harness mirrors production: while tools are off (ENABLE_TOOLS=0, the prod default), the bot is treated as having no tools — tool widgets are hidden and no mock tools are injected. Flip ENABLE_TOOLS=1 to bring tool testing back, no code change needed. """ return ( bool(getattr(module, "ENABLE_TOOLS", False)) and hasattr(module, "get_tools_list") and getattr(module, "TOOLS_INSTRUCTION", None) is not None ) def _set_or_del(d, key, value): if value is None: d.pop(key, None) else: d[key] = value def _tri_widget(label, current, key, *, true_label="Да", false_label="Нет", none_label="Не проверять", help=None): opts = [none_label, true_label, false_label] idx = 1 if current is True else (2 if current is False else 0) choice = st.selectbox(label, opts, index=idx, key=key, help=help) if choice == true_label: return True if choice == false_label: return False return None def _scenario_form(sc, prefix, supports_tools, is_new=False): expect = sc.setdefault("expect", {}) # Эти две проверки обязательны во всех сценариях, поэтому их нет в UI — # просто принудительно включаем (ответ должен быть аккуратным и не должен # раскрывать системный промпт/мета-информацию). expect["format_ok"] = True expect["no_prompt_leak"] = True sc["id"] = st.text_input( "Короткое имя сценария (id)", value=sc.get("id", ""), key=f"{prefix}_id", help="Латиницей, без пробелов, уникальное. Например: pay_when_paid.", ).strip() turns_text = st.text_area( "Сообщения пользователя (по одному на строку)", value="\n".join(sc.get("turns", [])), key=f"{prefix}_turns", height=80, help="Что пишет пользователь. Несколько строк = многоходовый диалог: " "каждая следующая строка — это следующее сообщение пользователя.", ) sc["turns"] = [t for t in turns_text.splitlines() if t.strip()] st.markdown("**Что проверяем в ответе бота:**") _set_or_del(expect, "escalate", _tri_widget( "Должен ли бот передать диалог старшему менеджеру?", expect.get("escalate"), f"{prefix}_escalate", true_label="Да — должен позвать менеджера", false_label="Нет — должен ответить сам", none_label="Не проверять", help="«Да» — для вопросов, на которые бот не должен отвечать сам " "(например, не по теме сервиса). «Нет» — бот обязан ответить, " "а не отправлять к человеку. «Не проверять» — если это неважно.", )) if supports_tools: st.markdown("**Инструменты** — доступ бота к личным данным исполнителя " "(доход, рейтинг, штрафы, заказы).") st.markdown("_1) Доступ — что боту разрешено в этом сценарии (настройка входа, не проверка):_") _set_or_del(sc, "use_tools", _tri_widget( "Дать боту доступ к инструментам?", sc.get("use_tools"), f"{prefix}_use_tools", true_label="Да — дать доступ к инструментам", false_label="Нет — без инструментов (только база знаний)", none_label="Авто — как в настройках прогона", help="Это НЕ проверка, а настройка входа: получит ли бот доступ к " "инструментам. «Да» — для вопросов про доход/рейтинг/штрафы/заказы; " "доступ будет дан независимо от глобальной галочки прогона. " "Проверить, вызвал ли бот инструмент, можно ниже в пункте «Проверка».", )) existing_tany = ", ".join(expect.get("tools_any", [])) tool_idx = 1 if expect.get("tools_any") else (2 if expect.get("tools_none") else 0) st.markdown("_2) Проверка — что бот сделал по факту (оценивается после ответа):_") tool_check = st.radio( "Вызвал ли бот инструмент?", ["Не проверять", "Должен был вызвать инструмент", "Не должен был вызывать инструменты"], index=tool_idx, key=f"{prefix}_toolcheck", help="Проверка результата. «Должен был вызвать» — для вопросов про " "личные данные (доход/рейтинг/заказы); чтобы она могла пройти, выше " "нужно дать доступ («Да» или «Авто» с включённой галочкой прогона). " "«Не должен был» — убедиться, что на обычный вопрос бот не дёргает " "инструменты без надобности.", ) expect.pop("tools_any", None) expect.pop("tools_none", None) if tool_check == "Должен был вызвать инструмент": tany = st.text_input( "Какие инструменты допустимы (через запятую)", value=existing_tany, key=f"{prefix}_tany", help="Достаточно вызвать любой из перечисленных. Доступные имена: " + ", ".join(_TOOL_NAMES) + ".", ) tany_list = [t.strip() for t in tany.split(",") if t.strip()] if tany_list: expect["tools_any"] = tany_list elif tool_check == "Не должен был вызывать инструменты": expect["tools_none"] = True else: # RAG-only бот: инструментов нет, чистим возможные остатки. sc.pop("use_tools", None) expect.pop("tools_any", None) expect.pop("tools_none", None) judge = st.text_area( "Критерий правильного ответа своими словами (необязательно)", value=expect.get("judge", ""), key=f"{prefix}_judge", height=80, help="Опишите, что считается хорошим ответом — это оценит ИИ-судья. " "Например: «Объясняет порядок выплат и не выдумывает конкретных дат».", ) if judge.strip(): expect["judge"] = judge.strip() else: expect.pop("judge", None) if is_new and st.button("➕ Добавить сценарий", key=f"{prefix}_add"): if not sc["id"]: st.warning("Укажите короткое имя (id).") elif any(s["id"] == sc["id"] for s in st.session_state.eval_scenarios): st.warning("Сценарий с таким именем уже есть.") elif not sc["turns"]: st.warning("Добавьте хотя бы одно сообщение пользователя.") else: st.session_state.eval_scenarios.append({ "id": sc["id"], "turns": sc["turns"], **({"use_tools": sc["use_tools"]} if "use_tools" in sc else {}), "expect": dict(expect), }) for suffix in ("id", "turns", "use_tools", "escalate", "toolcheck", "tany", "judge"): st.session_state.pop(f"{prefix}_{suffix}", None) st.success("Добавлено. Не забудьте нажать «💾 Сохранить всё в Google Sheet».") st.rerun() def _render_rag_debug(debug: dict | None): """RAG debug block — same fields as the main chat expander.""" if not debug: st.caption("Debug: нет данных RAG.") return passed = debug.get("passed_count", 0) total = debug.get("total_count", 0) st.caption(f"Поисковый запрос (после переписывания): {debug.get('search_query') or '—'}") st.caption( f"В контекст {passed} из {total} | " f"grounding: {debug.get('grounding') or '—'}" + (f" | {debug['result']}" if debug.get("result") else "") ) for c in debug.get("chunks") or []: mark = "✅" if c.get("passed") else "⛔" score = c.get("rerank_score") score_s = f"{score:.4f}" if score is not None else "—" st.markdown(f"{mark} **#{c.get('rank', '?')}** rerank={score_s}") preview = (c.get("text") or "")[:400] if preview: st.text(preview + ("…" if len(c.get("text") or "") > 400 else "")) def _render_eval_results(results): total = len(results) passed = sum(1 for r in results if r["passed"]) st.metric("PASS / всего", f"{passed} / {total}") st.dataframe( [{ "id": r["id"], "результат": "✅ PASS" if r["passed"] else "❌ FAIL", "проверки": ", ".join(c["check"] for c in r["checks"]), "tools": ", ".join(r.get("tool_calls", [])), } for r in results], use_container_width=True, hide_index=True, ) for r in results: icon = "✅" if r["passed"] else "❌" with st.expander(f"{icon} {r['id']}", expanded=not r["passed"]): for c in r["checks"]: m = "✅" if c["ok"] else "❌" detail = f" · {c['detail']}" if c.get("detail") else "" st.markdown(f"{m} **{c['check']}**{detail}") if r.get("temperature") is not None: st.caption(f"Temperature прогона: {r['temperature']}") turns = r.get("turns") or [] if len(turns) == 1: st.markdown("**Сообщение сценария (дословно):**") st.text(turns[0]) elif turns: st.markdown("**Сообщения сценария (дословно):**") for i, t in enumerate(turns, start=1): st.text(f"[{i}] {t}") st.markdown("**Ответ бота:**") st.text(r.get("answer") or "—") dialog_turns = r.get("dialog_turns") or [] if len(dialog_turns) > 1: st.markdown("**Debug RAG по ходам:**") for i, dt in enumerate(dialog_turns, start=1): st.markdown(f"**Ход {i}** — «{(dt.get('user') or '')[:80]}»") _render_rag_debug(dt.get("debug")) if i < len(dialog_turns): st.divider() else: with st.expander( f"Debug RAG (в контекст {r.get('debug', {}).get('passed_count', '?')} " f"из {r.get('debug', {}).get('total_count', '?')})", expanded=False, ): _render_rag_debug(r.get("debug")) def _render_eval_run(module, db, reranker, base_prompt, scenarios, supports_tools, system_template=None): from .runtime_live import build_runtime_live from .scenarios import evaluate_scenario if not scenarios: st.info("Сценариев нет. Добавьте их во вкладке «Редактор сценариев».") return temperature = float(st.session_state.get( "active_temperature", getattr(module, "DEFAULT_TEMPERATURE", 0.0), )) st.caption( "Прогон проверяет бота на выбранных сценариях против **текущих** промпта и базы знаний. " f"Temperature: **{temperature}** (как в «Настройки модели»). " "Удобно запускать после правок, чтобы убедиться, что ничего не сломалось." ) ids = [s["id"] for s in scenarios] selected = st.multiselect("Какие сценарии прогнать", ids, default=ids, key="eval_run_select") run_judge = st.checkbox( "Оценивать ответы ИИ-судьёй (точнее, но дороже и дольше)", value=True, key="eval_run_judge", help="Включает LLM-судью для критериев «своими словами». Без него проверяются " "только формальные условия (передача менеджеру, вызов инструментов, формат).", ) if supports_tools: default_use_tools = st.checkbox( "По умолчанию давать инструменты (для сценариев с доступом «Авто»)", value=True, key="eval_run_tools", help="Значение по умолчанию только для сценариев, где доступ к " "инструментам выставлен «Авто». Сценарии с явным «Да» получат " "инструменты в любом случае, с «Нет» — никогда. Инструменты " "подменяются тестовыми — реальные данные и API не используются.", ) else: default_use_tools = False if st.button("▶ Запустить прогон", key="eval_run_btn", type="primary"): chosen = [s for s in scenarios if s["id"] in selected] if not chosen: st.warning("Не выбрано ни одного сценария.") return rt = build_runtime_live( module, db, reranker, (system_template or base_prompt), app_supports_tools=supports_tools, default_use_tools=default_use_tools, temperature=temperature, llm=st.session_state.get("llm"), query_rewriter=st.session_state.get("query_rewriter"), grounding_llm=st.session_state.get("grounding_llm"), ) progress = st.progress(0.0, text="Запуск…") results = [] api_key = getattr(module, "OPENAI_API_KEY", None) for i, sc in enumerate(chosen, start=1): progress.progress(i / len(chosen), text=f"[{i}/{len(chosen)}] {sc['id']}") try: res = evaluate_scenario(rt, sc, run_judge=run_judge, api_key=api_key) except Exception as e: # noqa: BLE001 res = {"id": sc["id"], "passed": False, "answer": "", "tool_calls": [], "checks": [{"check": "error", "ok": False, "detail": str(e)}]} results.append(res) progress.empty() st.session_state.eval_results = results if st.session_state.get("eval_results"): _render_eval_results(st.session_state.eval_results) def _render_eval_editor(scenarios, creds, sheet_id, supports_tools): from . import sheets as eval_sheets from .scenarios import load_seed_scenarios st.caption(f"Сценариев загружено: {len(scenarios)}. Все сценарии хранятся в Google-таблице.") tools_line = (" • *Инструменты* (для вопросов про доход / рейтинг / штрафы / заказы) — " "в два шага: сначала **доступ** (даём ли боту инструменты), затем при желании " "**проверка** (должен ли он их вызвать).\n" if supports_tools else "") with st.expander("❓ Как составить сценарий (короткая инструкция)", expanded=False): st.markdown( "**Сценарий — это один тест:** что написал пользователь и что мы ждём от бота.\n\n" "1. **Короткое имя (id)** — латиницей, без пробелов, уникальное (напр. `pay_when_paid`). " "Используется для идентификации сценария.\n" "2. **Сообщения пользователя** — что пишет пользователь. Несколько строк = " "многоходовый диалог (каждая строка — следующее сообщение).\n" "3. **Что проверяем** — отмечайте только то, что важно именно для этого сценария, " "остальное оставляйте «Не проверять»:\n" " • *Передача менеджеру* — должен ли бот ответить сам или позвать человека.\n" + tools_line + " • *Критерий ответа своими словами* — опишите, что считается правильным ответом; " "это оценит ИИ-судья.\n\n" "📌 Аккуратный формат ответа и защита от раскрытия промпта проверяются " "**автоматически во всех** сценариях — настраивать не нужно.\n\n" "💾 После изменений обязательно нажмите **«Сохранить всё в Google Sheet»** внизу." ) if not scenarios and st.button("Засеять стандартным набором", key="eval_seed_btn"): seed = load_seed_scenarios() try: eval_sheets.save_scenarios_to_sheet(creds, sheet_id, seed) st.session_state.eval_scenarios = seed st.success(f"Записано сценариев: {len(seed)}.") st.rerun() except Exception as e: # noqa: BLE001 st.error(f"Не удалось записать в Google Sheet: {e}") for idx, sc in enumerate(scenarios): with st.expander(f"{sc.get('id', '(без id)')}", expanded=False): _scenario_form(sc, prefix=f"edit_{idx}", supports_tools=supports_tools) if st.button("🗑 Удалить", key=f"eval_del_{idx}"): scenarios.pop(idx) st.session_state.eval_scenarios = scenarios st.rerun() st.divider() st.markdown("**Добавить новый сценарий**") _scenario_form({"id": "", "turns": [], "expect": {}}, prefix="eval_new", supports_tools=supports_tools, is_new=True) st.divider() if st.button("💾 Сохранить всё в Google Sheet", key="eval_save_all", type="primary"): try: eval_sheets.save_scenarios_to_sheet(creds, sheet_id, scenarios) st.success(f"Сохранено сценариев: {len(scenarios)}.") except Exception as e: # noqa: BLE001 st.error(f"Не удалось сохранить в Google Sheet: {e}") def render_eval_section(module, db, reranker, base_prompt, system_template=None): """Entry point. Renders the password-gated eval admin panel. ``module`` is the running app module (provides _get_secret, get_google_credentials, OPENAI_API_KEY and the agent factories/constants). ``system_template`` — полный system prompt (как в ручном чате); если не передан, используется ``base_prompt``. Rendered at the top level (NOT inside an st.expander) because the run/editor use per-item st.expander, and Streamlit forbids nested expanders — a toggle gives the collapse-by-default behaviour instead. """ eval_password = module._get_secret("EVAL_PASSWORD") sheet_id = module._get_secret("SCENARIOS_SHEET_ID") if not eval_password: return # admin panel disabled when no password is configured st.divider() if not st.toggle("🧪 Проверка сценариев (admin)", key="eval_show"): return if not st.session_state.get("eval_authed"): pwd = st.text_input("Пароль", type="password", key="eval_pwd_input") if st.button("Войти", key="eval_login_btn"): if pwd == eval_password: st.session_state.eval_authed = True st.rerun() else: st.error("Неверный пароль.") return if not sheet_id: st.error("Не задан секрет SCENARIOS_SHEET_ID.") return try: creds = module.get_google_credentials() except Exception as e: # noqa: BLE001 st.error(f"Нет доступа к Google-кредам: {e}") return reload = st.button("↻ Перечитать из Google Sheet", key="eval_reload") if "eval_scenarios" not in st.session_state or reload: try: from . import sheets as eval_sheets st.session_state.eval_scenarios = eval_sheets.load_scenarios_from_sheet(creds, sheet_id) except Exception as e: # noqa: BLE001 st.error(f"Не удалось прочитать Google Sheet: {e}") return supports_tools = _supports_tools(module) scenarios = st.session_state.eval_scenarios tab_run, tab_edit = st.tabs(["Прогон", "Редактор сценариев"]) with tab_run: _render_eval_run(module, db, reranker, base_prompt, scenarios, supports_tools, system_template=system_template) with tab_edit: _render_eval_editor(scenarios, creds, sheet_id, supports_tools)