feat(lark): supports CardKit streaming output for feishu (#5777)

camera-2018 · bread-ovO · Soulter · web-flow · commit ec21cb13d3f6 · 2026-03-10T19:40:46.000+08:00
* feat(lark): 支持飞书 CardKit 流式输出

* refactor(lark): extract streaming fallback logic and deduplicate final text update

* fix(lark): 修复流式输出竞态条件及增强健壮性

- 修复 sender loop 中 delta 快照竟态: await 期间 delta 被 generator
  更新导致 last_sent 记录了未发送的值, 造成输出卡在最后一段
- send_streaming 入口增加 platform_meta 守卫, 未启用时直接回退
- _fallback_send_streaming 移除对已耗尽 generator 的 super() 调用,
  改为内联父类副作用 (Metric.upload + _has_send_oper)
- Metric.upload 统一改为 await, 确保指标上报在方法返回前完成
- 装饰器 support_streaming_message 改为 False, 与 meta() 动态配置对齐
- i18n hint 补充提示: 需在「AI 配置 → 其他配置」中开启流式输出

* chore(lark): 收口配置

* docs(lark): update streaming output instructions and client version requirements

---------

Co-authored-by: bread-ovo &lt;2570425204@qq.com&gt;
Co-authored-by: Soulter &lt;905617992@qq.com&gt;
diff --git a/astrbot/core/platform/sources/lark/lark_adapter.py b/astrbot/core/platform/sources/lark/lark_adapter.py
@@ -34,7 +34,7 @@
 
 
 @register_platform_adapter(
-    "lark", "飞书机器人官方 API 适配器", support_streaming_message=False
+    "lark", "飞书机器人官方 API 适配器", support_streaming_message=True
 )
 class LarkPlatformAdapter(Platform):
     def __init__(
@@ -491,7 +491,7 @@ def meta(self) -> PlatformMetadata:
             name="lark",
             description="飞书机器人官方 API 适配器",
             id=cast(str, self.config.get("id")),
-            support_streaming_message=False,
+            support_streaming_message=True,
         )
 
     async def convert_msg(self, event: lark.im.v1.P2ImMessageReceiveV1) -> None:
diff --git a/astrbot/core/platform/sources/lark/lark_event.py b/astrbot/core/platform/sources/lark/lark_event.py
@@ -1,10 +1,19 @@
+import asyncio
 import base64
 import json
 import os
 import uuid
 from io import BytesIO
 
 import lark_oapi as lark
+from lark_oapi.api.cardkit.v1 import (
+    ContentCardElementRequest,
+    ContentCardElementRequestBody,
+    CreateCardRequest,
+    CreateCardRequestBody,
+    SettingsCardRequest,
+    SettingsCardRequestBody,
+)
 from lark_oapi.api.im.v1 import (
     CreateFileRequest,
     CreateFileRequestBody,
@@ -28,6 +37,7 @@
     convert_video_format,
     get_media_duration,
 )
+from astrbot.core.utils.metrics import Metric
 
 
 class LarkMessageEvent(AstrMessageEvent):
@@ -555,15 +565,257 @@ async def react(self, emoji: str) -> None:
             logger.error(f"发送飞书表情回应失败({response.code}): {response.msg}")
             return
 
-    async def send_streaming(self, generator, use_fallback: bool = False):
+    async def _create_streaming_card(self) -> str | None:
+        """创建一个开启流式更新模式的卡片实体，返回 card_id。"""
+        if self.bot.cardkit is None:
+            logger.error("[Lark] API Client cardkit 模块未初始化")
+            return None
+
+        card_json = {
+            "schema": "2.0",
+            "header": {
+                "title": {"content": "", "tag": "plain_text"},
+            },
+            "config": {
+                "streaming_mode": True,
+                "summary": {"content": ""},
+                "streaming_config": {
+                    "print_frequency_ms": {"default": 50},
+                    "print_step": {"default": 2},
+                    "print_strategy": "fast",
+                },
+            },
+            "body": {
+                "elements": [
+                    {
+                        "tag": "markdown",
+                        "content": "",
+                        "element_id": "markdown_1",
+                    }
+                ]
+            },
+        }
+
+        request = (
+            CreateCardRequest.builder()
+            .request_body(
+                CreateCardRequestBody.builder()
+                .type("card_json")
+                .data(json.dumps(card_json, ensure_ascii=False))
+                .build()
+            )
+            .build()
+        )
+
+        try:
+            response = await self.bot.cardkit.v1.card.acreate(request)
+        except Exception as e:
+            logger.error(f"[Lark] 创建流式卡片实体失败: {e}")
+            return None
+
+        if not response.success():
+            logger.error(
+                f"[Lark] 创建流式卡片实体失败({response.code}): {response.msg}"
+            )
+            return None
+
+        if response.data is None or not response.data.card_id:
+            logger.error("[Lark] 创建流式卡片实体成功但未返回 card_id")
+            return None
+
+        card_id = response.data.card_id
+        logger.debug(f"[Lark] 创建流式卡片实体成功: {card_id}")
+        return card_id
+
+    async def _send_card_message(
+        self,
+        card_id: str,
+        reply_message_id: str | None = None,
+        receive_id: str | None = None,
+        receive_id_type: str | None = None,
+    ) -> bool:
+        """将卡片实体作为 interactive 消息发送。"""
+        content = json.dumps(
+            {"type": "card", "data": {"card_id": card_id}},
+            ensure_ascii=False,
+        )
+        return await self._send_im_message(
+            self.bot,
+            content=content,
+            msg_type="interactive",
+            reply_message_id=reply_message_id,
+            receive_id=receive_id,
+            receive_id_type=receive_id_type,
+        )
+
+    async def _update_streaming_text(
+        self,
+        card_id: str,
+        content: str,
+        sequence: int,
+    ) -> bool:
+        """调用 CardKit 流式更新文本接口，向 markdown_1 组件推送全量文本。"""
+        if self.bot.cardkit is None:
+            logger.error("[Lark] API Client cardkit 模块未初始化")
+            return False
+
+        request = (
+            ContentCardElementRequest.builder()
+            .card_id(card_id)
+            .element_id("markdown_1")
+            .request_body(
+                ContentCardElementRequestBody.builder()
+                .content(content)
+                .sequence(sequence)
+                .uuid(str(uuid.uuid4()))
+                .build()
+            )
+            .build()
+        )
+
+        try:
+            response = await self.bot.cardkit.v1.card_element.acontent(request)
+        except Exception as e:
+            logger.debug(f"[Lark] 流式更新文本失败 (ignored): {e}")
+            return False
+
+        if not response.success():
+            logger.debug(f"[Lark] 流式更新文本失败({response.code}): {response.msg}")
+            return False
+
+        return True
+
+    async def _close_streaming_mode(
+        self,
+        card_id: str,
+        sequence: int,
+    ) -> None:
+        """关闭卡片的流式更新模式，使其可正常转发、摘要恢复。"""
+        if self.bot.cardkit is None:
+            logger.error("[Lark] API Client cardkit 模块未初始化")
+            return
+
+        settings_json = json.dumps(
+            {"config": {"streaming_mode": False}},
+            ensure_ascii=False,
+        )
+
+        request = (
+            SettingsCardRequest.builder()
+            .card_id(card_id)
+            .request_body(
+                SettingsCardRequestBody.builder()
+                .settings(settings_json)
+                .sequence(sequence)
+                .uuid(str(uuid.uuid4()))
+                .build()
+            )
+            .build()
+        )
+
+        try:
+            response = await self.bot.cardkit.v1.card.asettings(request)
+        except Exception as e:
+            logger.error(f"[Lark] 关闭流式模式失败: {e}")
+            return
+
+        if not response.success():
+            logger.error(f"[Lark] 关闭流式模式失败({response.code}): {response.msg}")
+        else:
+            logger.debug(f"[Lark] 流式模式已关闭: {card_id}")
+
+    async def _fallback_send_streaming(self, generator, use_fallback: bool = False):
+        """回退到非流式发送：缓冲全部文本后一次性发送，并保留父类副作用。"""
         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)
