perf: streaming response for DingTalk

[jiangman202506]

钉钉消息回复流式卡片模板

Modifications / 改动点

1、修改 default.py (默认配置)
增加配置项:
* 在默认配置模板中添加了 card_template_id 。
* 意义：这样你在 AstrBot 的管理面板或者配置文件里就能看到这些选项，并进行填写了。
2、修改 dingtalk_adapter.py (钉钉适配器)

这个文件是 AstrBot 与钉钉平台交互的核心组件。

• 启用流式支持 (support_streaming_message=True):

  • 在 @register_platform_adapter 装饰器中，我们将 support_streaming_message 从
    False 改为了 True。
  • 意义：这告诉 AstrBot
    的核心调度系统（Pipeline），这个平台（钉钉）现在有能力接收流式数据（像打字机一
    样一个字一个字蹦出来），而不需要等待大模型完全生成完才发送整段话。

• 加载配置 (card_template_id):

  • 在 init 初始化方法中，增加了读取 card_template_id 的代码。
  • 意义：流式卡片需要一个在钉钉后台预先设计好的“壳子”（模板）。这里就是加载用户配
    置的这个模板 ID。

• 创建卡片 (create_message_card):

  • 新增了一个方法，使用 dingtalk_stream 库的 AICardReplier 工具。
  • 意义：当机器人准备回复时，先调用钉钉接口，“投放”一张空白的（或带有初始内容的）
    卡片到群聊或私聊中。这张卡片拥有一个唯一的 card_instance_id。

• 流式更新卡片 (send_card_message):

  • 新增了一个方法，用于不断更新上面那张卡片的内容。
  • 意义：大模型每生成几个字，就调用一次这个方法，把最新的完整内容推送到那张卡片上
    。用户看到的视觉效果就是卡片里的字在不断增加。
    3、修改 dingtalk_event.py (钉钉事件处理)

这个文件定义了 AstrBot 内部如何处理钉钉的消息事件。

• 接管流式发送 (send_streaming):
  • 重写了 send_streaming 方法。
  • 逻辑：
    1. 检查配置：首先看用户有没有配置
       card_template_id。如果没有配置，或者配置了但创建卡片失败了，就会回退
       (Fallback)
       到旧的模式——把流式内容攒起来，拼成一条完整的文本消息发送（就像普通的聊天机
       器人一样）。
    2. 创建卡片：如果配置了，就调用 Adapter 的 create_message_card 投放一张卡片。
    3. 循环更新：遍历大模型生成的流 (generator)，每收到一点新内容，就拼接到
       full_content 中，并调用 send_card_message
       刷新卡片显示。为了防止刷新太频繁触发限制，代码里做了一点优化（每收到2个片
       段刷新一次）。
    4. 结束：生成结束后，发送最后一次确定的内容。

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

---

Checklist / 检查清单

• 😊 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。/ If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
• 👀 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。/ My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
• 🤓 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到了 requirements.txt 和 pyproject.toml 文件相应位置。/ 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.txt and pyproject.toml.
• 😮 我的更改没有引入恶意代码。/ My changes do not introduce malicious code.

Summary by Sourcery

启用钉钉流式回复在配置了卡片模板时使用交互式消息卡片，并在需要时优雅地回退到纯文本流式回复。

新功能：

• 新增对钉钉流式回复的支持，可通过可配置的卡片模板 ID 驱动交互式消息卡片。

增强项：

• 在默认的聊天服务提供方配置模板中暴露钉钉的 card_template_id，以便于进行配置与使用。
• 扩展钉钉平台事件处理逻辑，以编排卡片创建及流式传输过程中的增量卡片内容更新；当卡片发送失败时，回退为缓冲后的文本输出。

Original summary in English

Summary by Sourcery

Enable DingTalk streaming replies to use interactive message cards when a card template is configured, with graceful fallback to plain text streaming.

New Features:

• Add support for DingTalk streaming replies via interactive message cards driven by a configurable card template ID.

Enhancements:

• Expose DingTalk card_template_id in the default chat provider configuration template for easier setup.
• Extend the DingTalk platform event handling to orchestrate card creation and incremental card content updates during streaming, falling back to buffered text when card delivery fails.

