fix(daily-paper): recover shorter notes and report partial briefs

This commit is contained in:
adarshh347 2026-10-27 18:12:59 +05:30
parent a7b9166c0e
commit 2ff331a338
9 changed files with 240 additions and 23 deletions

View file

@ -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 |

View file

@ -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` 理解为三篇论文全部成功。最终自动标签步骤仍可能报告自己的错误。

View file

@ -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)

View file

@ -31,3 +31,9 @@ analyze_user: |
# PDF 分页文本
{pdf_text}
analyze_short_retry: |
上一次结构化输出无效或被截断。请重新生成更短的完整笔记:body 不超过 800 个中文字符,
desc 不超过 100 个中文字符,保留 title、desc、body 三个字段。
只保留核心问题、主要方法、一个有依据的结果及局限;省略冗长背景和重复内容。
不得补充输入材料之外的事实。优先确保 JSON 完整,不要重复上一次的长篇输出。

View file

@ -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

View file

@ -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],

View file

@ -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,也不要重复一级标题。
- 总阅读时长约五分钟。
- 包含“今日一句话”“每篇一分钟读懂”“三篇论文之间的联系”“今天最值得关注什么”等部分。
- 包含“今日一句话”“每篇一分钟读懂”“今天最值得关注什么”等部分。多篇论文时比较它们的联系;只有一篇时不要虚构比较对象。

View file

@ -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

View file

@ -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