Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import html | |
| import io | |
| import json | |
| import os | |
| import re | |
| import time | |
| import uuid | |
| import zipfile | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Literal | |
| from urllib.parse import quote_plus | |
| from xml.etree import ElementTree | |
| import gradio as gr | |
| import httpx | |
| import spaces | |
| from huggingface_hub import InferenceClient | |
| from pydantic import BaseModel, Field, ValidationError, model_validator | |
| from pypdf import PdfReader | |
| BASE_DIR = Path(__file__).resolve().parent | |
| STATIC_DIR = BASE_DIR / "static" | |
| DATA_DIR = Path("/data/research-decks") if Path("/data").exists() else BASE_DIR / "output" | |
| DATA_DIR.mkdir(parents=True, exist_ok=True) | |
| MODEL_ID = os.getenv("HF_MODEL_ID", "deepseek-ai/DeepSeek-V4-Flash") | |
| HF_PROVIDER = os.getenv("HF_PROVIDER", "auto") | |
| SEARCH_CACHE_TTL = 3600 | |
| SEARCH_CACHE: dict[tuple[str, str, int], tuple[float, list[dict[str, Any]]]] = {} | |
| SEARCH_LAST_CALL: dict[str, float] = {} | |
| def zerogpu_runtime_probe() -> str: | |
| """Satisfy ZeroGPU Space validation; production inference uses HF Providers.""" | |
| return "ready" | |
| class ResearchRequest(BaseModel): | |
| topic: str = Field(min_length=2, max_length=180) | |
| purpose: str = Field(default="算法团队内部技术分享", max_length=500) | |
| audience: str = Field(default="熟悉机器学习基础的算法工程师", max_length=300) | |
| focus: str = Field(default="", max_length=1200) | |
| date_range: str = Field(default="经典工作至今", max_length=120) | |
| depth: str = Field(default="standard", pattern="^(quick|standard|deep)$") | |
| language: str = Field(default="中文", max_length=40) | |
| paper_urls: list[str] = Field(default_factory=list, max_length=12) | |
| class EvidenceCard(BaseModel): | |
| paper_id: str = Field(min_length=1) | |
| title: str = Field(min_length=2) | |
| year: str = Field(min_length=1) | |
| relevance: str = Field(min_length=3) | |
| problem: str = Field(min_length=10) | |
| core_idea: str = Field(min_length=10) | |
| delta: str = Field(min_length=10) | |
| evidence: list[str] = Field(min_length=1, max_length=4) | |
| limitations: list[str] = Field(min_length=1, max_length=4) | |
| class EvidenceCardsEnvelope(BaseModel): | |
| cards: list[EvidenceCard] = Field(min_length=1) | |
| SectionLayout = Literal[ | |
| "text", "timeline", "comparison", "process", "equation", | |
| "evidence", "paper-analysis", "evidence-gap", | |
| ] | |
| class SectionPlan(BaseModel): | |
| id: str = Field(min_length=2) | |
| title: str = Field(min_length=2, max_length=120) | |
| purpose: str = Field(min_length=10, max_length=500) | |
| layout: SectionLayout | |
| source_ids: list[str] = Field(default_factory=list) | |
| class ReportOutline(BaseModel): | |
| title: str = Field(min_length=2, max_length=160) | |
| subtitle: str = Field(min_length=2, max_length=240) | |
| research_type: str = Field(min_length=3, max_length=80) | |
| executive_summary: list[str] = Field(min_length=1, max_length=6) | |
| section_plan: list[SectionPlan] = Field(min_length=3, max_length=12) | |
| class ReportSection(BaseModel): | |
| id: str = Field(min_length=2) | |
| title: str = Field(min_length=2, max_length=120) | |
| purpose: str = Field(min_length=10, max_length=500) | |
| takeaway: str = Field(min_length=10, max_length=800) | |
| layout: SectionLayout | |
| content: dict[str, Any] | |
| source_ids: list[str] = Field(default_factory=list) | |
| def validate_layout_content(self) -> "ReportSection": | |
| content = self.content | |
| if not content or len(json.dumps(content, ensure_ascii=False).strip()) < 20: | |
| raise ValueError("section content must be substantive") | |
| if self.layout == "timeline" and len(content.get("nodes", [])) < 2: | |
| raise ValueError("timeline content requires at least two nodes") | |
| if self.layout == "timeline" and not all(isinstance(node, dict) for node in content.get("nodes", [])): | |
| raise ValueError("timeline nodes must be objects") | |
| if self.layout == "comparison" and ( | |
| len(content.get("columns", [])) < 2 or len(content.get("rows", [])) < 2 | |
| ): | |
| raise ValueError("comparison content requires columns and at least two rows") | |
| if self.layout == "comparison" and not all(isinstance(row, dict) for row in content.get("rows", [])): | |
| raise ValueError("comparison rows must be objects") | |
| if self.layout in {"paper-analysis", "evidence-gap"}: | |
| meaningful_blocks = [value for value in content.values() if value not in (None, "", [], {})] | |
| if len(meaningful_blocks) < 2: | |
| raise ValueError(f"{self.layout} requires at least two substantive information blocks") | |
| return self | |
| class SectionsEnvelope(BaseModel): | |
| sections: list[ReportSection] = Field(min_length=1, max_length=4) | |
| class ResearchReport(BaseModel): | |
| title: str = Field(min_length=2, max_length=160) | |
| subtitle: str = Field(min_length=2, max_length=240) | |
| research_type: str = Field(min_length=3, max_length=80) | |
| executive_summary: list[str] = Field(min_length=1, max_length=6) | |
| sections: list[ReportSection] = Field(min_length=3, max_length=12) | |
| references: list[str] = Field(default_factory=list) | |
| limitations: list[str] = Field(default_factory=list) | |
| english_understanding: dict[str, Any] = Field(default_factory=dict) | |
| DEPTHS = { | |
| "quick": {"discover": 24, "screen": 10, "deep": 2, "queries": 2}, | |
| "standard": {"discover": 40, "screen": 20, "deep": 4, "queries": 3}, | |
| "deep": {"discover": 60, "screen": 35, "deep": 8, "queries": 4}, | |
| } | |
| def safe_slug(value: str) -> str: | |
| slug = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower() | |
| return (slug or "algorithm-topic")[:64] | |
| def http_get_with_retries( | |
| url: str, | |
| *, | |
| headers: dict[str, str] | None = None, | |
| read_timeout: float = 90, | |
| attempts: int = 3, | |
| ) -> httpx.Response: | |
| last_error: Exception | None = None | |
| timeout = httpx.Timeout(connect=20, read=read_timeout, write=30, pool=30) | |
| for attempt in range(attempts): | |
| try: | |
| with httpx.Client( | |
| timeout=timeout, | |
| follow_redirects=True, | |
| headers=headers, | |
| http2=False, | |
| ) as client: | |
| response = client.get(url) | |
| response.raise_for_status() | |
| return response | |
| except httpx.HTTPStatusError as exc: | |
| last_error = exc | |
| retryable = exc.response.status_code == 429 or exc.response.status_code >= 500 | |
| if not retryable or attempt + 1 >= attempts: | |
| raise | |
| retry_after = exc.response.headers.get("retry-after", "") | |
| delay = float(retry_after) if retry_after.replace(".", "", 1).isdigit() else 3 + attempt * 2 | |
| time.sleep(min(delay, 8)) | |
| except (httpx.TimeoutException, httpx.NetworkError, httpx.RemoteProtocolError) as exc: | |
| last_error = exc | |
| if attempt + 1 < attempts: | |
| time.sleep(2 + attempt * 3) | |
| raise RuntimeError(f"External source did not respond after {attempts} attempts: {url}") from last_error | |
| def arxiv_search(topic: str, limit: int) -> list[dict[str, Any]]: | |
| query = quote_plus(f'all:"{topic}"') | |
| url = ( | |
| "https://export.arxiv.org/api/query" | |
| f"?search_query={query}&start=0&max_results={limit}" | |
| "&sortBy=submittedDate&sortOrder=descending" | |
| ) | |
| headers = {"User-Agent": f"algorithm-research-deck/1.0 ({os.getenv('CONTACT_EMAIL', 'contact-not-set')})"} | |
| response = http_get_with_retries(url, headers=headers, read_timeout=35, attempts=2) | |
| ns = {"a": "http://www.w3.org/2005/Atom"} | |
| root = ElementTree.fromstring(response.text) | |
| papers = [] | |
| for entry in root.findall("a:entry", ns): | |
| abs_url = (entry.findtext("a:id", default="", namespaces=ns) or "").strip() | |
| authors = [x.findtext("a:name", default="", namespaces=ns) for x in entry.findall("a:author", ns)] | |
| papers.append({ | |
| "id": abs_url.rsplit("/", 1)[-1], | |
| "title": " ".join((entry.findtext("a:title", default="", namespaces=ns) or "").split()), | |
| "summary": " ".join((entry.findtext("a:summary", default="", namespaces=ns) or "").split()), | |
| "authors": authors, | |
| "published": entry.findtext("a:published", default="", namespaces=ns), | |
| "updated": entry.findtext("a:updated", default="", namespaces=ns), | |
| "url": abs_url, | |
| "pdf_url": abs_url.replace("/abs/", "/pdf/") + ".pdf", | |
| }) | |
| return papers | |
| def openalex_search(query: str, limit: int) -> list[dict[str, Any]]: | |
| email = os.getenv("CONTACT_EMAIL", "").strip() | |
| api_key = os.getenv("OPENALEX_API_KEY", "").strip() | |
| url = f"https://api.openalex.org/works?search={quote_plus(query)}&per-page={min(limit, 100)}" | |
| if email: | |
| url += f"&mailto={quote_plus(email)}" | |
| if api_key: | |
| url += f"&api_key={quote_plus(api_key)}" | |
| response = http_get_with_retries(url, read_timeout=35, attempts=2) | |
| results = response.json().get("results", []) | |
| papers = [] | |
| for work in results: | |
| primary = work.get("primary_location") or {} | |
| source = primary.get("source") or {} | |
| open_access = work.get("open_access") or {} | |
| best_oa = work.get("best_oa_location") or {} | |
| landing = primary.get("landing_page_url") or work.get("doi") or work.get("id") or "" | |
| pdf_url = best_oa.get("pdf_url") or primary.get("pdf_url") or "" | |
| authors = [ | |
| ((item.get("author") or {}).get("display_name") or "") | |
| for item in work.get("authorships", []) | |
| ] | |
| inverted = work.get("abstract_inverted_index") or {} | |
| ordered_words = sorted( | |
| ((position, word) for word, positions in inverted.items() for position in positions), | |
| key=lambda item: item[0], | |
| ) | |
| abstract = " ".join(word for _, word in ordered_words) | |
| papers.append({ | |
| "id": (work.get("doi") or work.get("id") or landing).rsplit("/", 1)[-1], | |
| "title": work.get("title") or "Untitled", | |
| "summary": abstract, | |
| "authors": [x for x in authors if x], | |
| "published": work.get("publication_date") or "", | |
| "updated": "", | |
| "url": landing, | |
| "pdf_url": pdf_url, | |
| "venue": source.get("display_name") or "", | |
| "citation_count": work.get("cited_by_count") or 0, | |
| "open_access": open_access.get("is_oa", False), | |
| "source": "OpenAlex", | |
| }) | |
| return papers | |
| def normalize_user_papers(urls: list[str]) -> list[dict[str, Any]]: | |
| result = [] | |
| for raw in urls: | |
| match = re.search(r"arxiv\.org/(?:abs|pdf)/([\w.\-/]+)", raw) | |
| if match: | |
| paper_id = match.group(1).removesuffix(".pdf") | |
| result.append({ | |
| "id": paper_id, | |
| "title": f"User supplied arXiv paper {paper_id}", | |
| "summary": "User-supplied paper; prioritize during screening.", | |
| "authors": [], | |
| "published": "", | |
| "updated": "", | |
| "url": f"https://arxiv.org/abs/{paper_id}", | |
| "pdf_url": f"https://arxiv.org/pdf/{paper_id}.pdf", | |
| "user_supplied": True, | |
| }) | |
| return result | |
| def deduplicate(papers: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| seen = set() | |
| output = [] | |
| for paper in papers: | |
| normalized_title = re.sub(r"[^a-z0-9]+", "", paper.get("title", "").lower()) | |
| key = normalized_title or re.sub(r"v\d+$", "", paper["id"]) | |
| if key not in seen: | |
| seen.add(key) | |
| output.append(paper) | |
| return output | |
| def plan_search_queries(request: ResearchRequest) -> tuple[list[str], dict[str, Any]]: | |
| prompt = json.dumps({ | |
| "topic": request.topic, | |
| "purpose": request.purpose, | |
| "audience": request.audience, | |
| "focus": request.focus, | |
| "date_range": request.date_range, | |
| "language": request.language, | |
| "instruction": ( | |
| "First translate and normalize every user field into precise English technical meaning without dropping the original intent. " | |
| "Return english_understanding with topic, purpose, audience, focus, date_range, key_concepts, ambiguities, and scope. " | |
| "Also return search_queries containing 3 to 5 concise English academic queries. " | |
| "Cover historical/classical terminology, milestone method families, current terminology, and one limitation or benchmark angle. " | |
| "Queries must be suitable for arXiv and OpenAlex. Do not include site operators." | |
| ), | |
| }, ensure_ascii=False) | |
| result = chat_json( | |
| "You plan literature searches for an algorithm engineer. Expand ambiguous topic names into precise academic terminology.", | |
| prompt, | |
| 650, | |
| ) | |
| queries = [str(x).strip() for x in result.get("search_queries", []) if str(x).strip()] | |
| english_understanding = result.get("english_understanding") or {} | |
| fallback_query = str(english_understanding.get("topic") or request.topic).strip() | |
| if fallback_query and fallback_query not in queries: | |
| queries.append(fallback_query) | |
| return queries[:6], english_understanding | |
| def search_error_summary(exc: Exception) -> str: | |
| if exc.__cause__ is not None and isinstance(exc.__cause__, Exception): | |
| return search_error_summary(exc.__cause__) | |
| if isinstance(exc, httpx.HTTPStatusError): | |
| return f"HTTP {exc.response.status_code}" | |
| if isinstance(exc, httpx.TimeoutException) or "timed out" in str(exc).lower(): | |
| return "请求超时" | |
| return type(exc).__name__ | |
| def cached_search( | |
| provider: str, | |
| query: str, | |
| limit: int, | |
| search_fn: Any, | |
| ) -> list[dict[str, Any]]: | |
| key = (provider, query.strip().lower(), limit) | |
| cached = SEARCH_CACHE.get(key) | |
| if cached and time.monotonic() - cached[0] < SEARCH_CACHE_TTL: | |
| return [dict(paper) for paper in cached[1]] | |
| minimum_interval = 1.2 if provider == "arxiv" else 0.4 | |
| elapsed = time.monotonic() - SEARCH_LAST_CALL.get(provider, 0.0) | |
| if elapsed < minimum_interval: | |
| time.sleep(minimum_interval - elapsed) | |
| result = search_fn(query, limit) | |
| SEARCH_LAST_CALL[provider] = time.monotonic() | |
| SEARCH_CACHE[key] = (time.monotonic(), [dict(paper) for paper in result]) | |
| return result | |
| def search_literature(queries: list[str], total_limit: int) -> tuple[list[dict[str, Any]], list[str]]: | |
| per_query = max(5, min(20, total_limit // max(1, len(queries)))) | |
| buckets: list[list[dict[str, Any]]] = [] | |
| warnings: list[str] = [] | |
| failures = {"arXiv": 0, "OpenAlex": 0} | |
| for query in queries: | |
| query_papers: list[dict[str, Any]] = [] | |
| if failures["arXiv"] < 2: | |
| try: | |
| found = cached_search("arxiv", query, per_query, arxiv_search) | |
| for paper in found: | |
| paper["source"] = "arXiv" | |
| query_papers.extend(found) | |
| failures["arXiv"] = 0 | |
| except Exception as exc: | |
| failures["arXiv"] += 1 | |
| warnings.append(f"arXiv {search_error_summary(exc)}：{query}") | |
| if len(query_papers) < max(2, per_query // 4) and failures["OpenAlex"] < 2: | |
| try: | |
| query_papers.extend(cached_search("openalex", query, per_query, openalex_search)) | |
| failures["OpenAlex"] = 0 | |
| except Exception as exc: | |
| failures["OpenAlex"] += 1 | |
| warnings.append(f"OpenAlex {search_error_summary(exc)}：{query}") | |
| buckets.append(deduplicate(query_papers)) | |
| for provider, count in failures.items(): | |
| if count >= 2: | |
| warnings.append(f"{provider} 连续失败，已停止剩余重复请求") | |
| balanced: list[dict[str, Any]] = [] | |
| max_bucket = max((len(bucket) for bucket in buckets), default=0) | |
| for index in range(max_bucket): | |
| for bucket in buckets: | |
| if index < len(bucket): | |
| balanced.append(bucket[index]) | |
| balanced = deduplicate(balanced) | |
| if len(balanced) >= total_limit: | |
| break | |
| return deduplicate(balanced)[:total_limit], warnings | |
| def generate_text(system: str, prompt: str, max_tokens: int, structured: bool = True) -> str: | |
| token = os.getenv("HF_TOKEN") | |
| if not token: | |
| raise RuntimeError("Space Secret HF_TOKEN is missing") | |
| client = InferenceClient(provider=HF_PROVIDER, api_key=token, timeout=180) | |
| request_args = { | |
| "model": MODEL_ID, | |
| "messages": [ | |
| {"role": "system", "content": f"{system}\nUse non-thinking mode. Return only the requested final answer."}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| "temperature": 0.2, | |
| "max_tokens": max(2048, max_tokens), | |
| } | |
| if structured: | |
| try: | |
| response = client.chat.completions.create( | |
| **request_args, | |
| response_format={"type": "json_object"}, | |
| ) | |
| except Exception: | |
| response = client.chat.completions.create(**request_args) | |
| else: | |
| response = client.chat.completions.create(**request_args) | |
| message = response.choices[0].message | |
| content = message.content | |
| if isinstance(content, list): | |
| content = "".join( | |
| item.get("text", "") if isinstance(item, dict) else str(item) | |
| for item in content | |
| ) | |
| if not content: | |
| content = getattr(message, "reasoning_content", None) or "" | |
| return str(content).strip() | |
| def parse_json_object(content: str) -> dict[str, Any] | None: | |
| candidates = [content] | |
| fenced = re.sub(r"^```(?:json)?\s*|\s*```$", "", content, flags=re.IGNORECASE) | |
| if fenced != content: | |
| candidates.append(fenced) | |
| line_continuations_removed = re.sub(r"\\\r?\n", "", fenced) | |
| if line_continuations_removed != fenced: | |
| candidates.append(line_continuations_removed) | |
| match = re.search(r"\{[\s\S]*\}", content) | |
| if match: | |
| candidates.append(match.group(0)) | |
| for candidate in candidates: | |
| try: | |
| parsed = json.loads(candidate) | |
| if isinstance(parsed, dict): | |
| return parsed | |
| except (json.JSONDecodeError, TypeError): | |
| continue | |
| return None | |
| def chat_json(system: str, prompt: str, max_tokens: int = 6000) -> dict[str, Any]: | |
| last_content = "" | |
| for attempt in range(2): | |
| retry_rules = "" | |
| retry_prompt = prompt | |
| if attempt: | |
| retry_rules = ( | |
| " The previous response was malformed or degenerated. Regenerate from scratch as compact valid JSON." | |
| " Never put literal or escaped line breaks in string values. Do not repeat whitespace or characters." | |
| " Keep titles under 120 characters and every other string under 500 characters. Close every array and object." | |
| ) | |
| retry_prompt = ( | |
| f"{prompt}\n\nThe previous response could not be parsed. Its beginning was: " | |
| f"{last_content[:320]!r}\nRegenerate the requested object from scratch." | |
| ) | |
| last_content = generate_text(system + retry_rules, retry_prompt, max_tokens) | |
| parsed = parse_json_object(last_content) | |
| if parsed is not None: | |
| return parsed | |
| preview = last_content[:240].replace("\n", " ") if last_content else "<empty response>" | |
| raise RuntimeError( | |
| f"{MODEL_ID} did not return a JSON object. Provider={HF_PROVIDER}. Response preview: {preview}" | |
| ) | |
| def select_papers(request: ResearchRequest, papers: list[dict[str, Any]], screen_limit: int, deep_limit: int) -> list[str]: | |
| compact = [{k: p[k] for k in ("id", "title", "summary", "authors", "published", "url")} for p in papers[:screen_limit]] | |
| prompt = json.dumps({ | |
| "topic": request.topic, | |
| "purpose": request.purpose, | |
| "focus": request.focus, | |
| "papers": compact, | |
| "instruction": f"Select at most {deep_limit} pivotal papers. Return JSON with selected_ids and a short selection_reason per id.", | |
| }, ensure_ascii=False) | |
| result = chat_json("You select technically pivotal ML papers. Prefer coverage of the user's decision and strong evidence over recency alone.", prompt, 900) | |
| valid = {p["id"] for p in papers} | |
| return [x for x in result.get("selected_ids", []) if x in valid][:deep_limit] | |
| def extract_pdf(paper: dict[str, Any], max_chars: int = 14000) -> str: | |
| if not paper.get("pdf_url"): | |
| return "No open PDF URL was available from the academic index." | |
| response = http_get_with_retries(paper["pdf_url"], read_timeout=120, attempts=3) | |
| if len(response.content) > 20 * 1024 * 1024: | |
| return "PDF omitted because it exceeds the 20 MB limit." | |
| reader = PdfReader(io.BytesIO(response.content)) | |
| pages = [] | |
| indexes = list(range(min(8, len(reader.pages)))) | |
| if len(reader.pages) > 8: | |
| indexes += list(range(max(8, len(reader.pages) - 3), len(reader.pages))) | |
| for index in dict.fromkeys(indexes): | |
| text = reader.pages[index].extract_text() or "" | |
| pages.append(f"[Page {index + 1}]\n{text}") | |
| if sum(map(len, pages)) >= max_chars: | |
| break | |
| return "\n".join(pages)[:max_chars] | |
| def serializable_validation_errors(exc: ValidationError) -> list[dict[str, Any]]: | |
| """Remove exception objects and raw inputs before sending validation feedback to the model.""" | |
| return exc.errors(include_url=False, include_context=False, include_input=False) | |
| def validate_stage_with_repairs( | |
| result: dict[str, Any], | |
| model_type: type[BaseModel], | |
| stage_name: str, | |
| source_context: dict[str, Any], | |
| max_tokens: int, | |
| ) -> BaseModel: | |
| candidate = result | |
| for attempt in range(3): | |
| try: | |
| return model_type.model_validate(candidate) | |
| except ValidationError as exc: | |
| if attempt == 2: | |
| raise RuntimeError(f"{stage_name} remained invalid after two repairs: {exc}") from exc | |
| candidate = chat_json( | |
| f"Repair the {stage_name} JSON. Preserve valid content, fill missing fields, and return only the complete corrected object.", | |
| json.dumps({ | |
| "validation_errors": serializable_validation_errors(exc), | |
| "object_to_repair": candidate, | |
| "source_context": source_context, | |
| "schema": model_type.model_json_schema(), | |
| }, ensure_ascii=False), | |
| max_tokens, | |
| ) | |
| raise RuntimeError(f"{stage_name} validation failed") | |
| def align_sections_to_plan(result: dict[str, Any], plans: list[SectionPlan]) -> dict[str, Any]: | |
| """Order model sections by approved IDs and expose omissions to schema repair.""" | |
| candidates = [item for item in result.get("sections", []) if isinstance(item, dict)] | |
| by_id = {str(item.get("id")): item for item in candidates if item.get("id")} | |
| plan_ids = {plan.id for plan in plans} | |
| unused = [item for item in candidates if str(item.get("id", "")) not in plan_ids] | |
| aligned = [] | |
| for plan in plans: | |
| candidate = by_id.get(plan.id) | |
| if candidate is None and unused: | |
| candidate = unused.pop(0) | |
| candidate = dict(candidate or {}) | |
| candidate.update({ | |
| "id": plan.id, | |
| "title": plan.title, | |
| "purpose": plan.purpose, | |
| "layout": plan.layout, | |
| }) | |
| candidate.setdefault("takeaway", "") | |
| candidate.setdefault("content", {}) | |
| candidate.setdefault("source_ids", []) | |
| aligned.append(candidate) | |
| return {"sections": aligned} | |
| def prepare_evidence_inputs( | |
| papers: list[dict[str, Any]], deep_texts: dict[str, str], limit: int | |
| ) -> list[dict[str, Any]]: | |
| """Prioritize deeply read papers and bound evidence-card output size.""" | |
| selected_ids = set(deep_texts) | |
| ordered = sorted(papers, key=lambda paper: paper.get("id") not in selected_ids) | |
| evidence_inputs = [] | |
| for paper in ordered[:limit]: | |
| item = {k: paper.get(k) for k in ("id", "title", "summary", "authors", "published", "updated", "url")} | |
| if paper.get("id") in deep_texts: | |
| item["selected_text_extract"] = deep_texts[paper["id"]][:9000] | |
| evidence_inputs.append(item) | |
| return evidence_inputs | |
| def infer_section_source_ids(section: dict[str, Any], papers: list[dict[str, Any]]) -> list[str]: | |
| """Recover omitted citations only when the section clearly names a retrieved paper or its authors.""" | |
| source_ids = list(section.get("source_ids", [])) | |
| haystack = json.dumps( | |
| {key: section.get(key) for key in ("title", "takeaway", "content")}, ensure_ascii=False | |
| ).lower() | |
| compact_haystack = re.sub(r"\W+", "", haystack) | |
| for paper in papers: | |
| paper_id = paper.get("id") | |
| if not paper_id or paper_id in source_ids: | |
| continue | |
| title = re.sub(r"\W+", "", str(paper.get("title", "")).lower()) | |
| year = str(paper.get("published", ""))[:4] | |
| surnames = { | |
| re.sub(r"[^a-z0-9]", "", str(author).lower().split()[-1]) | |
| for author in paper.get("authors", []) if str(author).strip() | |
| } | |
| mentioned_authors = [name for name in surnames if len(name) >= 4 and name in haystack] | |
| title_match = len(title) >= 12 and title in compact_haystack | |
| author_match = len(mentioned_authors) >= 2 or ( | |
| len(mentioned_authors) == 1 and len(mentioned_authors[0]) >= 6 and year and year in haystack | |
| ) | |
| if title_match or author_match: | |
| source_ids.append(paper_id) | |
| return source_ids | |
| def build_report( | |
| request: ResearchRequest, | |
| papers: list[dict[str, Any]], | |
| deep_texts: dict[str, str], | |
| english_understanding: dict[str, Any], | |
| ) -> dict[str, Any]: | |
| evidence_limit = min(8, DEPTHS[request.depth]["deep"]) | |
| raw_evidence = prepare_evidence_inputs(papers, deep_texts, evidence_limit) | |
| if raw_evidence: | |
| card_payload = { | |
| "request": request.model_dump(), | |
| "papers": raw_evidence, | |
| "requirements": { | |
| "one_card_per_useful_paper": True, | |
| "paraphrase_instead_of_copy": True, | |
| "cite_only_supplied_paper_ids": True, | |
| "schema": EvidenceCardsEnvelope.model_json_schema(), | |
| }, | |
| } | |
| card_prompt = json.dumps(card_payload, ensure_ascii=False) | |
| try: | |
| cards_result = chat_json( | |
| "Create one compact evidence card per supplied paper. Keep only decision-relevant experimental conditions " | |
| "and limitations. Do not invent results, citations, or paper IDs. Return a complete JSON object; stay concise.", | |
| card_prompt, | |
| 3600, | |
| ) | |
| except RuntimeError as first_error: | |
| cards_result = chat_json( | |
| "The previous evidence-card response was incomplete or invalid JSON. Regenerate it more concisely. " | |
| "Return exactly one short card per supplied paper as a complete JSON object with no markdown or commentary.", | |
| json.dumps({ | |
| "previous_error": str(first_error), | |
| "request": request.model_dump(), | |
| "papers": [{k: item.get(k) for k in ("id", "title", "summary", "published")} for item in raw_evidence], | |
| "schema": EvidenceCardsEnvelope.model_json_schema(), | |
| }, ensure_ascii=False), | |
| 3000, | |
| ) | |
| cards = validate_stage_with_repairs( | |
| cards_result, | |
| EvidenceCardsEnvelope, | |
| "evidence cards", | |
| {"request": request.model_dump(), "papers": raw_evidence}, | |
| 3600, | |
| ).model_dump()["cards"] | |
| else: | |
| cards = [] | |
| shared_context = { | |
| "request": request.model_dump(), | |
| "english_understanding": english_understanding, | |
| "research_cutoff": datetime.now(timezone.utc).date().isoformat(), | |
| "evidence_cards": cards, | |
| "evidence_rule": ( | |
| "Use only supplied paper IDs for citations. State evidence gaps instead of inventing support." | |
| if cards else | |
| "Online evidence is disabled. Do not invent paper IDs or claim that recent literature was verified." | |
| ), | |
| } | |
| outline_result = chat_json( | |
| "Act as the editor of an algorithm-engineering research presentation. First infer the most useful research type " | |
| "from the user's topic and intent, then design a topic-specific outline. Return only JSON matching the schema. " | |
| "Do not force a timeline, comparison, equations, or paper review when they do not fit. Choose each section layout " | |
| "deliberately. For unfamiliar or weakly evidenced topics, include an evidence-gap section instead of fabricating facts.", | |
| json.dumps({ | |
| **shared_context, | |
| "requirements": { | |
| "research_type": "Choose a precise type such as concept survey, technical evolution, paper analysis, engineering decision, or exploratory evidence-gap report.", | |
| "executive_summary_items": "1 to 6 substantive conclusions appropriate to available evidence", | |
| "section_plan_items": "3 to 12; structure follows the topic rather than a fixed template", | |
| "allowed_layouts": list(SectionLayout.__args__), | |
| "schema": ReportOutline.model_json_schema(), | |
| }, | |
| }, ensure_ascii=False), | |
| 3000, | |
| ) | |
| outline = validate_stage_with_repairs( | |
| outline_result, | |
| ReportOutline, | |
| "report outline", | |
| shared_context, | |
| 3200, | |
| ) | |
| section_system = ( | |
| "Write presentation-ready sections for algorithm engineers. Use the layout selected in the approved plan and put " | |
| "layout-specific data in content. Text/evidence uses " | |
| "{points:[{heading,detail,subpoints}]} where each point keeps its detail and subpoints together; timeline uses " | |
| "{nodes:[{year,title,problem,innovation,limitation}]}; comparison uses " | |
| "{columns:[...],rows:[{name,values:[...]}]}; process uses {steps:[{name,description}]} or " | |
| "{common_steps:[...],branches:[{name,steps:[...]}]}; equation uses {equation,symbols,interpretation}; " | |
| "paper-analysis uses {method,evidence,limitations}; evidence-gap uses " | |
| "{interpretation,related_topics,hypotheses,required_inputs}. Equivalent topic-appropriate field names are allowed; " | |
| "never add empty placeholder fields merely to match an example. Write for slides, not an essay: one conclusion per " | |
| "section, concise points, explicit assumptions, comparable dimensions, and no repeated background. Distinguish " | |
| "retrieved facts from analysis or hypotheses. Never invent sources, papers, metrics, or recency claims." | |
| ) | |
| generated_sections: list[ReportSection] = [] | |
| plans = outline.section_plan | |
| for batch_index in range(0, len(plans), 4): | |
| batch = plans[batch_index:batch_index + 4] | |
| sections_result = chat_json( | |
| section_system, | |
| json.dumps({ | |
| **shared_context, | |
| "report_identity": { | |
| "title": outline.title, | |
| "research_type": outline.research_type, | |
| "executive_summary": outline.executive_summary, | |
| }, | |
| "approved_section_plan": [item.model_dump() for item in batch], | |
| "requirements": { | |
| "exactly_one_section_per_plan_item_in_order": True, | |
| "preserve_plan_id_title_purpose_layout_exactly": True, | |
| "source_ids": "Only IDs present in evidence_cards; otherwise []", | |
| "schema": SectionsEnvelope.model_json_schema(), | |
| }, | |
| }, ensure_ascii=False), | |
| 4200, | |
| ) | |
| sections_result = align_sections_to_plan(sections_result, batch) | |
| try: | |
| validated_batch = validate_stage_with_repairs( | |
| sections_result, | |
| SectionsEnvelope, | |
| f"report sections batch {batch_index // 4 + 1}", | |
| {**shared_context, "approved_section_plan": [item.model_dump() for item in batch]}, | |
| 4200, | |
| ) | |
| except RuntimeError: | |
| validated_batch = None | |
| if validated_batch is None or len(validated_batch.sections) != len(batch): | |
| recovered_sections: list[ReportSection] = [] | |
| for plan in batch: | |
| one_result = chat_json( | |
| section_system, | |
| json.dumps({ | |
| **shared_context, | |
| "report_identity": { | |
| "title": outline.title, | |
| "research_type": outline.research_type, | |
| "executive_summary": outline.executive_summary, | |
| }, | |
| "approved_section_plan": [plan.model_dump()], | |
| "requirements": { | |
| "return_exactly_one_section": True, | |
| "preserve_plan_identity_and_layout": True, | |
| "source_ids": "Only IDs present in evidence_cards; otherwise []", | |
| "schema": SectionsEnvelope.model_json_schema(), | |
| }, | |
| }, ensure_ascii=False), | |
| 2600, | |
| ) | |
| one_result = align_sections_to_plan(one_result, [plan]) | |
| recovered = validate_stage_with_repairs( | |
| one_result, | |
| SectionsEnvelope, | |
| f"missing report section {plan.id}", | |
| {**shared_context, "approved_section_plan": [plan.model_dump()]}, | |
| 2800, | |
| ) | |
| recovered_sections.append(recovered.sections[0]) | |
| validated_batch = SectionsEnvelope(sections=recovered_sections) | |
| for plan, section in zip(batch, validated_batch.sections): | |
| normalized = section.model_dump() | |
| normalized.update({ | |
| "id": plan.id, "title": plan.title, "purpose": plan.purpose, "layout": plan.layout, | |
| }) | |
| generated_sections.append(ReportSection.model_validate(normalized)) | |
| assembled = { | |
| "title": outline.title, | |
| "subtitle": outline.subtitle, | |
| "research_type": outline.research_type, | |
| "executive_summary": outline.executive_summary, | |
| "sections": [item.model_dump() for item in generated_sections], | |
| "references": sorted({source_id for section in generated_sections for source_id in section.source_ids}), | |
| "limitations": (["Online literature verification was disabled; current-work claims may be incomplete."] if not cards else []), | |
| "english_understanding": english_understanding, | |
| } | |
| allowed_ids = {paper["id"] for paper in papers} | |
| for section in assembled["sections"]: | |
| section["source_ids"] = [ | |
| paper_id for paper_id in infer_section_source_ids(section, papers) if paper_id in allowed_ids | |
| ] | |
| assembled["references"] = sorted({ | |
| paper_id for section in assembled["sections"] for paper_id in section["source_ids"] | |
| }) | |
| return validate_report_with_repairs(assembled, shared_context) | |
| def validate_report_with_repairs(report: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: | |
| candidate = report | |
| for attempt in range(3): | |
| try: | |
| validated = ResearchReport.model_validate(candidate).model_dump() | |
| allowed_ids = {card.get("paper_id") for card in context.get("evidence_cards", [])} | |
| for section in validated["sections"]: | |
| section["source_ids"] = [paper_id for paper_id in section["source_ids"] if paper_id in allowed_ids] | |
| validated["references"] = [paper_id for paper_id in validated["references"] if paper_id in allowed_ids] | |
| return validated | |
| except ValidationError as exc: | |
| if attempt == 2: | |
| raise RuntimeError(f"Report remained incomplete after two repairs: {exc}") from exc | |
| candidate = chat_json( | |
| "Repair an incomplete algorithm research report. Preserve its chosen research type, outline, layouts, and valid content. " | |
| "Fix only schema-invalid fields. Return the complete report as one JSON object and never invent source IDs.", | |
| json.dumps({ | |
| "validation_errors": serializable_validation_errors(exc), | |
| "report_to_repair": candidate, | |
| "evidence_cards": context.get("evidence_cards", []), | |
| "schema": ResearchReport.model_json_schema(), | |
| }, ensure_ascii=False), | |
| 5200, | |
| ) | |
| candidate["english_understanding"] = context.get("english_understanding", {}) | |
| raise RuntimeError("Report validation failed") | |
| def esc(value: Any) -> str: | |
| return html.escape(str(value or "")) | |
| def field_label(key: Any) -> str: | |
| labels = { | |
| "points": "要点", "details": "详细说明", "claim": "结论", "claims": "结论", | |
| "conditions": "适用条件", "evidence": "证据", "limitations": "局限", | |
| "method": "方法", "mechanism": "机制", "problem": "问题", "innovation": "创新", | |
| "tradeoffs": "权衡", "interpretation": "解释", "related_topics": "相关主题", | |
| "hypotheses": "待验证假设", "required_inputs": "所需补充信息", "findings": "发现", | |
| "known_evidence": "已有证据", "unknowns": "证据缺口", "next_steps": "下一步验证", | |
| "open_questions": "待回答问题", "method_summary": "方法概述", "experimental_results": "实验结果", | |
| } | |
| normalized = str(key).strip().lower().replace("-", "_") | |
| return labels.get(normalized, str(key).replace("_", " ").replace("-", " ").strip().title()) | |
| def render_value(value: Any, *, depth: int = 0) -> str: | |
| """Render arbitrary model JSON recursively instead of leaking JSON/Python repr.""" | |
| if value is None or value == "": | |
| return "" | |
| if isinstance(value, list): | |
| if not value: | |
| return "" | |
| return '<div class="content-list">' + "".join( | |
| f'<article class="content-card">{render_value(item, depth=depth + 1)}</article>' | |
| if isinstance(item, dict) else | |
| f'<div class="content-item">{render_value(item, depth=depth + 1)}</div>' | |
| for item in value | |
| ) + "</div>" | |
| if isinstance(value, dict): | |
| if not value: | |
| return "" | |
| heading = value.get("heading") or value.get("title") or value.get("name") | |
| detail = value.get("detail") or value.get("description") or value.get("text") | |
| subpoints = value.get("subpoints") or value.get("bullets") or value.get("items") | |
| if detail is not None or subpoints is not None: | |
| consumed = {"heading", "title", "name", "detail", "description", "text", "subpoints", "bullets", "items"} | |
| extras = {key: item for key, item in value.items() if key not in consumed} | |
| return ( | |
| (f'<h4 class="point-heading">{esc(heading)}</h4>' if heading else "") | |
| + (f'<div class="point-detail">{render_value(detail, depth=depth + 1)}</div>' if detail is not None else "") | |
| + (f'<div class="point-subpoints">{render_value(subpoints, depth=depth + 1)}</div>' if subpoints is not None else "") | |
| + (render_value(extras, depth=depth + 1) if extras else "") | |
| ) | |
| return '<div class="content-object">' + "".join( | |
| f'<div class="content-field"><h4>{esc(field_label(key))}</h4>{render_value(item, depth=depth + 1)}</div>' | |
| for key, item in value.items() if item is not None and item != "" | |
| ) + "</div>" | |
| if isinstance(value, bool): | |
| return f'<p>{"是" if value else "否"}</p>' | |
| return f"<p>{esc(value)}</p>" | |
| def normalize_comparison(content: dict[str, Any]) -> tuple[str, list[str], list[tuple[str, list[Any]]]]: | |
| columns = [str(column) for column in content.get("columns", [])] | |
| raw_rows = content.get("rows", []) | |
| row_keys = ("name", "dimension", "criterion", "metric", "title", "method") | |
| row_label = "对比维度" | |
| max_values = max( | |
| (len(row.get("values", [])) for row in raw_rows if isinstance(row, dict) and isinstance(row.get("values"), list)), | |
| default=0, | |
| ) | |
| first_is_label = bool(columns) and ( | |
| any(word in columns[0].lower() for word in ("维度", "指标", "dimension", "criterion", "metric")) | |
| or (max_values and len(columns) == max_values + 1) | |
| ) | |
| if first_is_label: | |
| row_label, columns = columns[0], columns[1:] | |
| normalized_rows: list[tuple[str, list[Any]]] = [] | |
| for row in raw_rows: | |
| if not isinstance(row, dict): | |
| continue | |
| label = next((str(row[key]) for key in row_keys if row.get(key) not in (None, "")), "") | |
| values = row.get("values") | |
| if isinstance(values, dict): | |
| cells = [values.get(column, "") for column in columns] | |
| elif isinstance(values, list): | |
| cells = list(values) | |
| if len(cells) == len(columns) + 1 and str(cells[0]) == label: | |
| cells = cells[1:] | |
| else: | |
| cells = [row.get(column, "") for column in columns] | |
| cells = (cells + [""] * len(columns))[:len(columns)] | |
| normalized_rows.append((label, cells)) | |
| return row_label, columns, normalized_rows | |
| def normalize_process(content: dict[str, Any]) -> tuple[list[Any], list[dict[str, Any]]]: | |
| if isinstance(content.get("branches"), list): | |
| common = content.get("common_steps") or ([content["start"]] if content.get("start") else []) | |
| return common, [branch for branch in content["branches"] if isinstance(branch, dict)] | |
| steps = content.get("steps", []) | |
| common: list[Any] = [] | |
| grouped: dict[str, list[Any]] = {} | |
| for step in steps: | |
| name = str(step.get("name", "")) if isinstance(step, dict) else str(step) | |
| match = re.match(r"^([^:：]{2,24})[:：]\s*(.+)$", name) | |
| if not match: | |
| common.append(step) | |
| continue | |
| lane, short_name = match.groups() | |
| normalized_step = dict(step) if isinstance(step, dict) else {"name": short_name} | |
| normalized_step["name"] = short_name | |
| grouped.setdefault(lane, []).append(normalized_step) | |
| if len(grouped) < 2: | |
| return [], [{"name": content.get("name", "流程"), "steps": steps}] | |
| return common, [{"name": name, "steps": lane_steps} for name, lane_steps in grouped.items()] | |
| def render_process_node(step: Any, index: str) -> str: | |
| if not isinstance(step, dict): | |
| return f'<article class="flow-node"><span>{esc(index)}</span><p>{esc(step)}</p></article>' | |
| name = step.get("name") or step.get("title") or f"步骤 {index}" | |
| details = {key: value for key, value in step.items() if key not in {"name", "title"}} | |
| return ( | |
| f'<details class="flow-node"><summary><span>{esc(index)}</span><strong>{esc(name)}</strong></summary>' | |
| f'<div class="flow-detail">{render_value(details)}</div></details>' | |
| ) | |
| def render_section_content(section: dict[str, Any]) -> str: | |
| content, layout = section.get("content", {}), section.get("layout", "text") | |
| if layout == "timeline": | |
| return '<div class="timeline">' + "".join( | |
| f'<article class="milestone"><span>{esc(n.get("year") or n.get("stage"))}</span>' | |
| f'<h3>{esc(n.get("title") or n.get("method") or n.get("name"))}</h3>' | |
| f'{render_value({k: v for k, v in n.items() if k not in {"year", "stage", "title", "method", "name"}})}</article>' | |
| for n in content.get("nodes", []) | |
| ) + "</div>" | |
| if layout == "comparison": | |
| row_label, columns, rows = normalize_comparison(content) | |
| head = "".join(f"<th>{esc(x)}</th>" for x in columns) | |
| body = "".join( | |
| f'<tr><th scope="row">{esc(label)}</th>{"".join(f"<td>{render_value(value)}</td>" for value in values)}</tr>' | |
| for label, values in rows | |
| ) | |
| return f'<div class="table-wrap"><table class="comparison-table"><thead><tr><th>{esc(row_label)}</th>{head}</tr></thead><tbody>{body}</tbody></table></div>' | |
| if layout == "process": | |
| common, branches = normalize_process(content) | |
| common_html = "".join(render_process_node(step, str(i)) for i, step in enumerate(common, 1)) | |
| branches_html = "".join( | |
| f'<section class="flow-lane"><h3>{esc(branch.get("name") or branch.get("title") or "分支")}</h3>' | |
| f'{"".join(render_process_node(step, f"{lane_index}.{step_index}") for step_index, step in enumerate(branch.get("steps", []), 1))}</section>' | |
| for lane_index, branch in enumerate(branches, 1) | |
| ) | |
| fork = '<div class="flow-fork" aria-hidden="true">分支</div>' if common and len(branches) > 1 else "" | |
| return f'<div class="process-flow"><div class="flow-common">{common_html}</div>{fork}<div class="flow-branches">{branches_html}</div></div>' | |
| if layout == "equation": | |
| return (f'<pre class="equation">{esc(content.get("equation"))}</pre><p>{esc(content.get("interpretation"))}</p>' | |
| f'{render_value(content.get("symbols"))}') | |
| wrapper = "aside" if layout == "evidence-gap" else "div" | |
| return f'<{wrapper} class="{esc(layout)}">' + "".join( | |
| f'<h3>{esc(field_label(k))}</h3>{render_value(v)}' for k, v in content.items() | |
| ) + f"</{wrapper}>" | |
| def render_presentation(report: dict[str, Any], papers: list[dict[str, Any]], cutoff: str) -> str: | |
| deck_css = "\n".join( | |
| (STATIC_DIR / name).read_text(encoding="utf-8") | |
| for name in ("deck.css", "dynamic-content.css") | |
| ) | |
| deck_js = (STATIC_DIR / "deck.js").read_text(encoding="utf-8").replace("</script", "<\\/script") | |
| citations = {p["id"]: p for p in papers} | |
| evidence_mode = (report.get("english_understanding") or {}).get("evidence_mode", "") | |
| evidence_note = "未联网核验：内容基于模型已有知识，最新进展未经外部验证。" if evidence_mode else "已联网检索，并基于当前可用论文证据生成。" | |
| evidence_label = "模型知识 · 未联网核验" if evidence_mode else f"论文证据 · {len(report.get('references', []))} 篇引用" | |
| summary_html = "".join( | |
| f'<article class="summary-card"><span>{i:02d}</span><p>{esc(item)}</p></article>' | |
| for i, item in enumerate(report.get("executive_summary", []), 1) | |
| ) | |
| agenda_items = "".join( | |
| f'<a href="#section-{i + 1}" data-slide-target="section-{i + 1}"><span>{i + 1:02d}</span>' | |
| f'<strong>{esc(section.get("title"))}</strong><small>{esc(section.get("layout"))}</small></a>' | |
| for i, section in enumerate(report.get("sections", [])) | |
| ) | |
| section_html = "".join( | |
| f'<section class="slide content-slide" id="section-{i + 1}" data-title="{esc(x.get("title"))}">' | |
| f'<header class="slide-header"><p class="eyebrow">{i + 3:02d} / {esc(x.get("layout"))}</p><h2>{esc(x.get("title"))}</h2>' | |
| f'<p class="takeaway">{esc(x.get("takeaway"))}</p></header><div class="slide-content">{render_section_content(x)}</div>' | |
| f'<footer class="slide-footer"><div class="source-chips">' | |
| f'{"".join(f"<a href=\"#ref-{esc(pid)}\" title=\"查看参考文献\">{esc(pid)}</a>" for pid in x.get("source_ids", [])) or "<span>本页未绑定外部论文</span>"}' | |
| f'</div><span>{i + 3:02d}</span></footer></section>' | |
| for i, x in enumerate(report.get("sections", [])) | |
| ) | |
| refs = "".join( | |
| f'<li id="ref-{esc(pid)}"><span class="ref-index">{i:02d}</span><div><a href="{esc(citations[pid].get("url"))}" target="_blank" rel="noopener noreferrer">{esc(citations[pid].get("title"))}</a>' | |
| f'<br><span>{esc(", ".join(citations[pid].get("authors", [])[:6]))} · {esc((citations[pid].get("published") or "")[:10])} · {esc(pid)}</span></div></li>' | |
| for i, pid in enumerate((pid for pid in report.get("references", []) if pid in citations), 1) | |
| ) | |
| limitations = "".join(f"<li>{esc(x)}</li>" for x in report.get("limitations", [])) | |
| title, subtitle = esc(report.get("title", "Algorithm Research")), esc(report.get("subtitle", "")) | |
| return f'''<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"> | |
| <title>{title}</title><style>{deck_css}</style></head><body> | |
| <nav class="deck-nav"><button id="menu" aria-label="打开目录" aria-expanded="false">☰</button><strong>{title}</strong><span id="counter"></span><button id="prev" aria-label="上一页">←</button><button id="next" aria-label="下一页">→</button><button id="print" title="打印或保存为 PDF">保存 PDF</button></nav> | |
| <aside id="overview" class="overview" aria-hidden="true"><header><strong>演示目录</strong><button id="close-overview" aria-label="关闭目录">×</button></header><a data-slide-target="cover">封面</a><a data-slide-target="summary">核心结论</a><a data-slide-target="agenda">研究结构</a>{agenda_items}<a data-slide-target="references">参考文献与局限</a></aside><div id="scrim" class="scrim"></div> | |
| <main><section class="slide cover active" id="cover" data-title="封面"><div class="cover-orb"></div><p class="eyebrow">ALGORITHM RESEARCH / {esc(cutoff)}</p><h1>{title}</h1><p class="subtitle">{subtitle}</p><div class="cover-meta"><span>{esc(report.get("research_type"))}</span><span>{esc(evidence_label)}</span></div><p class="evidence-note">{esc(evidence_note)}</p><p class="hint">方向键翻页 · 点击左上角查看目录 · 可打印保存</p></section> | |
| <section class="slide" id="summary" data-title="核心结论"><p class="eyebrow">01 / EXECUTIVE SUMMARY</p><h2>先看结论</h2><div class="summary-grid">{summary_html}</div></section> | |
| <section class="slide" id="agenda" data-title="研究结构"><p class="eyebrow">02 / RESEARCH MAP</p><h2>这份调研如何展开</h2><div class="agenda-grid">{agenda_items}</div></section>{section_html} | |
| <section class="slide references" id="references" data-title="参考文献与局限"><p class="eyebrow">SOURCES / {esc(cutoff)}</p><h2>参考文献与边界</h2><p class="evidence-note">{esc(evidence_note)}</p><ol>{refs or '<li class="empty-state">本报告未绑定外部论文来源。</li>'}</ol><ul class="limitations">{limitations}</ul></section></main> | |
| <div class="progress"><span id="progress-bar"></span></div><script>{deck_js}</script></body></html>''' | |
| def create_bundle(job_dir: Path, html_path: Path, report_path: Path) -> Path: | |
| bundle = job_dir / "research-package.zip" | |
| with zipfile.ZipFile(bundle, "w", zipfile.ZIP_DEFLATED) as archive: | |
| archive.write(html_path, html_path.name) | |
| archive.write(report_path, report_path.name) | |
| return bundle | |
| def preview_frame(content: str) -> str: | |
| source = html.escape(content, quote=True) | |
| return f'<iframe title="Interactive research presentation" srcdoc="{source}" style="width:100%;height:80vh;border:0;border-radius:14px;background:#071018;box-shadow:0 14px 40px #12283c2b" sandbox="allow-scripts allow-popups allow-modals"></iframe>' | |
| def run_research_ui( | |
| topic: str, | |
| purpose: str, | |
| audience: str, | |
| focus: str, | |
| date_range: str, | |
| online_search: bool, | |
| depth: str, | |
| language: str, | |
| paper_urls: str, | |
| progress=gr.Progress(), | |
| ): | |
| if not topic or len(topic.strip()) < 2: | |
| raise gr.Error("请先填写至少两个字符的调研主题。") | |
| request = ResearchRequest( | |
| topic=topic.strip(), | |
| purpose=purpose, | |
| audience=audience, | |
| focus=focus, | |
| date_range=date_range, | |
| depth=depth.lower(), | |
| language=language, | |
| paper_urls=[x.strip() for x in paper_urls.splitlines() if x.strip()], | |
| ) | |
| budget = DEPTHS[request.depth] | |
| progress(0.05, desc="模型正在理解主题并规划检索词") | |
| queries, english_understanding = plan_search_queries(request) | |
| queries = queries[:budget["queries"]] | |
| search_warnings: list[str] = [] | |
| deep_texts: dict[str, str] = {} | |
| if online_search: | |
| progress(0.16, desc="正在联网检索 arXiv 和 OpenAlex") | |
| user_papers = normalize_user_papers(request.paper_urls) | |
| discovered_papers, search_warnings = search_literature(queries, budget["discover"]) | |
| papers = deduplicate(user_papers + discovered_papers) | |
| screen = papers[: budget["screen"]] | |
| if screen: | |
| progress(0.28, desc="DeepSeek-V4-Flash 正在筛选关键论文") | |
| selected_ids = select_papers(request, screen, budget["screen"], budget["deep"]) | |
| if not selected_ids: | |
| selected_ids = [p["id"] for p in screen[: budget["deep"]]] | |
| progress(0.48, desc="正在读取关键论文") | |
| for index, paper in enumerate(screen): | |
| if paper["id"] in selected_ids: | |
| try: | |
| deep_texts[paper["id"]] = extract_pdf(paper) | |
| except Exception as exc: | |
| deep_texts[paper["id"]] = f"PDF extraction failed: {exc}" | |
| progress(0.48 + 0.16 * (index + 1) / len(screen), desc="正在提取论文证据") | |
| else: | |
| search_warnings.append("学术搜索未返回可用论文，已降级为模型知识模式") | |
| english_understanding["evidence_mode"] = "Academic search returned no usable papers; model knowledge only." | |
| progress(0.48, desc="检索无结果，已降级为模型知识模式") | |
| else: | |
| papers = [] | |
| screen = [] | |
| english_understanding["evidence_mode"] = "Model knowledge only; online paper search was disabled by the user." | |
| progress(0.45, desc="已关闭联网检索，使用模型已有知识生成") | |
| progress(0.68, desc="DeepSeek-V4-Flash 正在规划大纲并生成自适应章节") | |
| try: | |
| report = build_report(request, screen, deep_texts, english_understanding) | |
| except (ValidationError, RuntimeError) as exc: | |
| raise gr.Error(f"报告结构生成失败，模型在两次定向修复后仍未满足完整性要求：{exc}") from exc | |
| cutoff = datetime.now(timezone.utc).date().isoformat() | |
| content = render_presentation(report, screen, cutoff) | |
| job_dir = DATA_DIR / uuid.uuid4().hex | |
| job_dir.mkdir(parents=True, exist_ok=True) | |
| slug = safe_slug(request.topic) | |
| html_path = job_dir / f"{slug}-research.html" | |
| report_path = job_dir / f"{slug}-research.json" | |
| html_path.write_text(content, encoding="utf-8") | |
| report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") | |
| bundle_path = create_bundle(job_dir, html_path, report_path) | |
| progress(1, desc="完成") | |
| warning_text = f" 检索降级：{'；'.join(search_warnings)}。" if search_warnings else "" | |
| if online_search: | |
| status = ( | |
| f"✅ 已完成：模型规划 {len(queries)} 组检索词，发现 {len(papers)} 篇，" | |
| f"筛选 {len(screen)} 篇，深读 {len(deep_texts)} 篇。{warning_text}" | |
| ) | |
| else: | |
| status = "✅ 已完成：未联网检索，内容来自模型已有知识，论文与最新性未经过外部核验。" | |
| return status, preview_frame(content), str(html_path), str(report_path), str(bundle_path) | |
| CSS = """ | |
| .gradio-container{max-width:none!important;background:linear-gradient(145deg,#f7f9fc,#eef3f8)!important}.app-hero{margin:4px 4px 18px;padding:18px 22px;border:1px solid #dce5ef;border-radius:16px;background:linear-gradient(120deg,#0c1d2d,#153651);color:#fff;box-shadow:0 12px 35px #16334d1a}.app-hero h1{margin:0 0 5px;color:#fff!important;font-size:25px}.app-hero p{margin:0;color:#fff!important}.app-hero span{display:inline-block;margin-top:10px;padding:4px 8px;border:1px solid #ffffff2b;border-radius:999px;color:#82ead8;font-size:11px} | |
| .preview-column{min-width:0}.input-column{max-width:430px;padding:14px!important;border:1px solid #dce5ef;border-radius:14px;background:#ffffffb8;box-shadow:0 10px 30px #1a38500d} | |
| .preview-toolbar{align-items:center!important;justify-content:flex-end!important;gap:6px!important;min-height:38px} | |
| .preview-title{margin-right:auto!important;min-width:180px}.download-chip{flex:0 0 auto!important;min-width:42px!important;max-width:52px!important;position:relative} | |
| .download-chip button,.download-chip a{min-width:42px!important;padding:5px 8px!important;font-size:12px!important} | |
| .download-chip:hover:after{position:absolute;z-index:50;right:0;top:calc(100% + 7px);width:max-content;max-width:230px;padding:7px 9px;border-radius:7px;background:#161d2b;color:#f3f6fb;box-shadow:0 6px 20px #0006;font-size:12px;line-height:1.35;pointer-events:none} | |
| .download-html:hover:after{content:"下载可独立打开的交互式 HTML 演示"}.download-json:hover:after{content:"下载结构化研究内容与技术路线 JSON"}.download-zip:hover:after{content:"下载包含 HTML 和研究 JSON 的完整压缩包"} | |
| """ | |
| with gr.Blocks(title="Algorithm Research Deck") as demo: | |
| gr.HTML("""<div class="app-hero"><h1>Algorithm Research Deck</h1><p>从问题定义、论文证据到技术判断，生成可保存的交互式研究演示。</p><span>DeepSeek-V4-Flash · arXiv · OpenAlex</span></div>""") | |
| with gr.Row(): | |
| with gr.Column(scale=1, elem_classes="input-column"): | |
| topic = gr.Textbox(label="调研主题", placeholder="例如：Diffusion Transformer") | |
| purpose = gr.Textbox(label="用途", value="算法团队内部技术分享") | |
| audience = gr.Textbox(label="受众", value="熟悉机器学习基础的算法工程师") | |
| focus = gr.Textbox(label="重点关注", lines=3, placeholder="技术路线、推理加速、工程部署……") | |
| with gr.Row(): | |
| depth = gr.Radio(["Quick", "Standard", "Deep"], value="Quick", label="研究深度") | |
| language = gr.Dropdown(["中文", "English"], value="中文", label="输出语言") | |
| date_range = gr.Textbox(label="时间范围", value="经典工作至今") | |
| online_search = gr.Checkbox(label="联网搜索论文（arXiv + OpenAlex）", value=True) | |
| paper_urls = gr.Textbox(label="指定 arXiv 链接", lines=3, placeholder="每行一个，可选") | |
| submit = gr.Button("开始调研", variant="primary") | |
| status = gr.Markdown("等待任务") | |
| with gr.Column(scale=2, elem_classes="preview-column"): | |
| with gr.Row(elem_classes="preview-toolbar"): | |
| gr.Markdown("### 交互演示", elem_classes="preview-title") | |
| html_file = gr.DownloadButton("HTML", size="sm", elem_classes=["download-chip", "download-html"]) | |
| json_file = gr.DownloadButton("JSON", size="sm", elem_classes=["download-chip", "download-json"]) | |
| zip_file = gr.DownloadButton("ZIP", size="sm", elem_classes=["download-chip", "download-zip"]) | |
| preview = gr.HTML("<div style='height:80vh;display:grid;place-items:center;border-radius:14px;background:#071018;color:#758099'>交互演示将在这里生成</div>") | |
| submit.click( | |
| run_research_ui, | |
| inputs=[topic, purpose, audience, focus, date_range, online_search, depth, language, paper_urls], | |
| outputs=[status, preview, html_file, json_file, zip_file], | |
| ) | |
| demo.queue(default_concurrency_limit=1, max_size=8) | |
| if __name__ == "__main__": | |
| demo.launch(theme=gr.themes.Soft(), css=CSS, ssr_mode=False) | |