app.py

"""app.py — Gradio app for AI-powered System 1/2 review rating."""

import os
import json
import datetime
import gradio as gr
from fetcher import fetch_paper_reviews, get_bundled_ids
from rater import (
    rate_review, format_result_markdown,
    rate_metareview, format_metareview_result_markdown,
)
from analytics import load_all, fig_label_distribution, fig_rqs_by_decision, \
    fig_s1_s2_scatter, fig_bias_heatmap, fig_rqs_distribution, FINDINGS

_paper_cache: dict = {}
_last_result: dict = {}

FEEDBACK_FILE = os.path.join(os.path.dirname(__file__), "feedback.jsonl")

PROVIDER_MODELS = {
    "Anthropic (Recommended)": {
        "provider": "anthropic",
        "models": ["claude-sonnet-4-6", "claude-opus-4-6", "claude-haiku-4-5-20251001"],
        "placeholder": "sk-ant-...",
    },
    "OpenAI": {
        "provider": "openai",
        "models": ["gpt-4o", "gpt-4o-mini", "gpt-4-turbo", "o1", "o3-mini"],
        "placeholder": "sk-...",
    },
    "DeepSeek": {
        "provider": "deepseek",
        "models": ["deepseek-chat", "deepseek-reasoner"],
        "placeholder": "sk-...",
    },
}

# ── Section content ─────────────────────────────────────────────────────────────

SECTION_CONTENT = {
    "📖 Motivation": """### Motivation



Peer review is one of the central institutions governing scientific progress, yet most existing analysis focuses on outcomes such as scores, acceptance rates, disagreement levels, or textual sentiment. These signals are useful but incomplete. They do not directly capture **how reviewers think**.



Kahneman's dual-process framework provides a principled theoretical lens:



- **System 1** is rapid, associative, intuitive, and often relies on heuristics such as representativeness, familiarity, fluency, and global impressions.

- **System 2** is effortful, analytical, explicit, and more likely to engage in structured reasoning, evidence integration, and conditional judgment.



Applied to peer review, this distinction enables us to study whether a review is dominated by venue-fit heuristics, abstract "overall impression" judgments, or conclusion-first reasoning — or instead by falsifiable claims, methodological decomposition, comparative evidence, and belief updating.



This is not merely a stylistic distinction. It bears directly on questions of **review quality**, **rebuttal responsiveness**, **decision transparency**, and **cognitive bias in evaluation**.""",

    "🎯 Core Objectives": """### Core Objectives



The goal of Kahneman4Review is to build a robust framework for:



1. **Classifying** review text into cognitive reasoning modes inspired by Kahneman's theory;

2. **Characterizing** the effort structure of review reasoning, from low-effort impressionistic judgment to high-effort analytical synthesis;

3. **Diagnosing** cognitive biases in review and metareview, such as representativeness heuristics, question substitution, anchoring, confirmation bias, overconfidence, and narrative fallacy;

4. **Supporting** LLM-based judges that can assess the reasoning mode and epistemic quality of reviews in a structured, reproducible way.""",

    "📐 Academic Claim": """### Academic Claim



The central academic claim is that **review quality cannot be fully understood without reasoning structure**. A review may be long, harsh, polite, or even technically correct, yet still be cognitively shallow. Conversely, a review may be negative but high-quality if it exhibits strong System 2 properties such as precise falsifiability, explicit evidence chains, and principled updating under rebuttal.



This project sits at the intersection of:

- **Metascience**: understanding the scientific process itself;

- **AI for Science / AI for Institutions**: using language models to analyze scientific governance mechanisms;

- **Computational social science**: studying evaluation behavior through text;

- **LLM-as-a-Judge research**: moving beyond outcome scoring toward reasoning-aware judgment;

- **Cognitive science of decision-making**: operationalizing dual-process theory in institutional text.""",

    "🔑 Key Contributions": """### Key Contributions



**1. A cognitive taxonomy for peer review**

We operationalize Kahneman's theory into an annotation framework suitable for review text, including System 1, System 2, mixed / transitional reasoning, and non-evaluative administrative language.



**2. Effort-sensitive reasoning analysis**

Beyond binary labels, the framework distinguishes different levels of System 2 effort, separating shallow structured criticism from deeper falsification-oriented reasoning and meta-level synthesis.



**3. Bias diagnostics for review interpretation**

The framework explicitly identifies recurring bias pathways: venue-fit substitution, authority alignment, conclusion-first justification, selective evidence weighting, and failure to update after rebuttal.""",

    "💡 Why This Matters": """### Why This Matters



The significance of this project is not limited to review analytics. More broadly, it addresses a foundational problem in the evaluation of human and AI reasoning:



> *How can we distinguish genuine analysis from articulate intuition?*



In academic review, this distinction affects fairness, transparency, and the reliability of scientific gatekeeping. In LLM evaluation, it affects whether models merely mimic analytical language or actually detect structured reasoning.



By making the cognitive mode of review explicit, Kahneman4Review aims to support better review auditing, more interpretable LLM judges, stronger rebuttal strategies, and more scientifically grounded discussion of what constitutes a "good review." """,
}

