Blame for app.py

from __future__ import annotations

import os
from typing import Any, Literal

import gradio as gr
import httpx

# ── Configuration ─────────────────────────────────────────────────────────────

API_BASE_URL = os.environ.get("IDREF_API_URL", "https://idref-resolver.smartbiblia.fr/").rstrip("/")

# The credential is read from the environment only: never a tool argument, never
# echoed into a payload.
API_KEY = os.environ.get("IDREF_API_KEY", "")

# A Space has no command line: connector policy is constant here. An alignment
# fans out to as many as 41 upstream ABES requests, so the budget is generous.
REQUEST_TIMEOUT = 180.0

# Clamped harder than the canonical server (which allows 100 / 20): this endpoint
# is public and every call spends the operator's API budget.
MAX_CANDIDATES = 30
MAX_RETURNED = 10

# Mirrors the API's `EmbeddingModel` Literal. A closed enum there is a closed
# enum here, so a wrong value fails in the client instead of costing a round trip.
EMBEDDING_MODELS = (
    "lexical", "lexical-idf", "albert-bge-m3", "granite", "qwen", "minilm",
)
EmbeddingModel = Literal[
    "lexical", "lexical-idf", "albert-bge-m3", "granite", "qwen", "minilm"
]

# One module-level pooled client for the process.
HTTP = httpx.Client(timeout=REQUEST_TIMEOUT, follow_redirects=True)


def _post(path: str, payload: dict) -> tuple[dict | None, str | None]:
    """POST returning (data, error). Never raises — the demo answers with data."""
    headers = {"Content-Type": "application/json"}
    if API_KEY:
        headers["X-API-Key"] = API_KEY
    try:
        resp = HTTP.post(f"{API_BASE_URL}{path}", json=payload, headers=headers)
    except httpx.TimeoutException:
        return None, f"the API did not answer within {REQUEST_TIMEOUT:g}s"
    except httpx.HTTPError as exc:
        return None, f"cannot reach the API at {API_BASE_URL}: {exc}"
    if resp.status_code >= 400:
        # Surface the API's own detail; never echo the key back.
        return None, f"API {resp.status_code}: {resp.text[:300]}"
    try:
        return resp.json(), None
    except ValueError:
        return None, f"API returned a non-JSON body: {resp.text[:200]}"


# ── MCP tools (the only functions exposed with gr.api) ────────────────────────


def align_person(
    name: str,
    works: list[str] | None = None,
    field: str = "",
    affiliation: str = "",
    year: str = "",
    context: str = "",
    embedding_model: EmbeddingModel = "lexical-idf",
    max_candidates: int = 20,
    accept_threshold: float = 0.65,
    margin_threshold: float = 0.08,
    max_returned_candidates: int = 5,
) -> dict:
    """
    Align a person to an IdRef PPN from their name plus any disambiguating context.

    `best_ppn` is populated only when `status == "accepted"`. On any other status the
    alignment abstained: report the abstention, do not fall back to the top candidate's
    PPN. Raising a threshold makes the service stricter, i.e. more abstentions.

    Args:
        name: Full person name, e.g. "Valérie Robert". Required.
        works: Titles of documents the person is linked to. Often decisive.
        field: Discipline or subject area, e.g. "sociologie des sciences".
        affiliation: Institution, laboratory or place, e.g. "Université de Nancy".
        year: A relevant year as a string, e.g. "2003".
        context: Any other free text — a biographical note, an abstract.
        embedding_model: How texts are compared — lexical, lexical-idf, albert-bge-m3, granite, qwen or minilm.
        max_candidates: Candidate authorities to enrich and score, 1-30 on this demo endpoint.
        accept_threshold: Minimum final score for "accepted" (0.0-1.0).
        margin_threshold: Minimum lead over the runner-up for "accepted" (0.0-1.0).
        max_returned_candidates: Ranked candidates to include in the reply, 0-10 on this demo endpoint.

    Returns:
        {"source": "idref-resolver-api", "command": "align_person", "status": "accepted" | "ambiguous" | "low_confidence" | "not_found" | null, "best_ppn": str | null, "best_candidate": object | null, "candidates": [object], "candidates_scored": int, "similarity": object | null, "error": str | null}
    """
    reply: dict[str, Any] = {
        "source": "idref-resolver-api",
        "command": "align_person",
        "status": None,
        "best_ppn": None,
        "best_candidate": None,
        "candidates": [],
        "candidates_scored": 0,
        "similarity": None,
        "error": None,
    }

    if not (name or "").strip():
        reply["error"] = "name is required"
        return reply
    if embedding_model not in EMBEDDING_MODELS:
        reply["error"] = "embedding_model must be one of " + ", ".join(EMBEDDING_MODELS)
        return reply

    payload = {
        "name": name.strip(),
        "works": [w for w in (works or []) if w],
        "field": field or "",
        "affiliation": affiliation or "",
        "role": "",
        "year": str(year or ""),
        "context": context or "",
        "embedding_model": embedding_model,
        "max_candidates": max(1, min(int(max_candidates or 20), MAX_CANDIDATES)),
        "accept_threshold": float(accept_threshold),
        "margin_threshold": float(margin_threshold),
    }

    result, error = _post("/align/person", payload)
    if error:
        reply["error"] = error
        return reply

    ranked = result.get("candidates") or []
    reply.update(
        status=result.get("status"),
        best_ppn=result.get("best_ppn"),
        best_candidate=result.get("best_candidate"),
        candidates=ranked[: max(0, min(int(max_returned_candidates or 5), MAX_RETURNED))],
        candidates_scored=len(ranked),
        similarity=result.get("similarity"),
        error=result.get("error"),
    )
    return reply


# ── Presentation ──────────────────────────────────────────────────────────────

