Paper-Monitoring-SW/src/paper_monitor/ai_enrich.py

import json
import time


# ============================================================
# Gemini rate limiter
# ============================================================

_GEMINI_MIN_INTERVAL = 13.0
_GEMINI_LAST_CALL = 0.0


def _wait_for_gemini_rate_limit():
    global _GEMINI_LAST_CALL

    now = time.monotonic()
    elapsed = now - _GEMINI_LAST_CALL
    wait = _GEMINI_MIN_INTERVAL - elapsed

    if wait > 0:
        time.sleep(wait)

    _GEMINI_LAST_CALL = time.monotonic()


# ============================================================
# Common AI prompt
# ============================================================

SYSTEM = '''You are a technical literature triage assistant for an automotive radar engineer.

Analyze ONLY the supplied metadata or full text.
Do not invent experimental results, methods, conclusions, or claims that are not supported by the supplied information.

The primary technical scope is radar engineering, especially:
- automotive radar
- FMCW radar
- radar waveform design
- radar MIMO
- radar DOA / angle estimation
- radar super-resolution
- radar interference
- automotive radar antenna
- imaging radar / 4D radar
- radar target modeling
- radar signal processing
- radar hardware and RF impairments

Generic communications, acoustic DOA, sonar, medical sensing, or non-radar MIMO/antenna research should receive low relevance unless there is a clear technical connection to radar.

Return:
- relevance: integer from 1 to 5
- categories: technical categories
- summary: 1-2 concise Korean sentences
- reason: concise Korean explanation of why this paper is relevant

Preferred categories:
Antenna,
Waveform,
DOA,
MIMO,
Target Modeling,
Interference,
RF Impairment,
Imaging Radar,
Signal Processing,
Hardware,
Other.
'''


AI_RESPONSE_SCHEMA = {
    "type": "object",
    "properties": {
        "relevance": {
            "type": "integer",
            "minimum": 1,
            "maximum": 5,
        },
        "categories": {
            "type": "array",
            "items": {
                "type": "string"
            },
        },
        "summary": {
            "type": "string"
        },
        "reason": {
            "type": "string"
        },
    },
    "required": [
        "relevance",
        "categories",
        "summary",
        "reason",
    ],
}


# ============================================================
# Common helpers
# ============================================================

def _make_payload(paper):
    return {
        "title": paper.title,
        "abstract": paper.abstract,
        "venue": paper.venue,
        "year": paper.year,
        "current_categories": paper.categories,
        "current_relevance": paper.relevance,
    }


def _make_full_text_payload(paper, full_text):
    return {
        "title": paper.title,
        "abstract": paper.abstract,
        "venue": paper.venue,
        "year": paper.year,
        "full_text": full_text,
        "current_categories": paper.categories,
        "current_relevance": paper.relevance,
    }


def _apply_result(paper, data):
    paper.relevance = max(
        1,
        min(
            5,
            int(data.get("relevance", paper.relevance))
        )
    )

    if isinstance(data.get("categories"), list) and data["categories"]:
        paper.categories = [
            str(x)
            for x in data["categories"][:5]
        ]

    paper.summary = str(
        data.get("summary", "")
    ).strip()

    paper.ai_reason = str(
        data.get("reason", "")
    ).strip()

    return paper


def _parse_json_text(text):
    if not text:
        raise RuntimeError("AI returned empty response")

    text = text.strip()

    # Defensive handling for models that wrap JSON in Markdown fences.
    if text.startswith("```"):
        lines = text.splitlines()

        if lines:
            lines = lines[1:]

        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]

        text = "\n".join(lines).strip()

        if text.lower().startswith("json"):
            text = text[4:].strip()

    return json.loads(text)


# ============================================================
# Gemini - metadata / abstract
# ============================================================

def enrich_with_gemini(paper, api_key, model):
    if not api_key:
        return paper

    from google import genai
    from google.genai import types

    client = genai.Client(api_key=api_key)

    payload = _make_payload(paper)

    prompt = (
        SYSTEM
        + "\n\nPaper metadata:\n"
        + json.dumps(payload, ensure_ascii=False)
    )

    _wait_for_gemini_rate_limit()

    response = client.models.generate_content(
        model=model,
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
            response_schema=AI_RESPONSE_SCHEMA,
        ),
    )

    data = _parse_json_text(response.text)

    paper = _apply_result(paper, data)
    paper.ai_analysis_level = "abstract"

    return paper


# ============================================================
# Gemini - full text
# ============================================================

def enrich_with_gemini_full_text(
    paper,
    full_text,
    api_key,
    model,
):
    if not api_key:
        return paper

    if not full_text or not full_text.strip():
        return paper

    from google import genai
    from google.genai import types

    client = genai.Client(api_key=api_key)

    payload = _make_full_text_payload(
        paper,
        full_text,
    )

    prompt = (
        SYSTEM
        + """

You are given extracted full text from the paper.

Base the analysis primarily on the supplied full text.
Do not invent information that is not present in the supplied text.
If the extracted text appears incomplete or corrupted, be conservative.

The summary should describe the paper's main technical approach,
contribution, and relevance to automotive radar when supported by the text.
"""
        + "\n\nPaper data:\n"
        + json.dumps(payload, ensure_ascii=False)
    )

    _wait_for_gemini_rate_limit()

    response = client.models.generate_content(
        model=model,
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.1,
            response_mime_type="application/json",
            response_schema=AI_RESPONSE_SCHEMA,
        ),
    )

    data = _parse_json_text(response.text)

    paper = _apply_result(paper, data)
    paper.ai_analysis_level = "full_text"

    return paper


# ============================================================
# Groq common prompt
# ============================================================

def _make_groq_prompt(payload, full_text=False):
    if full_text:
        instruction = """
You are given extracted full text from the paper.

Base the analysis primarily on the supplied full text.
Do not invent information that is not present in the supplied text.
If the extracted text appears incomplete or corrupted, be conservative.

The summary should describe the paper's main technical approach,
contribution, and relevance to automotive radar when supported by the text.
"""
    else:
        instruction = """
Analyze the supplied paper metadata only.
Do not infer details that are not present in the metadata.
"""

    output_instruction = """
Return ONLY one valid JSON object.

The JSON object must have exactly these fields:

{
  "relevance": 1,
  "categories": ["Signal Processing"],
  "summary": "한국어 요약",
  "reason": "한국어 관련성 설명"
}

Rules:
- relevance must be an integer from 1 to 5
- categories must be a JSON array of strings
- summary must be written in Korean
- reason must be written in Korean
- do not use Markdown code fences
- do not include text before or after the JSON object
"""

    return (
        SYSTEM
        + instruction
        + output_instruction
        + "\n\nPaper data:\n"
        + json.dumps(payload, ensure_ascii=False)
    )


# ============================================================
# Groq - metadata / abstract
# ============================================================

def enrich_with_groq(
    paper,
    api_key,
    model,
):
    if not api_key:
        return paper

    from groq import Groq

    client = Groq(api_key=api_key)

    payload = _make_payload(paper)

    prompt = _make_groq_prompt(
        payload,
        full_text=False,
    )

    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": SYSTEM,
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
        temperature=0.1,
        response_format={
            "type": "json_object"
        },
    )

    content = response.choices[0].message.content

    data = _parse_json_text(content)

    paper = _apply_result(paper, data)
    paper.ai_analysis_level = "abstract"

    return paper


# ============================================================
# Groq - full text
# ============================================================

def enrich_with_groq_full_text(
    paper,
    full_text,
    api_key,
    model,
):
    if not api_key:
        return paper

    if not full_text or not full_text.strip():
        return paper

    from groq import Groq

    client = Groq(api_key=api_key)

    payload = _make_full_text_payload(
        paper,
        full_text,
    )

    prompt = _make_groq_prompt(
        payload,
        full_text=True,
    )

    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": SYSTEM,
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
        temperature=0.1,
        response_format={
            "type": "json_object"
        },
    )

    content = response.choices[0].message.content

    data = _parse_json_text(content)

    paper = _apply_result(paper, data)
    paper.ai_analysis_level = "full_text"

    return paper


# ============================================================
# OpenAI - legacy / optional
# ============================================================

def enrich_with_openai(
    paper,
    api_key,
    model,
):
    if not api_key:
        return paper

    from openai import OpenAI

    client = OpenAI(api_key=api_key)

    payload = _make_payload(paper)

    response = client.responses.create(
        model=model,
        instructions=SYSTEM,
        input=json.dumps(
            payload,
            ensure_ascii=False,
        ),
    )

    data = _parse_json_text(
        response.output_text
    )

    return _apply_result(
        paper,
        data,
    )