hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/reader/paper_reader_service.py

"""论文阅读服务 —— PDF 正文抽取、AI 导读生成、上下文对话与参考文献辅助."""

from __future__ import annotations

import logging
from typing import Any
from collections.abc import Iterable

from fastapi import BackgroundTasks, HTTPException
from starlette.concurrency import run_in_threadpool

from ...agents import get_paper_analysis_agent
from ...agents.support.reader_reference_lookup_tool import READER_RELATED_FROM_BIBLIOGRAPHY, READER_RELATED_FROM_PRE_SEARCH
from ...utils.common import suppress_exceptions_async

logger = logging.getLogger(__name__)

_OPENING_PROMPT = (
    "请用中文写一段不超过 380 字的导读：研究问题、核心方法、实验与结论的阅读要点。"
    "仅依据当前提供的摘要与摘录组织表述；勿单列「不确定处」「局限」或待查清单（用户追问时再说明材料范围即可）。"
)
_NO_HISTORY_PLACEHOLDER = "（尚无对话历史）"

def _update_memory_from_turn(*, store: Any, paper_id: int, user_message: str, assistant_reply: str) -> None:
    um = (user_message or "").strip()
    if not um:
        return
    store.add(scope="paper", paper_id=paper_id, kind="working", content=f"用户问：{um[:220]}", importance=0.5)
    try:
        store.extract_memory_via_llm(paper_id, um, assistant_reply)
    except Exception:
        logger.debug("extract_memory_via_llm failed for paper %s", paper_id, exc_info=True)

class PaperReaderService:
    def __init__(self, db: Any, agent: Any | None = None) -> None:
        self._db = db
        self._agent = agent or get_paper_analysis_agent()

    @property
    def db(self) -> Any:
        return self._db

    @staticmethod
    def _format_reader_history(turns: Iterable[Any]) -> str:
        lines: list[str] = []
        tail = list(turns or [])[-24:]
        for t in tail:
            role = (getattr(t, "role", None) or "").strip().lower()
            content = (getattr(t, "content", None) or "").strip()
            if not content:
                continue
            if role not in ("user", "assistant"):
                role = "user"
            label = "用户" if role == "user" else "助手"
            lines.append(f"{label}：{content}")
        return "\n\n".join(lines)

    async def _build_reader_context(self, paper_id: int, user_message: str = "") -> tuple[Any, str, str]:
        from ..memory.memory_store import MemoryStore
        from .paper_reader_context import build_reader_context_for_paper

        paper, base_ctx, pdf_ref_text, pdf_parsing = await run_in_threadpool(build_reader_context_for_paper, self._db, paper_id)
        if not paper:
            raise HTTPException(status_code=404, detail="文献不存在")

        mem = await run_in_threadpool(
            MemoryStore(self._db.db_path).build_context_block,
            paper_id=paper_id,
        )
        title_hint = str(getattr(paper, "title", None) or "")
        ctx = (base_ctx + ("\n\n" + mem if mem else "")).strip()
        return paper, ctx, title_hint, pdf_ref_text, pdf_parsing

    def _schedule_pdf_excerpt(self, paper_id: int, ctx: str, background_tasks: BackgroundTasks) -> None:
        from .paper_reader_context import compute_and_cache_excerpt

        try:
            pdf_path = self._db.get_library_pdf_abspath(paper_id)
            if pdf_path and "【PDF 正文摘录" not in ctx:
                background_tasks.add_task(compute_and_cache_excerpt, self._db.db_path, paper_id, pdf_path)
        except Exception as exc:
            logger.debug("paper_reader.schedule_pdf_excerpt_failed", extra={"paper_id": paper_id}, exc_info=exc)

    @suppress_exceptions_async(default_return=None, log_level="warning", log_message="paper_reader.ensure_opening_turn_failed")
    async def _ensure_opening_turn_safe(self, *, paper_id: int, opening_text: str) -> None:
        from .paper_reader_history import ensure_opening_turn

        await run_in_threadpool(
            ensure_opening_turn,
            self._db.db_path,
            paper_id=int(paper_id),
            opening_text=opening_text,
        )

    @suppress_exceptions_async(default_return=None, log_level="warning", log_message="paper_reader.append_history_failed")
    async def _append_history(self, *, paper_id: int, user_message: str, reply: str) -> None:
        from .paper_reader_history import append_turn

        await run_in_threadpool(
            append_turn,
            self._db.db_path,
            paper_id=int(paper_id),
            role="user",
            content=user_message,
        )
        await run_in_threadpool(
            append_turn,
            self._db.db_path,
            paper_id=int(paper_id),
            role="assistant",
            content=reply,
        )

    @suppress_exceptions_async(default_return=None, log_level="warning", log_message="paper_reader.memory_update_failed")
    async def _update_memory(self, *, store: Any, paper_id: int, user_message: str, reply: str) -> None:
        await run_in_threadpool(
            _update_memory_from_turn,
            store=store, paper_id=paper_id,
            user_message=user_message, assistant_reply=reply,
        )

    async def get_opening(self, *, paper_id: int, background_tasks: BackgroundTasks) -> dict:
        from .reader_opening_cache import get_cached_opening, set_cached_opening

        from .paper_reader_context import build_reader_snap

        paper, ctx, title_hint, pdf_ref_text, pdf_parsing = await self._build_reader_context(paper_id)

        reader_snap = build_reader_snap(paper, pdf_text_for_references=pdf_ref_text)
        try:
            pdf_path = self._db.get_library_pdf_abspath(paper_id)
            if pdf_path:
                reader_snap["_pdf_abspath"] = pdf_path
        except Exception:
            pass
        if pdf_parsing:
            self._schedule_pdf_excerpt(paper_id, ctx, background_tasks)

        cached, fresh = await run_in_threadpool(get_cached_opening, self._db.db_path, paper_id, 72)
        if cached and fresh:
            op = cached.strip()
            await self._ensure_opening_turn_safe(paper_id=paper_id, opening_text=op)
            return {"opening": op, "pdf_parsing": pdf_parsing}

        if cached and not fresh:
            def _refresh() -> None:
                try:
                    opening2, _, _ = self._agent.paper_reader_reply(
                        ctx, _NO_HISTORY_PLACEHOLDER, _OPENING_PROMPT, reader_snap
                    )
                    set_cached_opening(self._db.db_path, paper_id, opening2.strip())
                except Exception as exc:
                    logger.warning("paper_reader.opening_refresh_failed", extra={"paper_id": paper_id}, exc_info=exc)

            background_tasks.add_task(_refresh)
            op = cached.strip()
            await self._ensure_opening_turn_safe(paper_id=paper_id, opening_text=op)
            return {"opening": op, "pdf_parsing": pdf_parsing}

        opening, _, _ = await run_in_threadpool(
            lambda: self._agent.paper_reader_reply(ctx, _NO_HISTORY_PLACEHOLDER, _OPENING_PROMPT, reader_snap)
        )
        op = opening.strip()
        await run_in_threadpool(set_cached_opening, self._db.db_path, paper_id, op)
        await self._ensure_opening_turn_safe(paper_id=paper_id, opening_text=op)
        return {"opening": op, "pdf_parsing": pdf_parsing}

    async def process_chat(
        self,
        *,
        paper_id: int,
        messages: list[Any],
        user_message: str,
        background_tasks: BackgroundTasks,
    ) -> dict[str, Any]:
        from ..memory.memory_store import MemoryStore

        paper, ctx, title_hint, pdf_ref_text, pdf_parsing = await self._build_reader_context(paper_id, user_message)
        from .paper_reader_context import build_reader_snap

        reader_snap = build_reader_snap(paper, pdf_text_for_references=pdf_ref_text)
        try:
            pdf_path = self._db.get_library_pdf_abspath(paper_id)
            if pdf_path:
                reader_snap["_pdf_abspath"] = pdf_path
        except Exception:
            pass
        self._schedule_pdf_excerpt(paper_id, ctx, background_tasks)

        store = MemoryStore(self._db.db_path)
        rel_mem = await run_in_threadpool(
            store.get_context_for_query,
            paper_id=int(paper_id),
            query=user_message,
            limit=6,
        )
        if rel_mem:
            ctx += "\n\n" + rel_mem
        hist = self._format_reader_history(messages)

        reply, related_papers, related_sources = await run_in_threadpool(
            lambda: self._agent.paper_reader_reply(ctx, hist, user_message, reader_snap)
        )
        rs = list(related_sources or [])
        related_hints: list[dict[str, Any]] = [
            {
                "ref_idx": i,
                "title": getattr(p, "title", None),
                "reason": (
                    "来自当前文献参考文献题录（OpenAlex 解析）"
                    if i - 1 < len(rs) and rs[i - 1] == READER_RELATED_FROM_BIBLIOGRAPHY
                    else "基于论文主题相似度匹配"
                    if i - 1 < len(rs) and rs[i - 1] == READER_RELATED_FROM_PRE_SEARCH
                    else "来自用户给定英文短语或外部题名检索（OpenAlex）"
                ),
            }
            for i, p in enumerate(related_papers or [], start=1)
        ]

        await self._append_history(paper_id=paper_id, user_message=user_message, reply=reply)
        await self._update_memory(store=store, paper_id=paper_id, user_message=user_message, reply=reply)
        background_tasks.add_task(store.compress_working, scope="paper", paper_id=int(paper_id), min_entries=6)

        return {
            "reply": reply.strip(),
            "pdf_parsing": pdf_parsing,
            "related_papers": related_papers,
            "related_hints": related_hints,
            "kg_edges": [],
        }

    async def get_history(self, *, paper_id: int, limit: int) -> list[dict[str, Any]]:
        from .paper_reader_history import list_turns

        paper = await run_in_threadpool(self._db.get_paper_by_id, int(paper_id))
        if not paper:
            raise HTTPException(status_code=404, detail="文献不存在")
        return await run_in_threadpool(
            list_turns,
            self._db.db_path,
            paper_id=int(paper_id),
            limit=int(limit),
        )