laoxong/AstrBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: streaming response for DingTalk (#4590)

Browse Source 
closes: #4384

* #4384 钉钉消息回复卡片模板

* chore: ruff format

* chore: ruff format

---------

Co-authored-by: ManJiang <man.jiang@jg-robust.com>
Co-authored-by: Soulter <905617992@qq.com>
This commit is contained in:
jiangman202506
2026-01-21 12:48:45 +08:00
committed by Soulter
parent 4d28de6b4a
commit 93cc4cebe6
3 changed files with 124 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
astrbot/core/config/default.py
+6
View File
@@ -321,6 +321,7 @@ CONFIG_METADATA_2 = {
"enable": False,
"client_id": "",
"client_secret": "",
"card_template_id": "",
},
"Telegram": {
"id": "telegram",
@@ -582,6 +583,11 @@ CONFIG_METADATA_2 = {
"type": "string",
"hint": "可选：填写 Misskey 网盘中目标文件夹的 ID，上传的文件将放置到该文件夹内。留空则使用账号网盘根目录。",
},
"card_template_id": {
"description": "卡片模板 ID",
"type": "string",
"hint": "可选。钉钉互动卡片模板 ID。启用后将使用互动卡片进行流式回复。",
},
"telegram_command_register": {
"description": "Telegram 命令注册",
"type": "bool",
astrbot/core/platform/sources/dingtalk/dingtalk_adapter.py
+61 -2
View File
@@ -39,7 +39,7 @@ class MyEventHandler(dingtalk_stream.EventHandler):
@register_platform_adapter(
"dingtalk", "钉钉机器人官方 API 适配器", support_streaming_message=False
"dingtalk", "钉钉机器人官方 API 适配器", support_streaming_message=True
)
class DingtalkPlatformAdapter(Platform):
def __init__(
@@ -75,6 +75,8 @@ class DingtalkPlatformAdapter(Platform):
)
self.client_ = client # 用于 websockets 的 client
self._shutdown_event: threading.Event | None = None
self.card_template_id = platform_config.get("card_template_id")
self.card_instance_id_dict = {}
def _id_to_sid(self, dingtalk_id: str | None) -> str:
if not dingtalk_id:
@@ -96,9 +98,65 @@ class DingtalkPlatformAdapter(Platform):
name="dingtalk",
description="钉钉机器人官方 API 适配器",
id=cast(str, self.config.get("id")),
support_streaming_message=False,
support_streaming_message=True,
)
async def create_message_card(
self, message_id: str, incoming_message: dingtalk_stream.ChatbotMessage
):
if not self.card_template_id:
return False
card_instance = dingtalk_stream.AICardReplier(self.client_, incoming_message)
card_data = {"content": ""} # Initial content empty
try:
card_instance_id = await card_instance.async_create_and_deliver_card(
self.card_template_id,
card_data,
)
self.card_instance_id_dict[message_id] = (card_instance, card_instance_id)
return True
except Exception as e:
logger.error(f"创建钉钉卡片失败: {e}")
return False
async def send_card_message(self, message_id: str, content: str, is_final: bool):
if message_id not in self.card_instance_id_dict:
return
card_instance, card_instance_id = self.card_instance_id_dict[message_id]
content_key = "content"
try:
# 钉钉卡片流式更新
await card_instance.async_streaming(
card_instance_id,
content_key=content_key,
content_value=content,
append=False,
finished=is_final,
failed=False,
)
except Exception as e:
logger.error(f"发送钉钉卡片消息失败: {e}")
# Try to report failure
try:
await card_instance.async_streaming(
card_instance_id,
content_key=content_key,
content_value=content, # Keep existing content
append=False,
finished=True,
failed=True,
)
except Exception:
pass
if is_final:
self.card_instance_id_dict.pop(message_id, None)
async def convert_msg(
self,
message: dingtalk_stream.ChatbotMessage,
@@ -224,6 +282,7 @@ class DingtalkPlatformAdapter(Platform):
platform_meta=self.meta(),
session_id=abm.session_id,
client=self.client,
adapter=self,
)
self._event_queue.put_nowait(event)
astrbot/core/platform/sources/dingtalk/dingtalk_event.py
+57 -11
View File
@@ -1,5 +1,5 @@
import asyncio
from typing import cast
from typing import Any, cast
import dingtalk_stream
@@ -16,9 +16,11 @@ class DingtalkMessageEvent(AstrMessageEvent):
platform_meta,
session_id,
client: dingtalk_stream.ChatbotHandler,
adapter: "Any" = None,
):
super().__init__(message_str, message_obj, platform_meta, session_id)
self.client = client
self.adapter = adapter
async def send_with_client(
self,
@@ -83,14 +85,58 @@ class DingtalkMessageEvent(AstrMessageEvent):
await super().send(message)
async def send_streaming(self, generator, use_fallback: bool = False):
buffer = None
async for chain in generator:
if not self.adapter or not self.adapter.card_template_id:
logger.warning(
f"DingTalk streaming is enabled, but 'card_template_id' is not configured for platform '{self.platform_meta.id}'. Falling back to text streaming."
)
# Fallback to default behavior (buffer and send)
buffer = None
async for chain in generator:
if not buffer:
buffer = chain
else:
buffer.chain.extend(chain.chain)
if not buffer:
buffer = chain
else:
buffer.chain.extend(chain.chain)
if not buffer:
return None
buffer.squash_plain()
await self.send(buffer)
return await super().send_streaming(generator, use_fallback)
return None
buffer.squash_plain()
await self.send(buffer)
return await super().send_streaming(generator, use_fallback)
# Create card
msg_id = self.message_obj.message_id
incoming_msg = self.message_obj.raw_message
created = await self.adapter.create_message_card(msg_id, incoming_msg)
if not created:
# Fallback to default behavior (buffer and send)
buffer = None
async for chain in generator:
if not buffer:
buffer = chain
else:
buffer.chain.extend(chain.chain)
if not buffer:
return None
buffer.squash_plain()
await self.send(buffer)
return await super().send_streaming(generator, use_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: # Update every 2 chunks to be more responsive than 8
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}")
# Try to ensure final state is sent or cleaned up?
await self.adapter.send_card_message(msg_id, full_content, is_final=True)