xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-10-30 04:15:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(lark): stream replies after HITL submissions

Browse Source
This commit is contained in:
fdc310
2026-10-30 01:06:10 +08:00
parent 21e2187c6c
commit 983e12acdf
2 changed files with 109 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/langbot/pkg/platform/adapters/lark/adapter.py
+60 -15
View File
@@ -490,6 +490,31 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
self.reply_to_monitoring_msg.pop(next(iter(self.reply_to_monitoring_msg)), None)
return True
async def create_message_card_for_target(self, message_id: str, target_type: str, target_id: str) -> bool:
if target_type not in {'group', 'person'} or not target_id:
raise ValueError('Lark streaming card requires a group or person target')
card_id = await self.create_card_id(message_id)
content = {'type': 'card', 'data': {'card_id': card_id, 'template_variable': {'content': 'Thinking...'}}}
request = (
CreateMessageRequest.builder()
.receive_id_type('chat_id' if target_type == 'group' else 'open_id')
.request_body(
CreateMessageRequestBody.builder()
.receive_id(target_id)
.content(json.dumps(content))
.msg_type('interactive')
.uuid(str(uuid.uuid4()))
.build()
)
.build()
)
response: CreateMessageResponse = await self.api_client.im.v1.message.acreate(request)
if not response.success():
self.card_id_dict.pop(message_id, None)
self.card_sequence_dict.pop(card_id, None)
raise RuntimeError(f'Lark create_message_card_for_target failed: {response.code} {response.msg}')
return True
async def create_card_id(self, message_id) -> str:
while len(self.card_id_dict) >= 1000:
old_key = next(iter(self.card_id_dict))
@@ -552,13 +577,6 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
quote_origin: bool = False,
is_final: bool = False,
):
card_id = self.card_id_dict[bot_message.resp_message_id]
has_sent_update = self.card_sequence_dict.get(card_id, 0) > 0
now = time.monotonic()
last_update = self.card_last_update_dict.get(card_id, 0.0)
is_high_frequency_chunk = now - last_update < 1.0
if has_sent_update and is_high_frequency_chunk and bot_message.msg_sequence % 8 != 0 and not is_final:
return
cumulative_content = getattr(bot_message, 'all_content', None)
if isinstance(cumulative_content, str) and cumulative_content:
content = cumulative_content
@@ -568,8 +586,37 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
''.join(ele.get('text', '') for ele in paragraph if ele.get('tag') in {'text', 'md'})
for paragraph in text_elements
)
if (is_final and not bot_message.tool_calls) or card_id in self.closed_streaming_cards:
await self._replace_streaming_card(card_id, content, finished=is_final and not bot_message.tool_calls)
await self._update_streaming_card(
bot_message.resp_message_id,
content,
is_final=is_final,
tool_calls=bot_message.tool_calls,
msg_sequence=bot_message.msg_sequence,
tenant_key=self._tenant_key_from_source(message_source),
)
async def reply_message_chunk_for_target(self, *, message_id: str, content: str, is_final: bool) -> None:
await self._update_streaming_card(message_id, content, is_final=is_final, msg_sequence=1)
async def _update_streaming_card(
self,
message_id: str,
content: str,
*,
is_final: bool,
msg_sequence: int,
tool_calls: typing.Any = None,
tenant_key: str | None = None,
) -> None:
card_id = self.card_id_dict[message_id]
has_sent_update = self.card_sequence_dict.get(card_id, 0) > 0
now = time.monotonic()
last_update = self.card_last_update_dict.get(card_id, 0.0)
is_high_frequency_chunk = now - last_update < 1.0
if has_sent_update and is_high_frequency_chunk and msg_sequence % 8 != 0 and not is_final:
return
if (is_final and not tool_calls) or card_id in self.closed_streaming_cards:
await self._replace_streaming_card(card_id, content, finished=is_final and not tool_calls)
else:
sequence = self._next_card_sequence(card_id)
request = (
@@ -582,18 +629,16 @@ class LarkAdapter(LarkAPIMixin, abstract_platform_adapter.AbstractPlatformAdapte
.build()
)
response: ContentCardElementResponse = await self.api_client.cardkit.v1.card_element.acontent(
request, self.request_option(self._tenant_key_from_source(message_source))
request, self.request_option(tenant_key)
)
if not response.success():
if self._streaming_mode_closed(response):
await self._replace_streaming_card(
card_id, content, finished=is_final and not bot_message.tool_calls
)
await self._replace_streaming_card(card_id, content, finished=is_final and not tool_calls)
else:
raise RuntimeError(f'Lark card_element update failed: {response.code} {response.msg}')
self.card_last_update_dict[card_id] = now
if is_final and bot_message.tool_calls is None:
self.card_id_dict.pop(bot_message.resp_message_id, None)
if is_final and tool_calls is None:
self.card_id_dict.pop(message_id, None)
self.card_sequence_dict.pop(card_id, None)
self.card_last_update_dict.pop(card_id, None)
self.closed_streaming_cards.discard(card_id)
tests/unit_tests/platform/test_lark_eba_adapter.py
+49 -2
View File
@@ -7,6 +7,7 @@ import asyncio
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4
import pytest
import yaml
@@ -20,6 +21,7 @@ from langbot.pkg.platform.adapters.lark.interaction import (
interaction_delivery_capabilities,
interaction_event_from_callback,
)
from langbot.pkg.agent.runner.reply_stream import ReplyStreamRequest, ReplyStreamSession
from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
from langbot_plugin.api.entities.builtin.platform import entities as platform_entities
from langbot_plugin.api.entities.builtin.platform import events as platform_events
@@ -729,7 +731,10 @@ async def test_lark_streaming_card_uses_strictly_increasing_sequences():
final_card = json.loads(final_request.request_body.card.data)
assert final_card['body']['elements'][0] == {'tag': 'markdown', 'content': 'answer'}
assert 'header' not in final_card
assert final_card['body']['elements'][-1]['columns'][1]['elements'][0]['content'] == '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
assert (
final_card['body']['elements'][-1]['columns'][1]['elements'][0]['content']
== '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
)
assert not final_card['config'].get('streaming_mode', False)
assert 'response-1' not in adapter.card_id_dict
assert 'stream-card-1' not in adapter.card_sequence_dict
@@ -828,6 +833,43 @@ async def test_lark_streaming_card_falls_back_to_full_card_update_when_stream_cl
assert 'stream-card-1' not in adapter.closed_streaming_cards
@pytest.mark.asyncio
@pytest.mark.parametrize(
('target_type', 'target_id', 'receive_id_type'),
[
('person', 'user-1', 'open_id'),
('group', 'chat-1', 'chat_id'),
],
)
async def test_lark_hitl_reply_stream_creates_and_updates_new_card(target_type, target_id, receive_id_type):
adapter = make_adapter({'enable-stream-reply': True})
event = SimpleNamespace(
event_type='interaction.submitted',
delivery=SimpleNamespace(
surface='platform',
platform_capabilities={},
reply_target={'target_type': target_type, 'target_id': target_id, 'message_id': 'interaction-card'},
),
)
session = ReplyStreamSession(event, adapter)
stream_id = uuid4()
first = await session.apply(ReplyStreamRequest(stream_id=stream_id, operation='update', text='partial'))
final = await session.apply(ReplyStreamRequest(stream_id=stream_id, operation='finish', text='complete'))
assert first['delivery'] == final['delivery'] == 'streaming'
request = adapter.api_client.im.v1.message.acreate.await_args.args[0]
assert request.receive_id_type == receive_id_type
assert request.request_body.receive_id == target_id
assert request.request_body.msg_type == 'interactive'
assert adapter.api_client.im.v1.message.areply.await_count == 0
assert adapter.api_client.cardkit.v1.card_element.acontent.await_count == 1
final_card = json.loads(adapter.api_client.cardkit.v1.card.aupdate.await_args.args[0].request_body.card.data)
assert final_card['body']['elements'][0]['content'] == 'complete'
assert not adapter.card_id_dict
await session.close()
def test_lark_interaction_callback_builds_scoped_host_event():
callback = SimpleNamespace(
event=SimpleNamespace(
@@ -991,6 +1033,7 @@ async def test_lark_webhook_rejects_wrong_signature():
def test_lark_stream_card_separates_multiple_reasoning_rounds():
from langbot.pkg.platform.adapters.lark.stream_card import build_card, streaming_text
content = '<think>first thought</think>first answer<think>second thought</think>final answer'
preview = streaming_text(content)
assert '<think>' not in preview and '</think>' not in preview
@@ -1006,6 +1049,7 @@ def test_lark_stream_card_separates_multiple_reasoning_rounds():
def test_lark_live_preview_is_bounded_but_final_reasoning_is_complete():
from langbot.pkg.platform.adapters.lark.stream_card import build_card, streaming_text
thought = 'reasoning ' * 200
content = '<think>' + thought + '</thi'
assert len(streaming_text(content)) < 450
@@ -1022,6 +1066,9 @@ async def test_lark_stream_card_creation_is_async_and_branded():
request = adapter.api_client.cardkit.v1.card.acreate.await_args.args[0]
card = json.loads(request.request_body.data)
assert 'header' not in card
assert card['body']['elements'][-1]['columns'][1]['elements'][0]['content'] == '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
assert (
card['body']['elements'][-1]['columns'][1]['elements'][0]['content']
== '[LangBot](https://langbot.app?utm_source=feishu&utm_medium=bot_card&utm_campaign=langbot)'
)
assert card['config']['streaming_mode'] is True
assert card['body']['elements'][0]['element_id'] == 'streaming_txt'