[sourcery-ai]

Hey - 我发现了两个问题，并给出了一些整体反馈：

• DingtalkChatMessageEvent.send_streaming 中的回退分支重复了相同的缓冲并发送逻辑两次；建议将这部分提取到一个小的辅助函数中，以避免重复并让控制流更易理解。
• 在 send_streaming 中，你为了回退逻辑完全消费了 generator，然后又在已经被耗尽的 generator 上调用 super().send_streaming(generator, use_fallback)；如果基类实现也需要处理这个流，建议要么移除这次调用，要么重构逻辑，避免对同一流进行两次消费。
• DingtalkPlatformAdapter 中的 content_key = 'content' 以及使用单一 card_template_id 目前都是硬编码的；如果未来模板或字段有所变化，你可能希望将这些参数按平台或模板进行配置，而不是直接写死在实现中。

给 AI Agent 的提示词

Please address the comments from this code review:

## Overall Comments
- The fallback branches in `DingtalkChatMessageEvent.send_streaming` duplicate the same buffer-and-send logic twice; consider extracting this into a small helper to avoid repetition and make the control flow easier to follow.
- In `send_streaming`, you fully consume the `generator` for the fallback behavior and then call `super().send_streaming(generator, use_fallback)` on an already-exhausted generator; if the base implementation is expected to process the stream, consider either removing that call or restructuring so the stream is not consumed twice.
- The `content_key = 'content'` and the use of a single `card_template_id` in `DingtalkPlatformAdapter` are currently hardcoded; if future templates or fields differ, you may want to make these parameters configurable per platform or per template instead of embedding them in the implementation.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/dingtalk/dingtalk_event.py:126-128` </location>
<code_context>
+        seq = 0
+        try:
+            async for chain in generator:
+                for segment in chain.chain:
+                    if isinstance(segment, Comp.Plain):
+                        full_content += segment.text
+                
+                seq += 1
</code_context>

<issue_to_address>
**issue (bug_risk):** Only `Plain` segments are included in card streaming, which may drop other content types.

In this streaming path, only `Comp.Plain` segments are appended to `full_content`, so markdown, mentions, images, etc. will be dropped from the streamed card while still appearing in non-streaming or fallback flows. If DingTalk cards support richer content, consider mapping additional segment types to card fields so streamed vs non-streamed behavior stays aligned.
</issue_to_address>

### Comment 2
<location> `astrbot/core/platform/sources/dingtalk/dingtalk_event.py:87` </location>
<code_context>
         await super().send(message)

     async def send_streaming(self, generator, use_fallback: bool = False):
-        buffer = None
-        async for chain in generator:
</code_context>

<issue_to_address>
**issue (complexity):** Consider extracting the shared buffering fallback and card-streaming logic in `send_streaming` into separate helper methods to simplify the main method’s control flow and reduce duplication.

You can reduce the complexity and duplication in `send_streaming` by extracting the common “buffer and send” path and the card-streaming logic into small helpers. This keeps all current behavior but makes control flow clearer and easier to maintain.

For example:

```python
class DingtalkMessageEvent(AstrMessageEvent):
    ...

    async def _buffer_and_send_fallback(self, generator, use_fallback: bool):
        buffer = None
        async for chain in generator:
            if buffer is None:
                buffer = chain
            else:
                buffer.chain.extend(chain.chain)

        if buffer is None:
            return None

        buffer.squash_plain()
        await self.send(buffer)
        return await super().send_streaming(generator, use_fallback)

    def _can_use_card_streaming(self) -> bool:
        return bool(self.adapter and self.adapter.card_template_id)

    async def _stream_to_card(self, generator, msg_id: str, incoming_msg: Any):
        created = await self.adapter.create_message_card(msg_id, incoming_msg)
        if not created:
            return None  # caller decides how to fallback

        full_content = ""
        seq = 0
        try:
            async for chain in generator:
                for segment in chain.chain:
                    if isinstance(segment, Comp.Plain):
                        full_content += segment.text

                seq += 1
                if seq % 2 == 0:
                    await self.adapter.send_card_message(
                        msg_id, full_content, is_final=False
                    )

            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
        except Exception as e:
            logger.error(f"DingTalk streaming error: {e}")
            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
