Spaces:
Sleeping
Sleeping
File size: 7,053 Bytes
f473a91 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """
Basic rule-based formatting of API validation results without AI.
"""
import logging
from typing import Dict, List, Any, Optional
logger = logging.getLogger(__name__)
class BasicFormatter:
"""
Formats API validation results using simple rules without AI.
Provides clean, display-friendly output based on API scores and matches.
"""
def __init__(self):
"""Initialize the basic formatter."""
pass
def format_results(self, api_results: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Formats API results using rule-based logic without AI.
Args:
api_results: The results from the API validators.
Returns:
A dictionary containing formatted results.
"""
formatted_results = []
for result in api_results:
formatted = self._format_single_result(result)
formatted_results.append(formatted)
logger.info(f"Formatted {len(formatted_results)} results using basic rules")
return {"results": formatted_results, "candidates_grounding_metadata": []}
def _format_single_result(self, api_result: Dict[str, Any]) -> Dict[str, Any]:
"""
Formats a single API result.
Args:
api_result: Single API validation result.
Returns:
Formatted result dictionary.
"""
query = api_result.get("query", {})
exists = api_result.get("exists", False)
best_match = api_result.get("best_match", {})
# Score is stored in best_match as overall_score
score = best_match.get("overall_score", 0) if best_match else 0
source = api_result.get("source", "unknown")
# Check if there was an API error (critical difference from legitimate "not found")
error_type = api_result.get("error") # e.g., "rate_limit", "max_retries_exceeded", etc.
error_message = api_result.get("message", "")
# Extract data from query or best match
title = best_match.get("title") or query.get("title", "Unknown")
# Handle authors list - get first author
authors = best_match.get("authors", [])
first_author = authors[0] if authors else query.get("first_author", "Unknown")
year = best_match.get("year") or query.get("year", "Unknown")
journal = best_match.get("journal") or query.get("journal", "Unknown")
link = best_match.get("link") or best_match.get("url") or best_match.get("doi", "")
# Build explanation and identify issues (now includes error detection)
explanation, issues = self._build_explanation_and_issues(
exists, score, source, query, best_match, error_type, error_message
)
return {
"title": title,
"exists": exists,
"link": link,
"explanation": explanation,
"first_author": first_author,
"year": str(year),
"journal": journal,
"issues": issues if issues else None
}
def _build_explanation_and_issues(
self,
exists: bool,
score: float,
source: str,
query: Dict[str, Any],
best_match: Dict[str, Any],
error_type: Optional[str] = None,
error_message: str = ""
) -> tuple[str, List[str]]:
"""
Builds explanation and identifies issues based on API results.
Args:
exists: Whether reference exists in API
score: Match score
source: API source name
query: Original query data
best_match: Best match from API
error_type: Type of error if API failed (e.g., "rate_limit", "max_retries_exceeded")
error_message: Human-readable error message
Returns:
Tuple of (explanation string, list of issues)
"""
issues = []
# Handle API errors differently from legitimate "not found"
if not exists:
if error_type:
# API error occurred - don't claim "not found", explain the error
if error_type == "rate_limit":
explanation = "API rate limit exceeded - validation failed"
issues.append("API rate limit hit - try again later")
elif error_type == "max_retries_exceeded":
explanation = "API unavailable after retries - validation failed"
issues.append("API temporarily unavailable")
elif error_type == "API timeout":
explanation = "API request timed out - validation failed"
issues.append("Network timeout")
else:
# Generic error
explanation = f"Validation error: {error_message}"
issues.append(f"API error: {error_type}")
return explanation, issues
else:
# No error - legitimate "not found"
return "Not found in any API database", []
# Build explanation
if score >= 90:
explanation = f"Confirmed via {source.title()}"
elif score >= 70:
explanation = f"Likely match via {source.title()} (score: {score:.0f}%)"
issues.append("Medium confidence match")
else:
explanation = f"Weak match via {source.title()} (score: {score:.0f}%)"
issues.append("Low confidence match")
# Check for discrepancies
query_year = str(query.get("year", ""))
match_year = str(best_match.get("year", ""))
if query_year and match_year and query_year != match_year:
issues.append(f"Year mismatch: {query_year} vs {match_year}")
query_title = query.get("title", "").lower()
match_title = best_match.get("title", "").lower()
if query_title and match_title:
# Simple title similarity check
if query_title not in match_title and match_title not in query_title:
# Check if at least some words match
query_words = set(query_title.split())
match_words = set(match_title.split())
common_words = query_words & match_words
if len(common_words) < min(3, len(query_words) // 2):
issues.append("Title variation detected")
query_author = query.get("first_author", "").lower()
# Get first author from authors list
match_authors = best_match.get("authors", [])
match_author = match_authors[0].lower() if match_authors else ""
if query_author and match_author and query_author != match_author:
# Check if last names match (simple check)
query_last = query_author.split()[-1] if query_author else ""
match_last = match_author.split()[-1] if match_author else ""
if query_last != match_last:
issues.append(f"Author mismatch: {query_author} vs {match_author}")
return explanation, issues
|