siemens_ragas/rag_eval/advisor/writer.py

"""Write optimization advice to markdown file and emit log summary."""
from __future__ import annotations

import logging
from pathlib import Path

from .rules import Diagnosis

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

# Chinese display labels for each severity tier.
_SEVERITY_LABEL: dict[str, str] = {
    "critical": "严重",
    "warning": "警告",
    "low": "待优化",
}


def _format_log_summary(diagnoses: list[Diagnosis], advice_path: Path) -> str:
    """Return a single-line log summary of triggered diagnoses."""
    if not diagnoses:
        return "[advisor] 所有指标正常，无需优化建议。"
    parts = [
        f"{d.metric}({d.mean_score:.2f},{_SEVERITY_LABEL.get(d.severity, d.severity)})"
        for d in diagnoses
    ]
    triggered = " ".join(parts)
    return f"[advisor] 触发诊断 {len(diagnoses)} 项: {triggered}  →  {advice_path}"


def _build_fallback_report(diagnoses: list[Diagnosis]) -> str:
    """Build a rules-only report when LLM analysis is unavailable.

    Even without the LLM, embed each metric's worst sample(s) — question,
    answer, ground truth — so the advice still references concrete problems
    instead of reading as a purely generic template.
    """
    if not diagnoses:
        return ""
    lines = ["## 规则诊断（LLM 分析不可用，以下为规则引擎输出）\n"]
    for d in diagnoses:
        label = _SEVERITY_LABEL.get(d.severity, d.severity)
        lines.append(f"### {d.metric}  [{label}]  均值={d.mean_score:.4f}")
        lines.append("\n**可能原因：**")
        for cause in d.root_causes:
            lines.append(f"- {cause}")
        lines.append("\n**建议动作：**")
        for action in d.suggested_actions:
            lines.append(f"- {action}")
        if d.low_samples:
            lines.append("\n**低分样本举例拆解：**")
            for i, sample in enumerate(d.low_samples, 1):
                score = sample.get(d.metric, "N/A")
                question = str(sample.get("question", "")).strip() or "（无问题文本）"
                lines.append(f"\n- **样本 {i}**（{d.metric}={score}）问题：{question}")
                answer = str(sample.get("answer", "")).strip()
                if answer:
                    lines.append(f"  - 生成答案：{answer[:200]}")
                ground_truth = str(sample.get("ground_truth", "")).strip()
                if ground_truth:
                    lines.append(f"  - 标准答案：{ground_truth[:160]}")
                lines.append(
                    f"  - 拆解：该样本 {d.metric} 偏低，请对照上述「可能原因 / 建议动作」"
                    f"重点排查本问题的检索片段与生成答案。"
                )
        lines.append("")
    return "\n".join(lines)


def write_advice(
    diagnoses: list[Diagnosis],
    llm_markdown: str,
    advice_path: Path,
    scenario_name: str,
    run_id: str,
    judge_model: str,
) -> None:
    """Write optimization_advice.md and emit a log summary line.

    Args:
        diagnoses: List of Diagnosis from rules.diagnose().
        llm_markdown: LLM-generated Markdown body. Empty string triggers fallback.
        advice_path: Full path to write the .md file.
        scenario_name: Human-readable scenario identifier for the report header.
        run_id: Run identifier string.
        judge_model: Model used for LLM analysis (shown in header).
    """
    advice_path.parent.mkdir(parents=True, exist_ok=True)

    from rag_eval.shared.utils import utc_now_iso
    header_lines = [
        f"# 优化建议报告 — {scenario_name}",
        "",
        f"- run_id: `{run_id}`",
        f"- 生成时间: `{utc_now_iso()}`",
        f"- judge_model: `{judge_model}`",
        "",
        "---",
        "",
    ]

    if not diagnoses:
        body = "## ✅ 未发现明显指标异常\n\n所有指标均在正常范围内，当前 RAG 链路表现良好。\n"
    elif llm_markdown:
        body = llm_markdown
    else:
        body = _build_fallback_report(diagnoses)

    content = "\n".join(header_lines) + body
    advice_path.write_text(content, encoding="utf-8")

    summary = _format_log_summary(diagnoses, advice_path)
    logger.info(summary)
    logger.info("[advisor] 优化建议已写出: %s", advice_path)