+
+        if buffer:
+            buffer.squash_plain()
+            await self.send(buffer)
+
+        await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        self._has_send_oper = True
+
+    async def send_streaming(self, generator, use_fallback: bool = False):
+        """使用 CardKit 流式卡片实现打字机效果。
+
+        流程：创建卡片实体 → 发送消息 → 流式更新文本 → 关闭流式模式。
+        使用解耦发送循环，LLM token 到达时只更新 buffer 并唤醒发送协程，
+        发送频率由网络 RTT 自然限流。
+        """
+        # Step 1: 创建流式卡片实体
+        card_id = await self._create_streaming_card()
+        if not card_id:
+            logger.warning("[Lark] 无法创建流式卡片，回退到非流式发送")
+            await self._fallback_send_streaming(generator, use_fallback)
+            return
+
+        # Step 2: 发送卡片消息
+        sent = await self._send_card_message(
+            card_id,
+            reply_message_id=self.message_obj.message_id,
+        )
+        if not sent:
+            logger.error("[Lark] 发送流式卡片消息失败，回退到非流式发送")
+            await self._fallback_send_streaming(generator, use_fallback)
+            return
+
+        logger.info("[Lark] 流式输出: 使用 CardKit 流式卡片")
+
+        # Step 3: 解耦发送循环 (Event-driven, 参考 Telegram Draft 路径)
+        sequence = 0
+        delta = ""
+        last_sent = ""
+        done = False
+        text_changed = asyncio.Event()
+
+        async def _sender_loop() -> None:
+            """信号驱动的文本发送循环，有新内容就发，RTT 自然限流。"""
+            nonlocal sequence, last_sent
+            while not done:
+                await text_changed.wait()
+                text_changed.clear()
+                snapshot = delta
+                if snapshot and snapshot != last_sent:
+                    sequence += 1
+                    ok = await self._update_streaming_text(card_id, snapshot, sequence)
+                    if ok:
+                        last_sent = snapshot
+                    if delta != snapshot:
+                        text_changed.set()
+
+        sender_task = asyncio.create_task(_sender_loop())
+
+        try:
+            async for chain in generator:
+                if not isinstance(chain, MessageChain):
+                    continue
+
+                if chain.type == "break":
+                    # 飞书卡片不支持分段，忽略 break
+                    continue
+
+                for comp in chain.chain:
+                    if isinstance(comp, Plain):
+                        delta += comp.text
+                        text_changed.set()
+        finally:
+            done = True
+            text_changed.set()
+            await sender_task
+
+        # Step 4: 必要时补发最终文本 + 关闭流式模式
+        if delta and delta != last_sent:
+            sequence += 1
+            await self._update_streaming_text(card_id, delta, sequence)
+
+        sequence += 1
+        await self._close_streaming_mode(card_id, sequence)
+
+        # Step 5: 内联父类 send_streaming 的副作用
+        await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
+        self._has_send_oper = True
diff --git a/dashboard/src/i18n/locales/en-US/features/config-metadata.json b/dashboard/src/i18n/locales/en-US/features/config-metadata.json
@@ -1521,4 +1521,4 @@
     "helpMiddle": "or",
     "helpSuffix": "."
   }
