实战 5 · ★★★★★ 自动追踪目标算法论文的科研 agent

目标

做一个 agent，每周自动：

1. 在 arxiv 搜指定关键词的新论文
2. 用 Semantic Scholar 补引用数
3. 下载 PDF，解析前 3 页
4. 用 Claude 生成中文摘要
5. 按相关度 + 引用数排序
6. 邮件推送 Top 10

架构

关键词 → schedule 周触发
         ↓
    arXiv 搜索 → 下载 PDF → PyMuPDF 抽前 3 页
         ↓                          ↓
    Semantic Scholar          Claude 生成中文摘要
    补引用数排序                    ↓
         ↓ ←───────────────────────┘
    Resend 邮件推送 Top10

准备

pip install arxiv semanticscholar pymupdf anthropic resend schedule python-dotenv

.env：

ANTHROPIC_API_KEY=sk-ant-...
RESEND_API_KEY=re_...
EMAIL_FROM=agent@yourdomain.com
EMAIL_TO=you@example.com

代码

import os
import schedule
import time
import anthropic
import resend
import arxiv
import fitz  # PyMuPDF
from semanticscholar import SemanticScholar
from dotenv import load_dotenv
from pathlib import Path

load_dotenv()
client = anthropic.Anthropic()
resend.api_key = os.getenv("RESEND_API_KEY")
sch = SemanticScholar()

KEYWORDS = ["visual tracking", "object tracking", "SOT"]
MAX_RESULTS = 20
PAPERS_DIR = Path("papers")
PAPERS_DIR.mkdir(exist_ok=True)

def search_arxiv(keyword, max_results=10):
    """搜 arxiv，返回论文列表。"""
    search = arxiv.Search(
        query=keyword,
        max_results=max_results,
        sort_by=arxiv.SortCriterion.SubmittedDate
    )
    return list(arxiv.Client().results(search))

def get_citations(title):
    """从 Semantic Scholar 查引用数。"""
    try:
        results = sch.search_paper(title, limit=1, fields=["citationCount"])
        if results:
            return results[0].citationCount or 0
    except Exception:
        pass
    return 0

def download_and_extract(paper, max_pages=3):
    """下载 PDF，抽前 N 页文字。"""
    pdf_path = PAPERS_DIR / f"{paper.entry_id.split('/')[-1]}.pdf"
    paper.download_pdf(dirpath=str(PAPERS_DIR), filename=pdf_path.name)
    doc = fitz.open(pdf_path)
    text = "\n".join(page.get_text() for page in doc[:max_pages])
    doc.close()
    return text[:8000]  # 截断防爆 context

def summarize_zh(text, title):
    """Claude 生成中文摘要。"""
    resp = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=512,
        messages=[{"role": "user", "content": f"""用中文 200 字总结这篇论文的核心贡献、方法、实验结果：

标题：{title}
内容：{text}
"""}]
    )
    return resp.content[0].text

def send_email(top_papers):
    """Resend 发邮件。"""
    html = "<h2>本周论文 Top 10</h2><ol>"
    for p in top_papers:
        html += f"<li><b>{p['title']}</b> (引用: {p['citations']})<br>{p['summary']}<br><a href='{p['url']}'>原文</a></li>"
    html += "</ol>"
    resend.Emails.send({
        "from": os.getenv("EMAIL_FROM"),
        "to": os.getenv("EMAIL_TO"),
        "subject": "本周 arxiv 论文 Top 10",
        "html": html
    })

def weekly_job():
    """每周跑一次的主任务。"""
    all_papers = []
    for kw in KEYWORDS:
        all_papers.extend(search_arxiv(kw, max_results=MAX_RESULTS))
    
    # 去重
    seen = set()
    unique = []
    for p in all_papers:
        if p.title not in seen:
            seen.add(p.title)
            unique.append(p)
    
    # 处理前 30 篇
    processed = []
    for paper in unique[:30]:
        try:
            text = download_and_extract(paper)
            summary = summarize_zh(text, paper.title)
            citations = get_citations(paper.title)
            processed.append({
                "title": paper.title,
                "url": paper.entry_id,
                "summary": summary,
                "citations": citations
            })
        except Exception as e:
            print(f"处理 {paper.title} 失败: {e}")
    
    # 按引用数排序，取 Top 10
    processed.sort(key=lambda x: x["citations"], reverse=True)
    top10 = processed[:10]
    
    send_email(top10)
    print(f"已发送 {len(top10)} 篇论文摘要")

# 每周一早上 9 点跑
schedule.every().monday.at("09:00").do(weekly_job)

if __name__ == "__main__":
    print("Paper tracker 已启动，每周一 9:00 推送")
    # weekly_job()  # 立即跑一次测试
    while True:
        schedule.run_pending()
        time.sleep(60)

部署

本地常驻：python paper_tracker.py

服务器：用 nohup 或 systemd 跑成服务。

云端（推荐）：用 GitHub Actions 定时跑，不用维护服务器。.github/workflows/weekly-papers.yml：

name: Weekly Paper Tracker
on:
  schedule:
    - cron: '0 1 * * 1'  # 每周一 UTC 1:00（北京 9:00）
jobs:
  run:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r requirements.txt
      - run: python paper_tracker.py
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          RESEND_API_KEY: ${{ secrets.RESEND_API_KEY }}
          EMAIL_FROM: ${{ secrets.EMAIL_FROM }}
          EMAIL_TO: ${{ secrets.EMAIL_TO }}

安全提醒

• arxiv API 限流 1 req/3s，本代码已通过 arxiv 库自动限速
• Resend 免费额度 100 封/天，足够周更
• PDF 解析大文件可能爆 context，text[:8000] 截断保护
• API key 全部走环境变量，不要入库

练习

1. 加关键词管理（用 JSON 文件存，方便增删）
2. 加历史去重（记录已推送的论文 ID，不重复推）
3. 加相关度评分（用 embedding 算关键词和论文摘要的相似度）
4. 改成 Slack/飞书 推送