Skip to content

fix(lark): Defer card creation and renew on tool call break #6743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RC-CHN merged 4 commits into from
Mar 21, 2026
Merged

fix(lark): Defer card creation and renew on tool call break #6743

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
554be88
fix(lark): defer streaming card creation and renew card on tool call …
camera-2018 Mar 21, 2026
6634518
style: format ruff
camera-2018 Mar 21, 2026
55c90bc
fix: cr bug
camera-2018 Mar 21, 2026
df31a04
fix: cr
camera-2018 Mar 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions astrbot/core/astr_agent_run_util.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,8 +165,13 @@ async def run_agent(
# 对于其他情况，暂时先不处理
continue
elif resp.type == "tool_call":
if agent_runner.streaming:
# 用来标记流式响应需要分节
if agent_runner.streaming and show_tool_use:
# 向下游平台发送 "break" 分段信号（空 MessageChain，不携带数据）。
# 平台适配器收到后会关闭当前流式消息，并在后续文本到来时创建新消息。
# 仅在 show_tool_use 为 True 时才发送：此时紧接着会通过
# astr_event.send() 独立发送工具状态消息（如"🔨 调用工具: xxx"），
# 需要分段才能保证消息顺序正确。
# 若 show_tool_use 为 False，不会有独立消息插入，无需分段。
yield MessageChain(chain=[], type="break")

tool_info = _extract_chain_json_data(resp.data["chain"])
Expand Down
117 changes: 85 additions & 32 deletions astrbot/core/platform/sources/lark/lark_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -743,35 +743,20 @@ async def _fallback_send_streaming(self, generator, use_fallback: bool = False):
async def send_streaming(self, generator, use_fallback: bool = False):
"""使用 CardKit 流式卡片实现打字机效果。

流程：创建卡片实体 → 发送消息 → 流式更新文本 → 关闭流式模式。
流程：首字到来时创建卡片实体 → 发送消息 → 流式更新文本 → 关闭流式模式。
卡片创建延迟到第一个文本 token 到达时，避免工具调用阶段就渲染空卡片。
使用解耦发送循环，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 路径)
# Lazy-init: card & sender loop created on first text token
Comment thread
camera-2018 marked this conversation as resolved.
Resolved
card_id = None
sequence = 0
delta = ""
last_sent = ""
done = False
text_changed = asyncio.Event()
sender_task = None
fallback_used = False # 回退路径已处理 Metric，避免重复上报

async def _sender_loop() -> None:
"""信号驱动的文本发送循环，有新内容就发，RTT 自然限流。"""
Expand All @@ -788,34 +773,102 @@ async def _sender_loop() -> None:
if delta != snapshot:
text_changed.set()

sender_task = asyncio.create_task(_sender_loop())
async def _consume_rest_and_fallback(gen, initial_text: str) -> None:
"""Card creation failed; consume remaining chunks and send non-streaming."""
nonlocal fallback_used
fallback_used = True
buffer = MessageChain().message(initial_text) if initial_text else None
async for chain in gen:
if not isinstance(chain, MessageChain):
continue
if buffer is None:
buffer = chain
else:
buffer.chain.extend(chain.chain)
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 _flush_and_close_card() -> None:
"""补发最终文本并关闭当前卡片的流式模式。"""
nonlocal sequence
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)

try:
async for chain in generator:
if not isinstance(chain, MessageChain):
continue

if chain.type == "break":
# 飞书卡片不支持分段，忽略 break
# Tool call boundary: close current card, next text
# token will lazily create a new one below the tool
# status message.
if card_id and sender_task:
done = True
text_changed.set()
await sender_task
await _flush_and_close_card()
# Reset for lazy new-card creation
card_id = None
sequence = 0
delta = ""
last_sent = ""
done = False
sender_task = None
continue

for comp in chain.chain:
if isinstance(comp, Plain):
delta += comp.text

# Lazy card creation on first text token
if card_id is None:
card_id = await self._create_streaming_card()
if not card_id:
logger.warning(
"[Lark] 无法创建流式卡片，回退到非流式发送"
)
await _consume_rest_and_fallback(generator, delta)
return

sent = await self._send_card_message(
card_id,
reply_message_id=self.message_obj.message_id,
)
if not sent:
logger.error(
"[Lark] 发送流式卡片消息失败，回退到非流式发送"
)
await _consume_rest_and_fallback(generator, delta)
return

logger.info("[Lark] 流式输出: 使用 CardKit 流式卡片")
sender_task = asyncio.create_task(_sender_loop())

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)
if sender_task:
await sender_task

# If no text was produced at all, no card was created
if card_id is None:
if not fallback_used:
await Metric.upload(
msg_event_tick=1, adapter_name=self.platform_meta.name
)
self._has_send_oper = True
return

sequence += 1
await self._close_streaming_mode(card_id, sequence)
await _flush_and_close_card()

# Step 5: 内联父类 send_streaming 的副作用
# 内联父类 send_streaming 的副作用
await Metric.upload(msg_event_tick=1, adapter_name=self.platform_meta.name)
self._has_send_oper = True
Loading