SECTION_LABELS = list(SECTION_CONTENT.keys())


# ── Callbacks ───────────────────────────────────────────────────────────────────

def _get_api_key(user_key: str, provider_label: str) -> str:
    k = (user_key or "").strip()
    if k:
        return k
    # Fall back to env vars per provider
    provider = PROVIDER_MODELS.get(provider_label, {}).get("provider", "anthropic")
    if provider == "anthropic":
        return os.environ.get("ANTHROPIC_API_KEY", "")
    elif provider == "openai":
        return os.environ.get("OPENAI_API_KEY", "")
    elif provider == "deepseek":
        return os.environ.get("DEEPSEEK_API_KEY", "")
    return ""


def _get_provider(provider_label: str) -> str:
    return PROVIDER_MODELS.get(provider_label, {}).get("provider", "anthropic")


def toggle_section(label, current_label):
    if label == current_label:
        return "", gr.update(visible=False), ""
    return SECTION_CONTENT.get(label, ""), gr.update(visible=True), label


def update_provider(provider_label: str):
    info = PROVIDER_MODELS.get(provider_label, {})
    models = info.get("models", [])
    placeholder = info.get("placeholder", "API key...")
    return (
        gr.update(choices=models, value=models[0] if models else None),
        gr.update(placeholder=placeholder),
    )


def load_paper(paper_id: str):
    paper_id = (paper_id or "").strip()
    if not paper_id:
        return gr.update(choices=[], value=None), "Please enter a paper ID.", "", ""
    try:
        paper = fetch_paper_reviews(paper_id)
        _paper_cache[paper_id] = paper
        reviewers = [r["reviewer_id"] for r in paper["reviews"]]
        decision = paper.get("decision", "")
        info = f"**{paper.get('title', paper_id)}**\n\n{paper.get('conference', '')}"
        if decision:
            info += f"  ·  **Decision:** {decision}"
        info += f"  ·  {len(reviewers)} reviewer(s)"
        metareview = paper.get("metareview", "")
        meta_md = f"**Area Chair Meta-Review:**\n\n{metareview}" if metareview else "*No meta-review available.*"
        return gr.update(choices=reviewers, value=reviewers[0] if reviewers else None), info, meta_md, ""
    except Exception as e:
        return gr.update(choices=[], value=None), f"Error: {e}", "", ""


def show_review(paper_id: str, reviewer_id: str):
    paper = _paper_cache.get((paper_id or "").strip())
    if not paper or not reviewer_id:
        return ""
    for r in paper["reviews"]:
        if r["reviewer_id"] == reviewer_id:
            return f"**Initial:** {r['initial_rating']}  **Final:** {r['final_rating']}\n\n{r['review_content']}"
    return ""