```

Then `send_streaming` becomes a short orchestration method:

```python
    async def send_streaming(self, generator, use_fallback: bool = False):
        if not self._can_use_card_streaming():
            logger.warning(
                "DingTalk streaming is enabled, but 'card_template_id' is not "
                f"configured for platform '{self.platform_meta.id}'. "
                "Falling back to text streaming."
            )
            return await self._buffer_and_send_fallback(generator, use_fallback)

        msg_id = self.message_obj.message_id
        incoming_msg = self.message_obj.raw_message

        # try card streaming
        result = await self._stream_to_card(generator, msg_id, incoming_msg)
        if result is None:
            # card creation failed -> fallback to original behavior
            return await self._buffer_and_send_fallback(generator, use_fallback)

        return result
```

This removes the duplicated buffering logic, isolates the card-specific details, and keeps `send_streaming` focused on high-level decision-making without changing any functionality.
</issue_to_address>

---

Sourcery 对开源项目是免费的——如果你觉得这些评审有帮助，欢迎分享 ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{请帮我变得更有用！欢迎在每条评论上点 👍 或 👎，我会根据你的反馈改进评审质量。}

Original comment in English

Hey - I've found 2 issues, and left some high level feedback:

• The fallback branches in DingtalkChatMessageEvent.send_streaming duplicate the same buffer-and-send logic twice; consider extracting this into a small helper to avoid repetition and make the control flow easier to follow.
• In send_streaming, you fully consume the generator for the fallback behavior and then call super().send_streaming(generator, use_fallback) on an already-exhausted generator; if the base implementation is expected to process the stream, consider either removing that call or restructuring so the stream is not consumed twice.
• The content_key = 'content' and the use of a single card_template_id in DingtalkPlatformAdapter are currently hardcoded; if future templates or fields differ, you may want to make these parameters configurable per platform or per template instead of embedding them in the implementation.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The fallback branches in `DingtalkChatMessageEvent.send_streaming` duplicate the same buffer-and-send logic twice; consider extracting this into a small helper to avoid repetition and make the control flow easier to follow.
- In `send_streaming`, you fully consume the `generator` for the fallback behavior and then call `super().send_streaming(generator, use_fallback)` on an already-exhausted generator; if the base implementation is expected to process the stream, consider either removing that call or restructuring so the stream is not consumed twice.
- The `content_key = 'content'` and the use of a single `card_template_id` in `DingtalkPlatformAdapter` are currently hardcoded; if future templates or fields differ, you may want to make these parameters configurable per platform or per template instead of embedding them in the implementation.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/dingtalk/dingtalk_event.py:126-128` </location>
<code_context>
+        seq = 0
+        try:
+            async for chain in generator:
+                for segment in chain.chain:
+                    if isinstance(segment, Comp.Plain):
+                        full_content += segment.text
+                
+                seq += 1
</code_context>

<issue_to_address>
**issue (bug_risk):** Only `Plain` segments are included in card streaming, which may drop other content types.

In this streaming path, only `Comp.Plain` segments are appended to `full_content`, so markdown, mentions, images, etc. will be dropped from the streamed card while still appearing in non-streaming or fallback flows. If DingTalk cards support richer content, consider mapping additional segment types to card fields so streamed vs non-streamed behavior stays aligned.
</issue_to_address>

### Comment 2
<location> `astrbot/core/platform/sources/dingtalk/dingtalk_event.py:87` </location>
<code_context>
         await super().send(message)

     async def send_streaming(self, generator, use_fallback: bool = False):
-        buffer = None
-        async for chain in generator:
</code_context>

<issue_to_address>
**issue (complexity):** Consider extracting the shared buffering fallback and card-streaming logic in `send_streaming` into separate helper methods to simplify the main method’s control flow and reduce duplication.

You can reduce the complexity and duplication in `send_streaming` by extracting the common “buffer and send” path and the card-streaming logic into small helpers. This keeps all current behavior but makes control flow clearer and easier to maintain.

