Fetch paper sources concurrently instead of one at a time

The daily pull was almost entirely blocked on network waits taken in series.
Measured on the same query (4 arXiv categories + 12 journals): 488.57s before,
98.95s after -- 4.9x, about six and a half minutes saved per pull. Through the
GUI a real pull now completes in 29s.

Four levels were serial:

- The five sources (arXiv, OpenReview, journals, Semantic Scholar, custom RSS)
  ran one after another, so wall clock was their sum -- journals alone took 91s
  while arXiv sat idle. They now run together, each still failing in isolation.
- arXiv categories were fetched one per request in a loop: 13.13s for four
  against ~1.07s for one, now 0.73s (18x). The fetcher holds no shared mutable
  state and already degrades on 429.
- Journals were fetched one publisher at a time (91.66s for twelve).
- Within a single journal, every article's detail page was fetched serially,
  which dominated that journal's runtime (Science 48s, Science Advances 60s for
  ten papers). Now 25s and 45s. The remainder is the publishers' own latency:
  a single detail fetch has a ~5.1s median, so this is near the floor rather
  than something more workers would fix.

Embedding batches were also serial: 120 uncached texts took 126.52s, now 34.54s
(3.7x). The failure path is unchanged -- a raising request still falls back to
hash embeddings for every missing index and flips provider/model, verified
directly, because silently returning holes would corrupt scoring.

Ordering is preserved everywhere by using executor.map rather than as_completed:
downstream dedup keeps the first occurrence, so arXiv must still win over a later
journal copy of the same paper. A test finishes work in reverse order to catch a
regression back to completion-ordered collection.

Concurrency is bounded and tunable rather than unlimited, since these are
polite-use endpoints: PAPERFLOW_JOURNAL_CONCURRENCY (6, across publishers),
PAPERFLOW_JOURNAL_DETAIL_CONCURRENCY (4, deliberately lower because it hits one
host), PAPERFLOW_EMBED_CONCURRENCY (4, since 429s here would trip the hash
fallback and quietly degrade recommendations), and the existing
PAPERFLOW_MAX_CONCURRENCY for arXiv. Single-item work skips the pool entirely.

Journal entry collection now gathers candidates before fetching details, so the
per-journal limit is applied to accepted papers after _skip filtering, with a
small headroom so concurrent detail fetching cannot leave a result short.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: chengtan9907

deployments/feishu/daily-push-agent/main.py

@@ -17,7 +17,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, date
 from pathlib import Path
-from typing import Any, Callable, List, Dict, Optional, Set
+from typing import Any, Callable, List, Dict, Optional, Set, Tuple
 from dataclasses import dataclass
 from collections import defaultdict
 
@@ -1203,9 +1203,9 @@ def fetch_arxiv_papers(fetch_days: int) -> List[Dict]:
             return []
         start_day = (today - timedelta(days=max(1, int(fetch_days or 1)) - 1)).date()
         start_date = start_day.strftime("%Y%m%d")
-        papers: List[Dict] = []
         per_category_limit = max(1, int(math.ceil(limit_per_source / max(1, len(arxiv_categories)))))
-        for category in arxiv_categories:
+
+        def fetch_one_category(category: str) -> List[Dict]:
             print(f"Fetching from arXiv {category} ({start_date} to {end_date})...")
             category_papers = arxiv_fetch_by_date(
                 start_date=start_date,
@@ -1227,6 +1227,20 @@ def fetch_arxiv_papers(fetch_days: int) -> List[Dict]:
                     paper for paper in category_papers
                     if _paper_date_in_window(paper, start_day, end_day)
                 ]
+            return category_papers
+
+        papers: List[Dict] = []
+        # One independent API call per category; the serial loop this replaces
+        # measured 13.1s for 4 categories against ~1.1s for one. executor.map
+        # preserves category order so downstream dedup stays deterministic.
+        max_workers = min(_env_positive_int("PAPERFLOW_MAX_CONCURRENCY", default=5), len(arxiv_categories))
+        if len(arxiv_categories) > 1 and max_workers > 1:
+            with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                results = list(executor.map(fetch_one_category, arxiv_categories))
+        else:
+            results = [fetch_one_category(category) for category in arxiv_categories]
+
+        for category_papers in results:
             papers.extend(category_papers)
             if category_papers:
                 emit_fetched("arxiv", [{**paper, "source": "arxiv"} for paper in category_papers])