def run_rating(paper_id: str, reviewer_id: str, api_key: str, provider_label: str, model: str):
    global _last_result
    paper = _paper_cache.get((paper_id or "").strip())
    if not paper:
        yield "Please load a paper first.", gr.update(visible=False)
        return
    if not reviewer_id:
        yield "Please select a reviewer.", gr.update(visible=False)
        return
    key = _get_api_key(api_key, provider_label)
    if not key:
        yield "No API key found. Enter your API key above.", gr.update(visible=False)
        return
    review = next((r for r in paper["reviews"] if r["reviewer_id"] == reviewer_id), None)
    if not review:
        yield f"Reviewer {reviewer_id} not found.", gr.update(visible=False)
        return
    provider = _get_provider(provider_label)
    yield f"Calling {model} to rate **{reviewer_id}**…", gr.update(visible=False)
    try:
        result = rate_review(
            review_content=review["review_content"],
            initial_rating=review["initial_rating"],
            final_rating=review["final_rating"],
            conference=paper.get("conference", ""),
            api_key=key,
            provider=provider,
            model=model,
        )
        _last_result = {"paper_id": paper_id, "reviewer_id": reviewer_id, "result": result}
        yield format_result_markdown(reviewer_id, result), gr.update(visible=True)
    except Exception as e:
        yield f"Error: {e}", gr.update(visible=False)


def run_rating_all(paper_id: str, api_key: str, provider_label: str, model: str):
    paper = _paper_cache.get((paper_id or "").strip())
    if not paper:
        yield "Please load a paper first.", gr.update(visible=False)
        return
    key = _get_api_key(api_key, provider_label)
    if not key:
        yield "No API key found. Enter your API key above.", gr.update(visible=False)
        return
    provider = _get_provider(provider_label)
    accumulated = ""
    for i, review in enumerate(paper["reviews"]):
        rid = review["reviewer_id"]
        marker = f"\n\n---\n\n*Rating {i+1}/{len(paper['reviews'])}: {rid}…*"
        accumulated += marker
        yield accumulated, gr.update(visible=False)
        try:
            result = rate_review(
                review_content=review["review_content"],
                initial_rating=review["initial_rating"],
                final_rating=review["final_rating"],
                conference=paper.get("conference", ""),
                api_key=key,
                provider=provider,
                model=model,
            )
            accumulated = accumulated[: -len(marker)]
            accumulated += "\n\n---\n\n" + format_result_markdown(rid, result)
        except Exception as e:
            accumulated = accumulated[: -len(marker)]
            accumulated += f"\n\n---\n\n**{rid}** — Error: {e}"
        yield accumulated, gr.update(visible=False)
    yield accumulated + "\n\n---\n\n*Done.*", gr.update(visible=False)


def run_metareview_rating(paper_id: str, api_key: str, provider_label: str, model: str):
    paper = _paper_cache.get((paper_id or "").strip())
    if not paper:
        yield "Please load a paper first."
        return
    metareview = paper.get("metareview", "").strip()
    if not metareview:
        yield "No meta-review available for this paper."
        return
    key = _get_api_key(api_key, provider_label)
    if not key:
        yield "No API key found. Enter your API key above."
        return
    provider = _get_provider(provider_label)
    yield f"Calling {model} to rate the meta-review…"
    try:
        result = rate_metareview(
            metareview_content=metareview,
            decision=paper.get("decision", ""),
            conference=paper.get("conference", ""),
            api_key=key,
            provider=provider,
            model=model,
        )
        yield format_metareview_result_markdown(result)
    except Exception as e:
        yield f"Error: {e}"


def run_manual_review_rating(review_text: str, api_key: str, provider_label: str, model: str):
    if not (review_text or "").strip():
        yield "Please enter a review text."
        return
    key = _get_api_key(api_key, provider_label)
    if not key:
        yield "No API key found. Enter your API key above."
        return
    provider = _get_provider(provider_label)
    yield f"Calling {model}…"
    try:
        result = rate_review(
            review_content=review_text.strip(),
            initial_rating="unknown",
            final_rating="unknown",
            conference="",
            api_key=key,
            provider=provider,
            model=model,
        )
        yield format_result_markdown("Manual Input", result)
    except Exception as e:
        yield f"Error: {e}"


