graph/langgraph_pipeline.py

from typing import Any, TypedDict

from langgraph.graph import END, StateGraph

from agents.claim_extractor import ClaimExtractor
from agents.evidence_retriever import EvidenceRetriever
from agents.explainer import Explainer
from graph.jury_subgraph import run_jury_subgraph


class ClaimResult(TypedDict):
    claim: str
    evidence: list[dict[str, Any]]
    verdict: dict[str, Any]
    explanation: dict[str, Any]


class PipelineState(TypedDict):
    input_text: str
    claims: list[str]
    claim_results: list[ClaimResult]
    explanation: dict[str, Any] | None


claim_extractor = ClaimExtractor()
evidence_retriever = EvidenceRetriever()
explainer = Explainer()


VERDICT_LABELS = {
    "SUPPORTED": "مدعوم بالأدلة",
    "REFUTED": "مدحوض",
    "PARTIALLY_TRUE": "صحيح جزئيا",
    "UNVERIFIABLE": "لا يمكن التحقق منه",
}


def _to_plain_dict(value: Any) -> dict[str, Any]:
    if hasattr(value, "model_dump"):
        return value.model_dump()
    if isinstance(value, dict):
        return value
    return {}


def _overall_verdict(claim_results: list[ClaimResult]) -> str:
    if not claim_results:
        return "UNVERIFIABLE"

    verdicts = {
        result.get("verdict", {}).get("verdict", "UNVERIFIABLE")
        for result in claim_results
    }
    verdicts.discard("")

    if len(verdicts) == 1:
        return next(iter(verdicts))

    return "PARTIALLY_TRUE"


def _average_confidence(claim_results: list[ClaimResult]) -> float:
    if not claim_results:
        return 0.0

    confidences = []
    for result in claim_results:
        try:
            confidences.append(
                float(result.get("verdict", {}).get("confidence", 0.0)))
        except (TypeError, ValueError):
            confidences.append(0.0)

    return round(sum(confidences) / len(confidences), 3) if confidences else 0.0


def _dedupe_citations(claim_results: list[ClaimResult]) -> list[dict[str, str]]:
    citations: list[dict[str, str]] = []
    seen: set[tuple[str, str]] = set()

    for result in claim_results:
        for citation in result.get("explanation", {}).get("citations", []):
            if not isinstance(citation, dict):
                continue
            title = str(citation.get("title", "")).strip()
            url = str(citation.get("url", "")).strip()
            key = (title, url)
            if key in seen or (not title and not url):
                continue
            seen.add(key)
            citations.append({"title": title, "url": url})

    return citations


def _build_article_explanation(
    claims: list[str],
    claim_results: list[ClaimResult],
) -> dict[str, Any]:
    if not claims:
        return {
            "arabic_explanation": "لم يتم استخراج أي ادعاءات قابلة للتحقق من النص المرسل.",
            "verdict": "UNVERIFIABLE",
            "confidence": 0.0,
            "citations": [],
        }

    if len(claim_results) == 1:
        explanation = dict(claim_results[0]["explanation"])
        explanation.setdefault("citations", [])
        return explanation

    overall_verdict = _overall_verdict(claim_results)
    overall_confidence = _average_confidence(claim_results)
    overview = (
        f"تم استخراج {len(claim_results)} ادعاءات قابلة للتحقق من النص. "
        f"الحكم الإجمالي على مستوى المقال هو: {VERDICT_LABELS.get(overall_verdict, overall_verdict)}."
    )

    details = []
    for index, result in enumerate(claim_results, start=1):
        verdict = result.get("verdict", {}).get("verdict", "UNVERIFIABLE")
        explanation = str(result.get("explanation", {}).get(
            "arabic_explanation", "")).strip()
        details.append(
            f"{index}. الادعاء: \"{result['claim']}\"\n"
            f"الحكم: {VERDICT_LABELS.get(verdict, verdict)}.\n"
            f"{explanation}"
        )

    return {
        "arabic_explanation": overview + "\n\n" + "\n\n".join(details),
        "verdict": overall_verdict,
        "confidence": overall_confidence,
        "citations": _dedupe_citations(claim_results),
    }


async def run_claim_extractor(state: PipelineState) -> PipelineState:
    claims = await claim_extractor.run(state["input_text"])
    return {**state, "claims": claims}