@@ -1309,54 +1323,62 @@ def fetch_cached_arxiv_papers() -> List[Dict]:
             return []
 
     all_papers = []
-    arxiv_papers = []
-    openreview_papers = []
-    journal_papers = []
-    semantic_papers = []
-    custom_rss_papers = []
-
-    # 1. 从 arXiv 获取论文
-    try:
-        arxiv_papers = fetch_arxiv_papers(days)
-        all_papers.extend(arxiv_papers)
-    except Exception as e:
-        print(f"  arXiv fetch error: {e}")
 
-    # 2. 从 OpenReview 获取会议论文
-    try:
-        openreview_papers = fetch_openreview_papers(days)
-        all_papers.extend(openreview_papers)
-    except Exception as e:
-        print(f"  OpenReview fetch error: {e}")
+    # The five sources are independent network calls, so running them one after
+    # another made the wall clock their sum -- journals alone measured 91s while
+    # arXiv sat idle. Each still fails in isolation: a raising source contributes
+    # nothing and the others complete.
+    def run_semantic_scholar() -> List[Dict]:
+        papers = fetch_semantic_scholar_papers(
+            semantic_queries or [], days, limit_per_source, target_date=target_date
+        )
+        if papers:
+            emit_fetched("semantic_scholar", papers)
+        emit({"phase": "source_complete", "source": "semantic_scholar", "count": len(papers)})
+        return papers
 
-    # 3. 从期刊获取论文
-    try:
-        journal_papers = fetch_journal_papers(days)
-        all_papers.extend(journal_papers)
-    except Exception as e:
-        print(f"  Journal fetch error: {e}")
+    def run_custom_rss() -> List[Dict]:
+        papers = fetch_custom_rss_papers(custom_rss_urls or [], limit_per_source)
+        if papers:
+            emit_fetched("custom_rss", papers)
+        emit({"phase": "source_complete", "source": "custom_rss", "count": len(papers)})
+        return papers
 
-    # 4. Optional Semantic Scholar search, enabled from desktop settings.
+    source_jobs: List[Tuple[str, Callable[[], List[Dict]]]] = [
+        ("arXiv", lambda: fetch_arxiv_papers(days)),
+        ("OpenReview", lambda: fetch_openreview_papers(days)),
+        ("Journal", lambda: fetch_journal_papers(days)),
+    ]
     if enable_semantic_scholar:
-        try:
-            semantic_papers = fetch_semantic_scholar_papers(semantic_queries or [], days, limit_per_source, target_date=target_date)
-            all_papers.extend(semantic_papers)
-            if semantic_papers:
-                emit_fetched("semantic_scholar", semantic_papers)
-            emit({"phase": "source_complete", "source": "semantic_scholar", "count": len(semantic_papers)})
-        except Exception as e:
-            print(f"  Semantic Scholar fetch error: {e}")
-
-    # 5. Optional user-provided RSS/Atom feeds.
+        source_jobs.append(("Semantic Scholar", run_semantic_scholar))
     if enable_custom_rss:
+        source_jobs.append(("Custom RSS", run_custom_rss))
+
+    def run_source(job: Tuple[str, Callable[[], List[Dict]]]) -> List[Dict]:
+        label, fetcher = job
         try:
-            custom_rss_papers = fetch_custom_rss_papers(custom_rss_urls or [], limit_per_source)
-            all_papers.extend(custom_rss_papers)
-            if custom_rss_papers:
-                emit_fetched("custom_rss", custom_rss_papers)
-            emit({"phase": "source_complete", "source": "custom_rss", "count": len(custom_rss_papers)})
-        except Exception as e:
-            print(f"  Custom RSS fetch error: {e}")
+            return fetcher() or []
+        except Exception as exc:
+            print(f"  {label} fetch error: {exc}")
+            return []
+
+    if len(source_jobs) > 1:
+        with ThreadPoolExecutor(max_workers=len(source_jobs)) as executor:
+            source_results = list(executor.map(run_source, source_jobs))
+    else:
+        source_results = [run_source(job) for job in source_jobs]
+
+    # Preserve the original source ordering in all_papers: dedup downstream keeps
+    # the first occurrence, so arXiv must still win over a later journal copy.
+    by_label = dict(zip([label for label, _ in source_jobs], source_results))
+    # arXiv / OpenReview / journal results are consulted by the stale-fallback
+    # widening below, so they stay bound. Semantic Scholar and custom RSS have no
+    # such fallback and are already folded into all_papers.
+    arxiv_papers = by_label.get("arXiv", [])
+    openreview_papers = by_label.get("OpenReview", [])
+    journal_papers = by_label.get("Journal", [])
+    for papers in source_results:
+        all_papers.extend(papers)
 
     if not all_papers:
         all_papers.extend(fetch_cached_arxiv_papers())

skills/embedding/scripts/embed.py

@@ -14,8 +14,9 @@
 import json
 import os
 import re
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import requests
 
@@ -507,6 +508,23 @@ def embed_text(self, text: str) -> List[float]:
         self._save_cached(normalized_text, embedding)
         return embedding
 
+    def _batch_concurrency(self, batch_count: int) -> int:
+        """Concurrent embedding requests to issue.
+
+        Kept modest by default: these all hit one provider endpoint, and
+        over-parallelising invites 429s that would trip the hash fallback and
+        silently degrade recommendation quality. Override with
+        PAPERFLOW_EMBED_CONCURRENCY.
+        """
+        if batch_count <= 1:
+            return 1
+        raw = str(os.environ.get("PAPERFLOW_EMBED_CONCURRENCY", "")).strip()
+        try:
+            configured = int(raw) if raw else 4
+        except ValueError:
+            configured = 4
+        return max(1, min(configured, batch_count, 8))
+
     def embed_batch(self, texts: List[str], batch_size: int = 32) -> List[List[float]]:
         normalized_texts = [self._normalize_text(text) for text in texts]
         results: List[Optional[List[float]]] = [None] * len(normalized_texts)
