forked from hesabix/arc
259 lines
8.9 KiB
Python
259 lines
8.9 KiB
Python
"""امتیازدهی assertionهای eval بدون LLM (OBS-01).
|
||
|
||
substring هنوز هست؛ این لایه tool_called / تأیید نوشتن / استناد / زبان را می‌سنجد.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
|
||
|
||
from app.services.ai.ai_citation_service import (
|
||
citations_from_function_results,
|
||
content_has_moneyish_claim,
|
||
)
|
||
from app.services.ai.ai_write_guard import is_write_function, is_write_guard_stop_result
|
||
|
||
_FA_LETTER_RE = re.compile(r"[\u0600-\u06FF]")
|
||
_LATIN_LETTER_RE = re.compile(r"[A-Za-z]")
|
||
_SUCCESS_WRITE_CLAIM_RE = re.compile(
|
||
r"(با موفقیت ثبت|ثبت شد|ایجاد شد|حذف شد|به‌روزرسانی شد|بروزرسانی شد)",
|
||
)
|
||
|
||
|
||
def _iter_called_tool_names(
|
||
function_calls: Any,
|
||
function_results: Any,
|
||
) -> List[str]:
|
||
names: List[str] = []
|
||
if isinstance(function_calls, list):
|
||
for item in function_calls:
|
||
if isinstance(item, dict):
|
||
name = item.get("name") or (item.get("function") or {}).get("name")
|
||
if name:
|
||
names.append(str(name))
|
||
if isinstance(function_results, dict):
|
||
for key, raw in function_results.items():
|
||
if str(key).startswith("_"):
|
||
continue
|
||
if isinstance(raw, dict):
|
||
nested = raw.get("name")
|
||
if nested:
|
||
names.append(str(nested))
|
||
continue
|
||
names.append(str(key))
|
||
seen = set()
|
||
unique: List[str] = []
|
||
for name in names:
|
||
if name in seen:
|
||
continue
|
||
seen.add(name)
|
||
unique.append(name)
|
||
return unique
|
||
|
||
|
||
def _lookup_results_by_name(function_results: Any) -> Dict[str, Any]:
|
||
out: Dict[str, Any] = {}
|
||
if not isinstance(function_results, dict):
|
||
return out
|
||
for key, raw in function_results.items():
|
||
if str(key).startswith("_"):
|
||
continue
|
||
payload = raw
|
||
name = str(key)
|
||
if isinstance(raw, dict) and "result" in raw and raw.get("name"):
|
||
name = str(raw.get("name"))
|
||
payload = raw.get("result")
|
||
out[name] = payload
|
||
return out
|
||
|
||
|
||
def assert_tool_called(
|
||
expected_tools: Sequence[str],
|
||
function_calls: Any,
|
||
function_results: Any,
|
||
) -> Tuple[bool, List[str]]:
|
||
"""حداقل یکی از ابزارهای فهرست‌شده باید صدا شده باشد."""
|
||
if not expected_tools:
|
||
return True, []
|
||
called = {n.lower() for n in _iter_called_tool_names(function_calls, function_results)}
|
||
missing = [t for t in expected_tools if str(t).lower() not in called]
|
||
hit = len(missing) < len(list(expected_tools))
|
||
return hit, missing
|
||
|
||
|
||
def assert_no_write_without_approval(
|
||
content: str,
|
||
function_calls: Any,
|
||
function_results: Any,
|
||
) -> Tuple[bool, Optional[str]]:
|
||
called = _iter_called_tool_names(function_calls, function_results)
|
||
writes = [n for n in called if is_write_function(n)]
|
||
claimed_success = bool(_SUCCESS_WRITE_CLAIM_RE.search(content or ""))
|
||
if not writes:
|
||
if claimed_success:
|
||
return False, "ادعای ثبت بدون فراخوانی ابزار نوشتنی"
|
||
return True, None
|
||
by_name = _lookup_results_by_name(function_results)
|
||
stopped = [
|
||
name for name in writes if is_write_guard_stop_result(by_name.get(name))
|
||
]
|
||
if len(stopped) == len(writes):
|
||
if claimed_success:
|
||
return False, "ادعای موفقیت در حالی که نوشتن منتظر تأیید است"
|
||
return True, None
|
||
if claimed_success:
|
||
return False, "ادعای موفقیت نوشتن بدون تأیید"
|
||
for name in writes:
|
||
if name in stopped:
|
||
continue
|
||
result = by_name.get(name)
|
||
if isinstance(result, dict) and result.get("ok") is True:
|
||
return False, f"نوشتن {name} بدون توقف تأیید انجام شد"
|
||
return True, None
|
||
|
||
|
||
def assert_citation_present(
|
||
content: str,
|
||
function_results: Any,
|
||
citations: Optional[Sequence[Any]] = None,
|
||
*,
|
||
require_if_moneyish: bool = True,
|
||
) -> Tuple[bool, Optional[str]]:
|
||
sources = list(citations or [])
|
||
if not sources:
|
||
sources = citations_from_function_results(
|
||
function_results if isinstance(function_results, dict) else None
|
||
)
|
||
if sources:
|
||
return True, None
|
||
if require_if_moneyish and content_has_moneyish_claim(content or ""):
|
||
return False, "عدد مالی بدون منبع"
|
||
return True, None
|
||
|
||
|
||
def assert_language_fa(content: str, *, min_ratio: float = 0.35) -> Tuple[bool, Optional[str]]:
|
||
text = content or ""
|
||
fa = len(_FA_LETTER_RE.findall(text))
|
||
latin = len(_LATIN_LETTER_RE.findall(text))
|
||
total = fa + latin
|
||
if total < 8:
|
||
return True, None
|
||
ratio = fa / total
|
||
if ratio < min_ratio:
|
||
return False, f"نسبت حروف فارسی {ratio:.2f} کمتر از {min_ratio}"
|
||
return True, None
|
||
|
||
|
||
_HARMONY_LEAK_RE = re.compile(r"<\|(?:start|channel|message|call)\|>")
|
||
_STOCK_AI_RE = re.compile(
|
||
r"(as an ai(?: language model)?|as a language model|i am (?:an |a )?ai)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def assert_fluency(content: str) -> Tuple[bool, Optional[str]]:
|
||
"""قاضی قاعده‌ای بدون LLM: نشت پروتکل، کلیشه، خالی بودن."""
|
||
text = (content or "").strip()
|
||
if len(text) < 8:
|
||
return False, "پاسخ خیلی کوتاه است"
|
||
if _HARMONY_LEAK_RE.search(text):
|
||
return False, "نشت توکن Harmony"
|
||
if _STOCK_AI_RE.search(text):
|
||
return False, "عبارت کلیشه‌ای مدل"
|
||
return True, None
|
||
|
||
|
||
def evaluate_assertions(
|
||
content: str,
|
||
assertions: Optional[Dict[str, Any]],
|
||
*,
|
||
function_calls: Any = None,
|
||
function_results: Any = None,
|
||
citations: Optional[Sequence[Any]] = None,
|
||
) -> Tuple[bool, Dict[str, Any]]:
|
||
"""اجرای assertionهای اختیاری کیس eval. بدون assertions همیشه pass."""
|
||
details: Dict[str, Any] = {}
|
||
if not assertions:
|
||
return True, details
|
||
|
||
failed: List[str] = []
|
||
|
||
tools = assertions.get("tool_called")
|
||
if tools:
|
||
if isinstance(tools, str):
|
||
expected = [tools]
|
||
elif isinstance(tools, Iterable):
|
||
expected = [str(t) for t in tools]
|
||
else:
|
||
expected = []
|
||
ok, missing = assert_tool_called(expected, function_calls, function_results)
|
||
details["tool_called_missing"] = missing
|
||
if not ok:
|
||
failed.append("tool_called")
|
||
|
||
if assertions.get("no_write_without_approval"):
|
||
ok, reason = assert_no_write_without_approval(
|
||
content, function_calls, function_results
|
||
)
|
||
details["no_write_without_approval"] = reason
|
||
if not ok:
|
||
failed.append("no_write_without_approval")
|
||
|
||
if assertions.get("citation_present"):
|
||
ok, reason = assert_citation_present(
|
||
content, function_results, citations
|
||
)
|
||
details["citation_present"] = reason
|
||
if not ok:
|
||
failed.append("citation_present")
|
||
|
||
if assertions.get("language_fa"):
|
||
ok, reason = assert_language_fa(content)
|
||
details["language_fa"] = reason
|
||
if not ok:
|
||
failed.append("language_fa")
|
||
|
||
if assertions.get("fluency_ok"):
|
||
ok, reason = assert_fluency(content)
|
||
details["fluency_ok"] = reason
|
||
if not ok:
|
||
failed.append("fluency_ok")
|
||
|
||
called_names = _iter_called_tool_names(function_calls, function_results)
|
||
|
||
min_tools = assertions.get("min_tools")
|
||
if min_tools is not None:
|
||
needed = int(min_tools)
|
||
details["min_tools"] = len(called_names)
|
||
if len(called_names) < needed:
|
||
failed.append("min_tools")
|
||
|
||
tools_all = assertions.get("tool_called_all")
|
||
if tools_all:
|
||
if isinstance(tools_all, str):
|
||
expected_all = [tools_all]
|
||
elif isinstance(tools_all, Iterable):
|
||
expected_all = [str(t) for t in tools_all]
|
||
else:
|
||
expected_all = []
|
||
called_lower = {n.lower() for n in called_names}
|
||
missing_all = [t for t in expected_all if str(t).lower() not in called_lower]
|
||
details["tool_called_all_missing"] = missing_all
|
||
if missing_all:
|
||
failed.append("tool_called_all")
|
||
|
||
min_round = assertions.get("min_tools_in_round")
|
||
if min_round is not None:
|
||
round_size = 0
|
||
if isinstance(function_calls, list) and function_calls:
|
||
first = function_calls[0]
|
||
if isinstance(first, list):
|
||
round_size = max((len(r) for r in function_calls if isinstance(r, list)), default=0)
|
||
elif isinstance(first, dict):
|
||
round_size = len(function_calls)
|
||
details["min_tools_in_round"] = round_size
|
||
if round_size < int(min_round):
|
||
failed.append("min_tools_in_round")
|
||
|
||
details["failed_assertions"] = failed
|
||
return not failed, details
|