diff --git a/citations.py b/citations.py index f56ec44..3627244 100644 --- a/citations.py +++ b/citations.py @@ -153,6 +153,23 @@ def score(self) -> float | None: return round(len(self.citations) / self.attempted, 4) +class ClaimVerification(BaseModel): + """Verification result for a single extracted claim.""" + + claim: str + matched_citations: list[Citation] = [] + supported: bool = False + strength: Literal["strong", "moderate", "weak", "unsupported"] = "unsupported" + reasons: list[str] = [] + + +class EvidenceVerificationReport(BaseModel): + """Claims checked and the citations that supported them.""" + + claims: list[ClaimVerification] = [] + overall_score: float | None = None + + def _coerce_int(value: Any) -> int | None: """Accept 2 and "2" alike; reject everything else without raising.""" if isinstance(value, bool): @@ -369,6 +386,73 @@ def extract_citations(text: str | None) -> tuple[str, CitationExtraction]: return text, CitationExtraction() +_CLAIM_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+") + + +def _extract_claims(prose: str) -> list[str]: + """Split prose into candidate claims, keeping sentences that cite anything.""" + claims = [] + for sentence in _CLAIM_SENTENCE_RE.split(prose): + sentence = sentence.strip() + if not sentence: + continue + if re.search(r"(Qur'?an|hadith|surah|Surah|reported|narrated|said|wrote|states?)\b", sentence): + claims.append(sentence) + return claims + + +def _match_citations(claim: str, citations: list[Citation]) -> list[Citation]: + """Link a claim to the citations whose references appear in its text.""" + matched = [] + for citation in citations: + ref = citation.reference if isinstance(citation, QuranCitation) else None + if ref and ref in claim: + matched.append(citation) + continue + if isinstance(citation, HadithCitation) and citation.collection and citation.collection in claim: + matched.append(citation) + continue + if isinstance(citation, ScholarlyReference) and citation.work and citation.work in claim: + matched.append(citation) + continue + return matched + + +def verify_evidence(text: str | None) -> EvidenceVerificationReport: + """Verify claims in *text* against the citations it carries.""" + prose, extraction = extract_citations(text) + report = EvidenceVerificationReport() + for claim in _extract_claims(prose): + matched = _match_citations(claim, extraction.citations) + if not matched: + report.claims.append( + ClaimVerification( + claim=claim, + supported=False, + reasons=["no structured citation found for this claim"], + ) + ) + continue + weak = any( + isinstance(c, HadithCitation) and c.grading and "sahih" not in c.grading + for c in matched + ) + report.claims.append( + ClaimVerification( + claim=claim, + matched_citations=matched, + supported=True, + strength="moderate" if weak else "strong", + reasons=[f"matched {len(matched)} structured citation(s)"], + ) + ) + if report.claims: + report.overall_score = round( + sum(1 for c in report.claims if c.supported) / len(report.claims), 4 + ) + return report + + class CitationStreamFilter: """Keep the citation block out of SSE deltas. diff --git a/evidence_agent.py b/evidence_agent.py new file mode 100644 index 0000000..6bbec86 --- /dev/null +++ b/evidence_agent.py @@ -0,0 +1,4 @@ +from verifier import extract_ana_verify_all + +def run_evidence_agent(text: str) -> dict: + return {'results': extract_and_verify_all(text)} diff --git a/hadith.py b/hadith.py index f2068e6..7d14c39 100644 --- a/hadith.py +++ b/hadith.py @@ -503,6 +503,87 @@ def build_caution_note(text: str, references: list[HadithReference]) -> str | No return "\n".join(lines) +# --------------------------------------------------------------------------- +# Evidence verification agent +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class EvidenceVerdict: + claim: str + references: list[HadithReference] + supported: bool + strength: Strength + notes: list[str] = field(default_factory=list) + + +class EvidenceVerificationAgent: + """Validates scholarly claims by tracing cited hadith evidence to its + bundled primary-source grading record. + """ + + def __init__(self, source: GradingSource | None = None): + self._source = source or get_default_source() + + def extract_claims(self, text: str) -> list[str]: + """Extract candidate scholarly claims from generated response text.""" + claims = [] + for sentence in re.split(r"(?<=[.!?])\s+", text.strip()): + sentence = sentence.strip() + if sentence and re.search( + r"\b(narrated|reported|said|states|according to)\b", + sentence, + re.IGNORECASE, + ): + claims.append(sentence) + return claims + + def verify_claims(self, text: str) -> list[EvidenceVerdict]: + """Verify each extracted claim, falling back to the full response.""" + claims = self.extract_claims(text) or [text.strip()] + return [self._verify(claim, text) for claim in claims] + + def _verify(self, claim: str, full_text: str) -> EvidenceVerdict: + refs = annotate(claim, self._source) or annotate(full_text, self._source) + notes: list[str] = [] + if not refs: + return EvidenceVerdict(claim, [], False, Strength.UNKNOWN, ["no primary-source evidence cited"]) + if not any(ref.raw.lower() in claim.lower() for ref in refs): + notes.append("cited evidence is not explicitly connected to this claim") + verified = [ref for ref in refs if ref.verified] + if not verified: + notes.append("cited references could not be verified against primary sources") + return EvidenceVerdict(claim, refs, False, Strength.UNKNOWN, notes) + strengths = [Strength(ref.grade) for ref in verified] + strength = aggregate_strength(strengths) + if strength in (Strength.DAIF, Strength.MAWDU): + notes.append("cited evidence is weak or fabricated; do not rely on it") + if any(ref.flagged for ref in verified): + notes.append("cited evidence carries an unstated authenticity caveat") + supported = ( + bool(refs) + and strength not in (Strength.DAIF, Strength.MAWDU, Strength.UNKNOWN) + and not any(ref.flagged for ref in verified) + and not any("not explicitly connected" in note for note in notes) + ) + return EvidenceVerdict(claim, refs, supported, strength, notes) + + def trace_provenance(self, text: str) -> list[dict[str, str]]: + """Return a verification audit trail for every citation in *text*.""" + return [ + { + "collection": ref.collection or "", + "hadith_number": str(ref.hadith_number or ""), + "grade": ref.grade, + "grader": ref.grader or "", + "chain_type": ref.chain_type or "", + "verified": str(ref.verified).lower(), + "flagged": str(ref.flagged).lower(), + } + for ref in annotate(text, self._source) + ] + + HADITH_ADAB_CONTEXT = """ HADITH CITATION RULES: diff --git a/main.py b/main.py index e49609f..2fdacfe 100644 --- a/main.py +++ b/main.py @@ -148,6 +148,8 @@ genai.configure(api_key=GEMINI_API_KEY) +EVIDENCE_VERIFICATION_ENABLED = os.getenv("EVIDENCE_VERIFICATION_ENABLED", "true").lower() not in {"0", "false", "off"} + app = FastAPI(title="DeenBridge AI API") # --- Service API-key authentication --- @@ -344,6 +346,26 @@ class CitationVerificationResult(BaseModel): reason: str | None = None +class EvidenceClaim(BaseModel): + claim: str + evidence: list[str] = Field(default_factory=list) + source_type: str | None = None + verification_status: str = "unverified" + confidence: float | None = None + reason: str | None = None + alternative_evidence: list[str] = Field(default_factory=list) + + +class EvidenceVerificationResult(BaseModel): + claims: list[EvidenceClaim] = Field(default_factory=list) + overall_score: float = 0.0 + audit_trail: list[dict[str, Any]] = Field(default_factory=list) + + +class EvidenceVerifyRequest(BaseModel): + text: str = Field(..., min_length=1, max_length=30000) + + class ChatRequest(BaseModel): prompt: str = Field(..., max_length=CHAT_PROMPT_MAX_LENGTH) chat_id: str | None = None @@ -387,6 +409,8 @@ class ChatResponse(BaseModel): # Structured references parsed out of the answer (#15). Empty when the # answer cited nothing, or when nothing it cited could be validated. citations: list[Citation] = [] + # Evidence verification agent audit (#evidence-verification). + evidence_verification: EvidenceVerificationResult | None = None class FeedbackRequest(BaseModel): @@ -781,6 +805,87 @@ async def run_strict_corrective_loop( return safe_text or original_text +def _citation_to_dict(item: Any) -> dict[str, Any]: + if hasattr(item, "model_dump"): + return item.model_dump() + if hasattr(item, "dict"): + return item.dict() + return {} + + +def _evidence_label(d: dict[str, Any]) -> str: + if d.get("source") == "quran" and d.get("surah") is not None: + return f"Quran {d['surah']}:{d.get('ayah', '')}" + if d.get("collection"): + return f"{d['collection']} {d.get('number', '')}".strip() + return str(d.get("source") or "cited source") + + +def verify_evidence_claims( + response_text: str, + citation_extraction: CitationExtraction, + hadith_refs: list[HadithReference], +) -> EvidenceVerificationResult: + """Deterministic claim-evidence audit using already-verified citations.""" + result = EvidenceVerificationResult() + claims = [ + s.strip() + for s in response_text.replace("\n", " ").split(".") + if len(s.strip()) >= 25 + ] + evidence: list[tuple[str, str, bool]] = [] + + for c in citation_extraction.citations: + d = _citation_to_dict(c) + label = _evidence_label(d) + status = str(d.get("status") or d.get("verification") or "unverified") + evidence.append((label, status, status in {"verified", "authentic", "sahih", "hasan", "mutawatir"})) + + for h in hadith_refs: + d = _citation_to_dict(h) + label = _evidence_label(d) + grade = str(d.get("grade") or d.get("authenticity") or d.get("status") or "unverified").lower() + evidence.append((label, grade, grade not in {"daif", "weak", "munkar", "fabricated", "mawdu", "unverified"})) + + for claim in claims: + matched = [e for e in evidence if e[0].lower() in claim.lower()] + if not matched and len(claims) == 1 and len(evidence) == 1: + matched = evidence[:] + + if not matched: + result.claims.append( + EvidenceClaim( + claim=claim, + verification_status="unsupported", + confidence=0.1, + reason="No verifiable citation is present for this claim.", + alternative_evidence=["Provide a precise Quranic/Hadith citation for this claim."], + ) + ) + result.audit_trail.append({"claim": claim, "status": "unsupported"}) + continue + + supported = any(e[2] for e in matched) + status = "supported" if supported else "weak" + result.claims.append( + EvidenceClaim( + claim=claim, + evidence=[e[0] for e in matched], + verification_status=status, + confidence=0.9 if supported else 0.4, + reason="Evidence linked and verified." if supported else "Evidence cited but could not be verified.", + ) + ) + result.audit_trail.append({"claim": claim, "evidence": [e[0] for e in matched], "status": status}) + + result.overall_score = ( + sum(1 for c in result.claims if c.verification_status == "supported") / len(result.claims) + if result.claims + else 0.0 + ) + return result + + @app.post("/chat", response_model=ChatResponse) @limiter.limit(f"{CHAT_RATE_LIMIT_MAX}/{CHAT_RATE_LIMIT_WINDOW_SECONDS} seconds") async def chat(body: ChatRequest, request: Request, fastapi_response: Response) -> ChatResponse: @@ -1080,6 +1185,13 @@ async def generate(safety_prompt: str) -> str: # Baked into response_text *before* the cache write so a cached hit # replays the same caution the user originally saw. hadith_refs = annotate_hadith(response_text) + # --- Evidence verification agent --- + evidence_verification = None + if EVIDENCE_VERIFICATION_ENABLED: + try: + evidence_verification = verify_evidence_claims(response_text, citation_extraction, hadith_refs) + except Exception as exc: # noqa: BLE001 - verification is best-effort + logger.warning("Evidence verification failed; continuing without it: %s", exc) caution = build_caution_note(response_text, hadith_refs) if caution: response_text = f"{response_text.rstrip()}\n\n{caution}" @@ -1202,6 +1314,7 @@ async def generate(safety_prompt: str) -> str: purchases=purchase_info, language=effective_language, citations=citation_extraction.citations, + evidence_verification=evidence_verification, ) _finalize() _succeeded = True @@ -1526,6 +1639,12 @@ async def event_generator() -> AsyncGenerator[str, None]: # --- Hadith authenticity grading --- hadith_refs = annotate_hadith(combined_text) + evidence_verification = None + if EVIDENCE_VERIFICATION_ENABLED: + try: + evidence_verification = verify_evidence_claims(combined_text, citation_extraction, hadith_refs) + except Exception as exc: # noqa: BLE001 - verification is best-effort + logger.warning("Streaming evidence verification failed: %s", exc) caution = build_caution_note(combined_text, hadith_refs) if caution: combined_text = f"{combined_text.rstrip()}\n\n{caution}" @@ -1604,6 +1723,7 @@ async def event_generator() -> AsyncGenerator[str, None]: "tafsir": tafsir_info.model_dump() if tafsir_info else None, "zakat": zakat_info.model_dump() if zakat_info else None, "citations": [c.model_dump() for c in citation_extraction.citations], + "evidence_verification": evidence_verification.model_dump() if evidence_verification else None, }, ensure_ascii=False, ) @@ -1787,6 +1907,26 @@ async def delete_chat(chat_id: str, user_id: str | None = None) -> dict[str, str ) from e +@app.post("/evidence/verify", response_model=EvidenceVerificationResult) +async def verify_evidence_endpoint(body: EvidenceVerifyRequest) -> EvidenceVerificationResult: + """Run the evidence verification agent on a generated answer. + + Extracts any citation block, parses Quran/Hadith references, links them to + claims, and returns a structured audit of support strength. + """ + try: + cleaned_text, citation_extraction = extract_citations(body.text) + hadith_refs = annotate_hadith(cleaned_text) + return verify_evidence_claims(cleaned_text, citation_extraction, hadith_refs) + except Exception as exc: + logger.error("Evidence verification endpoint failed: %s", exc) + raise APIException( + status_code=500, + detail="Evidence verification failed.", + hint="The evidence verification agent could not process the supplied text. Verify the text is well-formed and retry.", + ) from exc + + # --------------------------------------------------------------------------- # Feedback: capture and admin views # --------------------------------------------------------------------------- diff --git a/reasoning_chains.py b/reasoning_chains.py index 53fa984..1f2c229 100644 --- a/reasoning_chains.py +++ b/reasoning_chains.py @@ -212,6 +212,36 @@ class EvidenceRef(BaseModel): reference: str = Field(..., description="Category-level pointer, e.g. 'fiqh:worship' or 'Hadith (sunnah.com)'") note: str | None = Field(None, description="Optional clarifying note") +class EvidenceStrength(str, Enum): + """Strength classification for evidence supporting a claim.""" + + STRONG = "strong" + MODERATE = "moderate" + WEAK = "weak" + UNVERIFIED = "unverified" + + +class Claim(BaseModel): + """A scholarly claim extracted from a reasoning step.""" + + text: str + step_id: str | None = None + evidence: list[EvidenceRef] = Field(default_factory=list) + suggested_evidence: list[EvidenceRef] = Field(default_factory=list) + + +class EvidenceVerificationReport(BaseModel): + """Result of tracing a claim's evidence back to primary sources.""" + + claims: list[Claim] = Field(default_factory=list) + verified_count: int = 0 + unsupported_count: int = 0 + weak_evidence_count: int = 0 + overall_strength: EvidenceStrength = EvidenceStrength.UNVERIFIED + audit_trail: list[str] = Field(default_factory=list) + attribution_issues: list[str] = Field(default_factory=list) + logical_issues: list[str] = Field(default_factory=list) + class ReasoningStep(BaseModel): """One link in the chain, individually addressable by ``id``.""" @@ -270,6 +300,7 @@ class ReasoningChain(BaseModel): branches: list[ReasoningBranch] = Field(default_factory=list) conclusion: str validation: ValidationReport + evidence_verification: EvidenceVerificationReport | None = Field(None, description="Optional evidence verification report") class ReasoningTemplate(BaseModel): @@ -601,6 +632,148 @@ def validate_chain(steps: list[ReasoningStep], threshold: float = WEAK_CONFIDENC ) +# --------------------------------------------------------------------------- +# Evidence verification +# --------------------------------------------------------------------------- + + +def verify_citation(ref: EvidenceRef) -> tuple[bool, str]: + """Trace one evidence reference to the primary-source category it points at.""" + lowered = ref.reference.lower() + if ref.source_type == SourceType.QURAN: + if "quran.com" in lowered: + return True, "Qur'anic source is available at quran.com" + return False, "Qur'an citation is not in the verified category-level format" + if ref.source_type == SourceType.HADITH: + if "sunnah.com" in lowered or "collection" in lowered: + return True, "Hadith source is available at sunnah.com" + return False, "Hadith citation is not in the verified category-level format" + if ref.source_type == SourceType.TAFSIR: + if "tafsir" in lowered: + return True, "Tafsir source is listed as a secondary scholarly work" + return False, "Tafsir citation could not be verified" + if ref.source_type == SourceType.FIQH: + if "fiqh:" in lowered: + return True, "Fiqh reference points to a recognised fiqh category" + return False, "Fiqh citation could not be verified" + return False, "General reference cannot be traced to a primary source" + + +def evidence_relevance(claim_text: str, ref: EvidenceRef) -> float: + """Deterministic relevance score between a claim and an evidence reference.""" + claim_type = classify_facet(claim_text) + if ref.source_type == claim_type: + return 0.9 + if claim_type == SourceType.FIQH and ref.source_type == SourceType.QURAN: + return 0.6 + if claim_type == SourceType.TAFSIR and ref.source_type == SourceType.QURAN: + return 0.7 + if claim_type == SourceType.QURAN and ref.source_type == SourceType.TAFSIR: + return 0.7 + if claim_type == SourceType.HADITH and ref.source_type == SourceType.QURAN: + return 0.4 + overlap = len(_tokens(claim_text) & _tokens(ref.reference)) + if overlap: + return round(min(0.8, 0.3 + (0.1 * overlap)), 4) + return 0.2 + + +def assess_evidence_strength(scores: list[float]) -> EvidenceStrength: + """Classify evidence strength from claim-evidence relevance scores.""" + if not scores: + return EvidenceStrength.UNVERIFIED + best = max(scores) + if best >= 0.8: + return EvidenceStrength.STRONG + if best >= 0.5: + return EvidenceStrength.MODERATE + return EvidenceStrength.WEAK + + +def _attribution_issues(claim: Claim) -> list[str]: + """Heuristic check that named scholarly attributions carry a fiqh/tafsir reference.""" + lowered = claim.text.lower() + named_scholars = ("imam", "ibn", "sheikh", "shaykh", "abu hanifa", "ahmad ibn hanbal", "ibn taymiyyah", "ibn qayyim") + if any(name in lowered for name in named_scholars) and not any( + ref.source_type in (SourceType.FIQH, SourceType.TAFSIR) for ref in claim.evidence + ): + return [f"{claim.step_id}: scholarly attribution is not supported by a fiqh or tafsir reference"] + return [] + + +def extract_claims(chain: ReasoningChain) -> list[Claim]: + """Extract one claim per reasoning step with its linked evidence.""" + claims: list[Claim] = [] + for step in chain.steps: + claim = Claim( + text=step.intermediate_conclusion, + step_id=step.id, + evidence=step.evidence, + ) + if not step.evidence: + claim.suggested_evidence = _evidence_for(step.source_type, step.facet or step.intermediate_conclusion) + claims.append(claim) + for branch in chain.branches: + for step in branch.steps: + claim = Claim( + text=step.intermediate_conclusion, + step_id=step.id, + evidence=step.evidence, + ) + if not step.evidence: + claim.suggested_evidence = _evidence_for(step.source_type, step.facet or step.intermediate_conclusion) + claims.append(claim) + return claims + + +def verify_chain_evidence(chain: ReasoningChain) -> EvidenceVerificationReport: + """Return an audit trail of citation checks for every step in the chain.""" + report = EvidenceVerificationReport() + claims = extract_claims(chain) + report.claims = claims + strengths: list[EvidenceStrength] = [] + for claim in claims: + if not claim.evidence: + report.unsupported_count += 1 + report.audit_trail.append(f"{claim.step_id}: no evidence source attached") + continue + verified_any = False + scores: list[float] = [] + for ref in claim.evidence: + verified, provenance = verify_citation(ref) + relevance = evidence_relevance(claim.text, ref) + scores.append(relevance) + if verified: + verified_any = True + report.audit_trail.append( + f"{claim.step_id}: verified {ref.source_type.value} '{ref.reference}' -> {provenance}" + ) + else: + report.audit_trail.append( + f"{claim.step_id}: unverified {ref.source_type.value} '{ref.reference}' -> {provenance}" + ) + if verified_any: + report.verified_count += 1 + strength = assess_evidence_strength(scores) + if strength == EvidenceStrength.WEAK: + report.weak_evidence_count += 1 + if strength == EvidenceStrength.WEAK and not claim.suggested_evidence: + claim.suggested_evidence = _evidence_for(classify_facet(claim.text), claim.text) + strengths.append(strength) + report.attribution_issues.extend(_attribution_issues(claim)) + for issue in find_contradictions(chain.steps): + report.logical_issues.append(issue.detail) + if strengths: + rank = { + EvidenceStrength.STRONG: 3, + EvidenceStrength.MODERATE: 2, + EvidenceStrength.WEAK: 1, + EvidenceStrength.UNVERIFIED: 0, + } + report.overall_strength = max(strengths, key=lambda strength: rank[strength]) + return report + + # --------------------------------------------------------------------------- # Assembly and rendering # --------------------------------------------------------------------------- @@ -620,7 +793,7 @@ def build_chain(question: str, madhhab: str | None = None) -> ReasoningChain: steps = decompose(question, branch=madhhab) branches = build_branches(question) if _needs_branching(question, madhhab) else [] validation = validate_chain(steps) - return ReasoningChain( + chain = ReasoningChain( id=f"chain-{abs(hash(question)) % 1_000_000:06d}", question=question, madhhab=madhhab, @@ -629,6 +802,8 @@ def build_chain(question: str, madhhab: str | None = None) -> ReasoningChain: conclusion=_summarize_conclusion(steps, branches), validation=validation, ) + chain.evidence_verification = verify_chain_evidence(chain) + return chain def render_markdown(chain: ReasoningChain) -> str: @@ -653,6 +828,12 @@ def render_markdown(chain: ReasoningChain) -> str: if chain.validation.weak_points: weakest = chain.validation.weak_points[0] lines.append(f"**Scrutinise first:** {weakest.step_id} — {weakest.reason}") + if chain.evidence_verification is not None: + verifier = chain.evidence_verification + lines.append( + f"**Evidence verification:** {verifier.verified_count} verified, " + f"{verifier.unsupported_count} unsupported, {verifier.weak_evidence_count} weak" + ) return "\n".join(lines) @@ -682,6 +863,12 @@ async def create_chain(request: ChainRequest) -> ChainResponse: return ChainResponse(chain=chain, outline=render_markdown(chain)) +@router.post("/verify", response_model=EvidenceVerificationReport) +async def verify_question(request: ChainRequest) -> EvidenceVerificationReport: + """Verify the evidence behind a generated reasoning chain.""" + return verify_chain_evidence(build_chain(request.question, madhhab=request.madhhab)) + + @router.get("/templates", response_model=TemplatesResponse) async def get_templates() -> TemplatesResponse: """Reasoning-step templates for common Islamic question patterns.""" diff --git a/verifier.py b/verifier.py index 965d129..97001d6 100644 --- a/verifier.py +++ b/verifier.py @@ -120,14 +120,65 @@ def verify_hadith_citation(collection: str, number: str | None = None, quote: st "status": VerificationStatus.UNVERIFIED, "reason": "Hadith corpus not available for verification.", } - # Future expansion for #24 when Hadith corpus lands - return { - "source": "hadith", - "collection": collection, - "number": number, - "status": VerificationStatus.UNVERIFIED, - "reason": "Hadith verification not implemented.", - } + # Attempt to retrieve hadith text from corpus + get_hadith = getattr(corpus, "get_hadith", None) + if not callable(get_hadith): + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.UNVERIFIED, + "reason": "Hadith corpus getter not available.", + } + + hadith_data = get_hadith(collection, number) + if hadith_data is None: + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.MISMATCH, + "reason": f"Hadith {collection} #{number} not found in corpus.", + } + + if not quote or not quote.strip(): + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.NOT_QUOTED, + "reason": "Reference exists; no quote provided for verification.", + } + + corpus_text = hadith_data.get("english", "") or hadith_data.get("text", "") + if not corpus_text: + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.UNVERIFIED, + "reason": "Hadith text not available in corpus.", + } + + similarity = calculate_similarity(quote, corpus_text) + if similarity >= 0.70: + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.VERIFIED, + "similarity": round(similarity, 2), + } + else: + return { + "source": "hadith", + "collection": collection, + "number": number, + "status": VerificationStatus.MISMATCH, + "similarity": round(similarity, 2), + "correct_text": corpus_text, + "reason": f"Quote does not match {collection} #{number} text in corpus.", + } def extract_and_verify_all(text: str) -> list[dict[str, Any]]: @@ -151,3 +202,63 @@ def extract_and_verify_all(text: str) -> list[dict[str, Any]]: results.append(res) return results + + +def verify_claim(claim: str, evidence: str) -> dict[str, Any]: + """Verify a scholarly claim against provided evidence text. + + Args: + claim: The scholarly claim being made. + evidence: Text containing citations (Quran, Hadith) supporting the claim. + + Returns: + A verification report including status, support score, and audit trail. + """ + # Extract and verify all citations from the evidence + citation_results = extract_and_verify_all(evidence) + + # If no citations are found, the claim is unsupported + if not citation_results: + return { + "claim": claim, + "status": VerificationStatus.UNVERIFIED, + "reason": "No primary source citations found in evidence.", + "evidence": [], + "support_score": 0.0, + "audit_trail": ["No citations extracted from evidence."] + } + + # Determine overall status based on citation verification results + verified = [r for r in citation_results if r["status"] == VerificationStatus.VERIFIED] + mismatched = [r for r in citation_results if r["status"] == VerificationStatus.MISMATCH] + unverified = [r for r in citation_results if r["status"] in (VerificationStatus.UNVERIFIED, VerificationStatus.NOT_QUOTED)] + + if verified and not mismatched and not unverified: + overall_status = VerificationStatus.VERIFIED + reason = "All cited evidence verified successfully." + elif mismatched: + overall_status = VerificationStatus.MISMATCH + reason = f"{len(mismatched)} citation(s) failed verification." + else: + overall_status = VerificationStatus.UNVERIFIED + reason = "No citation could be fully verified." + + support_score = len(verified) / len(citation_results) + + # Build audit trail + audit_trail = [] + for r in citation_results: + source = r.get("source", "unknown") + ref = r.get("surah", r.get("collection", "")) + detail = r.get("ayah", r.get("number", "")) + audit_trail.append(f"{source} {ref} {detail}: {r['status']} - {r.get('reason', '')}") + + return { + "claim": claim, + "status": overall_status, + "reason": reason, + "evidence": citation_results, + "support_score": round(support_score, 2), + "audit_trail": audit_trail + } +