perf(pipeline): LLM timeout+concurrency, bulk AI persist, off-loop parse, batched events [修复组⑤] - #28
Conversation
…ed parse, batched events [C8,C21,C22,C24,C25] - C8/C25: openai/claude/local processors get an explicit per-request LLM timeout (new Settings.llm_request_timeout_seconds, default 120s, config["timeout"]-overridable) and bound per-record concurrency via asyncio.gather + Semaphore (new Settings.llm_max_concurrency, default 4) instead of a sequential await-in-a-for-loop. Order is preserved (gather), and each record's LLM call keeps its own try/except so one failure can't abort the batch. - C21: pipeline.py's AI-enrichment persist step replaces one session.get() per record with a single bulk SELECT ... WHERE id IN (...) + in-memory id->row map before one commit. Same field writes (ai_enrichment, status="ai_processed"). - C22: rss_channel's feedparser.parse() (collect() and fetch()) and web_scraper_channel's BeautifulSoup parse now run via asyncio.to_thread instead of blocking the event loop inline. - C24: new events.emit_many(run_id, events) writes a whole step trace in one session + bulk insert + one commit; skill_channel's per-step loop (the only tight-loop emit() caller) now uses it. All other emit() call sites are untouched. Tests: LLM timeout/concurrency/order/failure-isolation per processor, AI-persist query-count spy, feedparser/BeautifulSoup off-thread checks, emit_many one-commit behavior. Fakes/monkeypatch only, no real network/DB.
|
✅ Health: 7.8 📋 At a glance Files & modules (2)
🚨 Change risk: 9.6/10 (high)
🔎 More signals (2)🔥 Hotspots touched (5)
2 more
💀 Dead code (7 findings)
4 more
👀 Suggested reviewers @xujinghua 📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-18 18:14 UTC |
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (18)
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe PR moves blocking channel parsing to worker threads, batches skill event writes, adds shared LLM timeout and concurrency settings, parallelizes enrichment requests with failure isolation, and replaces per-record enrichment persistence with bulk selection and updates. ChangesAsync processing and persistence
Sequence Diagram(s)sequenceDiagram
participant Channel
participant WorkerThread
participant SkillChannel
participant EventStore
Channel->>WorkerThread: parse RSS or HTML content
WorkerThread-->>Channel: return parsed content
SkillChannel->>EventStore: emit_many(run_id, event payloads)
EventStore-->>SkillChannel: commit batched task events
Possibly related PRs
Poem
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces several performance and concurrency optimizations across the backend. Key changes include offloading synchronous parsing tasks (BeautifulSoup and feedparser) to thread pools using 'asyncio.to_thread', batching event emissions via a new 'emit_many' database helper, optimizing database updates to avoid N+1 queries, and implementing bounded concurrency and configurable timeouts for LLM processors. Feedback on these changes suggests chunking the bulk query in 'pipeline.py' to prevent hitting SQLite's 999 parameter limit, and using '.get()' with defaults in 'emit_many' to avoid 'KeyError' exceptions that could fail an entire batch of events.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if enriched_ids: | ||
| async with AsyncSessionLocal() as session: | ||
| db_recs = ( | ||
| await session.execute( | ||
| select(CollectedRecord).where(CollectedRecord.id.in_(enriched_ids)) | ||
| ) | ||
| ).scalars().all() |
There was a problem hiding this comment.
在 SQLite 中，单个 SQL 查询中的参数（变量）数量默认限制为 999 个。如果 enriched_ids 的数量超过 999，执行 CollectedRecord.id.in_(enriched_ids) 查询时会抛出 OperationalError: too many SQL variables 异常。
为了提高代码的健壮性，建议对 enriched_ids 进行分批（Chunking）查询，以确保在处理大批量数据时不会触发 SQLite 的参数限制。
| if enriched_ids: | |
| async with AsyncSessionLocal() as session: | |
| db_recs = ( | |
| await session.execute( | |
| select(CollectedRecord).where(CollectedRecord.id.in_(enriched_ids)) | |
| ) | |
| ).scalars().all() | |
| if enriched_ids: | |
| async with AsyncSessionLocal() as session: | |
| db_recs = [] | |
| for i in range(0, len(enriched_ids), 999): | |
| chunk = enriched_ids[i : i + 999] | |
| res = await session.execute( | |
| select(CollectedRecord).where(CollectedRecord.id.in_(chunk)) | |
| ) | |
| db_recs.extend(res.scalars().all()) |
| 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 | ||
| ]) |
There was a problem hiding this comment.
在 emit_many 批量写入事件时，直接使用 event["step"] 和 event["message"] 进行字典取值。如果传入的事件列表中有任何一个事件字典缺失了 "step" 或 "message" 键，将会抛出 KeyError 异常。
虽然整个操作被包裹在 try...except 块中，但一个事件的格式错误会导致整批事件全部写入失败。建议使用 .get() 方法并提供合理的默认值，以提高公共工具函数的容错性和健壮性。
| 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 | |
| ]) | |
| session.add_all([ | |
| TaskRunEvent( | |
| run_id=run_id, | |
| level=event.get("level", "info"), | |
| step=event.get("step", "unknown"), | |
| message=event.get("message", ""), | |
| detail=event.get("detail"), | |
| elapsed_ms=event.get("elapsed_ms"), | |
| ) | |
| for event in events | |
| ]) |
1 participant
修复组⑤ — 吞吐 (账本 C8/C21/C22/C24/C25)
Sonnet 实施, Fable 审计通过 (diff 结构 + 主 repo 真跑测试)。纯性能, 行为不变。
timeout; per-record 循环从 await-in-for 改asyncio.gather+Semaphore限流。gather 保序, per-task try/except 失败隔离 (单条炸不断批), process_with_ai 富化计数契约 (C3) 不变session.get→ 一次 bulkSELECT ... WHERE id IN(...)+ 内存 id→row map, 消 N+1feedparser.parse/BeautifulSoup包asyncio.to_thread, 移出事件循环emit_many(run_id, events)单 session + add_all + 单 commit; skill step-loop 切过去; 单 emit() 调用点不动llm_request_timeout_seconds=120(local 原硬编码默认, 零行为变化; openai/claude 之前无超时=真修复),llm_max_concurrency=4Test
pytest tests/unit→ 1275 passed, 2 failed, 1 skipped。2 failed = 账本 P3-6 已知存量 flake (nodes_install netbird/ssh GBK), 与本改动无关合并顺序
④⑤⑥ 共同热点 pipeline.py。本 PR 改 C21 区 (AI 落库 ~404-426)。先后合任意, 后者 rebase。