hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/retrieval/ranking_prompt.py

"""LLM 精排 Prompt 构建 —— 根据 profile 生成不同排序策略的系统提示与用户提示."""

from __future__ import annotations

from typing import Any

from .method_acronym import is_method_acronym_token

RANKER_SYSTEM_PROMPT = (
    "你是学术文献评估专家。根据候选论文列表评估与排序；不得虚构论文。\n\n"
    "排序原则：\n"
    "1. 优先选择与用户查询语义最相关的论文\n"
    "2. 识别经典/里程碑论文：引用量极高(>500)且发表≥5年的开创性工作应排在前面\n"
    "3. 顶会/顶刊论文(Nature/Science/NeurIPS/CVPR/ICML/ICLR等)优先\n"
    "4. 在相关性接近时，被广泛引用的论文优先于新发论文\n"
    "5. 平衡新颖性：若用户明显在找最新方法，可适当降低经典论文权重\n\n"
    "请严格按照要求的JSON格式输出。"
)

RANKER_SYSTEM_PROMPT_RETRY = (
    "你是学术文献评估专家。根据候选论文列表评估与排序；不得虚构论文。\n\n"
    "排序原则：优先语义相关，识别经典/里程碑论文，顶会论文优先，被广泛引用的论文优先。\n"
    "请严格按照要求的JSON格式输出。"
)


def ranker_short_focus_query(query: str) -> bool:
    q = (query or "").strip()
    if not q:
        return False
    if is_method_acronym_token(q):
        return True
    return len(q.split()) <= 2 and len(q) <= 24


def build_retrieval_constraints_block(
    *,
    target_titles: list[str] | None,
    authors: list[str] | None,
    venues: list[str] | None,
    year_from: int | None,
    year_to: int | None,
    method_acronym: str | None = None,
) -> str:
    parts: list[str] = []
    ma = (method_acronym or "").strip()
    if ma:
        parts.append(
            f"- 方法缩写 **{ma}**：优先标题/摘要含「{ma}」的原始论文；"
            f"若用户指某会议上的该方法，应匹配该方法的正式论文（锚定标题优先）"
        )
    tt = [str(t).strip() for t in (target_titles or []) if str(t).strip()][:4]
    if tt:
        parts.append(f"- 目标论文标题（优先精确匹配）：{'; '.join(tt)}")
    au = [str(a).strip() for a in (authors or []) if str(a).strip()][:6]
    if au:
        parts.append(f"- 目标作者：{', '.join(au)}")
    vv = [str(v).strip() for v in (venues or []) if str(v).strip()][:4]
    if vv:
        parts.append(f"- 会议/期刊约束：{', '.join(vv)}")
    if year_from is not None or year_to is not None:
        yf = year_from if year_from is not None else "?"
        yt = year_to if year_to is not None else yf
        parts.append(f"- 年份范围：{yf}–{yt}")
    if not parts:
        return ""
    return "\n## 检索约束（必须遵守）\n" + "\n".join(parts) + "\n"


def _profile_task_and_dims(
    profile: str,
    *,
    n_papers: int,
    top_k: int,
    target_venue: str | None,
) -> tuple[str, str]:
    if profile == "novelty":
        return (
            f"从 {n_papers} 篇中选出 top {top_k}（「近期进展 / 新工作」），"
            "按「新且与查询相关」降序排列；相关度接近时优先更新、更前瞻的工作。",
            """## 评估维度（novelty）
1. **时效与趋势**（35%）：年份更新；反映该方向最新设定或基准
2. **主题相关性**（30%）：与查询任务一致（可略宽于 accuracy）
3. **新意与贡献**（25%）：架构/目标/数据/结论上相比既有方法有明确新点
4. **可信度底线**（10%）：实验充分；无关或空壳工作后排""",
        )
    if profile == "classic":
        return (
            f"从 {n_papers} 篇中选出 top {top_k}（「原始奠基 / 里程碑式经典工作」），"
            "优先**开创性论文**（查询所指方法/架构的原始提出）；近年引用/综述后排，除非用户明确要综述。",
            """## 评估维度（classic）
1. **开创性与匹配**（40%）：是否为查询所指方法/架构的**原始提出论文**或公认首作
2. **引用与影响力**（35%）：总引用与领域地位；仅讨论该方法的 survey 后排
3. **权威出处**（15%）：顶会/期刊正式收录
4. **时效**（10%）：开创性相近时优先更早的奠基论文""",
        )
    dims = """## 评估维度（accuracy）
1. **主题相关性**（40%）：论文主题与查询匹配程度
2. **方法创新性**（25%）：方法新颖与创新点
3. **结果质量**（20%）：实验充分性、结果可靠性
4. **权威与可引用性**（15%）：顶会/期刊与引用表现；经典工作可优先于纯新文"""
    if target_venue:
        dims = """## 评估维度（accuracy · 会议检索）
1. **主题相关性**（35%）：论文主题与查询匹配程度
2. **届次与年份**（25%）：相关度接近时**优先最近一届**（年份更大者优先）
3. **方法创新性**（20%）：方法新颖与创新点
4. **结果质量**（10%）：实验充分性、结果可靠性
5. **权威与可引用性**（10%）：正式 proceedings 与引用表现"""
    return (
        f"从 {n_papers} 篇中选出最相关的 top {top_k}，按与检索意图匹配程度降序排列。",
        dims,
    )


