siemens_ragas/rag_eval/advisor/llm_analyzer.py

"""LLM-powered analysis of rule diagnostics and low-score samples.

The analyzer issues a single, direct OpenAI-compatible chat call (no langchain,
no instructor structured output) to produce a detailed Chinese optimization
report. It deliberately uses plain text generation so it works with any
OpenAI-compatible gateway and never trips the structured-output response-shape
issues seen with the scoring path.
"""
from __future__ import annotations

import logging
from typing import Any

from .rules import Diagnosis

logger = logging.getLogger("rag_eval.advisor")

# Worked-example oriented prompt: forces per-question decomposition (为什么低 →
# 拆解 → 怎么改) plus cross-metric causal reasoning, so the report is concrete
# instead of a generic restatement of the static rule templates.
_PROMPT_TEMPLATE = """\
你是一位资深的 RAG（检索增强生成）系统优化专家，正在分析西门子医疗 CT 文档问答系统的 RAGAS 评测结果。
请基于以下诊断数据与低分样本，用中文撰写一份**详细、具体、可落地**的优化建议报告（Markdown 格式）。

## 评测诊断摘要

{diagnosis_summary}

## 低分样本明细（含检索片段 contexts，用于定位问题出在检索还是生成环节）

{low_sample_text}

## 撰写要求

1. **按指标分节**：每个指标一个 `## 指标名 [严重程度]` 小节。
2. **每节必须包含「举例拆解」**：从该指标的低分样本中挑 1-2 个最典型的，逐条按如下结构拆解：
   - **问题**：简述该样本的 question
   - **当前得分**：该样本在此指标上的分数
   - **为什么低**：结合该样本的 answer / 检索片段 contexts / 标准答案 ground_truth，**具体指出**问题所在
     （例如：答案里哪句话没有被检索片段支持；检索片段里缺了哪个关键信息；答案偏离了问题的哪个点）
   - **拆解定位**：判断问题出在「检索环节」「生成环节」还是「两者兼有」
   - **优化动作**：针对这个具体样本，给出 1-3 条**可操作**的改法，不要泛泛而谈
3. **跨指标关联分析**：若多个指标同时偏低，分析其因果关系。例如：
   - faithfulness 低且 context_recall 低 → 多半是检索缺失关键信息，导致模型臆造（幻觉）
   - context_recall 正常但 context_precision 低 → 检索引入噪声，稀释了有效信息
   - faithfulness 低但 context_recall 高 → 生成环节 grounding 不足，需收紧生成 prompt
4. 最后写一节 `## 优先优化次序`：按性价比排序（不增加 LLM 调用次数的优化优先；critical/warning 优先于 low）。
5. 语言简洁，面向工程师，重点是「具体、可操作」。不要复述本提示词，不要无意义的客套。

严重程度说明：critical=严重（远低于阈值），warning=警告（低于阈值），low=待优化（达标但低于 0.85，仍有提升空间）。

只输出 Markdown 报告正文，不要任何前置说明或代码块包裹。
"""


_SEVERITY_LABEL_ZH: dict[str, str] = {
    "critical": "严重",
    "warning": "警告",
    "low": "待优化",
}

# Per-sample text limits keep the prompt bounded regardless of sample size.
_ANSWER_LIMIT = 400
_GT_LIMIT = 300
_CONTEXT_LIMIT = 600
_CONTEXT_SEPARATOR = " |||| "


def _build_diagnosis_summary(diagnoses: list[Diagnosis]) -> str:
    """Render the per-metric diagnosis block fed to the LLM."""
    lines = []
    for d in diagnoses:
        direction = "（越低越好）" if d.metric == "noise_sensitivity" else ""
        label = _SEVERITY_LABEL_ZH.get(d.severity, d.severity)
        lines.append(
            f"- **{d.metric}** {direction} 均值={d.mean_score:.4f}，"
            f"阈值={d.threshold}，严重程度={label}"
        )
        lines.append(f"  - 可能原因：{'; '.join(d.root_causes)}")
        lines.append(f"  - 建议动作：{'; '.join(d.suggested_actions)}")
    return "\n".join(lines)


def _format_contexts(raw: Any) -> str:
    """Render a sample's contexts (stored as a joined string) as a short list."""
    text = str(raw or "").strip()
    if not text:
        return "（无检索片段）"
    parts = [p.strip() for p in text.split(_CONTEXT_SEPARATOR) if p.strip()]
    if not parts:
        return "（无检索片段）"
    rendered = "; ".join(f"[{i + 1}] {p}" for i, p in enumerate(parts))
    if len(rendered) > _CONTEXT_LIMIT:
        rendered = rendered[:_CONTEXT_LIMIT] + "…"
    return rendered


