From 2ff331a33819ef7abd8e032023a6cdb3941a4991 Mon Sep 17 00:00:00 2001 From: adarshh347 Date: Tue, 27 Oct 2026 18:12:59 +0530 Subject: [PATCH] fix(daily-paper): recover shorter notes and report partial briefs --- plugins/daily_paper/README.md | 17 ++++ plugins/daily_paper/README_ZH.md | 15 ++++ .../src/reme_daily_paper/analyze.py | 38 +++++--- .../src/reme_daily_paper/analyze.yaml | 6 ++ .../src/reme_daily_paper/collect.py | 18 +++- .../src/reme_daily_paper/digest.py | 20 ++++- .../src/reme_daily_paper/digest.yaml | 16 ++-- plugins/daily_paper/tests/test_daily_paper.py | 47 ++++++++++ .../tests/test_structured_recovery.py | 86 +++++++++++++++++++ 9 files changed, 240 insertions(+), 23 deletions(-) create mode 100644 plugins/daily_paper/tests/test_structured_recovery.py diff --git a/plugins/daily_paper/README.md b/plugins/daily_paper/README.md index 2f60ca25..970ece76 100644 --- a/plugins/daily_paper/README.md +++ b/plugins/daily_paper/README.md @@ -86,6 +86,23 @@ three source notes, and rebuilds the daily index. The workflow then runs `auto_t frontmatter of all three analyses and the final brief. The normal background file watcher observes those source-file changes and refreshes derived indexes. +## Partial results and recovery + +Each selected paper is analyzed independently. An ordinary failure is recorded and later papers are still attempted; +cancellation propagates. Invalid structured output gets one additional request asking for a shorter note. This is a +prompt-level size target, not a provider output-token guarantee; the SDK may try its own structured-output strategies +within each request. Downloads and PDF extraction are not repeated by this recovery. + +With one or two successful notes, the workflow writes a partial brief using only those notes and lists the failed +arXiv IDs. A generated partial brief has `success=true` and `metadata.partial=true`; `analysis_count`, `failure_count`, +`failed_papers`, `note_paths`, and `pdf_paths` report the outcome. An all-failed batch has `success=false` and makes no +brief-generation call. If brief generation itself fails, completed note paths remain in response metadata. + +Saved partial briefs follow the existing skip policy. A later call reports `partial=true` and the saved failed IDs; +use `reme daily_paper date=YYYY-MM-DD force=true` to regenerate. This starts the usual selection pipeline again; it is +not a failed-paper-only resume. Hosts should inspect `partial` rather than treating `success=true` as three successful +analyses. The final tagging step can still report its own failure after a brief is written. + ## Parameters | Parameter | Default | Purpose | diff --git a/plugins/daily_paper/README_ZH.md b/plugins/daily_paper/README_ZH.md index cb809870..51d57f0a 100644 --- a/plugins/daily_paper/README_ZH.md +++ b/plugins/daily_paper/README_ZH.md @@ -121,3 +121,18 @@ python -m pytest plugins/daily_paper -v ``` 单元测试 mock Hugging Face、arXiv 和 AgentScope 边界，不访问外部服务。 + + +## 部分结果与恢复 + +每篇论文独立处理，普通错误会被记录，后续论文仍会继续；任务取消会向上传播。结构化输出无效时， +最多额外请求一次更短的笔记，不重复下载和 PDF 提取。这只是提示词中的长度目标，不是服务端输出 token +限制；每次请求内部仍可能使用 SDK 的结构化输出策略回退。 + +成功一篇或两篇时，仅根据成功的笔记生成部分简报，并列出失败的 arXiv ID。此时 `success=true`， +`metadata.partial=true`；`analysis_count`、`failure_count`、`failed_papers`、`note_paths` 和 `pdf_paths` +提供结果明细。全部失败时 `success=false`，不调用简报模型。若简报本身生成失败，已完成笔记的路径仍保留在响应元数据中。 + +已保存的部分简报沿用跳过策略；再次调用会返回 `partial=true` 和保存的失败 ID。使用 +`reme daily_paper date=YYYY-MM-DD force=true` 重新生成，会重新运行正常选题流程，而非仅恢复失败论文。 +调用方应检查 `partial`，不要把 `success=true` 理解为三篇论文全部成功。最终自动标签步骤仍可能报告自己的错误。 diff --git a/plugins/daily_paper/src/reme_daily_paper/analyze.py b/plugins/daily_paper/src/reme_daily_paper/analyze.py index 5fc10ecb..78dc196b 100644 --- a/plugins/daily_paper/src/reme_daily_paper/analyze.py +++ b/plugins/daily_paper/src/reme_daily_paper/analyze.py @@ -4,6 +4,8 @@ import asyncio import json from pathlib import Path +from pydantic import ValidationError + from .arxiv import ArxivPdfClient from .base import ( PAPER_COUNT, @@ -99,19 +101,33 @@ class DailyPaperAnalyzeStep(DailyPaperStep): f"chars={len(pdf_text)} truncated={truncated}", ) self.logger.info(f"[{self.name}] agent start arxiv_id={paper.arxiv_id}") - result = await self.agent_wrapper.reply( - self.prompt_format( - "analyze_user", - paper_info=json.dumps(paper.model_dump(), ensure_ascii=False, indent=2), - selection_reason=selected.reasoning, - page_count=page_count, - truncated=str(truncated).lower(), - pdf_text=pdf_text, - ), - output_schema=DailyPaperMarkdownOutput, + prompt = self.prompt_format( + "analyze_user", + paper_info=json.dumps(paper.model_dump(), ensure_ascii=False, indent=2), + selection_reason=selected.reasoning, + page_count=page_count, + truncated=str(truncated).lower(), + pdf_text=pdf_text, ) + # AgentScope is optional: other wrappers can report Pydantic validation failures. + recoverable = (ValidationError,) + try: + from agentscope.exception import StructuredOutputError + except ImportError: + pass + else: + recoverable += (StructuredOutputError,) + for attempt in range(2): + try: + result = await self.agent_wrapper.reply(prompt, output_schema=DailyPaperMarkdownOutput) + output = structured_output(result, DailyPaperMarkdownOutput) + break + except recoverable: + if attempt: + raise + self.logger.warning(f"[{self.name}] retry shorter structured note arxiv_id={paper.arxiv_id}") + prompt += "\n\n" + self.prompt_format("analyze_short_retry") self.logger.info(f"[{self.name}] agent done arxiv_id={paper.arxiv_id}") - output = structured_output(result, DailyPaperMarkdownOutput) title = normalize_chinese_title(output.title, f"论文解读-{paper.arxiv_id}") day_dir = self.workspace_path / daily_dir / day existing_note = self._find_existing_note(day_dir, paper.arxiv_id) diff --git a/plugins/daily_paper/src/reme_daily_paper/analyze.yaml b/plugins/daily_paper/src/reme_daily_paper/analyze.yaml index 65b869f4..28f411f7 100644 --- a/plugins/daily_paper/src/reme_daily_paper/analyze.yaml +++ b/plugins/daily_paper/src/reme_daily_paper/analyze.yaml @@ -31,3 +31,9 @@ analyze_user: | # PDF 分页文本 {pdf_text} + +analyze_short_retry: | + 上一次结构化输出无效或被截断。请重新生成更短的完整笔记：body 不超过 800 个中文字符， + desc 不超过 100 个中文字符，保留 title、desc、body 三个字段。 + 只保留核心问题、主要方法、一个有依据的结果及局限；省略冗长背景和重复内容。 + 不得补充输入材料之外的事实。优先确保 JSON 完整，不要重复上一次的长篇输出。 diff --git a/plugins/daily_paper/src/reme_daily_paper/collect.py b/plugins/daily_paper/src/reme_daily_paper/collect.py index e6d5832e..3f25abf3 100644 --- a/plugins/daily_paper/src/reme_daily_paper/collect.py +++ b/plugins/daily_paper/src/reme_daily_paper/collect.py @@ -4,6 +4,8 @@ import asyncio import datetime as dt from pathlib import Path +import frontmatter + from .arxiv import ARXIV_ID_PATTERN from .base import DailyPaperStep, iter_note_metadata, now from .huggingface_papers import HuggingFacePapersClient @@ -101,9 +103,21 @@ class DailyPaperCollectStep(DailyPaperStep): self._set_state("skip", True) self._set_state("digest_path", digest_rel) self.context.response.success = True - self.context.response.answer = f"Skipped: daily paper brief already exists at {digest_rel}" + saved = frontmatter.load(digest_path) + partial = bool(saved.metadata.get("partial", False)) + qualifier = "partial " if partial else "" + self.context.response.answer = ( + f"Skipped: {qualifier}daily paper brief already exists at {digest_rel}" + + ("; use force=true to regenerate" if partial else "") + ) self.context.response.metadata.update( - {"date": day, "digest_path": digest_rel, "skipped": True}, + { + "date": day, + "digest_path": digest_rel, + "skipped": True, + "partial": partial, + "failed_arxiv_ids": saved.metadata.get("failed_arxiv_ids", []), + }, ) self.logger.info(f"[{self.name}] skip existing digest path={digest_rel}") return self.context.response diff --git a/plugins/daily_paper/src/reme_daily_paper/digest.py b/plugins/daily_paper/src/reme_daily_paper/digest.py index 09484b37..6e0a80bd 100644 --- a/plugins/daily_paper/src/reme_daily_paper/digest.py +++ b/plugins/daily_paper/src/reme_daily_paper/digest.py @@ -72,7 +72,13 @@ class DailyPaperDigestStep(DailyPaperStep): if self.agent_wrapper is None: raise RuntimeError("An agent_wrapper is required for the daily brief") analyses: list[AnalyzedPaper] = self._state("analyses") or [] - if len(analyses) != PAPER_COUNT: + failures = self._state("failures") or [] + if not analyses and failures: + self.context.response.success = False + self.context.response.answer = "Daily paper analysis failed for every selected paper; no brief generated" + self.context.response.metadata["partial"] = False + return self.context.response + if not analyses or len(analyses) + len(failures) != PAPER_COUNT: raise RuntimeError( "Detailed paper notes are missing before digest generation", ) @@ -88,6 +94,7 @@ class DailyPaperDigestStep(DailyPaperStep): "digest_user", documents=json.dumps(documents, ensure_ascii=False, indent=2), daily_dir=daily_dir, + paper_count=len(analyses), ), output_schema=DailyPaperMarkdownOutput, job_tools=list(self.kwargs.get("job_tools") or []), @@ -121,6 +128,8 @@ class DailyPaperDigestStep(DailyPaperStep): body = self._validate_historical_wikilinks(body, run_day, digest_path) body += "\n\n## 详细论文\n\n" + "\n".join(f"- {link}" for link in wikilinks) selected_ids = [item.arxiv_id for item in analyses] + if failures: + body += "\n\n## 未完成的论文\n\n" + "\n".join(f"- {item['arxiv_id']}" for item in failures) await write_markdown( digest_path, body, @@ -131,6 +140,8 @@ class DailyPaperDigestStep(DailyPaperStep): "kind": "daily-paper-brief", "date": day, "arxiv_ids": selected_ids, + "partial": bool(failures), + "failed_arxiv_ids": [item["arxiv_id"] for item in failures], "selection_reasoning": [item.reasoning for item in analyses], "source_notes": wikilinks, "generated_at": utc_now_iso(), @@ -156,10 +167,15 @@ class DailyPaperDigestStep(DailyPaperStep): self.logger.info(f"[{self.name}] refresh index done date={day}") self.context.response.success = True - self.context.response.answer = f"Generated daily paper brief: {digest_rel}" + qualifier = f"partial ({len(analyses)}/{PAPER_COUNT} papers) " if failures else "" + self.context.response.answer = f"Generated {qualifier}daily paper brief: {digest_rel}" self.context.response.metadata.update( { "date": day, + "partial": bool(failures), + "analysis_count": len(analyses), + "failure_count": len(failures), + "failed_papers": list(failures), "week": self._state("week"), "month": self._state("month"), "selection_reasoning": [item.reasoning for item in analyses], diff --git a/plugins/daily_paper/src/reme_daily_paper/digest.yaml b/plugins/daily_paper/src/reme_daily_paper/digest.yaml index dd6c3b84..fad7d1fa 100644 --- a/plugins/daily_paper/src/reme_daily_paper/digest.yaml +++ b/plugins/daily_paper/src/reme_daily_paper/digest.yaml @@ -1,22 +1,22 @@ # Prompt bundled with the Daily Paper plugin distribution. digest_user: | - 你是中文 AI 研究资讯主编。请忠实综合下面三篇详细论文解读，生成一篇普通读者五分钟可以读懂的每日论文速读。 + 你是中文 AI 研究资讯主编。请忠实综合下面{paper_count} 篇详细论文解读，生成一篇普通读者五分钟可以读懂的每日论文速读。 内容只能依据输入文档，不得补充文档中没有提供的事实。 - 保留技术准确性，同时解释三篇论文为什么值得关注，以及它们之间有什么联系。 + 保留技术准确性，同时解释{paper_count} 篇论文为什么值得关注，以及它们之间有什么联系。 - 下面提供的三篇详细论文解读是本次写作的主要输入，首要任务是基于它们生成每日论文速读。 - 历史搜索只用于帮助构建 Wikilink，最多调用两次 `search`：围绕三篇论文共同的核心问题、方法、关键词和 + 下面提供的{paper_count} 篇详细论文解读是本次写作的主要输入，首要任务是基于它们生成每日论文速读。 + 历史搜索只用于帮助构建 Wikilink，最多调用两次 `search`：围绕{paper_count} 篇论文共同的核心问题、方法、关键词和 同义表达组织一次综合查询，搜索结果不必局限于 `{daily_dir}/`。只有 `{daily_dir}/` 下日期早于今天、 且与本期内容确实相似或互补的 Markdown 文章才可作为正文中的历史链接候选；必要时调用 `read` 核验全文， 不要仅凭标题判断。如果没有搜索到合适的历史文章，说明已有记忆中暂无相似文章，不添加 Wikilink 即可， - 仍须基于三篇论文解读完成每日论文速读；不得将搜索无结果误判为论文输入缺失，也不得要求用户重新提供文档。 + 仍须基于{paper_count} 篇论文解读完成每日论文速读；不得将搜索无结果误判为论文输入缺失，也不得要求用户重新提供文档。 将确认相关的旧文章以 Wikilink 自然织入正文，并用句子说明关联（延续、对比、补充或方法相似）； 链接必须采用带 `.md` 的完整 workspace-relative 路径，例如 `[[{daily_dir}/2026-07-01/旧文章.md|此前的相关解读]]`。不要输出裸链接、独立关系字段，也不要虚构搜索未命中的路径。 旧文章只用于判断关联和建立链接，不得用来补充本期事实。如果没有真正相关的旧文章，不要强行添加； - 当日三篇详细解读的链接会由系统统一附在文末。 + 当日{paper_count} 篇详细解读的链接会由系统统一附在文末。 - # 三篇论文解读 + # {paper_count} 篇论文解读 {documents} @@ -24,4 +24,4 @@ digest_user: | - title 必须是简洁准确的中文标题，不要包含 Markdown 标题符号、路径或 .md 后缀。 - 只返回结构化结果中的 title、desc 和 body；body 不要包含 YAML frontmatter，也不要重复一级标题。 - 总阅读时长约五分钟。 - - 包含“今日一句话”“每篇一分钟读懂”“三篇论文之间的联系”“今天最值得关注什么”等部分。 + - 包含“今日一句话”“每篇一分钟读懂”“今天最值得关注什么”等部分。多篇论文时比较它们的联系；只有一篇时不要虚构比较对象。 diff --git a/plugins/daily_paper/tests/test_daily_paper.py b/plugins/daily_paper/tests/test_daily_paper.py index 77c83dc2..ceb8a786 100644 --- a/plugins/daily_paper/tests/test_daily_paper.py +++ b/plugins/daily_paper/tests/test_daily_paper.py @@ -623,6 +623,7 @@ def test_digest_prompt_uses_configured_daily_directory(tmp_path: Path): prompt = step.prompt_format( "digest_user", documents="[]", + paper_count=3, daily_dir=str(step.config_value("daily_dir")).strip("/"), ) @@ -1079,3 +1080,49 @@ async def test_analysis_does_not_swallow_cancellation(tmp_path, monkeypatch): await step(context) assert analyze_one.await_count == 1 assert context["daily_paper_failures"] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("count", [0, 1, 2, 3]) +async def test_digest_reports_partial_and_total_failure(tmp_path, count): + """Generate a useful brief from any successes and avoid a model call on total failure.""" + analyses = [ + AnalyzedPaper( + arxiv_id=f"2607.1000{i}", + reasoning="test", + title=f"论文{i}", + desc="test", + body="evidence", + note_path=f"daily/2026-07-21/论文{i}.md", + pdf_path=f"resource/papers/{i}.pdf", + ) + for i in range(count) + ] + failures = [{"arxiv_id": f"2607.1000{i}", "error": "invalid output"} for i in range(count, 3)] + context = RuntimeContext( + daily_paper_run_date="2026-07-21", + daily_paper_analyses=analyses, + daily_paper_failures=failures, + ) + agent = _QueuedAgentWrapper([{"title": "今日简报", "desc": "summary", "body": "Grounded brief"}]) + await DailyPaperDigestStep( + app_context=ApplicationContext(workspace_dir=str(tmp_path)), + agent_wrapper=agent, + )(context) + assert context.response.success == bool(count) + if not count: + assert not agent.calls + assert not list(tmp_path.rglob("*.md")) + return + assert context.response.metadata["partial"] == (count < 3) + assert context.response.metadata["failure_count"] == 3 - count + assert context.response.metadata["note_paths"] == [item.note_path for item in analyses] + brief = frontmatter.load(tmp_path / context.response.metadata["digest_path"]) + assert brief.metadata["failed_arxiv_ids"] == [item["arxiv_id"] for item in failures] + assert all(item["arxiv_id"] in brief.content for item in failures) + assert f"{count} 篇" in agent.calls[0]["inputs"] + if count < 3: + rerun = RuntimeContext(date="2026-07-21") + await DailyPaperCollectStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))(rerun) + assert rerun.response.metadata["partial"] is True + assert "force=true" in rerun.response.answer diff --git a/plugins/daily_paper/tests/test_structured_recovery.py b/plugins/daily_paper/tests/test_structured_recovery.py new file mode 100644 index 00000000..139ba205 --- /dev/null +++ b/plugins/daily_paper/tests/test_structured_recovery.py @@ -0,0 +1,86 @@ +"""Offline recovery tests through the actual AgentScope structured-output parser.""" + +# pylint: disable=protected-access + +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from pydantic import BaseModel + +from reme_daily_paper.analyze import DailyPaperAnalyzeStep +from reme_daily_paper.schema import PaperInfo, PaperPick +from reme.components import ApplicationContext, RuntimeContext +from reme.components.agent_wrapper import AsAgentWrapper + +# The plugin also supports wrappers without the optional AgentScope installation. +ChatModelBase = pytest.importorskip("agentscope.model").ChatModelBase +ChatResponse = pytest.importorskip("agentscope.model").ChatResponse +ChatUsage = pytest.importorskip("agentscope.model").ChatUsage +OpenAIChatFormatter = pytest.importorskip("agentscope.formatter").OpenAIChatFormatter +TextBlock = pytest.importorskip("agentscope.message").TextBlock +ToolCallBlock = pytest.importorskip("agentscope.message").ToolCallBlock + + +class _Parameters(BaseModel): + """No provider parameters are needed for this offline model.""" + + +class _TruncatedModel(ChatModelBase): + """Emit malformed tool arguments until the plugin asks for a shorter note.""" + + def __init__(self, recover): + super().__init__(credential=None, model="offline", parameters=_Parameters(), stream=False) + self.formatter = OpenAIChatFormatter() + self.recover = recover + self.analysis_calls = 0 + self.structured_calls = 0 + + async def _call_api(self, *_args, **kwargs): + """Replace provider I/O while retaining real agent and parser behavior.""" + tools = kwargs.get("tools") or [] + structured = any(item.get("function", {}).get("name") == "generate_structured_output" for item in tools) + if structured: + self.structured_calls += 1 + shortened = "800" in str(kwargs.get("messages")) + raw = json.dumps({"title": "论文摘要", "desc": "summary", "body": "Supported result"}) + if not (self.recover and shortened): + raw = '{"body":"' + "unfinished note " * 2000 + content = [ToolCallBlock(id="offline-call", name="generate_structured_output", input=raw)] + else: + self.analysis_calls += 1 + content = [TextBlock(text="Supported analysis")] + return ChatResponse( + content=content, + is_last=True, + usage=ChatUsage(input_tokens=10, output_tokens=10, time=0.01), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("recover", [True, False]) +async def test_shorter_retry_uses_real_structured_parser(tmp_path, monkeypatch, recover): + """A shorter second request recovers or fails without starting a third attempt.""" + app = ApplicationContext(workspace_dir=str(tmp_path)) + model = _TruncatedModel(recover) + agent = AsAgentWrapper(app_context=app, as_llm="") + agent.as_llm = SimpleNamespace(model=model) + step = DailyPaperAnalyzeStep(app_context=app, agent_wrapper=agent) + step.context = RuntimeContext(daily_paper_run_date="2026-07-21") + monkeypatch.setattr(step, "_extract_pdf_text_sync", lambda *_: ("Supported result [p. 1]", 1, False)) + downloader = SimpleNamespace(download=AsyncMock()) + paper = PaperInfo(arxiv_id="2607.10001") + pick = PaperPick(arxiv_id=paper.arxiv_id, reasoning="test") + if recover: + result = await step._analyze_one(downloader, paper, pick, set()) + assert (tmp_path / result.note_path).is_file() + else: + from agentscope.exception import StructuredOutputError + + with pytest.raises(StructuredOutputError): + await step._analyze_one(downloader, paper, pick, set()) + assert not list(tmp_path.rglob("*.md")) + assert model.analysis_calls == 2 + assert model.structured_calls >= 2 + assert downloader.download.await_count == 1