def run_manual_metareview_rating(metareview_text: str, api_key: str, provider_label: str, model: str):
    if not (metareview_text or "").strip():
        yield "Please enter a meta-review text."
        return
    key = _get_api_key(api_key, provider_label)
    if not key:
        yield "No API key found. Enter your API key above."
        return
    provider = _get_provider(provider_label)
    yield f"Calling {model}…"
    try:
        result = rate_metareview(
            metareview_content=metareview_text.strip(),
            decision="",
            conference="",
            api_key=key,
            provider=provider,
            model=model,
        )
        yield format_metareview_result_markdown(result)
    except Exception as e:
        yield f"Error: {e}"


def submit_feedback(satisfaction: str, correct_label: str, comment: str):
    if not _last_result:
        return "No rating to give feedback on yet."
    entry = {
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "paper_id": _last_result.get("paper_id", ""),
        "reviewer_id": _last_result.get("reviewer_id", ""),
        "predicted_label": _last_result.get("result", {}).get("main_label", ""),
        "reasoning_quality_score": _last_result.get("result", {}).get("reasoning_quality_score"),
        "derived": _last_result.get("result", {}).get("derived", {}),
        "satisfaction": satisfaction,
        "correct_label": correct_label if correct_label else None,
        "comment": comment.strip() if comment else None,
    }
    with open(FEEDBACK_FILE, "a") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return "✅ Feedback saved. Thank you!"


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

_default_provider = "Anthropic (Recommended)"
_default_models = PROVIDER_MODELS[_default_provider]["models"]

