perf(pipeline): LLM timeout+concurrency, bulk AI persist, off-loop parse, batched events [修复组⑤]

backend/channels/rss_channel.py

@@ -1,5 +1,6 @@
 """RSS channel using feedparser."""
 
+import asyncio
 from typing import Any
 
 import feedparser
@@ -75,7 +76,12 @@ async def collect(
                 f"Failed to fetch RSS feed: {exc}", error_type=type(exc).__name__
             )
 
-        parsed = feedparser.parse(content)
+        # AUDIT C22: feedparser.parse() is a synchronous, potentially
+        # multi-second call for a large feed — running it inline would freeze
+        # the whole event loop (every other request/task on this process)
+        # for the duration. asyncio.to_thread runs it in the default executor
+        # instead; the parsed result is used identically either way.
+        parsed = await asyncio.to_thread(feedparser.parse, content)
         if parsed.bozo and not parsed.entries:
             return ChannelResult.fail(
                 f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}"
@@ -153,7 +159,9 @@ async def fetch(self, ctx: FetchContext) -> FetchResult:
             return FetchResult(items=[], next_cursor=(cursor or None), has_more=False)
         response.raise_for_status()
 
-        parsed = feedparser.parse(response.text)
+        # AUDIT C22: see collect()'s twin comment — off-load the synchronous
+        # parse instead of blocking the event loop.
+        parsed = await asyncio.to_thread(feedparser.parse, response.text)
         if parsed.bozo and not parsed.entries:
             raise ChannelFetchError(
                 f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}"

backend/channels/skill_channel.py

