Spaces:
Sleeping
Sleeping
A newer version of the Gradio SDK is available: 6.26.0
ResearchRadar 项目任务书
读者：甲、乙（研一，均为 Python 入门水平） 用途：第一周演示汇报 + 后续 6 周开发的执行依据 汇报对象：导师 + 投资人 | 时长：6 分钟
一、项目定义
1.1 我们要解决什么问题
AI 研究者读到一篇论文后，想在该方向开展研究、写论文跑对比实验，面临三个递进的痛点：
| 痛点 | 具体描述 | 人工耗时 |
|---|---|---|
| 摸清方向难 | 读完一篇论文，想知道这个研究方向有哪些主流方法族、各自有什么开源实现、目前前沿在哪里——信息分散在 GitHub、PapersWithCode、知乎、各实验室主页，拼凑全景图极费时间 | 1~2 天 |
| 找对代码难 | 即便找到一堆仓库，也分不清哪个是"官方实现"、哪个是"社区最热复现"、哪个只是"只有 README 的空壳" | 半天~1 天 |
| 评估能不能跑难 | 看中一个仓库想拿来跑对比实验，装了两天环境发现依赖崩了、数据集没公开、训练脚本跑不通 | 半天~1 天 |
ResearchRadar 做的事情：输入一篇论文的 arxiv 链接，自动完成「摸清研究方向全景 → 找到高质量开源实现 → 评估可复现性 → 推荐最适合拿来做对比实验的仓库」。
1.2 输入与输出
输入：一篇论文的 arxiv URL（如 https://arxiv.org/abs/1706.03762）
↓
输出：一份"研究方向全景研报"，包含：
1. 研究方向解读：这篇论文属于什么子领域、该领域有哪些主流方法族
2. 开源实现全景图：每个方法族下有哪些高质量仓库（含 star、维护状态、方法类型）
3. 每个仓库的可复现性评分（0-100）+ 五维度明细 + 判定理由
4. 对比实验推荐：哪些仓库可以直接拿来跑 baseline / 对比实验，为什么
1.3 核心价值链（三层）
论文输入
↓
┌─────────────────────────────────────────────┐
│ 第一层：研究方向解析 │
│ LLM 理解论文 → 识别子领域 → 梳理方法族谱系 │
│ 输出：该方向的"研究地图" │
├─────────────────────────────────────────────┤
│ 第二层：开源代码发现 │
│ 针对每个方法族搜索 GitHub → 去重 → 初筛 │
│ 输出：按方法族分类的候选仓库列表 │
├─────────────────────────────────────────────┤
│ 第三层：可复现性评估 + 对比实验适配度 │
│ 抓 README + 依赖 → LLM 多维度评分 │
│ 输出：排名 + 推荐 + 风险提示 │
└─────────────────────────────────────────────┘
↓
研报输出（研究方向全景 + 代码推荐 + 复现风险评估）
三层缺一不可：不解析方向就搜不准代码，不评估就不知道能不能跑，最终目标是为研究者写论文跑对比实验提供可靠的后备代码库。
二、系统架构
2.1 总体架构图
┌──────────────────────────────────────────┐
│ ResearchRadar 系统 │
│ │
用户输入 arxiv URL ──→ │ ┌──────────────────────────────────┐ │
│ │ 第一层：研究方向解析（Agent） │ │
│ │ │ │
│ │ 1. arxiv API → 论文元信息 │ │
│ │ 2. LLM → 子领域识别 + 方法族梳理 │ │
│ │ 3. LLM → 为每个方法族生成搜索词 │ │
│ │ 输出：研究方向 "地图" │ │
│ └──────────────┬───────────────────┘ │
│ ↓ │
│ ┌──────────────────────────────────┐ │
│ │ 第二层：开源代码发现（Workflow） │ │
│ │ │ │
│ │ 1. GitHub Search → 多关键词搜索 │ │
│ │ 2. 去重 + 按方法族归类 │ │
│ │ 3. 初筛（star/更新时间/语言） │ │
│ │ 输出：按方法族分类的候选仓库列表 │ │
│ └──────────────┬───────────────────┘ │
│ ↓ │
│ ┌──────────────────────────────────┐ │
│ │ 第三层：评估与推荐（Agent） │ │
│ │ │ │
│ │ 对每个候选仓库： │ │
│ │ 1. 抓 README + 依赖文件（Workflow）│ │
│ │ 2. LLM → 可复现性五维度评分 │ │
│ │ 3. LLM → 对比实验适配度评估 │ │
│ │ 4. 综合排序 + 生成推荐理由 │ │
│ └──────────────┬───────────────────┘ │
│ ↓ │
│ 研究方向全景研报（Markdown） │
└──────────────────────────────────────────┘
2.2 架构中的两类组件
按照路线大纲的约定，整个系统包含两类组件：
| 类型 | 定义 | 本系统中的对应 |
|---|---|---|
| Workflow（路径写死） | 确定性流程，不涉及 LLM 决策 | 论文信息抓取、GitHub 搜索、依赖文件扫描、仓库元信息抓取、去重归类 |
| Agent（LLM 决策） | 需要 LLM 理解和判断的节点 | 研究方向解析、方法族梳理、搜索词生成、可复现性评分、对比实验适配度评估 |
Week 1 只实现 Workflow + 最简单的 Agent（baseline 方案，单次 LLM 调用出评分，不涉及回环）。对于三层架构，Week 1 三个层都会实现但每层只保留最简版本：
- 第一层：1 次 LLM 调用完成方向解析 + 关键词生成
- 第二层：直接用关键词搜 GitHub，不加方法族归类（归类逻辑 Week 3 加）
- 第三层：1 次 LLM 调用完成可复现性评分 + 对比实验适配度（合并为一个 prompt）
三、技术选型
| 项 | 选型 | 理由 |
|---|---|---|
| 编程语言 | Python 3.10+ | 两人都会，生态最好 |
| LLM 调用 | DeepSeek API（OpenAI SDK） | 你们有无限额度的 key，中文能力强，便宜 |
| 论文信息 | arxiv REST API | 免费、无需认证、覆盖所有 AI 论文 |
| 代码搜索 | GitHub Search REST API | 免费（带 token 30次/分钟），可按关键词、语言、star 筛选 |
| 数据抓取 | requests + 标准库 |
不引入爬虫框架 |
| UI | Gradio | 路线大纲已定死，轻量够用 |
| 状态管理 | Python dict | 不引入任何状态管理框架 |
| 包管理 | pip + requirements.txt | 两人都在 Windows 上，简单第一 |
四、第一周 MVP 范围
4.1 MVP 目标
一句话：python run.py <arxiv_url> 能跑通完整链路——从论文到推荐仓库。
4.2 MVP 包含什么（7 个模块）
模块 1: 论文信息抓取 arxiv API → 标题、摘要、作者、分类
模块 2: 研究方向解析 + 关键词 LLM → 子领域识别 + 方法族梳理 + 搜索词生成（1 次调用完成）
模块 3: GitHub 代码搜索 GitHub Search API → 多关键词搜索，去重，按 star 排序
模块 4: 仓库详情抓取 对 top 5 候选仓库：抓 README + 依赖文件 + 元信息
模块 5: 可复现性 + 适配度评估 LLM → 六维度评分（可复现性 5 维 + 对比实验适配度）
模块 6: 排序推荐 综合加权排序，生成推荐理由
模块 7: Gradio 研报输出 结构化 Markdown：方向全景 + 仓库排行 + 推荐建议
4.3 MVP 不包含什么
- ❌ Issue 抓取与分析（模块过大，Week 2 加）
- ❌ Agent 回环机制（Week 4 加）
- ❌ 流式输出（Week 5 加）
- ❌ 多轮并发（始终串行，先跑通再优化）
- ❌ 复杂的错误恢复（网络挂了就报错，不吞异常）
五、甲的任务（数据层）
甲负责 所有与外部 API 打交道 的模块：论文信息、代码搜索、仓库详情抓取、评测集建设。
5.1 模块 1：论文信息抓取 paper_fetcher.py
做什么：输入 arxiv URL → 输出论文标题、摘要、作者、分类标签。
技术要点：
- arxiv API 端点：
http://export.arxiv.org/api/query?id_list={arxiv_id}&max_results=1 - 返回 Atom XML 格式，用标准库
xml.etree.ElementTree解析，不引入额外依赖 - 必须加 User-Agent 请求头，否则 arxiv 可能拒绝请求
- arxiv ID 从 URL 中提取：
https://arxiv.org/abs/1706.03762→1706.03762，兼容abs、pdf等前缀
# paper_fetcher.py
import urllib.request
import urllib.error
import xml.etree.ElementTree as ET
import re
def extract_arxiv_id(url: str) -> str:
"""从 arxiv URL 提取论文 ID
支持格式:
- https://arxiv.org/abs/1706.03762
- https://arxiv.org/abs/1706.03762v2
- https://arxiv.org/pdf/1706.03762.pdf
- arxiv:1706.03762
- 1706.03762
"""
# 尝试匹配 URL 格式
match = re.search(r'arxiv\.org/(?:abs|pdf)/(\d{4}\.\d{4,5})(?:v\d+)?(?:\.pdf)?', url)
if match:
return match.group(1)
# 尝试匹配 arxiv:ID 或直接 ID 格式
match = re.search(r'(?:arxiv:)?(\d{4}\.\d{4,5})', url)
if match:
return match.group(1)
raise ValueError(f"无法从 URL 提取 arxiv ID: {url}")
def fetch_paper_info(arxiv_url: str) -> dict:
"""从 arxiv 获取论文元信息
返回: {"arxiv_id": str, "title": str, "abstract": str,
"authors": [str, ...], "categories": [str, ...], "published": str}
"""
arxiv_id = extract_arxiv_id(arxiv_url)
api_url = f"http://export.arxiv.org/api/query?id_list={arxiv_id}&max_results=1"
req = urllib.request.Request(api_url, headers={"User-Agent": "ResearchRadar/1.0"})
try:
with urllib.request.urlopen(req, timeout=15) as resp:
xml_text = resp.read().decode("utf-8")
except urllib.error.HTTPError as e:
raise RuntimeError(f"arxiv API 请求失败 (HTTP {e.code})，请检查论文 ID 是否正确")
except urllib.error.URLError as e:
raise RuntimeError(f"无法连接 arxiv API，请检查网络: {e.reason}")
root = ET.fromstring(xml_text)
ns = {
"atom": "http://www.w3.org/2005/Atom",
"arxiv": "http://arxiv.org/schemas/atom",
}
entry = root.find("atom:entry", ns)
if entry is None:
raise ValueError(f"arxiv 未找到论文 {arxiv_id}，请检查 ID 是否正确")
title_el = entry.find("atom:title", ns)
summary_el = entry.find("atom:summary", ns)
title = title_el.text.strip().replace("\n", " ") if title_el is not None and title_el.text else ""
abstract = summary_el.text.strip().replace("\n", " ") if summary_el is not None and summary_el.text else ""
authors = [
a.find("atom:name", ns).text
for a in entry.findall("atom:author", ns)
if a.find("atom:name", ns) is not None
]
categories = [
c.get("term", "") for c in entry.findall("atom:category", ns)
]
published_el = entry.find("atom:published", ns)
published = published_el.text if published_el is not None and published_el.text else ""
return {
"arxiv_id": arxiv_id,
"title": title,
"abstract": abstract,
"authors": authors,
"categories": categories,
"published": published,
}
5.2 模块 3：GitHub 代码搜索 repo_searcher.py
做什么：给定一个关键词列表 → 搜索 GitHub 返回相关仓库列表。
技术要点：
- GitHub Search API：
GET /search/repositories?q={query}&sort=stars&order=desc&per_page=5 - 搜索 query 构造：用关键词 + 限定条件（language:python、stars 门槛）
- 搜索接口限速更严：带 token 30 次/分钟，不带 token 10 次/分钟。你们两人各申请一个 token，在代码里轮换
- 返回结果去重（同一个 repo 可能被不同关键词搜到多次）
# repo_searcher.py
import requests
import os
GITHUB_TOKENS = [
os.getenv("GITHUB_TOKEN_1", ""),
os.getenv("GITHUB_TOKEN_2", ""),
]
GITHUB_TOKENS = [t for t in GITHUB_TOKENS if t] # 过滤空值
_token_index = 0
def _get_token():
"""轮换使用两个 token，降低限速风险"""
global _token_index
if not GITHUB_TOKENS:
return ""
token = GITHUB_TOKENS[_token_index]
_token_index = (_token_index + 1) % len(GITHUB_TOKENS)
return token
def _get_headers():
token = _get_token()
if token:
return {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github.v3+json"}
return {"Accept": "application/vnd.github.v3+json"}
def search_repos(keywords: list[str], max_per_keyword: int = 5, language: str = "python") -> list[dict]:
"""用关键词列表搜索 GitHub 仓库
返回: 去重后的仓库列表，按 star 降序排列
"""
seen = set()
results = []
for kw in keywords:
# 构造查询：关键词 + Python + 至少 5 个 star（过滤空壳仓库）
query = f"{kw} language:{language} stars:>=5"
url = "https://api.github.com/search/repositories"
params = {"q": query, "sort": "stars", "order": "desc", "per_page": max_per_keyword}
try:
resp = requests.get(url, headers=_get_headers(), params=params, timeout=15)
if resp.status_code == 403 and "rate limit" in resp.text.lower():
print(f"[警告] GitHub 搜索限速，跳过关键词: {kw}")
continue
resp.raise_for_status()
data = resp.json()
except Exception as e:
print(f"[警告] 搜索关键词 '{kw}' 失败: {e}")
continue
for item in data.get("items", []):
full_name = item["full_name"]
if full_name not in seen:
seen.add(full_name)
results.append({
"full_name": full_name,
"html_url": item["html_url"],
"description": item.get("description", ""),
"stars": item.get("stargazers_count", 0),
"language": item.get("language", ""),
"topics": item.get("topics", []),
"updated_at": item.get("updated_at", ""),
"match_keyword": kw,
})
# 按 star 降序
results.sort(key=lambda r: r["stars"], reverse=True)
return results
5.3 模块 5（的一部分）：仓库详情抓取 repo_fetcher.py
做什么：给定一个 GitHub 仓库的 owner/repo → 抓 README、依赖文件、元信息。
这与之前方案的内容一致，但需要注意：现在是批量抓取（一个论文对应 5 个候选仓库），需要处理部分仓库抓取失败的情况——不能因为一个仓库失败了就让整个流程崩溃。
# repo_fetcher.py
import os
import base64
import requests
GITHUB_TOKENS = [
os.getenv("GITHUB_TOKEN_1", ""),
os.getenv("GITHUB_TOKEN_2", ""),
]
GITHUB_TOKENS = [t for t in GITHUB_TOKENS if t]
_token_index = 0
def _get_token():
global _token_index
if not GITHUB_TOKENS:
return ""
token = GITHUB_TOKENS[_token_index]
_token_index = (_token_index + 1) % len(GITHUB_TOKENS)
return token
API_BASE = "https://api.github.com"
DEPENDENCY_FILENAMES = [
"requirements.txt", "environment.yml", "environment.yaml",
"setup.py", "setup.cfg", "pyproject.toml",
"Pipfile", "Pipfile.lock", "Dockerfile", "Makefile", "CMakeLists.txt",
]
def _github_get(owner: str, repo: str, endpoint: str) -> dict | None:
"""统一的 GitHub API GET 请求"""
headers = {"Accept": "application/vnd.github.v3+json"}
token = _get_token()
if token:
headers["Authorization"] = f"Bearer {token}"
url = f"{API_BASE}/repos/{owner}/{repo}/{endpoint}" if endpoint else f"{API_BASE}/repos/{owner}/{repo}"
resp = requests.get(url, headers=headers, timeout=15)
if resp.status_code == 404:
return None
if resp.status_code in (403, 429):
raise RuntimeError(f"GitHub API 限速或拒绝访问: {resp.status_code}")
resp.raise_for_status()
return resp.json()
def fetch_readme(owner: str, repo: str) -> str | None:
"""抓取 README 内容（已解码的纯文本）"""
data = _github_get(owner, repo, "readme")
if not data:
return None
content = data.get("content", "")
try:
return base64.b64decode(content).decode("utf-8", errors="replace")
except Exception:
return None
def fetch_dependencies(owner: str, repo: str) -> dict[str, str]:
"""抓取所有依赖文件，返回 {文件名: 内容}"""
result = {}
contents = _github_get(owner, repo, "contents") or []
for item in contents:
name = item.get("name", "")
if name in DEPENDENCY_FILENAMES:
file_data = _github_get(owner, repo, f"contents/{name}")
if file_data and file_data.get("content"):
try:
decoded = base64.b64decode(file_data["content"]).decode("utf-8", errors="replace")
result[name] = decoded
except Exception:
pass
return result
def fetch_issues(owner: str, repo: str, max_count: int = 30) -> list[dict]:
"""抓取最近的 Issues（为 Week 2 预留接口，Week 1 可以不调用）"""
issues = []
page = 1
while len(issues) < max_count:
params = {"state": "all", "sort": "created", "direction": "desc",
"per_page": min(30, max_count - len(issues)), "page": page}
resp = requests.get(
f"{API_BASE}/repos/{owner}/{repo}/issues",
headers=_get_headers(),
params=params,
timeout=15,
)
if resp.status_code != 200:
break
page_items = resp.json()
if not page_items:
break
for item in page_items:
if "pull_request" not in item: # GitHub Issues API 会混入 PR，需要过滤
issues.append({
"title": item.get("title", ""),
"state": item.get("state", ""),
"created_at": item.get("created_at", ""),
"body": (item.get("body") or "")[:500], # 截断，只要前 500 字符
"comments": item.get("comments", 0),
})
page += 1
return issues
def _get_headers():
token = _get_token()
headers = {"Accept": "application/vnd.github.v3+json"}
if token:
headers["Authorization"] = f"Bearer {token}"
return headers
5.4 评测集建设
这是整个项目最重要的资产。评测集分两部分：
Part A：代码发现评测集（10 篇论文）
每篇论文标注：已知的最佳开源实现（ground truth 仓库）。
{
"paper_arxiv_id": "1706.03762",
"paper_title": "Attention Is All You Need",
"paper_keywords_human": ["transformer", "attention mechanism", "seq2seq", "neural machine translation"],
"known_implementations": [
{"full_name": "huggingface/transformers", "relevance": "official"},
{"full_name": "tensorflow/tensor2tensor", "relevance": "official"},
{"full_name": "jadore801120/attention-is-all-you-need-pytorch", "relevance": "community"}
]
}
Part B：可复现性评测集（20 个仓库）
沿用路线大纲的方案：10 个正样本 + 10 个负样本。
{
"repo_url": "https://github.com/xxx/yyy",
"ground_truth": "reproducible",
"evidence": ["证据1", "证据2"],
"rationale": "为什么这样标注"
}
Week 1 最低要求：Part A 标完 3 篇论文（用于 demo），Part B 标完 6-8 个仓库。不需要在 demo 时展示评测集，但标注工作是整个项目的基准。
六、乙的任务（LLM + 集成层）
乙负责 所有与 LLM 交互 的模块：关键词提取、可复现性评估、主流程编排、Gradio 界面。
6.1 模块 2：研究方向解析 + 搜索词生成 direction_analyzer.py
做什么：输入论文标题 + 摘要 → LLM 一次性完成三件事：(1) 识别该论文所属子领域，(2) 梳理该领域的主流方法族，(3) 生成针对每个方法族的 GitHub 搜索词。
为什么比单纯提取关键词更好：
- 研究者不只需要"一个官方实现"，他要知道这个方向有几条技术路线、每条路线有哪些代表实现
- 比如输入 Transformer 论文，应该输出「vanilla Transformer / BERT 系 / GPT 系 / T5 系 / ViT 系」等方法族
- 这直接对应研究者的实际需求——写论文跑对比实验时需要覆盖多个 baseline
# direction_analyzer.py
import json
import re
from openai import OpenAI
client = OpenAI(
api_key="your-deepseek-api-key",
base_url="https://api.deepseek.com",
)
DIRECTION_SYSTEM_PROMPT = """你是一个 AI 研究领域的导航专家。你的任务是：给定一篇论文，帮助研究者快速了解该方向的前沿开源生态。
## 你的分析框架
请完成以下三个层次的分析：
### 层次 1：子领域定位
- 这篇论文属于 AI/ML 的哪个具体子领域？（如 "Transformer-based Neural Machine Translation" 而非笼统的 "Deep Learning"）
- 该子领域在 2024-2025 年的主流趋势是什么？（1-2 句话）
### 层次 2：方法族谱系
- 该子领域目前有哪些主流方法族/变体？
- 每个方法族的代表工作是什么？
- 输出 3-6 个方法族，每个包含方法名 + 一句话说明
### 层次 3：搜索策略
- 为每个方法族生成 2-3 个 GitHub 搜索查询
- 查询应使用英文，包含：方法名 + 关键词 + "implementation" 或 "pytorch" 或 "official"
- 额外生成 2-3 个覆盖全领域的宽泛搜索
## 输出格式（严格 JSON，不要任何额外文字）
{
"subfield": "子领域名称（中英文均可）",
"subfield_trend": "该子领域当前主流趋势（1-2 句中文）",
"method_families": [
{
"family_name": "方法族名称",
"description": "一句话说明",
"representative_work": "代表论文或工作",
"search_queries": ["搜索词1", "搜索词2"]
}
],
"broad_queries": ["全领域宽泛搜索词1", "全领域宽泛搜索词2"]
}
"""
def analyze_direction(title: str, abstract: str) -> dict:
"""从论文标题和摘要中解析研究方向并生成搜索策略"""
user_prompt = f"""## 论文标题
{title}
## 论文摘要
{abstract[:2000]}
请分析该论文的研究方向并生成搜索策略。"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": DIRECTION_SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0.4, # 稍高温度让方法族梳理更有创造力
max_tokens=1500,
)
raw = response.choices[0].message.content.strip()
return _parse_json(raw)
def _parse_json(raw: str) -> dict:
"""安全解析 JSON，容忍 LLM 的格式问题"""
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
match = re.search(r'```(?:json)?\s*([\s\S]*?)```', raw)
if match:
try:
return json.loads(match.group(1).strip())
except json.JSONDecodeError:
pass
match = re.search(r'\{[\s\S]*\}', raw)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass
raise ValueError(f"方向解析 LLM 返回格式无法解析: {raw[:300]}")
6.2 模块 5：可复现性 + 对比实验适配度评估 repo_evaluator.py
做什么：输入一个仓库的 README + 依赖文件 + 元信息 + 所属方法族 → LLM 输出六维度评分（可复现性 5 维 + 对比实验适配度 1 维）。
为什么新增「对比实验适配度」维度：很多仓库能跑（可复现），但不适合直接拿来跑对比实验——比如它只实现了推理没有训练、数据集用的是私有数据、或者 benchmark 指标不是领域标准指标。研究者需要的是"能直接加入论文 experiment section 的代码"。
# repo_evaluator.py
import json
import re
from openai import OpenAI
client = OpenAI(
api_key="your-deepseek-api-key",
base_url="https://api.deepseek.com",
)
EVAL_SYSTEM_PROMPT = """你是一个 AI 论文代码可复现性与对比实验适配度评估专家。
你的任务是评估一个 GitHub 仓库在以下两个方面的表现：
- **可复现性**：这个仓库的代码能否被他人成功跑通？
- **对比实验适配度**：这个仓库适不适合直接拿来写论文做对比实验（baseline / comparison）？
## 评估框架（六维度）
### 可复现性（五维度，合计 0-80 分）
1. **环境配置完整性（0-15 分）**
- 15 分：提供了完整的依赖文件（requirements.txt / environment.yml），版本号明确
- 8-14 分：有依赖文件但部分版本号缺失或模糊
- 0-7 分：没有依赖文件或严重不完整
2. **文档质量（0-20 分）**
- 16-20 分：有清晰的安装步骤、训练/推理命令、数据集获取方式
- 8-15 分：有基本说明但关键步骤缺失
- 0-7 分：README 空洞，没有可执行的命令
3. **代码可用性（0-20 分）**
- 16-20 分：有完整的训练/推理脚本、配置文件，入口清晰
- 8-15 分：有核心代码但缺少关键组件（如只有推理没有训练）
- 0-7 分：代码残缺
4. **社区活跃度（0-10 分）**
- 8-10 分：Star > 500，最近半年有更新
- 4-7 分：有一定关注度但更新不频繁
- 0-3 分：长期不更新，Star < 10
5. **依赖健康度（0-15 分）**
- 12-15 分：依赖全是主流活跃包，无已知兼容性问题
- 6-11 分：有少量过时依赖但可升级
- 0-5 分：依赖已停止维护（如 TF 1.x）或已知无法安装
### 对比实验适配度（一维度，0-20 分）
6. **对比实验适配度（0-20 分）**
- 16-20 分：提供标准 benchmark 评估脚本 + 预训练权重 + 公开数据集，可以直接跑出论文常用指标
- 10-15 分：有训练脚本和评估代码，但需要自己准备数据或微调配接
- 0-9 分：只有推理、使用私有数据集、没有标准化评估流程
## 输出格式（严格 JSON，不要任何额外文字）
{
"reproducibility_score": 整数（0-80，可复现性五项求和）,
"benchmark_fitness_score": 整数（0-20，对比实验适配度）,
"overall_score": 整数（0-100，六项总和）,
"verdict": "reproducible" | "partially" | "not_reproducible",
"env_score": 整数,
"doc_score": 整数,
"code_score": 整数,
"community_score": 整数,
"dep_score": 整数,
"benchmark_score": 整数,
"reasoning": "200字以内的中文分析，必须区分'能不能跑'和'适不适合跑对比实验'两个层面",
"risks": ["风险1", "风险2"],
"benchmark_readiness": "ready" | "partial" | "not_ready",
"suggested_use": "可以直接作为对比实验 baseline" | "需要少量修改后可用于对比实验" | "仅适合参考代码实现，不适合直接跑对比实验"
}
## 判定规则
- overall_score >= 70 → verdict = "reproducible"
- 40 <= overall_score < 70 → verdict = "partially"
- overall_score < 40 → verdict = "not_reproducible"
- benchmark_score >= 16 → benchmark_readiness = "ready"
- 10 <= benchmark_score < 16 → benchmark_readiness = "partial"
- benchmark_score < 10 → benchmark_readiness = "not_ready"
"""
def evaluate_repo(repo: dict, readme: str | None, dependencies: dict[str, str],
method_family: str = "") -> dict:
"""评估单个仓库的可复现性和对比实验适配度"""
readme_text = (readme or "README 未找到")[:2500]
deps_text = "\n\n".join(
f"### {fname}\n```\n{content[:1000]}\n```"
for fname, content in dependencies.items()
) or "未找到依赖文件"
family_hint = f"\n该方法属于: {method_family}" if method_family else ""
user_prompt = f"""## 仓库信息
- 名称: {repo.get('full_name', '')}
- URL: {repo.get('html_url', '')}
- Stars: {repo.get('stars', 0)}
- 描述: {repo.get('description', '')}
- 最后更新: {repo.get('updated_at', '')}
- 语言: {repo.get('language', '')}{family_hint}
## README
{readme_text}
## 依赖文件
{deps_text}
请评估该仓库的可复现性和对比实验适配度。"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": EVAL_SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0.3,
max_tokens=1500,
)
raw = response.choices[0].message.content.strip()
return _parse_json(raw)
def _parse_json(raw: str) -> dict:
"""安全 JSON 解析"""
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
match = re.search(r'```(?:json)?\s*([\s\S]*?)```', raw)
if match:
try:
return json.loads(match.group(1).strip())
except json.JSONDecodeError:
pass
match = re.search(r'\{[\s\S]*\}', raw)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass
raise ValueError(f"评估 LLM 返回格式无法解析: {raw[:300]}")
def evaluate_repo(repo: dict, readme: str | None, dependencies: dict[str, str]) -> dict:
"""评估单个仓库的可复现性"""
readme_text = (readme or "README 未找到")[:2500]
deps_text = "\n\n".join(
f"### {fname}\n\n{content[:1000]}\n"
for fname, content in dependencies.items()
) or "未找到依赖文件"
user_prompt = f"""## 仓库信息
- 名称: {repo.get('full_name', '')}
- URL: {repo.get('html_url', '')}
- Stars: {repo.get('stars', 0)}
- 描述: {repo.get('description', '')}
- 最后更新: {repo.get('updated_at', '')}
- 语言: {repo.get('language', '')}
README
{readme_text}
依赖文件
{deps_text}
请评估该仓库的可复现性。"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": EVAL_SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0.3,
max_tokens=1200,
)
raw = response.choices[0].message.content.strip()
return _parse_json(raw)
def _parse_json(raw: str) -> dict:
"""安全 JSON 解析，同 direction_analyzer 中的实现"""
try:
return json.loads(raw)
except json.JSONDecodeError:
pass
match = re.search(r'(?:json)?\s*([\s\S]*?)', raw)
if match:
try:
return json.loads(match.group(1).strip())
except json.JSONDecodeError:
pass
match = re.search(r'{[\s\S]*}', raw)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass
raise ValueError(f"评估 LLM 返回格式无法解析: {raw[:300]}")
### 6.3 模块 6：主流程 `run.py`
**这是整个项目的中央调度器**。按三层架构串联所有模块。
```python
# run.py
import sys
from paper_fetcher import fetch_paper_info
from direction_analyzer import analyze_direction
from repo_searcher import search_repos
from repo_fetcher import fetch_readme, fetch_dependencies
from repo_evaluator import evaluate_repo
def run(arxiv_url: str, top_n: int = 5) -> dict:
"""主入口：从论文到"研究方向全景 + 代码推荐"的完整链路
返回 dict:
- paper: 论文信息
- direction: 研究方向解析结果（子领域 + 方法族 + 搜索词）
- repos: 评估后的仓库列表（按 overall_score 降序）
"""
# ===== 第一层：研究方向解析 =====
print(f"[1/5] 正在获取论文信息...")
paper = fetch_paper_info(arxiv_url)
print(f" 论文: {paper['title'][:80]}...")
print(f"[2/5] 正在解析研究方向...")
direction = analyze_direction(paper["title"], paper["abstract"])
print(f" 子领域: {direction.get('subfield', '未知')}")
print(f" 方法族: {len(direction.get('method_families', []))} 个")
# 汇总所有搜索词（方法族搜索 + 宽泛搜索）
all_queries = []
for mf in direction.get("method_families", []):
all_queries.extend(mf.get("search_queries", []))
all_queries.extend(direction.get("broad_queries", []))
print(f" 生成 {len(all_queries)} 个搜索查询")
# ===== 第二层：开源代码发现 =====
print(f"[3/5] 正在搜索 GitHub 仓库...")
candidates = search_repos(all_queries, max_per_keyword=5)
candidates = candidates[:top_n]
print(f" 去重后找到 {len(candidates)} 个候选仓库")
if not candidates:
return {
"paper": paper,
"direction": direction,
"repos": [],
"error": "未找到相关开源仓库",
}
# ===== 第三层：可复现性 + 对比实验适配度评估 =====
print(f"[4/5] 正在评估 {len(candidates)} 个候选仓库...")
evaluated = []
for i, repo in enumerate(candidates):
full_name = repo["full_name"]
owner, name = full_name.split("/", 1)
# 尝试将仓库匹配到对应的方法族（简单规则：描述中包含方法族名称）
matched_family = _match_family(repo, direction.get("method_families", []))
print(f" ({i+1}/{len(candidates)}) 评估 {full_name} [{matched_family or '未归类'}]...")
try:
readme = fetch_readme(owner, name)
deps = fetch_dependencies(owner, name)
evaluation = evaluate_repo(repo, readme, deps, matched_family)
except Exception as e:
evaluation = {
"overall_score": 0,
"reproducibility_score": 0,
"benchmark_fitness_score": 0,
"verdict": "error",
"reasoning": f"评估失败: {str(e)[:100]}",
"env_score": 0, "doc_score": 0, "code_score": 0,
"community_score": 0, "dep_score": 0, "benchmark_score": 0,
"risks": [], "benchmark_readiness": "not_ready",
"suggested_use": "评估失败，请手动检查",
}
evaluated.append({
**repo,
"method_family": matched_family,
"evaluation": evaluation,
})
# 按综合评分降序
evaluated.sort(key=lambda r: r["evaluation"].get("overall_score", 0), reverse=True)
print(f"[5/5] 生成研报...")
return {
"paper": paper,
"direction": direction,
"repos": evaluated,
}
def _match_family(repo: dict, families: list[dict]) -> str:
"""简单的仓库→方法族匹配：基于描述和 topics"""
desc = (repo.get("description") or "").lower()
topics = " ".join(repo.get("topics", [])).lower()
text = desc + " " + topics
for mf in families:
family_name = mf.get("family_name", "").lower()
if family_name and family_name in text:
return mf["family_name"]
return ""
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python run.py <arxiv_url>")
print("示例: python run.py https://arxiv.org/abs/1706.03762")
sys.exit(1)
result = run(sys.argv[1])
if result.get("error"):
print(f"\n❌ {result['error']}")
else:
# 使用 app.py 中的 format_report 输出
from app import format_report
print(format_report(result))
6.4 Gradio 界面 app.py
# app.py
import gradio as gr
from run import run
def diagnose(arxiv_url: str) -> str:
"""Gradio 回调：输入 arxiv URL → 输出研究方向全景研报"""
url = arxiv_url.strip()
if not url:
return "## 请输入 arxiv 论文链接"
try:
result = run(url)
except ValueError as e:
return f"## 输入格式错误\n\n{str(e)}"
except RuntimeError as e:
return f"## API 请求失败\n\n{str(e)}"
except Exception as e:
return f"## 运行出错\n\n```\n{str(e)}\n```"
if result.get("error"):
return f"## 未找到相关代码\n\n{result['error']}"
return format_report(result)
def format_report(result: dict) -> str:
"""将 run() 的结果格式化为"研究方向全景研报" Markdown"""
paper = result["paper"]
direction = result["direction"]
repos = result["repos"]
# ===== 研报头部 =====
md = f"""# ResearchRadar 研究方向全景研报
---
## 一、论文信息
| 项目 | 内容 |
|------|------|
| **标题** | {paper['title']} |
| **arXiv ID** | [{paper['arxiv_id']}](https://arxiv.org/abs/{paper['arxiv_id']}) |
| **作者** | {', '.join(paper['authors'][:5])}{'...' if len(paper['authors']) > 5 else ''} |
| **发表时间** | {paper.get('published', 'N/A')[:10]} |
| **分类** | {', '.join(paper['categories']) if paper.get('categories') else 'N/A'} |
---
## 二、研究方向全景
### 子领域定位
**{direction.get('subfield', '未知')}**
### 当前趋势
{direction.get('subfield_trend', '暂无分析')}
### 方法族谱系
| # | 方法族 | 说明 | 代表工作 |
|---|--------|------|----------|
"""
for i, mf in enumerate(direction.get("method_families", [])):
md += f"| {i+1} | **{mf.get('family_name', '')}** | {mf.get('description', '')} | {mf.get('representative_work', '')} |\n"
md += """
---
## 三、开源实现评估排名
"""
# ===== 仓库排名 =====
for i, repo in enumerate(repos):
ev = repo["evaluation"]
verdict_icon = {
"reproducible": "✅", "partially": "⚠️",
"not_reproducible": "❌", "error": "💥"
}.get(ev.get("verdict", ""), "❓")
readiness_icon = {
"ready": "🟢", "partial": "🟡", "not_ready": "🔴"
}.get(ev.get("benchmark_readiness", ""), "⚪")
family_tag = f" `[{repo.get('method_family', '未归类')}]`" if repo.get("method_family") else ""
md += f"""### {i+1}. [{repo['full_name']}]({repo['html_url']}){family_tag}
{verdict_icon} 可复现性: **{ev.get('reproducibility_score', ev.get('overall_score', 'N/A'))}/80** | {readiness_icon} 对比实验适配度: **{ev.get('benchmark_fitness_score', ev.get('benchmark_score', 'N/A'))}/20** | 综合: **{ev.get('overall_score', 'N/A')}/100**
| 维度 | 得分 | 满分 | 说明 |
|------|:----:|:----:|------|
| 环境配置 | {ev.get('env_score', 'N/A')} | 15 | 依赖文件完整度 |
| 文档质量 | {ev.get('doc_score', 'N/A')} | 20 | 安装/训练/数据说明 |
| 代码可用性 | {ev.get('code_score', 'N/A')} | 20 | 训练+推理脚本完整性 |
| 社区活跃 | {ev.get('community_score', 'N/A')} | 10 | Star + 更新频率 |
| 依赖健康 | {ev.get('dep_score', 'N/A')} | 15 | 依赖包可用性 |
| **对比实验** | **{ev.get('benchmark_score', 'N/A')}** | **20** | **标准化评估 + 预训练权重** |
⭐ Star: {repo.get('stars', 0)} | 匹配: `{repo.get('match_keyword', '')}`
**分析**: {ev.get('reasoning', 'N/A')}
**对比实验建议**: {ev.get('suggested_use', 'N/A')}
"""
if ev.get("risks"):
md += "**风险点**: " + " | ".join(f"`{r}`" for r in ev["risks"]) + "\n"
md += "\n---\n\n"
# ===== 对比实验推荐总结 =====
ready_repos = [r for r in repos if r["evaluation"].get("benchmark_readiness") == "ready"]
partial_repos = [r for r in repos if r["evaluation"].get("benchmark_readiness") == "partial"]
md += """## 四、对比实验推荐总结
"""
if ready_repos:
md += "### 🟢 可直接用于对比实验\n\n"
for r in ready_repos:
md += f"- **[{r['full_name']}]({r['html_url']})** — {r['evaluation'].get('suggested_use', '')}\n"
md += "\n"
if partial_repos:
md += "### 🟡 需要少量修改后可用\n\n"
for r in partial_repos:
md += f"- **[{r['full_name']}]({r['html_url']})** — {r['evaluation'].get('suggested_use', '')}\n"
md += "\n"
not_ready = [r for r in repos if r["evaluation"].get("benchmark_readiness") not in ("ready", "partial")]
if not_ready:
md += "### 🔴 仅适合参考代码实现\n\n"
for r in not_ready:
md += f"- **[{r['full_name']}]({r['html_url']})** — {r['evaluation'].get('suggested_use', r['evaluation'].get('reasoning', 'N/A')[:80])}...\n"
md += "\n"
md += """
---
> ResearchRadar · 研究方向全景 + 开源代码评估 + 对比实验推荐
"""
return md
DEMO_EXAMPLES = [
["https://arxiv.org/abs/1706.03762"], # Attention Is All You Need (Transformer)
["https://arxiv.org/abs/2010.11929"], # An Image is Worth 16x16 Words (ViT)
["https://arxiv.org/abs/2006.11239"], # Denoising Diffusion Probabilistic Models (DDPM)
]
demo = gr.Interface(
fn=diagnose,
inputs=gr.Textbox(
label="论文 arxiv 链接",
placeholder="https://arxiv.org/abs/1706.03762",
lines=1,
),
outputs=gr.Markdown(label="研究方向全景研报"),
title="ResearchRadar — 研究方向全景 + 开源代码评估",
description="输入一篇 AI 论文的 arxiv 链接，AI 帮你摸清研究方向全景，找到最适配的开源实现，评估能否直接用于对比实验。",
examples=DEMO_EXAMPLES,
theme="soft",
)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", share=False)
七、数据流与控制流
7.1 完整链路（三层架构）
用户输入 arxiv URL
│
▼
┌──────────────────────────────────────────┐
│ paper_fetcher.fetch_paper_info() │ ← arxiv API
│ 输出: {title, abstract, authors, ...} │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ direction_analyzer.analyze_direction() │ ← DeepSeek API (1 次)
│ 输出: {subfield, method_families[], │
│ broad_queries[]} │
│ ※ 同时完成子领域识别 + 方法族梳理 + │
│ 搜索词生成 │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ repo_searcher.search_repos() │ ← GitHub Search API (N 次)
│ 输入: 所有方法族搜索词 + 宽泛搜索词 │
│ 输出: 去重排序后的候选仓库 top 5 │
└──────────────┬───────────────────────────┘
│
▼ (对每个候选仓库循环)
┌──────────────────────────────────────────┐
│ repo_fetcher.fetch_readme() │ ← GitHub Content API
│ repo_fetcher.fetch_dependencies() │
│ 输出: {readme, dependencies} │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ repo_evaluator.evaluate_repo() │ ← DeepSeek API (每个仓库 1 次)
│ 输出: 六维度评分（可复现性 5 维 + │
│ 对比实验适配度 1 维） │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ 按 overall_score 降序 │
│ format_report() → 研究方向全景研报 │
│ 包含：方向解读 + 方法族谱系 + 仓库排行 │
│ + 对比实验推荐 │
└──────────────────────────────────────────┘
7.2 API 调用次数估算（单次运行）
| 步骤 | API | 调用次数 |
|---|---|---|
| 论文信息 | arxiv API | 1 |
| 研究方向解析 | DeepSeek | 1 |
| GitHub 搜索 | GitHub Search | N（方法族×2 + 宽泛×2 ≈ 10 次） |
| 仓库详情（×5） | GitHub Content | 5 |
| 可复现性+适配度评估（×5） | DeepSeek | 5 |
单次运行总耗时估算：约 25-45 秒（串行执行）。主要耗时在 GitHub 搜索（10 次 × 每次约 1-2 秒）和 5 个仓库的评估（5 次 LLM 调用 × 每次 3-5 秒）。
八、关键节点与里程碑
| 周次 | 目标 | 甲（数据层） | 乙（LLM+集成层） | 交付物 |
|---|---|---|---|---|
| Week 1 | 地基 + 三层 MVP | paper_fetcher.py + repo_searcher.py + repo_fetcher.py + 评测集 Part A 标 3 篇 + Part B 标 6 个 |
direction_analyzer.py + repo_evaluator.py + run.py + app.py + 架构图 |
python run.py <arxiv_url> 完整链路 + 研究方向全景研报 |
| Week 2 | 端到端链路 | Workflow 优化（LLM 判断依赖是否还能装）、加 Tavily 社区搜索、加 Issue 抓取 | Agent C 初版（水贴/真Bug 分类，不做回环）、完善方法族归类逻辑 | 完整链路 + 评测集基线准确率 |
| Week 3 | 准确率攻关 | 持续扩充评测集（论文→仓库 ground truth）、优化数据采集质量 | Prompt 调优、方法族识别准确率、六维权重用评测集反推 | 研究方向识别准确率 ≥ 80%、可复现性准确率 ≥ 75% |
| Week 4 | 回环 + 增强搜索 | Gradio 界面升级（方法族分块展示）、增加 PapersWithCode 数据源 | Agent C 回环机制、Agent D 边界分数再迭代 | 带回环版，准确率 ≥ 80% |
| Week 5 | UI 升级 + 结营准备 | 录制 demo 视频、整理汇报材料（准备 2 篇论文的完整全景研报对比） | 流式输出、按方法族分块展示 | 结营答辩材料 |
| Week 6 | 缓冲 | 查漏补缺、评测集新问题修复 | 同上 | 最终答辩 |
九、文件结构
ResearchRadar/
├── paper_fetcher.py # 甲：arxiv 论文信息抓取
├── repo_searcher.py # 甲：GitHub 代码搜索（多关键词去重）
├── repo_fetcher.py # 甲：仓库详情抓取（README + 依赖 + Issues）
├── direction_analyzer.py # 乙：LLM 研究方向解析 + 方法族梳理 + 搜索词生成
├── repo_evaluator.py # 乙：LLM 可复现性（5维）+ 对比实验适配度（1维）评估
├── run.py # 乙：三层主流程调度
├── app.py # 乙：Gradio 界面 + 研究方向全景研报格式化
├── eval.py # 甲：评测脚本（方向识别准确率 + 可复现性准确率）
├── test_set/
│ ├── papers.json # Part A：论文→研究方向 ground truth + 已知实现仓库
│ └── repos.json # Part B：仓库→可复现性 ground truth
├── requirements.txt # 项目依赖：requests, gradio, openai
└── README.md
**requirements.txt**：
requests>=2.31
gradio>=4.0
openai>=1.0
十、第一周五天时间线
| 天 | 甲 | 乙 |
|---|---|---|
| 周一 | 写 paper_fetcher.py，在 5 篇论文上测试 arxiv API。申请 GitHub Token |
写 direction_analyzer.py prompt，用 3-5 篇论文测试：子领域识别是否合理？方法族是否覆盖全？搜索词质量如何？ |
| 周二 | 写 repo_searcher.py + repo_fetcher.py，联调搜索→抓取链路。两个 token 都配好轮换 |
写 repo_evaluator.py prompt，用 5 个已知仓库测试六维评分是否合理，特别注意"可复现"和"适配度"有没有区分开 |
| 周三 | 准备 3 篇 demo 论文 + 标 Part A 评测集（方向 + 方法族 + 已知实现仓库），确保每篇搜到的仓库有代表性 | 写 run.py 完整三层链路 + app.py 全景研报格式化，下午与甲联调 |
| 周四 | 标 Part B 评测集（6-8 个仓库），处理边界 case（论文搜不到、仓库没 README、方向识别错误） | debug 整条链路：时间消耗、JSON 解析失败、API 异常处理。准备 PPT 三层架构图 |
| 周五上午 | 完整走一遍 demo 流程 + 准备讲稿（3 篇论文各讲什么故事：方向全景→代码发现→对比实验推荐） | 同上，联调 + 答辩预演 |
| 周五下午 | 汇报 | 汇报 |
十一、你们需要决定的事情
在开始写代码之前，下面几个问题需要甲和乙坐下来对清楚。不用一次回答完，可以边写边决定：
甲和乙各自申请 GitHub Token 了吗？ 搜索接口限速很严（30次/分钟），两个 token 轮换是最低配置。如果没有，周一第一件事就去 github.com/settings/tokens 各申请一个 Fine-grained token（只勾选 public repo 读权限）。
DeepSeek API key 现在就能用吗？ 谁拿着 key？周一上午先用最简单的
curl或 Python 脚本验证 key 有效。这件事阻塞所有 LLM 模块。3 篇 demo 论文选什么？ 建议选一篇经典论文（如 Transformer、ResNet、DDPM，方法族丰富、开源实现多、演示效果好），再从你们各自方向各选一篇。关键是要确保每篇论文都能搜到 3+ 个仓库，否则 demo 时没东西展示。
方向解析（direction_analyzer）的评测标准怎么定？ 这是整个系统最"主观"的模块。子领域定位对不对？方法族全不全？搜索词好不好？你们需要定一个简单的判定规则，比如"子领域由人工确认正确"、"方法族至少覆盖 3 个该领域的公认主流方向"。
"对比实验适配度"的 ground truth 怎么标？ 这个比"可复现性"更难标注。可复现性可以试着装环境跑一下，适配度需要判断"这个仓库能直接跑出论文里的指标吗"。建议先用你们自己熟悉的仓库标 5 个，再扩展到 20 个。
论文太新搜不到代码怎么处理？ 新论文（如 2025 年的）往往没有开源实现，但这恰恰是工具的价值——提前告诉你"这个方向还没有成熟代码，你可能要从头实现"。演示时如果遇到这种情况，你应该怎么讲这个故事？
最后一句：你们是研一小白，这完全不丢人。6 分钟演示，评委不会被花哨 UI 打动，但会被「清楚的逻辑 + 真实的数据 + 诚实的局限」说服。别伪装，别造假。你们的价值是"帮研究者节省从论文到跑通对比实验的时间"，这个价值一句话就能讲清楚。