with gr.Blocks(title="Kahneman4Review", theme=gr.themes.Soft()) as demo:

    gr.Markdown("""# 🧠 Kahneman4Review



Kahneman4Review is a research-oriented framework for analyzing the cognitive structure of academic peer review through the lens of Daniel Kahneman's dual-process theory in *Thinking, Fast and Slow*. The project studies whether review statements are primarily driven by **System 1** reasoning (fast, intuitive, impression-based judgment) or by **System 2** reasoning (slow, deliberate, evidence-based analysis).



Rather than treating reviews only as scalar signals of acceptance or rejection, this project asks a deeper scientific question:



> *What kinds of cognition are reflected in peer review text, and how do those cognitive modes shape review quality, fairness, and decision reliability?*



This perspective reframes peer review as a **reasoning process** rather than merely an evaluative outcome.

""")

    # ── Expandable sections ────────────────────────────────────────────────────
    _current_section = gr.State("")

    with gr.Row():
        sec_btns = [
            gr.Button(label, size="sm", variant="secondary")
            for label in SECTION_LABELS
        ]

    section_box = gr.Markdown("", visible=False)

    for btn in sec_btns:
        btn.click(
            fn=toggle_section,
            inputs=[btn, _current_section],
            outputs=[section_box, section_box, _current_section],
        )

    gr.Markdown("""---

> *"A review should be judged not only by what it concludes, but by how it reaches that conclusion."*



---""")

    # ── API / Model / Paper loader — all in one row ────────────────────────────
    with gr.Row():
        provider_dd = gr.Dropdown(
            choices=list(PROVIDER_MODELS.keys()),
            value=_default_provider,
            label="Provider",
            scale=2,
        )
        model_dd = gr.Dropdown(
            choices=_default_models,
            value=_default_models[0],
            label="Model",
            scale=2,
        )
        api_key_box = gr.Textbox(
            label="API Key",
            placeholder=PROVIDER_MODELS[_default_provider]["placeholder"],
            type="password",
            scale=3,
        )
        paper_id_box = gr.Textbox(
            label="OpenReview Paper ID",
            placeholder="e.g. B1e3OlStPB",
            scale=2,
        )
        load_btn = gr.Button("Load Paper", variant="primary", scale=1)

    paper_info = gr.Markdown("")

    with gr.Tabs():
        with gr.Tab("Reviews"):
            with gr.Row():
                reviewer_dd = gr.Dropdown(choices=[], label="Select Reviewer", interactive=True, scale=2)
                rate_one_btn = gr.Button("AI Rate This Reviewer", variant="primary", scale=1)
                rate_all_btn = gr.Button("AI Rate All Reviewers", variant="secondary", scale=1)

            review_display = gr.Markdown("")
            gr.Markdown("---")
            gr.Markdown("### Rating Results")
            result_display = gr.Markdown("")

            with gr.Group(visible=False) as feedback_panel:
                gr.Markdown("#### 💬 Feedback on this rating")
                with gr.Row():
                    satisfaction = gr.Radio(
                        choices=["👍 Agree", "🤔 Partially agree", "👎 Disagree"],
                        label="Do you agree with this classification?",
                        scale=2,
                    )
                    correct_label = gr.Dropdown(
                        choices=["System 1", "System 2", "Mixed", "Non-evaluative"],
                        label="What should the correct label be? (optional)",
                        allow_custom_value=False,
                        scale=1,
                    )
                comment = gr.Textbox(
                    label="Additional comments (optional)",
                    placeholder="e.g. The reviewer clearly cited specific equations, so System 2 seems more appropriate...",
                    lines=2,
                )
                with gr.Row():
                    submit_fb_btn = gr.Button("Submit Feedback", variant="primary")
                feedback_status = gr.Markdown("")

        with gr.Tab("Meta-Review"):
            meta_display = gr.Markdown("*Load a paper to see the meta-review.*")
            gr.Markdown("---")
            rate_meta_btn = gr.Button("AI Rate Meta-Review", variant="primary")
            gr.Markdown("### Meta-Review Analysis")
            meta_result_display = gr.Markdown("")

        with gr.Tab("Manual Input"):
            gr.Markdown("Paste any review or meta-review text directly for evaluation — no paper ID needed.")
            with gr.Tabs():
                with gr.Tab("Review"):
                    manual_review_box = gr.Textbox(
                        label="Review text",
                        placeholder="Paste the review text here…",
                        lines=10,
                    )
                    manual_review_btn = gr.Button("AI Rate This Review", variant="primary")
                    manual_review_result = gr.Markdown("")

                with gr.Tab("Meta-Review"):
                    manual_meta_box = gr.Textbox(
                        label="Meta-review text",
                        placeholder="Paste the meta-review text here…",
                        lines=10,
                    )
                    manual_meta_btn = gr.Button("AI Rate This Meta-Review", variant="primary")
                    manual_meta_result = gr.Markdown("")

        with gr.Tab("📊 Analytics"):
            gr.Markdown(FINDINGS)
            gr.Markdown("---")
            _adata = load_all()
            with gr.Row():
                gr.Plot(value=fig_label_distribution(_adata))
                gr.Plot(value=fig_rqs_by_decision(_adata))
            with gr.Row():
                gr.Plot(value=fig_rqs_distribution(_adata))
                gr.Plot(value=fig_bias_heatmap(_adata))
            gr.Plot(value=fig_s1_s2_scatter(_adata))

    # ── Wire events ────────────────────────────────────────────────────────────
    provider_dd.change(update_provider, [provider_dd], [model_dd, api_key_box])

    load_btn.click(
        load_paper,
        [paper_id_box],
        [reviewer_dd, paper_info, meta_display, result_display],
    )
    reviewer_dd.change(show_review, [paper_id_box, reviewer_dd], [review_display])

    rate_one_btn.click(
        run_rating,
        [paper_id_box, reviewer_dd, api_key_box, provider_dd, model_dd],
        [result_display, feedback_panel],
    )
    rate_all_btn.click(
        run_rating_all,
        [paper_id_box, api_key_box, provider_dd, model_dd],
        [result_display, feedback_panel],
    )
    rate_meta_btn.click(
        run_metareview_rating,
        [paper_id_box, api_key_box, provider_dd, model_dd],
        [meta_result_display],
    )
    manual_review_btn.click(
        run_manual_review_rating,
        [manual_review_box, api_key_box, provider_dd, model_dd],
        [manual_review_result],
    )
    manual_meta_btn.click(
        run_manual_metareview_rating,
        [manual_meta_box, api_key_box, provider_dd, model_dd],
        [manual_meta_result],
    )
    submit_fb_btn.click(
        submit_feedback,
        [satisfaction, correct_label, comment],
        [feedback_status],
    )


if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, show_api=False)