For example:

```python
class DingtalkMessageEvent(AstrMessageEvent):
    ...

    async def _buffer_and_send_fallback(self, generator, use_fallback: bool):
        buffer = None
        async for chain in generator:
            if buffer is None:
                buffer = chain
            else:
                buffer.chain.extend(chain.chain)

        if buffer is None:
            return None

        buffer.squash_plain()
        await self.send(buffer)
        return await super().send_streaming(generator, use_fallback)

    def _can_use_card_streaming(self) -> bool:
        return bool(self.adapter and self.adapter.card_template_id)

    async def _stream_to_card(self, generator, msg_id: str, incoming_msg: Any):
        created = await self.adapter.create_message_card(msg_id, incoming_msg)
        if not created:
            return None  # caller decides how to fallback

        full_content = ""
        seq = 0
        try:
            async for chain in generator:
                for segment in chain.chain:
                    if isinstance(segment, Comp.Plain):
                        full_content += segment.text

                seq += 1
                if seq % 2 == 0:
                    await self.adapter.send_card_message(
                        msg_id, full_content, is_final=False
                    )

            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
        except Exception as e:
            logger.error(f"DingTalk streaming error: {e}")
            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
```

Then `send_streaming` becomes a short orchestration method:

```python
    async def send_streaming(self, generator, use_fallback: bool = False):
        if not self._can_use_card_streaming():
            logger.warning(
                "DingTalk streaming is enabled, but 'card_template_id' is not "
                f"configured for platform '{self.platform_meta.id}'. "
                "Falling back to text streaming."
            )
            return await self._buffer_and_send_fallback(generator, use_fallback)

        msg_id = self.message_obj.message_id
        incoming_msg = self.message_obj.raw_message

        # try card streaming
        result = await self._stream_to_card(generator, msg_id, incoming_msg)
        if result is None:
            # card creation failed -> fallback to original behavior
            return await self._buffer_and_send_fallback(generator, use_fallback)

        return result
```

This removes the duplicated buffering logic, isolates the card-specific details, and keeps `send_streaming` focused on high-level decision-making without changing any functionality.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue (bug_risk): 目前卡片流式发送中只包含 Plain 段落，这可能会导致其他类型内容被丢弃。

在这条流式处理路径中，只有 Comp.Plain 段被追加到 full_content，因此 markdown、@提及、图片等内容会从流式卡片中被丢弃，但在非流式或回退流程中仍然存在。如果钉钉卡片支持更丰富的内容类型，建议将更多的 segment 类型映射到卡片字段，以便保持流式与非流式行为的一致性。

Original comment in English

issue (bug_risk): Only Plain segments are included in card streaming, which may drop other content types.

In this streaming path, only Comp.Plain segments are appended to full_content, so markdown, mentions, images, etc. will be dropped from the streamed card while still appearing in non-streaming or fallback flows. If DingTalk cards support richer content, consider mapping additional segment types to card fields so streamed vs non-streamed behavior stays aligned.

[sourcery-ai]

issue (complexity): 建议将 send_streaming 中共享的缓冲回退逻辑和卡片流式逻辑抽取为独立的辅助方法，以简化主方法的控制流并减少重复。

你可以通过将通用的“缓冲并发送”路径和卡片流式逻辑抽取到几个小的辅助方法中，来降低 send_streaming 的复杂度和重复度。这样既能保留当前所有行为，又能让控制流更清晰、更易维护。

例如：

class DingtalkMessageEvent(AstrMessageEvent):
    ...

    async def _buffer_and_send_fallback(self, generator, use_fallback: bool):
        buffer = None
        async for chain in generator:
            if buffer is None:
                buffer = chain
            else:
                buffer.chain.extend(chain.chain)

        if buffer is None:
            return None

        buffer.squash_plain()
        await self.send(buffer)
        return await super().send_streaming(generator, use_fallback)

    def _can_use_card_streaming(self) -> bool:
        return bool(self.adapter and self.adapter.card_template_id)

    async def _stream_to_card(self, generator, msg_id: str, incoming_msg: Any):
        created = await self.adapter.create_message_card(msg_id, incoming_msg)
        if not created:
            return None  # caller decides how to fallback

        full_content = ""
        seq = 0
        try:
            async for chain in generator:
                for segment in chain.chain:
                    if isinstance(segment, Comp.Plain):
                        full_content += segment.text

                seq += 1
                if seq % 2 == 0:
                    await self.adapter.send_card_message(
                        msg_id, full_content, is_final=False
                    )

            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
        except Exception as e:
            logger.error(f"DingTalk streaming error: {e}")
            await self.adapter.send_card_message(msg_id, full_content, is_final=True)