async def run_evidence_retriever(state: PipelineState) -> PipelineState:
    claim_results: list[ClaimResult] = []

    for claim in state["claims"]:
        evidence = await evidence_retriever.run(claim)
        claim_results.append(
            {
                "claim": claim,
                "evidence": evidence,
                "verdict": {},
                "explanation": {},
            }
        )

    return {**state, "claim_results": claim_results}


async def run_jury(state: PipelineState) -> PipelineState:
    updated_results: list[ClaimResult] = []

    for result in state["claim_results"]:
        verdict = await run_jury_subgraph(result["claim"], result["evidence"])
        updated_results.append({**result, "verdict": verdict})

    return {**state, "claim_results": updated_results}


async def run_explainer(state: PipelineState) -> PipelineState:
    updated_results: list[ClaimResult] = []

    for result in state["claim_results"]:
        verdict_block = result.get("verdict", {})

        explainer_verdict = {
            "verdict":            verdict_block.get("verdict", "UNVERIFIABLE"),
            "confidence":         verdict_block.get("confidence", 0.0),
            "reasoning":          verdict_block.get("reasoning", ""),
            "jury_outputs":       verdict_block.get("jury_outputs", []),
            "debate_log":         verdict_block.get("debate_log", []),
            "needs_human_review": verdict_block.get("needs_human_review", False),
        }

        explanation = await explainer.run(
            result["claim"],
            result["evidence"],
            explainer_verdict,
        )
        updated_results.append(
            {
                **result,
                "explanation": _to_plain_dict(explanation),
            }
        )

    article_explanation = _build_article_explanation(
        state["claims"], updated_results)
    return {
        **state,
        "claim_results": updated_results,
        "explanation": article_explanation,
    }


def build_pipeline() -> StateGraph:
    graph = StateGraph(PipelineState)
    graph.add_node("claim_extractor",    run_claim_extractor)
    graph.add_node("evidence_retriever", run_evidence_retriever)
    graph.add_node("jury",               run_jury)
    graph.add_node("explainer",          run_explainer)

    graph.set_entry_point("claim_extractor")
    graph.add_edge("claim_extractor",    "evidence_retriever")
    graph.add_edge("evidence_retriever", "jury")
    graph.add_edge("jury",               "explainer")
    graph.add_edge("explainer",          END)

    return graph.compile()


pipeline = build_pipeline()


def _serialize_claim_results(claim_results: list[ClaimResult]) -> list[dict[str, Any]]:
    out = []
    for result in claim_results:
        exp = result.get("explanation", {})
        verdict_block = result.get("verdict", {})
        out.append({
            "claim":       result["claim"],
            "verdict":     verdict_block.get("verdict", "UNVERIFIABLE"),
            "confidence":  round(float(verdict_block.get("confidence", 0.0)), 3),
            "explanation": exp.get("arabic_explanation", ""),
            "citations": [
                {"title": c.get("title", ""), "url": c.get("url", "")}
                for c in exp.get("citations", [])
                if isinstance(c, dict)
            ],
            "needs_human_review": verdict_block.get("needs_human_review", False),
            "jury_outputs":       verdict_block.get("jury_outputs", []),
            "debate_log":         verdict_block.get("debate_log", []),
        })
    return out


async def run_pipeline(text: str) -> dict[str, Any]:
    initial_state: PipelineState = {
        "input_text": text,
        "claims": [],
        "claim_results": [],
        "explanation": None,
    }
    final_state = await pipeline.ainvoke(initial_state)

    article = final_state.get("explanation") or {
        "arabic_explanation": "لم يتمكن النظام من توليد نتيجة نهائية.",
        "verdict": "UNVERIFIABLE",
        "confidence": 0.0,
        "citations": [],
    }

    claim_results = final_state.get("claim_results", [])
    needs_human_review = any(
        r.get("verdict", {}).get("needs_human_review", False)
        for r in claim_results
    )

    return {
        "verdict":            article.get("verdict", "UNVERIFIABLE"),
        "confidence":         round(float(article.get("confidence", 0.0)), 3),
        "arabic_explanation": article.get("arabic_explanation", ""),
        "citations":          article.get("citations", []),
        "needs_human_review": needs_human_review,
        "claims":             _serialize_claim_results(claim_results),
    }