VERDICTS = {
    "accepted": "✅ **Accepté** — un PPN est retenu.",
    "ambiguous": "🤔 **Ambigu** — plusieurs autorités se tiennent ; l'outil s'abstient.",
    "low_confidence": "⚠️ **Confiance insuffisante** — le meilleur score reste sous le seuil.",
    "not_found": "❌ **Aucune autorité candidate** pour ce nom.",
}


def _render(payload: dict) -> str:
    lines: list[str] = []
    status = payload.get("status")
    lines.append(VERDICTS.get(status, f"**Statut** — `{status}`"))
    if payload.get("best_ppn"):
        ppn = payload["best_ppn"]
        lines.append(f"\n**PPN retenu** — [{ppn}](https://www.idref.fr/{ppn})")
    elif status:
        lines.append(
            "\n_Abstention : ne retenez pas le premier candidat comme s'il avait "
            "été accepté. Une abstention est une réponse._"
        )
    if payload.get("error"):
        lines.append(f"\n> ⚠️ {payload['error']}")

    candidates = payload.get("candidates") or []
    if candidates:
        lines += [
            "",
            f"**{len(candidates)} candidats affichés sur {payload.get('candidates_scored', '?')} évalués**",
            "",
            "| PPN | Score final | Forme retenue |",
            "|---|---|---|",
        ]
        for c in candidates:
            ppn = c.get("ppn") or "—"
            score = ((c.get("score") or {}).get("final"))
            score_txt = f"{score:.3f}" if isinstance(score, (int, float)) else "—"
            forms = ((c.get("evidence") or {}).get("preferred_forms") or [])
            form = (forms[0] if forms else "—").replace("|", "\\|")
            link = f"[{ppn}](https://www.idref.fr/{ppn})" if ppn != "—" else "—"
            lines.append(f"| {link} | {score_txt} | {form} |")
    return "\n".join(lines)


def _run(name, works_text, field, affiliation, year, context,
         embedding_model, max_candidates, accept_threshold, margin_threshold):
    works = [line.strip() for line in (works_text or "").splitlines() if line.strip()]
    payload = align_person(
        name=name,
        works=works,
        field=field,
        affiliation=affiliation,
        year=year,
        context=context,
        embedding_model=embedding_model,
        max_candidates=max_candidates,
        accept_threshold=accept_threshold,
        margin_threshold=margin_threshold,
    )
    if payload.get("error") and payload.get("status") is None:
        raise gr.Error(payload["error"])
    return _render(payload), payload


# ── UI ────────────────────────────────────────────────────────────────────────

with gr.Blocks(title="IdRef resolver MCP demo") as demo:
    gr.Markdown(
        "# IdRef resolver — démo MCP\n"
        "Démo autonome du serveur MCP "
        "[`idref-resolver-api`](https://github.com/smartbiblia-solutions/agentic-stack/tree/main/mcp/idref-resolver-api) "
        ", aligner une autoritéé personne extraite d'un document sur son PPN IdRef, ou "
        "s'abstenir quand l'indice est trop faible. "
    )

    with gr.Row():
        with gr.Column():
            name = gr.Textbox(label="Nom de la personne", placeholder="Valérie Robert")
            works_text = gr.Textbox(
                label="Titres d'œuvres (un par ligne)",
                lines=3,
                placeholder="Nous n'avons jamais été modernes",
            )
            field = gr.Textbox(label="Domaine", placeholder="sociologie des sciences")
            affiliation = gr.Textbox(label="Affiliation", placeholder="Université de Nancy")
            with gr.Row():
                year = gr.Textbox(label="Année", placeholder="2003")
                embedding_model = gr.Dropdown(
                    list(EMBEDDING_MODELS), value="lexical-idf", label="Comparaison"
                )
            context = gr.Textbox(label="Contexte libre", lines=2)
            with gr.Accordion("Seuils", open=False):
                max_candidates = gr.Slider(
                    1, MAX_CANDIDATES, value=20, step=1, label="Candidats évalués"
                )
                accept_threshold = gr.Slider(
                    0.0, 1.0, value=0.65, step=0.01, label="Seuil d'acceptation"
                )
                margin_threshold = gr.Slider(
                    0.0, 1.0, value=0.08, step=0.01, label="Écart minimal au second"
                )
            align_btn = gr.Button("Aligner", variant="primary")
        with gr.Column():
            verdict_out = gr.Markdown()
            raw_out = gr.JSON(label="Sortie brute de l'outil")

    inputs = [name, works_text, field, affiliation, year, context,
              embedding_model, max_candidates, accept_threshold, margin_threshold]

    gr.Examples(
        examples=[
            ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences",
             "Sciences Po", "1991", "", "lexical-idf", 20, 0.65, 0.08],
            ["Michel Foucault", "Surveiller et punir : naissance de la prison",
             "philosophie", "Collège de France", "1975", "", "lexical-idf", 20, 0.65, 0.08],
            ["Valérie Robert", "", "études germaniques",
             "Université Sorbonne Nouvelle", "", "", "lexical-idf", 20, 0.65, 0.08],
            ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences",
             "Sciences Po", "1991", "", "albert-bge-m3", 20, 0.65, 0.08],
            ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences",
             "Sciences Po", "1991", "", "lexical-idf", 20, 0.85, 0.20],
        ],
        inputs=inputs,
        label="Titre décisif, autre titre décisif, indices sans œuvre, comparaison "
              "sémantique, puis le même cas avec des seuils sévères",
    )
    align_btn.click(_run, inputs=inputs, outputs=[verdict_out, raw_out], api_name=False)

    # The only declared MCP tool. The name matches the canonical server's.
    gr.api(align_person, api_name="align_person")


demo.launch(theme=gr.themes.Soft(), mcp_server=True)