def build_ranking_prompt(
    papers: list[Any],
    query: str,
    top_k: int,
    ranking_profile: str = "accuracy",
    *,
    abstract_max_chars: int = 500,
    target_venue: str | None = None,
    main_conference_proceedings_only: bool = False,
    intent_source_message: str | None = None,
    target_titles: list[str] | None = None,
    authors: list[str] | None = None,
    venues: list[str] | None = None,
    year_from: int | None = None,
    year_to: int | None = None,
    method_acronym: str | None = None,
) -> str:
    max_abs = max(120, int(abstract_max_chars or 500))
    papers_desc: list[str] = []
    for i, rp in enumerate(papers, 1):
        paper = rp.paper
        title = getattr(paper, "title", "N/A")
        abstract = getattr(paper, "abstract", "") or ""
        if len(abstract) > max_abs:
            abstract = abstract[:max_abs] + "..."
        year = getattr(paper, "year", "N/A")
        venue = getattr(paper, "venue", "") or getattr(paper, "journal", "N/A")
        citations = getattr(paper, "citations", 0) or 0
        papers_desc.append(
            f"\n【论文 {i}】\n标题：{title}\n年份：{year}\n会议/期刊：{venue}\n"
            f"引用数：{citations}\n摘要：{abstract}\n"
        )

    constraint_hint = build_retrieval_constraints_block(
        target_titles=target_titles,
        authors=authors,
        venues=venues or ([target_venue] if target_venue else None),
        year_from=year_from,
        year_to=year_to,
        method_acronym=method_acronym,
    )

    venue_hint = ""
    if target_venue:
        venue_hint = f"""
## 会场约束
用户限定了 **{target_venue}** 会议。按以下优先级判断：
1. **优先**：会议/期刊字段明确标注 {target_venue} 或其 proceedings 全称
2. **降级**：仅标题/摘要提及 {target_venue} 但会议字段不明（arXiv预印本）
3. **末位**：会议字段明确为其他会议
4. 同相关度时**发表年份更近**优先；主会优先于 workshop/symposium。
每条 reason 注明关联判断依据。

"""

    main_track_hint = ""
    if main_conference_proceedings_only and target_venue:
        um = (intent_source_message or "").strip()
        um_block = f"\n### 用户原始表述\n{um[:700]}\n" if um else ""
        main_track_hint = f"""
## 主会议录用
仅保留 **{target_venue}** 主会正式论文；排除 workshop、卫星会等。{um_block}
依据「会议/期刊」字段判断，**非主会论文不得进入前 {top_k}**（不足则少填）。
每条 reason 说明认定为主会的依据。

"""

    profile = (ranking_profile or "accuracy").strip().lower()
    if profile not in ("accuracy", "novelty", "classic"):
        profile = "accuracy"

    short_disambig = ""
    if ranker_short_focus_query(query):
        short_disambig = """
## 短查询消歧
同名缩写论文：优先副标题更匹配 ML 顶会主流问题且会议为高等级 proceedings 的论文；
下调任务/数据形态与用户意图明显不符的论文。每条 reason 说明区分依据。

"""

    task_line, dims = _profile_task_and_dims(
        profile, n_papers=len(papers), top_k=top_k, target_venue=target_venue
    )
    papers_block = "\n---\n".join(papers_desc)

    return f"""你是一位学术文献评估专家，请根据用户检索需求对以下论文精排序。

## 用户检索需求
{query}
{constraint_hint}{short_disambig}{venue_hint}{main_track_hint}
## 候选论文列表
{papers_block}

## 排序任务
{task_line}

{dims}

## 输出格式
JSON 格式：
```json
{{"rankings": [
  {{"rank": 1, "paper_index": 1, "fine_score": 9.2, "reason": "排序理由：..."}},
  ...
]}}
```

要求（必须遵守）：
1. paper_index 对应序号 1-{len(papers)}，不得虚构论文
2. fine_score 范围 0-10，保留一位小数
3. reason 用中文 2-3 句话简洁说明
4. 只输出 JSON，无其他内容
5. 仅依据上方列表判断，不得编造不存在的论文、会议或结果
"""