-}
+}
diff --git a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
@@ -1524,4 +1524,4 @@
     "helpMiddle": "或",
     "helpSuffix": "。"
   }
-}
+}
diff --git a/docs/en/platform/lark.md b/docs/en/platform/lark.md
@@ -14,6 +14,10 @@
 
 Proactive message push: Supported.
 
+Streaming output: Supported. You must enable the `Create and update cards (cardkit:card:write)` permission for your app in the Lark Developer Console.
+
+The Lark client version must be >= 7.20. Lower versions only display the title and an upgrade prompt.
+
 ## Creating a Bot
 
 Navigate to the [Developer Console](https://open.feishu.cn/app) and create a custom enterprise application.
@@ -88,6 +92,8 @@ Next, click on "Permission Management," click "Enable Permissions," and enter `i
 
 Enter `im:resource:upload,im:resource` again to enable image upload permissions.
 
+If you want to use streaming output, additionally enable `Create and update cards (cardkit:card:write)`.
+
 The final set of permissions should look like this:
 
 ![Final Permissions](https://files.astrbot.app/docs/source/images/lark/image-11.png)
diff --git a/docs/zh/platform/lark.md b/docs/zh/platform/lark.md
@@ -14,6 +14,10 @@
 
 主动消息推送：支持。
 
+流式输出：支持。需要在飞书开发者后台为应用开通 `创建与更新卡片(cardkit:card:write)` 权限。
+
+飞书客户端版本需 >= 7.20。低版本客户端将只显示标题和升级提示。
+
 ## 创建机器人
 
 前往 [开发者后台](https://open.feishu.cn/app) ，创建企业自建应用。
@@ -88,6 +92,8 @@
 
 再次输入 `im:resource:upload,im:resource` 开通上传图片相关的权限。
 
+如果需要使用流式输出，请额外开通 `创建与更新卡片(cardkit:card:write)` 权限。
+
 最终开通的权限如下图：
 
 ![最终开通的权限](https://files.astrbot.app/docs/source/images/lark/image-11.png)

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@`
`34`	`34`
`35`	`35`
`36`	`36`	`@register_platform_adapter(`
`37`		`- "lark", "飞书机器人官方 API 适配器", support_streaming_message=False`
	`37`	`+ "lark", "飞书机器人官方 API 适配器", support_streaming_message=True`
`38`	`38`	`)`
`39`	`39`	`class LarkPlatformAdapter(Platform):`
`40`	`40`	`def __init__(`
`@@ -491,7 +491,7 @@ def meta(self) -> PlatformMetadata:`
`491`	`491`	`name="lark",`
`492`	`492`	`description="飞书机器人官方 API 适配器",`
`493`	`493`	`id=cast(str, self.config.get("id")),`
`494`		`- support_streaming_message=False,`
	`494`	`+ support_streaming_message=True,`
`495`	`495`	`)`
`496`	`496`
`497`	`497`	`async def convert_msg(self, event: lark.im.v1.P2ImMessageReceiveV1) -> None:`
Original file line number	Diff line number	Diff line change
`@@ -1521,4 +1521,4 @@`
`1521`	`1521`	`"helpMiddle": "or",`
`1522`	`1522`	`"helpSuffix": "."`
`1523`	`1523`	`}`
`1524`		`-}`
	`1524`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -1524,4 +1524,4 @@`
`1524`	`1524`	`"helpMiddle": "或",`
`1525`	`1525`	`"helpSuffix": "。"`
`1526`	`1526`	`}`
`1527`		`-}`
	`1527`	`+}`