def _build_low_sample_text(diagnoses: list[Diagnosis]) -> str:
    """Render low-score samples (now including contexts) for grounding analysis."""
    lines = []
    for d in diagnoses:
        if not d.low_samples:
            continue
        lines.append(f"### {d.metric} 低分样本（最多 3 条）")
        for i, s in enumerate(d.low_samples, 1):
            score = s.get(d.metric, "N/A")
            lines.append(f"\n**样本 {i}**（{d.metric}={score}）")
            lines.append(f"- 问题 question：{s.get('question', '')}")
            lines.append(f"- 生成答案 answer：{str(s.get('answer', ''))[:_ANSWER_LIMIT]}")
            lines.append(f"- 检索片段 contexts：{_format_contexts(s.get('contexts'))}")
            lines.append(f"- 标准答案 ground_truth：{str(s.get('ground_truth', ''))[:_GT_LIMIT]}")
    return "\n".join(lines)


def _is_reasoning_model(model: str) -> bool:
    """Return True for OpenAI reasoning models (gpt-5+/o-series/codex-mini).

    These require `max_completion_tokens` instead of `max_tokens` and do not
    accept a custom temperature. Mirrors RAGAS's own detection logic so advice
    generation stays consistent with the scoring path.
    """
    m = (model or "").lower()
    # O-series: o1..o9 optionally followed by - or _
    if len(m) >= 2 and m[0] == "o" and m[1] in "123456789":
        if len(m) == 2 or m[2] in ("-", "_"):
            return True
    # GPT-5 through GPT-19 generation
    if m.startswith("gpt-"):
        version_str = m[4:].split("-")[0].split("_")[0].split(".")[0]
        try:
            if 5 <= int(version_str) <= 19:
                return True
        except ValueError:
            pass
    if m == "codex-mini":
        return True
    return False


def _chat_token_kwargs(model: str, max_tokens: int) -> dict[str, Any]:
    """Return the completion-budget kwargs appropriate for the model family."""
    if _is_reasoning_model(model):
        # Reasoning models: only max_completion_tokens; temperature must stay default (1).
        return {"max_completion_tokens": max_tokens}
    return {"max_tokens": max_tokens, "temperature": 0.2}


async def analyze(
    diagnoses: list[Diagnosis],
    scenario_name: str,
    judge_model: str,
    settings: Any,
    *,
    chat_client: Any | None = None,
) -> str:
    """Call the judge LLM directly to generate a Chinese optimization report.

    Args:
        diagnoses: Non-empty list of Diagnosis from rules.diagnose().
        scenario_name: Used only for logging.
        judge_model: Model name to call (also selects the matching LLM profile).
        settings: EvaluationSettings (supplies client kwargs + token budget).
        chat_client: Optional pre-built AsyncOpenAI-compatible client (for tests).

    Returns:
        LLM-generated Markdown string, or "" on failure (triggers writer fallback).
    """
    if not diagnoses:
        return ""

    prompt = _PROMPT_TEMPLATE.format(
        diagnosis_summary=_build_diagnosis_summary(diagnoses),
        low_sample_text=_build_low_sample_text(diagnoses),
    )

    try:
        logger.info("[advisor] calling LLM for optimization analysis  scenario=%s", scenario_name)
        client = chat_client
        owns_client = False
        if client is None:
            from openai import AsyncOpenAI

            from rag_eval.metrics.factory import attach_usage_hook, resolve_openai_client_kwargs

            client = AsyncOpenAI(**resolve_openai_client_kwargs(judge_model, settings))
            attach_usage_hook(client)
            owns_client = True

        try:
            # Advice is a longer document than per-metric scoring; give it headroom.
            max_tokens = max(2048, int(getattr(settings, "ragas_llm_max_tokens", 4096) or 4096))
            token_kwargs = _chat_token_kwargs(judge_model, max_tokens)

            response = await client.chat.completions.create(
                model=judge_model,
                messages=[{"role": "user", "content": prompt}],
                **token_kwargs,
            )
            text = (response.choices[0].message.content or "").strip()
            logger.info("[advisor] LLM analysis complete  chars=%d", len(text))
            return text
        finally:
            # Close the httpx connection pool inside THIS event loop. run_advisor
            # drives analyze() via asyncio.run(), which closes the loop on return;
            # a later GC-time aclose() on the dead loop would otherwise log
            # "RuntimeError: Event loop is closed". Only close clients we created.
            if owns_client:
                try:
                    await client.close()
                except Exception:  # noqa: BLE001
                    pass
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "[advisor] LLM analysis failed (%s: %s) — falling back to rule report",
            type(exc).__name__, exc,
        )
        return ""