@@ -259,57 +259,62 @@ async def model_call(
     return model_call
 
 
-def _emit_loop_events(run_id: str, result: LoopResult) -> list[Any]:
-    """Build the per-step ``events.emit`` coroutines for a finished loop.
+def _emit_loop_events(result: LoopResult) -> list[dict[str, Any]]:
+    """Build the per-step ``TaskRunEvent`` payloads for a finished loop.
 
     The loop is *pure of the spine* (it only self-emits ``awaiting_confirm`` on a
     gate block); spine event emission is this channel's job. We walk the ordered
-    ``result.steps`` and emit one event each — ``skill_extract`` for ``extract``
-    verbs, ``skill_step`` for everything else — bracketed by a leading
-    ``skill_perceive`` and a trailing ``skill_done`` carrying the outcome. Every
-    ``emit`` is best-effort and never raises (see ``events.emit``).
+    ``result.steps`` and build one event payload each — ``skill_extract`` for
+    ``extract`` verbs, ``skill_step`` for everything else — bracketed by a
+    leading ``skill_perceive`` and a trailing ``skill_done`` carrying the
+    outcome.
+
+    AUDIT C24: this used to return a list of already-built ``events.emit(...)``
+    coroutines, awaited one at a time by the caller (one session + INSERT +
+    commit/fsync per step). It now returns plain dicts so the caller can hand
+    the whole trace to ``events.emit_many`` — one session, one bulk insert, one
+    commit for the entire run's step trace.
     """
-    coros: list[Any] = []
-    coros.append(
-        events.emit(
-            run_id, STEP_PERCEIVE,
-            f"开始执行技能 | 步数={len(result.steps)}",
-            detail={"step_count": len(result.steps)},
-        )
-    )
+    payloads: list[dict[str, Any]] = [
+        {
+            "step": STEP_PERCEIVE,
+            "message": f"开始执行技能 | 步数={len(result.steps)}",
+            "detail": {"step_count": len(result.steps)},
+        }
+    ]
     for step in result.steps:
         verb = step.verb or "?"
         is_extract = verb == "extract"
-        coros.append(
-            events.emit(
-                run_id,
-                STEP_EXTRACT if is_extract else STEP_STEP,
-                f"步骤 {step.index} | {verb}" + (f" | 错误: {step.error}" if step.error else ""),
-                level="warning" if step.error else "info",
-                detail={
+        payloads.append(
+            {
+                "step": STEP_EXTRACT if is_extract else STEP_STEP,
+                "message": f"步骤 {step.index} | {verb}"
+                + (f" | 错误: {step.error}" if step.error else ""),
+                "level": "warning" if step.error else "info",
+                "detail": {
                     "index": step.index,
                     "verb": step.verb,
                     "target": step.target,
                     "error": step.error,
                     "result": step.result,
                 },
-                elapsed_ms=step.elapsed_ms,
-            )
+                "elapsed_ms": step.elapsed_ms,
+            }
         )
-    coros.append(
-        events.emit(
-            run_id, STEP_DONE,
-            f"技能执行结束 | 结果={result.outcome} 提取={len(result.extracts)}",
-            level="warning" if result.outcome in ("error", "done_failed") else "info",
-            detail={
+    payloads.append(
+        {
+            "step": STEP_DONE,
+            "message": f"技能执行结束 | 结果={result.outcome} 提取={len(result.extracts)}",
+            "level": "warning" if result.outcome in ("error", "done_failed") else "info",
+            "detail": {
                 "outcome": result.outcome,
                 "extract_count": len(result.extracts),
                 "awaiting_confirm": result.awaiting_confirm,
                 "summary": result.summary,
             },
-        )
+        }
     )
-    return coros
+    return payloads
 
 
 def _extracts_to_items(result: LoopResult) -> list[dict[str, Any]]:
@@ -616,9 +621,10 @@ async def collect(
                     await skill_page.aclose()
 
                 # Emit per-step events (best-effort; no-op when no run_id).
+                # AUDIT C24: one session + bulk insert + one commit for the
+                # whole step trace, instead of one commit (fsync) per step.
                 if run_id:
-                    for coro in _emit_loop_events(run_id, result):
-                        await coro
+                    await events.emit_many(run_id, _emit_loop_events(result))
 
                 items = _extracts_to_items(result)
 

backend/channels/web_scraper_channel.py

@@ -1,5 +1,6 @@
 """Web scraper channel using httpx + BeautifulSoup."""
 
+import asyncio
 import logging
 from typing import Any
 from urllib.parse import urlparse
@@ -117,7 +118,11 @@ async def fetch(self, ctx: FetchContext) -> FetchResult:
             async with client as opened_client:
                 response = await self._get(opened_client, url, timeout)
 
-        soup = BeautifulSoup(response.text, "lxml")
+        # AUDIT C22: BeautifulSoup's lxml parse is synchronous and can take
+        # seconds on a large page — run it off the event loop so it can't
+        # freeze every other request/task on this process. Same object,
+        # same parser, just executed in the default thread-pool executor.
+        soup = await asyncio.to_thread(BeautifulSoup, response.text, "lxml")
 
         if list_selector:
             containers = soup.select(list_selector)

backend/config.py

@@ -125,6 +125,18 @@ def cdp_endpoints(self) -> list[str]:
     # WS dispatch timeout when center sends a task over a reverse WS channel
     agent_ws_timeout: int = 130
 
+    # AI enrichment processors (processors/openai_processor.py, claude_processor.py,
+    # local_processor.py): explicit per-request timeout on the LLM API call itself
+    # (AUDIT C8) — the SDKs' own default is a 600s x 2-retry black hole that can
+    # otherwise pin a whole batch in ai_processing for hours behind a dead/slow
+    # gateway. A source's ai_config can still override this per call via
+    # config["timeout"]; this is only the fallback default.
+    llm_request_timeout_seconds: int = 120
+    # Bound how many per-record LLM calls run concurrently within one enrichment
+    # batch (AUDIT C25) — replaces a plain await-in-a-for-loop, where wall-clock
+    # cost was record_count x per-call latency.
+    llm_max_concurrency: int = 4
+
     # Webhooks
     webhook_secret: str = "change-me-webhook-secret"
 

backend/pipeline/events.py

@@ -30,3 +30,37 @@ async def emit(
             await session.commit()
     except Exception as exc:
         logger.warning("emit event failed: %s", exc)
+
+
+async def emit_many(run_id: str, events: list[dict[str, Any]]) -> None:
+    """Write multiple TaskRunEvent rows for one run_id in a single session +
+    bulk insert + one commit (AUDIT C24) — the batched counterpart to
+    :func:`emit` for a caller that already has a whole step trace in hand
+    (e.g. the skill channel's per-step spine events) instead of one commit
+    (fsync) per event in a tight loop.
+
+    Each item in ``events`` accepts the same keys as ``emit``'s kwargs:
+    ``step`` and ``message`` (required), ``level``/``detail``/``elapsed_ms``
+    (optional, same defaults as ``emit``). Best-effort: never raises, mirrors
+    ``emit``. A no-op for an empty list (no session is even opened).
+    """
+    if not events:
+        return
+    try:
+        from backend.database import AsyncSessionLocal
+        from backend.models.task import TaskRunEvent
+        async with AsyncSessionLocal() as session:
+            session.add_all([
+                TaskRunEvent(
+                    run_id=run_id,
+                    level=event.get("level", "info"),
+                    step=event["step"],
+                    message=event["message"],
+                    detail=event.get("detail"),
+                    elapsed_ms=event.get("elapsed_ms"),
+                )
+                for event in events
+            ])
+            await session.commit()
+    except Exception as exc:
+        logger.warning("emit_many event failed: %s", exc)

backend/pipeline/pipeline.py

@@ -6,6 +6,8 @@
 from email.utils import parsedate_to_datetime
 from typing import Any
 
+from sqlalchemy import select
+
 from backend.channels.base import ChannelFetchError
 from backend.control.error_kinds import map_error_type, map_exception
 from backend.control.recorder import FreshnessInfo, record_run_measurement
@@ -400,16 +402,28 @@ async def run_pipeline(
                 # in backend.pipeline.runner phase 2, so it's used as-is.
                 resolve_provider=agent_config is None,
             )
-            # Persist enrichments — new_records are detached after step3 session closed
+            # Persist enrichments — new_records are detached after step3 session
+            # closed. AUDIT C21: one bulk SELECT ... WHERE id IN (...) + an
+            # in-memory id->row map, instead of one `session.get` per record
+            # (N+1) — same field writes (ai_enrichment, status="ai_processed").
             from backend.models.record import CollectedRecord
-            async with AsyncSessionLocal() as session:
-                for rec in new_records:
-                    if rec.ai_enrichment is not None:
-                        db_rec = await session.get(CollectedRecord, rec.id)
+            enriched_ids = [rec.id for rec in new_records if rec.ai_enrichment is not None]
+            if enriched_ids:
+                async with AsyncSessionLocal() as session:
+                    db_recs = (
+                        await session.execute(
+                            select(CollectedRecord).where(CollectedRecord.id.in_(enriched_ids))
+                        )
+                    ).scalars().all()
+                    db_recs_by_id = {db_rec.id: db_rec for db_rec in db_recs}
+                    for rec in new_records:
+                        if rec.ai_enrichment is None:
+                            continue
+                        db_rec = db_recs_by_id.get(rec.id)
                         if db_rec:
                             db_rec.ai_enrichment = rec.ai_enrichment
                             db_rec.status = "ai_processed"
-                await session.commit()
+                    await session.commit()
             logger.info("[task:%s] step4/ai done | processed=%d", task_id, ai_count)
             if run_id:
                 # AUDIT C3: ai_count is now the real enrichment count (0 when

backend/processors/claude_processor.py

@@ -1,5 +1,6 @@
 """Claude (Anthropic) AI processor."""
 
+import asyncio
 import json
 import logging
 import os
@@ -41,12 +42,26 @@ async def process(
                 success=False, error="anthropic package not installed"
             )
 
+        from backend.config import get_settings
+
+        settings = get_settings()
+
         api_key = config.get("api_key") or os.environ.get("ANTHROPIC_API_KEY", "")
         model = config.get("model", "claude-haiku-4-5-20251001")
         max_tokens = config.get("max_tokens", 1024)
-
-        logger.info("claude processor | model=%s max_tokens=%d records=%d",
-                    model, max_tokens, len(records))
+        # AUDIT C8: explicit per-request timeout (see openai_processor's twin
+        # comment) — a source's ai_config can still override this per call
+        # via config["timeout"].
+        request_timeout = config.get("timeout", settings.llm_request_timeout_seconds)
+        # AUDIT C25: bound how many records are in flight at once instead of
+        # a plain await-in-a-for-loop (wall clock == record_count x latency).
+        max_concurrency = max(1, settings.llm_max_concurrency)
+
+        logger.info(
+            "claude processor | model=%s max_tokens=%d records=%d "
+            "timeout=%s max_concurrency=%d",
+            model, max_tokens, len(records), request_timeout, max_concurrency,
+        )
 
         # GOAL-6 PR-E: client construction consolidated through
         # backend.llm.anthropic.AnthropicAdapter (via
@@ -56,18 +71,24 @@ async def process(
         # no behavior change.
         adapter = build_anthropic_adapter(api_key=api_key)
         client = await adapter.get_client()
-        enrichments: list[dict[str, Any]] = []
 
-        try:
-            for i, record in enumerate(records):
-                prompt = _render(prompt_template, record.normalized_data)
-                logger.debug("claude req [%d/%d] | prompt_preview=%s",
-                             i + 1, len(records), prompt[:200])
+        semaphore = asyncio.Semaphore(max_concurrency)
+
+        async def _process_one(i: int, record: "CollectedRecord") -> dict[str, Any]:
+            # AUDIT C25: the semaphore (not the for-loop) is what bounds
+            # concurrency now — every record's coroutine is created up front
+            # and handed to gather, but only `max_concurrency` run their LLM
+            # call at once.
+            async with semaphore:
                 try:
+                    prompt = _render(prompt_template, record.normalized_data)
+                    logger.debug("claude req [%d/%d] | prompt_preview=%s",
+                                 i + 1, len(records), prompt[:200])
                     response = await client.messages.create(
                         model=model,
                         max_tokens=max_tokens,
                         messages=[{"role": "user", "content": prompt}],
+                        timeout=request_timeout,
                     )
                     text = response.content[0].text
                     usage = response.usage
@@ -76,13 +97,24 @@ async def process(
                                 usage.input_tokens, usage.output_tokens,
                                 text[:200])
                     try:
-                        enrichment = json.loads(text)
+                        return json.loads(text)
                     except json.JSONDecodeError:
-                        enrichment = {"analysis": text}
-                    enrichments.append(enrichment)
+                        return {"analysis": text}
                 except Exception as exc:
+                    # A single record's failure must not abort the batch —
+                    # it becomes an {"error": ...} enrichment, exactly like
+                    # the old sequential loop's inner except did.
                     logger.error("claude error [%d/%d] | %s", i + 1, len(records), exc)
-                    enrichments.append({"error": str(exc)})
+                    return {"error": str(exc)}
+
+        try:
+            # asyncio.gather returns results in the same order as the input
+            # awaitables (not completion order), so enrichments[i] still
+            # lines up with records[i] — process_with_ai's zip(records,
+            # enrichments) contract (and the C3 enriched-count fix) holds.
+            enrichments: list[dict[str, Any]] = list(await asyncio.gather(
+                *(_process_one(i, record) for i, record in enumerate(records))
+            ))
         finally:
             await adapter.aclose()