refactor(core): add type hints to pipeline scheduler - #9621
Merged
Soulter merged 2 commits intoAug 12, 2026
Conversation
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The
cast(AsyncGenerator[None], coroutine)line is currently usingAsyncGeneratorfromcollections.abc, which is not subscriptable; consider importingAsyncGeneratorfromtyping(with proper type parameters) or adjusting the cast to a compatible generic type so type checkers don't error on this annotation.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `cast(AsyncGenerator[None], coroutine)` line is currently using `AsyncGenerator` from `collections.abc`, which is not subscriptable; consider importing `AsyncGenerator` from `typing` (with proper type parameters) or adjusting the cast to a compatible generic type so type checkers don't error on this annotation.
## Individual Comments
### Comment 1
<location path="astrbot/core/pipeline/scheduler.py" line_range="53-54" />
<code_context>
if isinstance(coroutine, AsyncGenerator):
# 如果返回的是异步生成器, 实现洋葱模型的核心
- async for _ in coroutine:
+ agen = cast(AsyncGenerator[None], coroutine)
+ async for _ in agen:
# 此处是前置处理完成后的暂停点(yield), 下面开始执行后续阶段
if event.is_stopped():
</code_context>
<issue_to_address>
**issue (bug_risk):** Fix the AsyncGenerator type parameters in cast to avoid runtime/type-checking issues.
`collections.abc.AsyncGenerator` requires two type parameters `(YieldType, SendType)`, so `AsyncGenerator[None]` is invalid and can raise `TypeError` at runtime as well as fail static checks. Use something like `cast(AsyncGenerator[Any, Any], coroutine)` (or `AsyncGenerator[object, object]` if that’s more accurate), or drop the cast and rely on the `isinstance(..., AsyncGenerator)` check if your type checker supports that narrowing.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR aims to improve static typing for the pipeline scheduler by explicitly annotating PipelineScheduler.stages as list[Stage] and narrowing the inferred type of the async generator returned by Stage.process().
Changes:
- Add
Stagetyping forself.stagesto improve IDE/type-checker inference. - Introduce a
typing.castin_process_stagesto narrow the async generator type returned fromStage.process(). - Update imports to include
castandStage.
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Soulter
approved these changes
Aug 12, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
在astrbot\core\pipeline\scheduler.py中为self.stages添加类型标注,让 IDE 与类型检查器能正确推断list[Stage]类型,纯类型层面改动,无功能变化。
Modifications / 改动点
astrbot/core/pipeline/scheduler.pyself.stages标注为list[Stage]_process_stages中,使用cast(AsyncGenerator[None], coroutine)来收窄Stage.process()返回的生成器类型Stage和castScreenshots or Test Results / 运行截图或测试结果
uv run pytest tests/test_smoke.py -v

Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能,已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试,并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库,或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Refine the pipeline scheduler’s stage handling to use explicit typing for stage instances and their async generators.
Enhancements: