From 257eb5d5987421d135ce4f2e5f13565cd55a7261 Mon Sep 17 00:00:00 2001 From: AaronZ345 <34849476+AaronZ345@users.noreply.github.com> Date: Fri, 17 Jul 2026 12:54:26 +0800 Subject: [PATCH] feat(feishu): add rich progress cards --- channel/feishu/feishu_channel.py | 189 ++++++----------------- channel/feishu/feishu_progress_card.py | 204 +++++++++++++++++++++++++ docs/channels/feishu.mdx | 1 + docs/ja/channels/feishu.mdx | 1 + docs/zh/channels/feishu.mdx | 1 + tests/test_feishu_progress_card.py | 103 +++++++++++++ 6 files changed, 358 insertions(+), 141 deletions(-) create mode 100644 channel/feishu/feishu_progress_card.py create mode 100644 tests/test_feishu_progress_card.py diff --git a/channel/feishu/feishu_channel.py b/channel/feishu/feishu_channel.py index 9a9f3307..bb99cf97 100644 --- a/channel/feishu/feishu_channel.py +++ b/channel/feishu/feishu_channel.py @@ -28,6 +28,7 @@ from bridge.context import ContextType from bridge.reply import Reply, ReplyType from channel.chat_channel import ChatChannel, check_prefix from channel.feishu.feishu_message import FeishuMessage +from channel.feishu.feishu_progress_card import FeishuProgressState from common import utils from common.expired_dict import ExpiredDict from common.log import logger @@ -722,15 +723,15 @@ class FeiShuChanel(ChatChannel): 基于飞书官方"流式更新卡片"API 实现打字机回复。 流程: - 1. message_update 首次到达 → POST /cardkit/v1/cards 创建带 streaming_mode 的卡片实体, + 1. agent_start → POST /cardkit/v1/cards 创建带 streaming_mode 的状态卡片, 随后用 POST /im/v1/messages(或 reply)以 card_id 把卡片发出去 2. 后续 message_update → PUT /cardkit/v1/cards/{id}/elements/{eid}/content 传入"当前轮"的全量文本,飞书平台自动计算增量并以打字机效果上屏 (流式模式下不受 10 QPS 限制) - 3. message_end(一轮 LLM 输出结束,且本轮触发了工具调用)→ 把 current 累计到 committed - 并加入分隔符;下一轮 message_update 又从空白开始,避免多轮内容串到一起 - 4. agent_end → 用 final_response 强制覆盖卡片,再 PATCH /cardkit/v1/cards/{id}/settings - 关闭 streaming_mode,标记 context["feishu_streamed"]=True 让 chat_channel 跳过普通 send() + 3. message_end(本轮触发工具调用)→ 原地刷新 Reasoning / Tools 面板, + 后续 turn 继续复用同一张卡片 + 4. agent_end → 用 final_response、最终状态和耗时整卡更新,关闭 streaming_mode, + 标记 context["feishu_streamed"]=True 让 chat_channel 跳过普通 send() 前提条件: - 机器人已开通 cardkit:card:write 权限 @@ -740,21 +741,16 @@ class FeiShuChanel(ChatChannel): - 创建卡片实体失败(缺权限、网络等)→ 不设置 feishu_streamed 标记,让 chat_channel 走普通文本回复路径,用户收到完整回复但无打字机效果,并打 warning 日志 """ - # 共享状态(受 lock 保护) - # 多轮 agent 模式下,每个"中间过场消息"会作为一张独立卡片发送。 - # current_text 只承载当前正在流式渲染的那张卡片的内容;message_end / agent_end - # 时会把它定型并 reset。 - current_text = [""] # 当前卡片正在累加的 LLM 输出 - card_id = [None] # 当前流式卡片的实体 ID(每段独立) - message_id = [None] # 当前卡片发送后的消息 ID(仅日志用) + # 共享状态(受 lock 保护)。一个 agent run 始终复用同一张卡片; + # reasoning、tools、最终正文和状态头均由 progress_state 统一渲染。 + progress_state = FeishuProgressState() + card_id = [None] + message_id = [None] # 占位发送是同步进行的,但用一个 in-flight 标记防止并发的多条 message_update # 事件各自触发一次创建+发送,导致发出多张卡片。 init_in_flight = [False] # 一旦初始化失败就长期标记为 disabled,本次回复不再尝试任何流式调用 disabled = [False] - # True after agent_cancelled: agent_end stops rewriting the card - # with stale final_response and just finalizes current content. - cancelled = [False] lock = threading.Lock() # ---- 异步推送队列 ---------------------------------------------------- @@ -807,14 +803,6 @@ class FeiShuChanel(ChatChannel): is_group = context.get("isgroup", False) receiver = context.get("receiver") receive_id_type = context.get("receive_id_type", "open_id") - # 客户端打字机渲染参数(飞书 App 侧实际"出字"速度): - # - print_freq_ms:每次刷新的间隔 - # - print_step:每次刷新出多少个字符 - # 当前 40ms × 4 字 ≈ 100 字/秒,接近 ChatGPT/DeepSeek 网页端的节奏。 - print_freq_ms = 40 - print_step = 4 - print_strategy = "fast" - headers = { "Authorization": "Bearer " + access_token, "Content-Type": "application/json; charset=utf-8", @@ -825,34 +813,15 @@ class FeiShuChanel(ChatChannel): sequence = [0] def _next_sequence(): - sequence[0] += 1 - return sequence[0] + with lock: + sequence[0] += 1 + return sequence[0] def _build_card_json(): - """卡片 JSON 2.0 结构 + streaming_mode + 单 markdown 组件""" - return json.dumps({ - "schema": "2.0", - "config": { - "streaming_mode": True, - "summary": {"content": "[正在生成回复...]"}, - "streaming_config": { - "print_frequency_ms": {"default": print_freq_ms}, - "print_step": {"default": print_step}, - "print_strategy": print_strategy, - }, - }, - "body": { - "elements": [ - { - "tag": "markdown", - "content": "...", - "element_id": ELEMENT_ID, - } - ], - }, - # 注意:JSON 2.0 不支持自定义 fallback 字段(传入会报错)。 - # 客户端 < 7.20 时,飞书会自动展示"请升级客户端"占位,无需配置。 - }, ensure_ascii=False) + """Build the initial streaming Card 2.0 payload.""" + with lock: + card = progress_state.build_card(streaming=True) + return json.dumps(card, ensure_ascii=False) def _create_and_send_card(): """同步执行:创建卡片实体 → 发送消息。任意一步失败则 disabled=True 触发降级""" @@ -955,37 +924,13 @@ class FeiShuChanel(ChatChannel): except Exception as e: logger.warning(f"[FeiShu] Stream: update text exception: {e}") - def _close_streaming_mode(final_text: str = ""): - """关闭流式模式(卡片转入"普通"状态,可被转发)。 - - 同时通过整卡更新接口把 summary 改成最终内容的预览,否则飞书会话列表 - 会一直显示创建卡片时的占位摘要("[正在生成回复...]")。 - """ + def _update_full_card(streaming: bool): + """Refresh panels, header, footer and streaming state in one update.""" with lock: cid = card_id[0] + full_card = progress_state.build_card(streaming=streaming) if not cid: return - - # 1) 通过整卡更新接口把 streaming_mode 关掉,并改写 summary - # (settings 接口的 config 不接受 summary 字段,会报 code=2200) - preview_src = (final_text or "").strip().replace("\n", " ") - preview = preview_src[:30] if preview_src else "" - full_card = { - "schema": "2.0", - "config": { - "streaming_mode": False, - "summary": {"content": preview or " "}, - }, - "body": { - "elements": [ - { - "tag": "markdown", - "content": final_text or " ", - "element_id": ELEMENT_ID, - } - ], - }, - } put_url = f"https://open.feishu.cn/open-apis/cardkit/v1/cards/{cid}" put_body = { "card": {"type": "card_json", "data": json.dumps(full_card, ensure_ascii=False)}, @@ -996,11 +941,11 @@ class FeiShuChanel(ChatChannel): res_json = res.json() if res_json.get("code") != 0: logger.warning( - f"[FeiShu] Stream: finalize card (close+summary) failed: {res_json}" + f"[FeiShu] Stream: full card update failed: {res_json}" ) except Exception as e: logger.warning( - f"[FeiShu] Stream: finalize card exception: {e}" + f"[FeiShu] Stream: full card update exception: {e}" ) def on_event(event: dict): @@ -1012,6 +957,19 @@ class FeiShuChanel(ChatChannel): if disabled[0]: return + if event_type == "agent_start": + with lock: + progress_state.consume(event) + if card_id[0] is None and not init_in_flight[0]: + init_in_flight[0] = True + _create_and_send_card() + return + + if event_type in ("turn_start", "reasoning_update"): + with lock: + progress_state.consume(event) + return + if event_type == "message_update": delta = data.get("delta", "") if not delta: @@ -1037,87 +995,37 @@ class FeiShuChanel(ChatChannel): snapshot = "" should_push = False with lock: - current_text[0] += delta + progress_state.consume(event) if card_id[0]: - snapshot = current_text[0] + snapshot = progress_state.current_text should_push = True if should_push: push_queue.put(snapshot) elif event_type == "message_end": - # 一轮 LLM 输出结束。如果本轮触发了工具调用,说明当前轮的文本是 - # "中间过场消息"(如"来看看!"),应该作为独立卡片定型,然后为下一轮 - # 重新创建一张新卡片。这样最终用户看到的是: - # [卡片1: 中间过场1] - # [卡片2: 中间过场2] - # ... - # [卡片N: 最终回复] - # 与 wecom_bot 的多消息流式体验对齐。 + # 工具轮结束后原地刷新 Reasoning / Tools 面板,保留同一张卡片。 tool_calls = data.get("tool_calls", []) or [] + with lock: + progress_state.consume(event) if not tool_calls: - # 没有工具调用:本轮即最终回复,留给 agent_end 统一处理。 return - - with lock: - text_to_finalize = current_text[0].rstrip() - current_text[0] = "" - - if not text_to_finalize: - return - - # 等异步队列里堆积的快照都推完,避免它们晚于 final 文本到达把内容覆盖掉 _drain_push_queue() - # 用最终文本覆盖当前卡片并关闭流式模式(凝固成普通卡片, - # 同时把会话列表的 summary 改成预览,不再显示"正在生成回复...") - _stream_update_text(text_to_finalize) - _close_streaming_mode(text_to_finalize) - - # 重置卡片状态,下一段 message_update 会触发新卡片的创建 - with lock: - card_id[0] = None - message_id[0] = None - sequence[0] = 0 + _update_full_card(streaming=True) elif event_type == "agent_cancelled": - # Lock channel into "no-rewrite" mode: the subsequent - # agent_end's final_response is from the last *completed* - # turn (the user already saw it), so rewriting the card - # would duplicate it visually. with lock: - cancelled[0] = True + progress_state.consume(event) elif event_type == "agent_end": - # 最终回复:用 final_response 覆盖当前流式卡片,然后关闭流式模式。 - final_response = data.get("final_response", "") - # 标记 streamed 让 chat_channel 跳过 send() - context["feishu_streamed"] = True - + # Finalize the same card with a status header and elapsed footer. with lock: - was_cancelled = cancelled[0] + progress_state.consume(event) + final_text = progress_state.current_text has_card = card_id[0] is not None init_busy = init_in_flight[0] - pending_text = current_text[0] + context["feishu_streamed"] = True - if was_cancelled: - # Cancelled path: finalize the in-flight card with - # partial output (or a short marker if empty); drop - # stale final_response to avoid duplicating last turn. - if has_card: - _drain_push_queue() - partial = (pending_text or "").rstrip() - final_text = partial or "_(已中止)_" - _stream_update_text(final_text) - _close_streaming_mode(final_text) - push_queue.put(None) - return - - if not final_response: - return - final_text = str(final_response) - - # 罕见情况:agent_end 触发时还没创建过卡片(极快返回 / 没有 - # message_update),主动创建一张承载 final_text。 if not has_card and not init_busy: with lock: init_in_flight[0] = True @@ -1128,8 +1036,7 @@ class FeiShuChanel(ChatChannel): _drain_push_queue() _stream_update_text(final_text) - _close_streaming_mode(final_text) - # 通知 push worker 退出(本次回复彻底结束) + _update_full_card(streaming=False) push_queue.put(None) return on_event diff --git a/channel/feishu/feishu_progress_card.py b/channel/feishu/feishu_progress_card.py new file mode 100644 index 00000000..88116c75 --- /dev/null +++ b/channel/feishu/feishu_progress_card.py @@ -0,0 +1,204 @@ +"""State and Card 2.0 rendering for a Feishu agent run.""" + +from __future__ import annotations + +import time +from typing import Any, Dict, List, Optional + + +_MAX_PANEL_STEPS = 10 +_MAX_STEP_CHARS = 800 + + +class FeishuProgressState: + """Reduce CowAgent stream events into one renderable Feishu card state.""" + + def __init__(self, started_at: Optional[float] = None): + self.started_at = time.monotonic() if started_at is None else started_at + self.status = "running" + self.turns = 0 + self.current_text = "" + self._reasoning_buffer = "" + self.reasoning_steps: List[str] = [] + self.tool_steps: List[Dict[str, str]] = [] + self.cancelled = False + + def consume(self, event: Dict[str, Any]) -> None: + """Consume one event emitted by ``AgentStreamHandler``.""" + event_type = event.get("type") + data = event.get("data") or {} + + if event_type == "turn_start": + self._mark_running_tools_done() + turn = data.get("turn") + if isinstance(turn, int): + self.turns = max(self.turns, turn) + else: + self.turns += 1 + if self.turns > 1: + self.current_text = "" + return + + if event_type == "reasoning_update": + self._reasoning_buffer += str(data.get("delta") or "") + return + + if event_type == "message_update": + self.current_text += str(data.get("delta") or "") + return + + if event_type == "message_end": + self._commit_reasoning() + for tool_call in data.get("tool_calls") or []: + self.tool_steps.append( + { + "summary": _format_tool_call(tool_call), + "status": "running", + } + ) + return + + if event_type == "agent_cancelled": + self.cancelled = True + self.status = "stopped" + return + + if event_type == "agent_end": + self._commit_reasoning() + self._mark_running_tools_done() + cancelled = self.cancelled or bool(data.get("cancelled")) + if cancelled: + self.status = "stopped" + self.current_text = self.current_text.rstrip() or "_(stopped)_" + else: + self.status = "done" + final_response = data.get("final_response") + if final_response: + self.current_text = str(final_response) + + def build_card(self, streaming: bool, now: Optional[float] = None) -> Dict[str, Any]: + """Render the current state as a Feishu Card 2.0 object.""" + title, template = { + "running": ("Working", "blue"), + "done": ("Done", "green"), + "stopped": ("Stopped", "grey"), + "error": ("Error", "red"), + }.get(self.status, ("Working", "blue")) + + main_text = self.current_text or "..." + elements: List[Dict[str, Any]] = [] + + if self.reasoning_steps: + elements.append( + _panel( + "Reasoning ({})".format(len(self.reasoning_steps)), + [_text_row(step, muted=True) for step in self.reasoning_steps[-_MAX_PANEL_STEPS:]], + expanded=streaming, + ) + ) + elif streaming: + elements.append( + _panel("Reasoning", [_text_row("Thinking...", muted=True)], expanded=True) + ) + + if self.tool_steps: + elements.append( + _panel( + "Tools ({})".format(len(self.tool_steps)), + [ + _text_row("{} · {}".format(step["summary"], step["status"])) + for step in self.tool_steps[-_MAX_PANEL_STEPS:] + ], + expanded=streaming, + ) + ) + + elements.append( + { + "tag": "markdown", + "element_id": "stream_md", + "content": main_text, + } + ) + + elapsed = max(0.0, (time.monotonic() if now is None else now) - self.started_at) + turn_label = "turn" if self.turns == 1 else "turns" + elements.extend( + [ + {"tag": "hr"}, + { + "tag": "markdown", + "content": "{:.1f}s · {} {}".format(elapsed, self.turns, turn_label), + "text_size": "notation", + }, + ] + ) + + config: Dict[str, Any] = { + "streaming_mode": streaming, + "update_multi": True, + "enable_forward_interaction": True, + "summary": {"content": _summary(main_text, title)}, + } + if streaming: + config["streaming_config"] = { + "print_frequency_ms": {"default": 40}, + "print_step": {"default": 4}, + "print_strategy": "fast", + } + + return { + "schema": "2.0", + "config": config, + "header": { + "template": template, + "title": {"tag": "plain_text", "content": title}, + }, + "body": {"elements": elements}, + } + + def _commit_reasoning(self) -> None: + reasoning = self._reasoning_buffer.strip() + if reasoning: + self.reasoning_steps.append(reasoning[-_MAX_STEP_CHARS:]) + self._reasoning_buffer = "" + + def _mark_running_tools_done(self) -> None: + for step in self.tool_steps: + if step["status"] == "running": + step["status"] = "done" + + +def _format_tool_call(tool_call: Dict[str, Any]) -> str: + # Tool arguments can contain user data or credentials. The progress card + # only needs an activity label, so never echo raw arguments into chat. + return str(tool_call.get("name") or "tool") + + +def _panel(title: str, elements: List[Dict[str, Any]], expanded: bool) -> Dict[str, Any]: + return { + "tag": "collapsible_panel", + "expanded": expanded, + "background_color": "grey", + "header": {"title": {"tag": "plain_text", "content": title}}, + "border": {"color": "grey"}, + "vertical_spacing": "8px", + "padding": "4px 8px", + "elements": elements, + } + + +def _text_row(content: str, muted: bool = False) -> Dict[str, Any]: + text = { + "tag": "plain_text", + "content": content, + "text_size": "notation", + } + if muted: + text["text_color"] = "grey" + return {"tag": "div", "text": text} + + +def _summary(text: str, fallback: str) -> str: + preview = " ".join(text.strip().split()) + return preview[:60] or fallback diff --git a/docs/channels/feishu.mdx b/docs/channels/feishu.mdx index 1283d0c1..67e096cd 100644 --- a/docs/channels/feishu.mdx +++ b/docs/channels/feishu.mdx @@ -98,6 +98,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | Image messages | ✅ send/receive | | Voice messages | ✅ send/receive | | Streaming reply | ✅ (powered by Feishu cardkit streaming card) | +| Rich progress card | ✅ status header, reasoning/tool panels and elapsed time | Streaming reply requires the `cardkit:card:write` permission (already enabled by one-click creation) and Feishu client version ≥ 7.20. Older clients see an upgrade prompt; if the permission or version is not satisfied, replies fall back to plain text automatically. diff --git a/docs/ja/channels/feishu.mdx b/docs/ja/channels/feishu.mdx index d703954a..604fb914 100644 --- a/docs/ja/channels/feishu.mdx +++ b/docs/ja/channels/feishu.mdx @@ -95,6 +95,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | 画像メッセージ | ✅ 送受信 | | 音声メッセージ | ✅ 送受信 | | ストリーミング応答 | ✅(Feishu cardkit ストリーミングカードベース) | +| リッチ進捗カード | ✅ ステータス、推論/ツールパネル、経過時間 | ストリーミング応答には `cardkit:card:write` 権限(ワンクリック作成では自動付与)と Feishu クライアント 7.20 以上が必要です。古いクライアントではアップグレード案内が表示され、権限/バージョン未充足時は通常テキスト応答に自動フォールバックします。 diff --git a/docs/zh/channels/feishu.mdx b/docs/zh/channels/feishu.mdx index 5cb8fe80..04adf931 100644 --- a/docs/zh/channels/feishu.mdx +++ b/docs/zh/channels/feishu.mdx @@ -99,6 +99,7 @@ im:message,im:message.group_at_msg,im:message.group_at_msg:readonly,im:message.p | 图片消息 | ✅ 收发 | | 语音消息 | ✅ 收发 | | 流式回复 | ✅(通过 `feishu_stream_reply` 配置控制,默认开启) | +| 富进度卡片 | ✅ 状态头、思考/工具面板与耗时 | 流式回复需要机器人具备 `cardkit:card:write` 权限(一键创建已默认开通),且接收方飞书客户端版本 ≥ 7.20。低版本客户端会显示升级提示,权限或版本不满足时自动降级为普通文本回复。 diff --git a/tests/test_feishu_progress_card.py b/tests/test_feishu_progress_card.py new file mode 100644 index 00000000..91e76ba7 --- /dev/null +++ b/tests/test_feishu_progress_card.py @@ -0,0 +1,103 @@ +from channel.feishu.feishu_progress_card import FeishuProgressState + + +def _panels(card): + return [ + element + for element in card["body"]["elements"] + if element.get("tag") == "collapsible_panel" + ] + + +def test_running_card_has_status_reasoning_lane_and_stream_target(): + state = FeishuProgressState(started_at=100.0) + + card = state.build_card(streaming=True, now=102.0) + + assert card["header"]["template"] == "blue" + assert card["header"]["title"]["content"] == "Working" + assert card["config"]["streaming_mode"] is True + assert _panels(card)[0]["header"]["title"]["content"] == "Reasoning" + stream = next( + element + for element in card["body"]["elements"] + if element.get("element_id") == "stream_md" + ) + assert stream["content"] == "..." + + +def test_tool_turn_populates_reasoning_and_tools_lanes(): + state = FeishuProgressState(started_at=100.0) + state.consume({"type": "turn_start", "data": {"turn": 1}}) + state.consume({"type": "reasoning_update", "data": {"delta": "Need the latest files."}}) + state.consume({"type": "message_update", "data": {"delta": "Checking now."}}) + state.consume( + { + "type": "message_end", + "data": { + "tool_calls": [ + {"name": "read_file", "arguments": {"path": "README.md"}} + ] + }, + } + ) + + card = state.build_card(streaming=True, now=103.0) + panels = _panels(card) + assert [panel["header"]["title"]["content"] for panel in panels] == [ + "Reasoning (1)", + "Tools (1)", + ] + assert panels[0]["elements"][0]["text"]["content"] == "Need the latest files." + assert "read_file" in panels[1]["elements"][0]["text"]["content"] + assert "running" in panels[1]["elements"][0]["text"]["content"] + + state.consume({"type": "turn_start", "data": {"turn": 2}}) + done_card = state.build_card(streaming=True, now=104.0) + assert "done" in _panels(done_card)[1]["elements"][0]["text"]["content"] + + +def test_agent_end_finalizes_status_body_and_footer(): + state = FeishuProgressState(started_at=100.0) + state.consume({"type": "turn_start", "data": {"turn": 1}}) + state.consume( + { + "type": "agent_end", + "data": {"final_response": "**Finished**", "cancelled": False}, + } + ) + + card = state.build_card(streaming=False, now=105.2) + + assert card["header"]["template"] == "green" + assert card["header"]["title"]["content"] == "Done" + assert card["config"]["streaming_mode"] is False + assert next( + element["content"] + for element in card["body"]["elements"] + if element.get("element_id") == "stream_md" + ) == "**Finished**" + footer = card["body"]["elements"][-1] + assert footer["text_size"] == "notation" + assert footer["content"] == "5.2s · 1 turn" + + +def test_cancelled_agent_uses_partial_output_and_stopped_status(): + state = FeishuProgressState(started_at=100.0) + state.consume({"type": "message_update", "data": {"delta": "partial"}}) + state.consume({"type": "agent_cancelled", "data": {}}) + state.consume( + { + "type": "agent_end", + "data": {"final_response": "stale response", "cancelled": True}, + } + ) + + card = state.build_card(streaming=False, now=101.0) + + assert card["header"]["title"]["content"] == "Stopped" + assert next( + element["content"] + for element in card["body"]["elements"] + if element.get("element_id") == "stream_md" + ) == "partial"