# 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 ` 能跑通完整链路——从论文到推荐仓库。 ### 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` 等前缀 ```python # 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 可能被不同关键词搜到多次) ```python # 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 个候选仓库),需要处理部分仓库抓取失败的情况——不能因为一个仓库失败了就让整个流程崩溃。 ```python # 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 仓库)。 ```json { "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 个负样本。 ```json { "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 ```python # 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 的代码"。 ```python # 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 ") 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` ```python # 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 ` 完整链路 + 研究方向全景研报 | | **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 篇论文各讲什么故事:方向全景→代码发现→对比实验推荐) | 同上,联调 + 答辩预演 | | **周五下午** | **汇报** | **汇报** | --- ## 十一、你们需要决定的事情 在开始写代码之前,下面几个问题需要甲和乙坐下来对清楚。不用一次回答完,可以边写边决定: 1. **甲和乙各自申请 GitHub Token 了吗?** 搜索接口限速很严(30次/分钟),两个 token 轮换是最低配置。如果没有,周一第一件事就去 [github.com/settings/tokens](https://github.com/settings/tokens) 各申请一个 Fine-grained token(只勾选 public repo 读权限)。 2. **DeepSeek API key 现在就能用吗?** 谁拿着 key?周一上午先用最简单的 `curl` 或 Python 脚本验证 key 有效。这件事阻塞所有 LLM 模块。 3. **3 篇 demo 论文选什么?** 建议选一篇经典论文(如 Transformer、ResNet、DDPM,方法族丰富、开源实现多、演示效果好),再从你们各自方向各选一篇。关键是要确保每篇论文都能搜到 3+ 个仓库,否则 demo 时没东西展示。 4. **方向解析(direction_analyzer)的评测标准怎么定?** 这是整个系统最"主观"的模块。子领域定位对不对?方法族全不全?搜索词好不好?你们需要定一个简单的判定规则,比如"子领域由人工确认正确"、"方法族至少覆盖 3 个该领域的公认主流方向"。 5. **"对比实验适配度"的 ground truth 怎么标?** 这个比"可复现性"更难标注。可复现性可以试着装环境跑一下,适配度需要判断"这个仓库能直接跑出论文里的指标吗"。建议先用你们自己熟悉的仓库标 5 个,再扩展到 20 个。 6. **论文太新搜不到代码怎么处理?** 新论文(如 2025 年的)往往没有开源实现,但这恰恰是工具的价值——提前告诉你"这个方向还没有成熟代码,你可能要从头实现"。演示时如果遇到这种情况,你应该怎么讲这个故事? --- > **最后一句**:你们是研一小白,这完全不丢人。6 分钟演示,评委不会被花哨 UI 打动,但会被「清楚的逻辑 + 真实的数据 + 诚实的局限」说服。别伪装,别造假。你们的价值是"帮研究者节省从论文到跑通对比实验的时间",这个价值一句话就能讲清楚。