#!/usr/bin/env python3 """ research67/run.py — 简化版 Runner =================================== 用本地工具 stub 替代尚不存在的 MCP 服务,直接通过 lambdagent runtime 执行 research67 多智能体科研流程。 使用 Claude Opus 4.6 (1M context) 的原生能力: - web_search → 替代 scholar-mcp / arxiv-mcp - python_exec → 替代 code-sandbox-mcp / plotting-mcp - local fs → 替代 fs-mcp - local latex → 替代 latex-mcp 用法: export ANTHROPIC_API_KEY=sk-ant-... python run.py # 完整流程 python run.py --phase idea_analysis # 只跑某个阶段 python run.py --resume # 从上次检查点恢复 """ from __future__ import annotations import argparse import json import os import subprocess import sys import time import shutil from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional # ── 路径设置 ────────────────────────────────────────────── PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent # Paper08/ AGENT_DIR = Path(__file__).resolve().parent # research67/ sys.path.insert(0, str(PROJECT_ROOT)) from lambdagent.fromconfig import from_config from lambdagent.agentruntime.runtime import Runtime, RuntimeResult from lambdagent.agentruntime.config import RuntimeConfig from lambdagent.core import Context # ═══════════════════════════════════════════════════════════ # 1. 本地工具 Stubs — 替代 MCP 服务 # ═══════════════════════════════════════════════════════════ class LocalToolkit: """ 将 6 个 MCP 服务的功能用本地实现替代。 每个方法签名对齐 MCP tool 的 (input: str) -> str 接口。 """ def __init__(self, workspace: Path): self.workspace = workspace # ── fs-mcp 替代 ────────────────────────────────────── def read_file(self, path: str) -> str: p = self._resolve(path) if not p.exists(): return f"[ERROR] File not found: {p}" return p.read_text(encoding="utf-8") def write_file(self, args: str) -> str: """args = JSON: {"path": "...", "content": "..."}""" d = json.loads(args) p = self._resolve(d["path"]) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(d["content"], encoding="utf-8") return f"Written {len(d['content'])} bytes to {p}" def list_dir(self, path: str) -> str: p = self._resolve(path) if not p.is_dir(): return f"[ERROR] Not a directory: {p}" entries = sorted(p.iterdir()) return "\n".join(f"{'[DIR] ' if e.is_dir() else ''}{e.name}" for e in entries) def mkdir(self, path: str) -> str: p = self._resolve(path) p.mkdir(parents=True, exist_ok=True) return f"Created: {p}" def copy_file(self, args: str) -> str: d = json.loads(args) src, dst = self._resolve(d["src"]), self._resolve(d["dst"]) dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) return f"Copied {src} → {dst}" # ── scholar-mcp / arxiv-mcp 替代 ───────────────────── def semantic_scholar_search(self, query: str) -> str: """用 Semantic Scholar 公开 API 搜索论文 (无需 API key)""" import urllib.request import urllib.parse url = f"https://api.semanticscholar.org/graph/v1/paper/search?query={urllib.parse.quote(query)}&limit=10&fields=title,authors,year,citationCount,venue,abstract,externalIds" try: req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) papers = data.get("data", []) results = [] for p in papers: results.append({ "title": p.get("title", ""), "authors": [a.get("name", "") for a in (p.get("authors") or [])[:3]], "year": p.get("year"), "venue": p.get("venue", ""), "citationCount": p.get("citationCount", 0), "abstract": (p.get("abstract") or "")[:200], "arxivId": (p.get("externalIds") or {}).get("ArXiv", ""), }) return json.dumps({"total": data.get("total", 0), "papers": results}, ensure_ascii=False, indent=2) except Exception as e: return f"[SEARCH_ERROR] {e}" def get_paper_details(self, paper_id: str) -> str: import urllib.request url = f"https://api.semanticscholar.org/graph/v1/paper/{paper_id}?fields=title,abstract,authors,year,venue,citationCount,references,citations" try: req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) return json.dumps(data, ensure_ascii=False, indent=2)[:3000] except Exception as e: return f"[ERROR] {e}" def get_citations(self, paper_id: str) -> str: return self.get_paper_details(paper_id) def get_references(self, paper_id: str) -> str: return self.get_paper_details(paper_id) def arxiv_search(self, query: str) -> str: """用 arXiv API 搜索""" import urllib.request import urllib.parse url = f"http://export.arxiv.org/api/query?search_query=all:{urllib.parse.quote(query)}&start=0&max_results=10&sortBy=submittedDate&sortOrder=descending" try: req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"}) with urllib.request.urlopen(req, timeout=30) as resp: xml = resp.read().decode("utf-8") # Simple XML extraction entries = [] for entry in xml.split("")[1:]: title = _extract_xml(entry, "title").strip().replace("\n", " ") summary = _extract_xml(entry, "summary").strip()[:200] arxiv_id = _extract_xml(entry, "id").split("/abs/")[-1] published = _extract_xml(entry, "published")[:10] entries.append({"title": title, "arxiv_id": arxiv_id, "published": published, "summary": summary}) return json.dumps({"results": entries}, ensure_ascii=False, indent=2) except Exception as e: return f"[ARXIV_ERROR] {e}" def arxiv_download(self, arxiv_id: str) -> str: return f"[STUB] PDF download not implemented for {arxiv_id}. Use arxiv_search for metadata." def pdf_to_text(self, path: str) -> str: return f"[STUB] PDF extraction not implemented for {path}." # ── code-sandbox-mcp 替代 ──────────────────────────── def python_exec(self, code: str) -> str: """在本地沙箱执行 Python 代码""" try: result = subprocess.run( [sys.executable, "-c", code], capture_output=True, text=True, timeout=120, cwd=str(self.workspace), ) output = result.stdout if result.returncode != 0: output += f"\n[STDERR] {result.stderr}" return output[:5000] except subprocess.TimeoutExpired: return "[TIMEOUT] Python execution exceeded 120s" except Exception as e: return f"[EXEC_ERROR] {e}" def install_package(self, pkg: str) -> str: try: result = subprocess.run( [sys.executable, "-m", "pip", "install", pkg, "-q"], capture_output=True, text=True, timeout=60, ) return result.stdout + result.stderr except Exception as e: return f"[PIP_ERROR] {e}" def gpu_status(self, _: str = "") -> str: try: r = subprocess.run(["nvidia-smi", "--query-gpu=name,memory.used,memory.total", "--format=csv,noheader"], capture_output=True, text=True, timeout=10) return r.stdout if r.returncode == 0 else "No GPU detected (CPU mode)" except FileNotFoundError: return "nvidia-smi not found (CPU mode)" def disk_usage(self, path: str = ".") -> str: import shutil as sh total, used, free = sh.disk_usage(self._resolve(path)) return f"Total: {total // (1024**3)}GB, Used: {used // (1024**3)}GB, Free: {free // (1024**3)}GB" # ── latex-mcp 替代 ─────────────────────────────────── def compile_latex(self, tex_path: str) -> str: p = self._resolve(tex_path) if not p.exists(): return f"[ERROR] {p} not found" try: r = subprocess.run( ["pdflatex", "-interaction=nonstopmode", "-output-directory", str(p.parent), str(p)], capture_output=True, text=True, timeout=60, cwd=str(p.parent), ) # Run twice for references subprocess.run( ["pdflatex", "-interaction=nonstopmode", "-output-directory", str(p.parent), str(p)], capture_output=True, text=True, timeout=60, cwd=str(p.parent), ) pdf = p.with_suffix(".pdf") if pdf.exists(): return f"Compiled: {pdf} ({pdf.stat().st_size} bytes)" return f"Compilation output:\n{r.stdout[-2000:]}\n{r.stderr[-1000:]}" except FileNotFoundError: return "[STUB] pdflatex not found. LaTeX compilation skipped." except Exception as e: return f"[LATEX_ERROR] {e}" def latex_lint(self, tex_path: str) -> str: return f"[STUB] LaTeX lint for {tex_path}: no issues found." def bibtex_format(self, bib_path: str) -> str: return f"[STUB] BibTeX formatted: {bib_path}" # ── plotting-mcp 替代 ──────────────────────────────── def create_figure(self, args: str) -> str: """用 matplotlib 生成图表,args 是 JSON spec""" code = f""" import json spec = json.loads('''{args}''') # Placeholder: actual plotting code would be generated by the agent print(f"Figure created: {{spec.get('name', 'figure')}}.pdf") """ return self.python_exec(code) def create_table(self, args: str) -> str: return f"[STUB] Table created from spec" # ── 特殊本地工具 ───────────────────────────────────── def terminate(self, result: str) -> str: return result def read_idea(self, path: str = "") -> str: idea_path = AGENT_DIR / "paper" / "paper01" / "IDEA.md" if not idea_path.exists(): return f"[ERROR] IDEA.md not found at {idea_path}" return idea_path.read_text(encoding="utf-8") def save_checkpoint(self, data: str) -> str: ckpt_dir = self.workspace / ".checkpoints" ckpt_dir.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") path = ckpt_dir / f"checkpoint_{ts}.json" path.write_text(data, encoding="utf-8") return f"Checkpoint saved: {path}" def load_checkpoint(self, pattern: str = "latest") -> str: ckpt_dir = self.workspace / ".checkpoints" if not ckpt_dir.exists(): return "[ERROR] No checkpoints found" files = sorted(ckpt_dir.glob("checkpoint_*.json")) if not files: return "[ERROR] No checkpoints found" return files[-1].read_text(encoding="utf-8") def human_feedback(self, question: str) -> str: print(f"\n{'='*60}") print(f"[HUMAN FEEDBACK REQUESTED]") print(f"{question}") print(f"{'='*60}") try: answer = input("Your response (or press Enter to skip): ").strip() return answer if answer else "[NO FEEDBACK]" except (EOFError, KeyboardInterrupt): return "[NO FEEDBACK]" def dispatch_agent(self, args: str) -> str: """调度子智能体执行""" d = json.loads(args) agent_name = d.get("agent", "") agent_input = d.get("input", "") phase_dir = d.get("phase_dir", "") agent_yml = AGENT_DIR / "agents" / f"{agent_name}.yml" if not agent_yml.exists(): return f"[ERROR] Agent config not found: {agent_yml}" print(f"\n >> Dispatching sub-agent: {agent_name}") print(f" Config: {agent_yml}") print(f" Phase dir: {phase_dir}") try: result = Runtime.execute(str(agent_yml), agent_input) return result.result except Exception as e: return f"[AGENT_ERROR] {agent_name} failed: {e}" # ── 内部辅助 ───────────────────────────────────────── def _resolve(self, path: str) -> Path: p = Path(path) if p.is_absolute(): return p return self.workspace / p def get_tool_map(self) -> Dict[str, callable]: """返回工具名 → 函数的映射,用于注入 runtime""" return { # fs-mcp "read_file": self.read_file, "write_file": self.write_file, "list_dir": self.list_dir, "mkdir": self.mkdir, "copy_file": self.copy_file, # scholar-mcp "semantic_scholar_search": self.semantic_scholar_search, "get_paper_details": self.get_paper_details, "get_citations": self.get_citations, "get_references": self.get_references, # arxiv-mcp "arxiv_search": self.arxiv_search, "arxiv_download": self.arxiv_download, "pdf_to_text": self.pdf_to_text, # code-sandbox-mcp "python_exec": self.python_exec, "install_package": self.install_package, "gpu_status": self.gpu_status, "disk_usage": self.disk_usage, # latex-mcp "compile_latex": self.compile_latex, "latex_lint": self.latex_lint, "bibtex_format": self.bibtex_format, # plotting-mcp "create_figure": self.create_figure, "create_table": self.create_table, # local "terminate": self.terminate, "read_idea": self.read_idea, "save_checkpoint": self.save_checkpoint, "load_checkpoint": self.load_checkpoint, "human_feedback": self.human_feedback, "dispatch_agent": self.dispatch_agent, } # ═══════════════════════════════════════════════════════════ # 2. Runner — 驱动整个科研流程 # ═══════════════════════════════════════════════════════════ PHASES = [ ("01_idea_analysis", "idea-analyst"), ("02_literature_review", "lit-searcher"), ("03_experiment_plan", "exp-planner"), ("04_experiment_execution","exp-executor"), ("05_result_analysis", "result-analyzer"), ("06_paper_writing", "paper-writer"), ("07_data_verification", "data-verifier"), ("08_paper_review", "paper-reviewer"), ] def create_workspace() -> Path: ts = datetime.now().strftime("%Y%m%d_%H%M%S") ws = AGENT_DIR / "workspace" / f"run_{ts}" ws.mkdir(parents=True, exist_ok=True) return ws def run_phase( phase_name: str, agent_name: str, workspace: Path, round_num: int, toolkit: LocalToolkit, prev_reports: List[str], revision_context: str = "", ) -> Dict[str, Any]: """运行单个阶段的子智能体""" phase_dir = workspace / f"round_{round_num}" / phase_name phase_dir.mkdir(parents=True, exist_ok=True) (phase_dir / "artifacts").mkdir(exist_ok=True) agent_yml = AGENT_DIR / "agents" / f"{agent_name}.yml" # 构建输入上下文 context_parts = [ f"workspace: {workspace}", f"phase_dir: {phase_dir}", f"round: {round_num}", f"dependencies: {json.dumps(prev_reports)}", ] if revision_context: context_parts.append(f"revision_context: {revision_context}") # 读取 IDEA idea_content = toolkit.read_idea() context_parts.append(f"\n## IDEA.md Content:\n{idea_content}") # 附上前序阶段的关键结果摘要 for rp in prev_reports[-3:]: # 最多带最近 3 个阶段 rp_path = workspace / rp if rp_path.exists(): try: data = json.loads(rp_path.read_text(encoding="utf-8")) # 只取 _meta + 关键字段,避免太长 summary = {k: v for k, v in data.items() if k == "_meta" or not isinstance(v, (list, dict))} context_parts.append(f"\n## Previous report ({rp}):\n{json.dumps(summary, ensure_ascii=False, indent=2)[:2000]}") except Exception: pass input_text = "\n".join(context_parts) print(f"\n{'━'*60}") print(f" Phase: {phase_name}") print(f" Agent: {agent_name} → {agent_yml.name}") print(f" Output: {phase_dir}") print(f"{'━'*60}") t0 = time.time() try: result = Runtime.execute(str(agent_yml), input_text) duration = time.time() - t0 print(f" ✓ Completed in {duration:.1f}s, {len(result.trace)} trace steps") # 保存 trace trace_path = phase_dir / "trace.json" trace_data = [ {"step": t.step, "term": t.term_name, "type": t.term_type, "duration_ms": t.duration_ms, "output": t.output[:200]} for t in result.trace ] trace_path.write_text(json.dumps(trace_data, indent=2), encoding="utf-8") # 尝试解析 report.json report_path = phase_dir / "report.json" if report_path.exists(): report = json.loads(report_path.read_text(encoding="utf-8")) else: # Agent 可能直接返回了 JSON try: report = json.loads(result.result) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") except (json.JSONDecodeError, TypeError): report = {"_meta": {"status": "completed"}, "raw_output": result.result[:3000]} report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") return report except Exception as e: duration = time.time() - t0 print(f" ✗ Failed after {duration:.1f}s: {e}") error_report = { "_meta": {"status": "failed", "error": str(e), "duration_seconds": duration}, } (phase_dir / "report.json").write_text(json.dumps(error_report, indent=2), encoding="utf-8") return error_report def run_full_pipeline( start_phase: Optional[str] = None, max_rounds: int = 5, ): """运行完整的科研流程""" # 检查 API key if not os.environ.get("ANTHROPIC_API_KEY"): print("ERROR: ANTHROPIC_API_KEY environment variable not set.") print(" export ANTHROPIC_API_KEY=sk-ant-...") sys.exit(1) workspace = create_workspace() toolkit = LocalToolkit(workspace) print(f""" ╔══════════════════════════════════════════════════════════╗ ║ Research-467 Autonomous Research Agent ║ ║ Model: Claude Opus 4.6 (1M context) ║ ║ IDEA: paper/paper01/IDEA.md ║ ╚══════════════════════════════════════════════════════════╝ Workspace: {workspace} Max rounds: {max_rounds} """) # 写入项目计划 idea = toolkit.read_idea() project_plan = f"""# Research-467 Project Plan ## IDEA Source paper/paper01/IDEA.md ## Model Claude Opus 4.6 (1M context) — claude-opus-4-6 ## Phases {chr(10).join(f'{i+1}. {name} → {agent}' for i, (name, agent) in enumerate(PHASES))} ## Target CCF-A conference acceptance probability ≥ 50% ## Created {datetime.now(timezone.utc).isoformat()} """ (workspace / "project_plan.md").write_text(project_plan, encoding="utf-8") progress = { "project_id": workspace.name, "current_round": 1, "current_phase": PHASES[0][0], "max_rounds": max_rounds, "best_score": 0.0, "rounds": [], } for round_num in range(1, max_rounds + 1): print(f"\n{'═'*60}") print(f" ROUND {round_num} / {max_rounds}") print(f"{'═'*60}") progress["current_round"] = round_num round_reports = [] prev_reports = [] # 确定本轮需要跑哪些阶段 phases_to_run = PHASES if round_num > 1 and "revision_target" in progress: # 回退到指定阶段 target = progress["revision_target"] start_idx = next((i for i, (name, _) in enumerate(PHASES) if name == target), 0) phases_to_run = PHASES[start_idx:] print(f" Revision: starting from {target}") if start_phase and round_num == 1: start_idx = next((i for i, (name, _) in enumerate(PHASES) if start_phase in name), 0) phases_to_run = PHASES[start_idx:] for phase_name, agent_name in phases_to_run: progress["current_phase"] = phase_name _save_progress(workspace, progress) report = run_phase( phase_name, agent_name, workspace, round_num, toolkit, prev_reports, revision_context=progress.get("revision_context", ""), ) report_rel = f"round_{round_num}/{phase_name}/report.json" prev_reports.append(report_rel) round_reports.append({"phase": phase_name, "status": report.get("_meta", {}).get("status", "unknown")}) progress["rounds"].append({"round": round_num, "phases": round_reports}) # 检查评审结果 review_report_path = workspace / f"round_{round_num}" / "08_paper_review" / "report.json" if review_report_path.exists(): review = json.loads(review_report_path.read_text(encoding="utf-8")) score = review.get("acceptance_probability", 0) progress["best_score"] = max(progress.get("best_score", 0), score) print(f"\n Review Score: {score:.2%}") print(f" Best Score: {progress['best_score']:.2%}") if score >= 0.5: print(f"\n ✓ TARGET REACHED! Acceptance probability: {score:.2%}") _finalize(workspace, round_num, toolkit, progress) break # 决定回退目标 scores = review.get("scores", {}) if scores: lowest = min(scores.items(), key=lambda x: x[1].get("score", 10) if isinstance(x[1], dict) else 10) dimension = lowest[0] target_map = { "novelty": "01_idea_analysis", "soundness": "03_experiment_plan", "clarity": "06_paper_writing", "significance": "02_literature_review", "reproducibility": "04_experiment_execution", } progress["revision_target"] = target_map.get(dimension, "06_paper_writing") revision_items = review.get("revision_priority", []) progress["revision_context"] = json.dumps(revision_items[:3], ensure_ascii=False) print(f" Lowest dimension: {dimension} → Retrying from {progress['revision_target']}") else: print(f"\n [WARN] No review report found, ending pipeline") break else: # 达到 max_rounds print(f"\n Max rounds ({max_rounds}) reached. Finalizing best version.") _finalize(workspace, round_num, toolkit, progress) _save_progress(workspace, progress) print(f"\n{'═'*60}") print(f" Pipeline complete. Workspace: {workspace}") print(f"{'═'*60}\n") def _save_progress(workspace: Path, progress: dict): (workspace / "progress.json").write_text( json.dumps(progress, ensure_ascii=False, indent=2), encoding="utf-8" ) def _finalize(workspace: Path, last_round: int, toolkit: LocalToolkit, progress: dict): """归档最终产出到 final/ 目录""" final_dir = workspace / "final" final_dir.mkdir(exist_ok=True) round_dir = workspace / f"round_{last_round}" # 复制论文 for src_name, dst_name in [ ("06_paper_writing/artifacts/paper.tex", "paper.tex"), ("06_paper_writing/artifacts/paper.pdf", "paper.pdf"), ("06_paper_writing/artifacts/references.bib", "references.bib"), ]: src = round_dir / src_name if src.exists(): shutil.copy2(src, final_dir / dst_name) # 复制图表 figs_src = round_dir / "05_result_analysis" / "artifacts" / "figures" if figs_src.exists(): shutil.copytree(figs_src, final_dir / "figures", dirs_exist_ok=True) # 复制实验数据 data_src = round_dir / "04_experiment_execution" / "artifacts" / "raw_results" if data_src.exists(): shutil.copytree(data_src, final_dir / "experiment_data", dirs_exist_ok=True) # 生成投稿建议 recommendation = { "final_paper_path": "final/paper.pdf", "total_rounds": last_round, "best_score": progress.get("best_score", 0), "workspace": str(workspace), "completed_at": datetime.now(timezone.utc).isoformat(), } (final_dir / "submission_recommendation.json").write_text( json.dumps(recommendation, ensure_ascii=False, indent=2), encoding="utf-8" ) progress["final_output"] = str(final_dir) print(f"\n Final output archived to: {final_dir}") # ═══════════════════════════════════════════════════════════ # 3. XML 辅助 # ═══════════════════════════════════════════════════════════ def _extract_xml(text: str, tag: str) -> str: start = text.find(f"<{tag}>") end = text.find(f"") if start == -1 or end == -1: return "" return text[start + len(f"<{tag}>"):end] # ═══════════════════════════════════════════════════════════ # 4. CLI # ═══════════════════════════════════════════════════════════ def main(): parser = argparse.ArgumentParser(description="Research-467 Simplified Runner") parser.add_argument("--phase", type=str, default=None, help="Start from a specific phase (e.g., idea_analysis, paper_writing)") parser.add_argument("--max-rounds", type=int, default=5, help="Maximum iteration rounds (default: 5)") parser.add_argument("--test-tools", action="store_true", help="Test local toolkit connectivity and exit") args = parser.parse_args() if args.test_tools: _test_tools() return run_full_pipeline(start_phase=args.phase, max_rounds=args.max_rounds) def _test_tools(): """测试本地工具是否可用""" ws = Path("/tmp/research67_test") ws.mkdir(exist_ok=True) toolkit = LocalToolkit(ws) tests = [ ("read_idea", lambda: toolkit.read_idea()), ("semantic_scholar_search", lambda: toolkit.semantic_scholar_search("spectral analysis pretrained features")), ("arxiv_search", lambda: toolkit.arxiv_search("label-free feature quality")), ("python_exec", lambda: toolkit.python_exec("import sys; print(f'Python {sys.version}')")), ("write_file", lambda: toolkit.write_file(json.dumps({"path": "test.txt", "content": "hello"}))), ("read_file", lambda: toolkit.read_file("test.txt")), ("list_dir", lambda: toolkit.list_dir(".")), ("gpu_status", lambda: toolkit.gpu_status()), ] print("Testing local toolkit...\n") for name, fn in tests: try: t0 = time.time() result = fn() dt = time.time() - t0 ok = "[ERROR]" not in result and "[STUB]" not in result status = "✓" if ok else "⚠" print(f" {status} {name:30s} ({dt:.1f}s) {result[:80]}") except Exception as e: print(f" ✗ {name:30s} ERROR: {e}") shutil.rmtree(ws, ignore_errors=True) print("\nDone.") if __name__ == "__main__": main()