@@ -523,16 +541,36 @@ def embed_batch(self, texts: List[str], batch_size: int = 32) -> List[List[float
                 missing_indices.append(index)
 
         if missing_indices and self.provider == "openai" and self.client is not None:
+            batches = [
+                missing_indices[batch_start:batch_start + batch_size]
+                for batch_start in range(0, len(missing_indices), batch_size)
+            ]
+
+            def embed_one_batch(batch_indices: List[int]) -> List[Tuple[int, List[float]]]:
+                batch_texts = [normalized_texts[index] for index in batch_indices]
+                kwargs: Dict[str, Any] = {"model": self.model, "input": batch_texts}
+                if self.model.startswith("text-embedding-3"):
+                    kwargs["dimensions"] = self.dimensions
+                response = self.client.embeddings.create(**kwargs)
+                return [
+                    (index, self._resize_vector(list(item.embedding)))
+                    for index, item in zip(batch_indices, response.data)
+                ]
+
             try:
-                for batch_start in range(0, len(missing_indices), batch_size):
-                    batch_indices = missing_indices[batch_start:batch_start + batch_size]
-                    batch_texts = [normalized_texts[index] for index in batch_indices]
-                    kwargs: Dict[str, Any] = {"model": self.model, "input": batch_texts}
-                    if self.model.startswith("text-embedding-3"):
-                        kwargs["dimensions"] = self.dimensions
-                    response = self.client.embeddings.create(**kwargs)
-                    for index, item in zip(batch_indices, response.data):
-                        embedding = self._resize_vector(list(item.embedding))
+                # Batches are independent API round trips; issuing them serially
+                # meant 120 uncached texts took 126.5s. The OpenAI client is
+                # thread-safe and each cache entry is its own file, so the only
+                # shared state is the pre-sized results list, written by index.
+                workers = self._batch_concurrency(len(batches))
+                if len(batches) > 1 and workers > 1:
+                    with ThreadPoolExecutor(max_workers=workers) as executor:
+                        batch_results = list(executor.map(embed_one_batch, batches))
+                else:
+                    batch_results = [embed_one_batch(batch) for batch in batches]
+
+                for pairs in batch_results:
+                    for index, embedding in pairs:
                         results[index] = embedding
                         self._save_cached(normalized_texts[index], embedding)
             except Exception as exc:

skills/journal-fetcher/scripts/fetch_journal.py

@@ -31,6 +31,7 @@
 
 import requests
 import re
+from concurrent.futures import ThreadPoolExecutor
 from html import unescape
 from datetime import datetime, timedelta
 from typing import Dict, List, Optional, Any, Tuple
@@ -133,6 +134,11 @@
 
 JOURNAL_REQUEST_TIMEOUT = float(os.environ.get("JOURNAL_REQUEST_TIMEOUT", "20"))
 JOURNAL_DETAIL_FETCH_ENABLED = os.environ.get("JOURNAL_DETAIL_FETCH_ENABLED", "1").strip().lower() not in {"0", "false", "off", "no"}
+# Extra RSS entries collected beyond `limit` so that candidates dropped by the
+# detail fetch (_skip: corrections, editorials, paywalled stubs) do not leave the
+# result short. Detail pages are now fetched concurrently, so the cost of a few
+# extra is small; unbounded collection would not be.
+JOURNAL_DETAIL_SKIP_HEADROOM = 5
 HTTP_RETRY_TOTAL = int(os.environ.get("PAPERFLOW_HTTP_RETRIES", "2"))
 HTTP_RETRY_BACKOFF = float(os.environ.get("PAPERFLOW_HTTP_BACKOFF", "0.8"))
 DEFAULT_REQUEST_HEADERS = {
@@ -1206,14 +1212,22 @@ def fetch_journal_papers(
         return []
 
     papers = []
+    # Entries are collected first, then their article pages are fetched
+    # concurrently below.
+    candidates: List[Dict[str, Any]] = []
     cutoff_date = datetime.now() - timedelta(days=days)
 
     try:
         # 解析 RSS Feed
         feed = feedparser.parse(journal_info["rss_url"])
 
         for entry in feed.entries:
-            if len(papers) >= limit:
+            # Bound on collected candidates, not on accepted papers: detail
+            # fetching now happens after this loop. Some candidates are dropped
+            # later by _skip, so this can under-fill relative to the old
+            # behaviour; the extra headroom keeps the common case identical
+            # without fetching an unbounded number of article pages.
+            if len(candidates) >= limit + JOURNAL_DETAIL_SKIP_HEADROOM:
                 break
 
             # 提取论文信息
@@ -1227,7 +1241,7 @@ def fetch_journal_papers(
             if not _looks_like_research_article(entry, journal_key):
                 continue
 
-            paper = {
+            candidates.append({
                 "title": entry.get('title', 'Unknown'),
                 "abstract": _clean_abstract_text(entry.get('summary', entry.get('description', ''))),
                 "authors": extract_authors_from_rss(entry),
@@ -1238,12 +1252,33 @@ def fetch_journal_papers(
                 "publish_date": published.isoformat(),
                 "categories": [journal_key],
                 "source": "rss",
-            }
-            detail = _fetch_article_detail(paper["url"])
+            })
+
+        # Enrich from the article pages. This was one blocking request per paper,
+        # which dominated a single journal's runtime (Science measured 48s,
+        # Science Advances 60s for 10 papers). _http_get builds and closes its own
+        # session per call, so there is no shared state to guard. executor.map
+        # keeps feed order, and _skip filtering stays positional.
+        detail_workers = _journal_detail_concurrency(len(candidates))
+        if len(candidates) > 1 and detail_workers > 1:
+            def fetch_detail(candidate: Dict[str, Any]) -> Dict[str, Any]:
+                try:
+                    return _fetch_article_detail(candidate["url"])
+                except Exception as exc:
+                    print(f"  Detail fetch error for {candidate.get('url')}: {exc}")
+                    return {}
+
+            with ThreadPoolExecutor(max_workers=detail_workers) as executor:
+                details = list(executor.map(fetch_detail, candidates))
+        else:
+            details = [_fetch_article_detail(candidate["url"]) for candidate in candidates]
+
+        for candidate, detail in zip(candidates, details):
             if detail.get("_skip"):
                 continue
-            paper = _merge_detail_into_paper(paper, detail)
-            papers.append(paper)
+            papers.append(_merge_detail_into_paper(candidate, detail))
+            if len(papers) >= limit:
+                break
 
     except Exception as e:
         print(f"Error fetching {journal_info['name']}: {e}")
@@ -1350,6 +1385,40 @@ def fetch_by_date(
     return unique_papers[:limit]
 
 
+def _journal_detail_concurrency(candidate_count: int) -> int:
+    """Workers for per-article detail fetches within a single journal.
+
+    Unlike the journal-level pool these all hit the *same* publisher host, so the
+    cap is deliberately low. Override with PAPERFLOW_JOURNAL_DETAIL_CONCURRENCY.
+    """
+    if candidate_count <= 1:
+        return 1
+    raw = str(os.environ.get("PAPERFLOW_JOURNAL_DETAIL_CONCURRENCY", "")).strip()
+    try:
+        configured = int(raw) if raw else 4
+    except ValueError:
+        configured = 4
+    return max(1, min(configured, candidate_count, 8))
+
+
+def _journal_fetch_concurrency(source_count: int) -> int:
+    """Workers to use when fetching journals in parallel.
+
+    Each journal is a different publisher host, so this is not hammering a single
+    server. Capped conservatively anyway (and overridable via
+    PAPERFLOW_JOURNAL_CONCURRENCY) because these are polite-use RSS endpoints and
+    some publishers rate-limit aggressively.
+    """
+    if source_count <= 1:
+        return 1
+    raw = str(os.environ.get("PAPERFLOW_JOURNAL_CONCURRENCY", "")).strip()
+    try:
+        configured = int(raw) if raw else 6
+    except ValueError:
+        configured = 6
+    return max(1, min(configured, source_count, 12))
+
+
 def get_recent_papers(
     days: int = 7,
     journals: List[str] = None,
@@ -1379,13 +1448,32 @@ def get_recent_papers(
 
     all_papers = []
 
-    for journal in normalized_journals:
-        papers = fetch_journal_papers(
-            journal=journal,
-            limit=limit_per_journal,
-            days=days
-        )
-        all_papers.extend(papers)
+    # Each journal is an independent RSS fetch plus optional detail-page
+    # enrichment, so the serial loop this replaces spent its time waiting: 12
+    # journals measured 91.66s end to end. executor.map preserves input order,
+    # which keeps the dedup below deterministic -- the first journal listed still
+    # wins for a paper that appears in several feeds.
+    max_workers = _journal_fetch_concurrency(len(normalized_journals))
+    if len(normalized_journals) > 1 and max_workers > 1:
+        def fetch_one(journal: str) -> List[Dict[str, Any]]:
+            try:
+                return fetch_journal_papers(journal=journal, limit=limit_per_journal, days=days)
+            except Exception as exc:
+                # One unreachable feed must not abort the whole pull.
+                print(f"  Journal fetch error for {journal}: {exc}")
+                return []
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            for papers in executor.map(fetch_one, normalized_journals):
+                all_papers.extend(papers)
+    else:
+        for journal in normalized_journals:
+            papers = fetch_journal_papers(
+                journal=journal,
+                limit=limit_per_journal,
+                days=days
+            )
+            all_papers.extend(papers)
 
     # 去重
     seen = set()