然后 send_streaming 就可以简化为一个较短的调度方法：

    async def send_streaming(self, generator, use_fallback: bool = False):
        if not self._can_use_card_streaming():
            logger.warning(
                "DingTalk streaming is enabled, but 'card_template_id' is not "
                f"configured for platform '{self.platform_meta.id}'. "
                "Falling back to text streaming."
            )
            return await self._buffer_and_send_fallback(generator, use_fallback)

        msg_id = self.message_obj.message_id
        incoming_msg = self.message_obj.raw_message

        # try card streaming
        result = await self._stream_to_card(generator, msg_id, incoming_msg)
        if result is None:
            # card creation failed -> fallback to original behavior
            return await self._buffer_and_send_fallback(generator, use_fallback)

        return result

这样可以去除重复的缓冲逻辑，将卡片相关的细节隔离出来，并让 send_streaming 专注于高层决策，同时不改变任何现有功能。

Original comment in English

issue (complexity): Consider extracting the shared buffering fallback and card-streaming logic in send_streaming into separate helper methods to simplify the main method’s control flow and reduce duplication.

You can reduce the complexity and duplication in send_streaming by extracting the common “buffer and send” path and the card-streaming logic into small helpers. This keeps all current behavior but makes control flow clearer and easier to maintain.

For example:

class DingtalkMessageEvent(AstrMessageEvent):
    ...

    async def _buffer_and_send_fallback(self, generator, use_fallback: bool):
        buffer = None
        async for chain in generator:
            if buffer is None:
                buffer = chain
            else:
                buffer.chain.extend(chain.chain)

        if buffer is None:
            return None

        buffer.squash_plain()
        await self.send(buffer)
        return await super().send_streaming(generator, use_fallback)

    def _can_use_card_streaming(self) -> bool:
        return bool(self.adapter and self.adapter.card_template_id)

    async def _stream_to_card(self, generator, msg_id: str, incoming_msg: Any):
        created = await self.adapter.create_message_card(msg_id, incoming_msg)
        if not created:
            return None  # caller decides how to fallback

        full_content = ""
        seq = 0
        try:
            async for chain in generator:
                for segment in chain.chain:
                    if isinstance(segment, Comp.Plain):
                        full_content += segment.text

                seq += 1
                if seq % 2 == 0:
                    await self.adapter.send_card_message(
                        msg_id, full_content, is_final=False
                    )

            await self.adapter.send_card_message(msg_id, full_content, is_final=True)
        except Exception as e:
            logger.error(f"DingTalk streaming error: {e}")
            await self.adapter.send_card_message(msg_id, full_content, is_final=True)

Then send_streaming becomes a short orchestration method:

    async def send_streaming(self, generator, use_fallback: bool = False):
        if not self._can_use_card_streaming():
            logger.warning(
                "DingTalk streaming is enabled, but 'card_template_id' is not "
                f"configured for platform '{self.platform_meta.id}'. "
                "Falling back to text streaming."
            )
            return await self._buffer_and_send_fallback(generator, use_fallback)

        msg_id = self.message_obj.message_id
        incoming_msg = self.message_obj.raw_message

        # try card streaming
        result = await self._stream_to_card(generator, msg_id, incoming_msg)
        if result is None:
            # card creation failed -> fallback to original behavior
            return await self._buffer_and_send_fallback(generator, use_fallback)

        return result

This removes the duplicated buffering logic, isolates the card-specific details, and keeps send_streaming focused on high-level decision-making without changing any functionality.