昨天我們讓模型能自己選工具，從人工判斷變成自動決策，不過還有一個問題，那麼就是 :
它的答案到底好不好？準不準？有沒有亂講？
光靠肉眼看很難判斷，這時就要請出我們之前提過的 RAGAS！

今天的目標是：

• 讓 Agent 回答完後自動評分（用 RAGAS 的四個指標）
• 把分數寫進 JSONL，方便之後丟給 n8n 或 Grafana 畫趨勢圖
• 如果分數太低或答案太短，就自我修正重答一次

為什麼要寫進 JSONL？
因為 JSONL（每行一筆 JSON）很適合被各種工具自動讀取。未來我們就可以用外部系統來追蹤 Agent 的表現。
例如：今天回答平均分數多少？哪些問題最常出錯？RAGAS 分數有沒有越來越好？

n8n
它是一個「開源的自動化流程平台」，它能幫我們自動「每天讀取 JSONL → 分析 → 存進 Google Sheet 或資料庫」，也能設定「如果分數太低，就自動發通知到 Slack / LINE / Email」，總而言之就是會去自動整理與監控 Agent 的表現

檔案架構

一樣只有新增新的檔案，不要刪掉舊的資料夾窩><

project/
└─ utils/
   ├─ metrics.py    # 封裝 RAGAS 評分
   └─ guard.py      # 失敗保護：分數太低自我修正

1. 新增 utils/metrics.py

這邊主要負責呼叫 RAGAS 的四個評分指標，幫 Agent 打分數。
我們會封裝成一個簡單的 compute_metrics() 函式，之後只要在 Agent 回答完後呼叫一次就能拿到結果！

from __future__ import annotations
from typing import List, Dict, Any

from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

# LLM 與 Embeddings 包裝（RAGAS 要求物件，不是字串）
from ragas.llms import LangchainLLMWrapper
from langchain_ollama import ChatOllama, OllamaLLM
from ragas.embeddings import HuggingfaceEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings as LCHuggingFaceEmbeddings
from ragas.embeddings import LangchainEmbeddingsWrapper


def _results_to_scores(result_obj: Any) -> Dict[str, float]:
    """把 RAGAS evaluate 回傳的結果轉成 {metric: score}."""
    try:
        df = result_obj.to_pandas()  # 新版 RAGAS
    except AttributeError:
        df = result_obj              # 舊版可能直接是 DataFrame

    num = df.select_dtypes(include="number")
    if num.empty:
        return {}
    import math
    scores: Dict[str, float] = {}
    for col in num.columns:
        v = float(num[col].mean())
        if not (math.isnan(v) or math.isinf(v)):
            scores[col] = round(v, 4)
    return scores


def compute_metrics(
    query: str,
    hits: List[Dict[str, Any]],
    answer: str,
    reference: str | None = None,
    model: str = "mistral",
) -> Dict[str, float]:
    """
    舊路徑：用 Dataset 跑 RAGAS 四指標。hits：每個元素需含 'text'。
    """
    try:
        contexts = [h.get("text", "") for h in (hits or [])] or [""]
        ds = Dataset.from_dict({
            "question": [query],
            "contexts": [contexts],
            "answer": [answer],
            "reference": [reference or ""],
        })

        # 評分用 LLM（LangChain 物件 → RAGAS wrapper）
        ollama_raw = OllamaLLM(model=model, temperature=0, num_ctx=4096)
        llm = LangchainLLMWrapper(ollama_raw)

        # 評分用 Embeddings（用 LangChain 的 HF embeddings，再包 RAGAS wrapper）
        lc_hf = LCHuggingFaceEmbeddings(model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
        emb = LangchainEmbeddingsWrapper(lc_hf)

        result = evaluate(
            ds,
            metrics=[context_precision, context_recall, answer_relevancy, faithfulness],
            llm=llm,
            embeddings=emb,
            show_progress=False,
        )
        return _results_to_scores(result)
    except Exception as e:
        print(f"[metrics] 評分失敗：{e}")
        return {}


def evaluate_ragas(
    question: str,
    answer: str,
    contexts: List[str] | None,
    ground_truths: List[str] | None,
    llm_model: str = "mistral",
    emb_model: str = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
) -> Dict[str, float]:
    """
    輕量評分：直接餵 python dict list（新版 RAGAS 支援）。
    重要：llm / embeddings 必須是物件，不能是字串。
    """
    sample = {
        "question": question,
        "answer": answer,
        "contexts": contexts or [],
        "ground_truths": ground_truths or [],
    }

    # 這裡用 ChatOllama（chat 型），再用 RAGAS 的 Langchain wrapper 包起來
    e_llm = LangchainLLMWrapper(ChatOllama(model=llm_model, temperature=0))
    # 這裡用 RAGAS 內建的 HF embeddings（非 LangChain 版）
    e_emb = HuggingfaceEmbeddings(model_name=emb_model)

    result = evaluate(
        [sample],
        metrics=[context_precision, context_recall, answer_relevancy, faithfulness],
        llm=e_llm,
        embeddings=e_emb,
        show_progress=False,
    )
    return _results_to_scores(result)

2. 新增 utils/guard.py

這個是「失敗保護機制」，當答案太短或分數太低時，會讓 Agent 自我修正重答一次。

import re
from typing import Dict, Tuple

def is_too_short(ans: str, min_chars: int = 120) -> bool:
    """答案太短就視為可疑"""
    return len((ans or "").strip()) < min_chars

def is_low_score(scores: Dict[str, float], th: float = 0.6) -> bool:
    """任一關鍵指標低於門檻即視為品質不佳"""
    keys = ["faithfulness", "answer_relevancy", "context_precision"]
    vals = [scores.get(k, 1.0) for k in keys]
    return any(v < th for v in vals)

def must_retry_for_article(q: str, ans: str) -> bool:
    """
    若使用者問『第N條』，但答案中沒有清楚出現『第N條』或沒有條文原文區塊，則要求重答。
    """
    m = re.search(r"第\s*([0-9０-９]{1,3})\s*條", q)
    if not m:
        return False
    n = int(re.sub(r"[０-９]", lambda d: str(ord(d.group()) - 65248), m.group(1)))
    has_num = (f"第{n}條" in ans)
    has_block = ("【條文原文】" in ans and "【說明完畢】" in ans)
    return (not has_num) or (not has_block)

def need_retry(ans: str, scores: Dict[str, float], min_chars=120, th=0.6) -> Tuple[bool, str]:
    """
    一般重答規則：答案過短或分數過低。
    """
    if is_too_short(ans, min_chars):
        return True, "答案過短，請補充條列解釋、補齊定義與關鍵條文索引。"
    if is_low_score(scores, th):
        return True, "品質分數偏低，請更聚焦問題、引用正確片段並在句尾附來源索引。"
    return False, ""

3. 修改 agent/agent_autogen.py

import sys, json, argparse, asyncio, time, re
from pathlib import Path
from datetime import datetime

# import project/*
sys.path.append(str(Path(__file__).resolve().parents[1]))

from rag.connect import search_chunks, build_prompt, ask_ollama, article_lookup
from utils.memory import memory_lines, remember
from utils.metrics import compute_metrics, evaluate_ragas
from utils.guard import need_retry, must_retry_for_article


from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.ollama import OllamaChatCompletionClient


# === 工具：條文直接查詢 ===
def lookup_article_tool(q: str) -> str:
    """
    直查第N條全文；輸出固定模板：
    【條文原文】
    <章｜第N條>
    <逐字原文>
    【說明完畢】
    """
    text = article_lookup(q) or ""
    if not text:
        return "【條文原文】\n（沒有查到對應條文）\n【說明完畢】"
    # 保持原樣，避免模型「幫忙潤飾」
    return f"【條文原文】\n{text}\n【說明完畢】"


# === 工具：一般問答（RAG 檢索 + 生成） ===
def search_law_tool(query: str, k: int = 4) -> str:
    hits = search_chunks(query, k)
    prompt = build_prompt(query, hits, mem_lines=memory_lines())
    return ask_ollama(prompt).strip()


# === 取最終文字（避免印整串 trace） ===
def final_text(result):
    msgs = getattr(result, "messages", [])
    texts = [getattr(m, "content", "") for m in msgs if getattr(m, "type", "") == "TextMessage"]
    return texts[-1] if texts else getattr(result, "content", str(result))


# === 從答案擷取條文原文區塊，給 RAGAS 當 contexts ===
_ARTICLE_BLOCK_RE = re.compile(r"【條文原文】\s*(.*?)\s*【說明完畢】", re.S)

def extract_article_contexts(answer: str) -> list[dict]:
    m = _ARTICLE_BLOCK_RE.search(answer or "")
    if not m:
        return []
    raw = m.group(1).strip()
    # 直接當作唯一 context
    return [{"text": raw}]

async def main():
    ap = argparse.ArgumentParser(description="AgentChat（Tool Selection + RAGAS + Guard）")
    ap.add_argument("--q", required=True, help="問題內容")
    ap.add_argument("--model", default="mistral", help="Ollama 模型名（如 mistral/llama3）")
    ap.add_argument("--max_tool_iter", type=int, default=3, help="單次最多工具呼叫次數")
    ap.add_argument("--json_out", action="store_true", help="是否輸出 JSONL 到 logs/")
    args = ap.parse_args()

    model_client = OllamaChatCompletionClient(model=args.model)

    agent = AssistantAgent(
        name="law_agent",
        model_client=model_client,
        tools=[lookup_article_tool, search_law_tool],
        max_tool_iterations=args.max_tool_iter,
         system_message=(
        "你是台灣資安法規助理。"
        "【重要】所有輸出一律為純文字，不要用 JSON、不要用鍵值對、不要用表格語法。"
        "• 若偵測到『第N條』這類條文查詢，必須優先呼叫 lookup_article_tool。"
        "• 只要使用了 lookup_article_tool，請『逐字輸出工具回傳文本』，不得改寫或補充意見；"
        "  僅可在文本下方附一行極簡備註（例如：來源：第N條）。"
        "• 一般法規問題請用 search_law_tool（RAG 檢索），回答要條列、每條句末加來源索引。"
        "• 若工具回傳『沒有查到對應條文』，請直接回覆該訊息，不要憑空生成內容。"
        ),
        reflect_on_tool_use=False,
        model_client_stream=False,
    )

    # === 第一次回答 ===
    t0 = time.time()
    try:
        result = await agent.run(task=args.q)
        answer = final_text(result)
    except Exception as e:
        # 任何解析 / 執行錯誤，直接給空字串；下面條文覆蓋會接手
        print(f"[agent] 失敗但已攔截：{e}")
        answer = ""

    elapsed_ms = int((time.time() - t0) * 1000)

    # === 條文強制覆蓋 ===
    article_hits = []
    m = re.search(r"第\s*([0-9０-９]{1,3})\s*條", args.q)
    if m:
        raw = article_lookup(args.q) or ""
        if raw:
            answer = f"【條文原文】\n{raw}\n【說明完畢】\n（來源：第{int(re.sub(r'[０-９]', lambda d: str(ord(d.group())-65248), m.group(1)))}條）"
            article_hits = [{"text": raw}]
        else:
            answer = "【條文原文】\n（沒有查到對應條文）\n【說明完畢】"
            article_hits = []
    else:
        article_hits = extract_article_contexts(answer)
    
    # === 建立 RAGAS 樣本 ===
    # 有條文原文就用它當 contexts（article_hits 由前面「條文覆蓋」邏輯產生）
    scores = compute_metrics(
        query=args.q,
        hits=article_hits,          
        answer=answer,
        model=args.model,           # 評分也用你指定的 ollama 模型
    )

    # # === RAGAS 評分 ===
    # scores = compute_metrics(args.q, hits=article_hits, answer=answer, model=args.model)

    # === 失敗保護===
    retry = False
    reason = ""
    # 條文類檢查：問第N條但沒原文/沒條號
    if must_retry_for_article(args.q, answer):
        retry, reason = True, "條文直接查詢結果不完整，請直接逐字輸出該條原文（勿改寫），並附上章名與條號。"
    else:
        # 一般規則：過短 / 低分
        r2, reason2 = need_retry(answer, scores, min_chars=120, th=0.6)
        if r2:
            retry, reason = True, reason2

    # === 二次嘗試（只重答一次） ===
    if retry:
        improved_query = f"{args.q}\n\n[改進指示] {reason}"
        result2 = await agent.run(task=improved_query)
        improved_answer = final_text(result2)
        # 若更長就採用，並重算分數
        if len(improved_answer) > len(answer):
            answer = improved_answer
            article_hits = extract_article_contexts(answer)
            scores = compute_metrics(args.q, hits=article_hits, answer=answer, model=args.model)

    # === 輸出 ===
    print("\n=== 最終回答 ===")
    print(answer)
    print("\n=== 評分 ===")
    print(scores)

    # === 記憶 + 日誌 ===
    remember(args.q, answer)

    if args.json_out:
        logs_dir = Path("logs")
        logs_dir.mkdir(parents=True, exist_ok=True)
        outpath = logs_dir / f"run_{datetime.now():%Y%m%d}.jsonl"
        record = {
            "ts": datetime.now().isoformat(timespec="seconds"),
            "mode": "autogen-1agent",
            "query": args.q,
            "model": args.model,
            "elapsed_ms": elapsed_ms,
            "answer_chars": len(answer),
            "scores": scores,
            "retry": {"did_retry": retry, "reason": reason},
            # 把條文 context 也記一份（方便之後排除錯誤/評分對照）
            "contexts": [h["text"] for h in article_hits],
            "answer": answer,
        }
        outpath.open("a", encoding="utf-8").write(json.dumps(record, ensure_ascii=False) + "\n")
        print(f"\n[log] 已寫入：{outpath}")


if __name__ == "__main__":
    asyncio.run(main())

可以輸入這個指令做確認：

python agent/agent_autogen.py --q "第1條是什麼？" --json_out

提醒：
RAGAS 的四個指標中，context_precision 和 context_recall 是用來衡量「檢索結果與答案的關聯度」。
所以如果問題是「第 N 條是什麼？」這種直接查條文的情境（沒有經過檢索），這兩個分數可能會是 0 —— 這是正常的，不代表模型出錯！

今天就到這邊了，明天會做一個收尾！