diff --git a/src/langbot/libs/dingtalk_api/api.py b/src/langbot/libs/dingtalk_api/api.py index 9a7c2314f..5cc4c9571 100644 --- a/src/langbot/libs/dingtalk_api/api.py +++ b/src/langbot/libs/dingtalk_api/api.py @@ -712,13 +712,20 @@ class DingTalkClient: Returns ``(card state, out_track_id)`` for subsequent A2UI updates. """ + is_group = str(incoming_message.conversation_type) == '2' + return await self.create_and_card_for_target( + 'group' if is_group else 'person', + incoming_message.conversation_id if is_group else incoming_message.sender_staff_id, + ) + + async def create_and_card_for_target(self, target_type: str, target_id: str): + """Create a streaming card when the reply has no original chat message.""" + if target_type not in {'group', 'person'} or not target_id: + raise ValueError('DingTalk streaming card requires a group or person target') out_track_id = uuid.uuid4().hex surface_id = f'langbot-{out_track_id}' - is_group = str(incoming_message.conversation_type) == '2' - if is_group: - open_space_id = f'dtv1.card//im_group.{incoming_message.conversation_id}' - else: - open_space_id = f'dtv1.card//im_robot.{incoming_message.sender_staff_id}' + is_group = target_type == 'group' + open_space_id = f'dtv1.card//im_group.{target_id}' if is_group else f'dtv1.card//im_robot.{target_id}' await self.create_and_deliver_ai_card( out_track_id=out_track_id, diff --git a/src/langbot/pkg/agent/runner/reply_stream.py b/src/langbot/pkg/agent/runner/reply_stream.py index aa2f3500f..b9beb8817 100644 --- a/src/langbot/pkg/agent/runner/reply_stream.py +++ b/src/langbot/pkg/agent/runner/reply_stream.py @@ -86,9 +86,24 @@ class ReplyStreamSession: raise ValueError('This run has no platform delivery adapter') # Set before I/O: an uncertain create must never be retried as another send. reply.started = True - reply.native = self.source is not None and await self.adapter.is_stream_output_supported() + target_streaming = ( + self.source is None + and callable(getattr(self.adapter, 'create_message_card_for_target', None)) + and callable(getattr(self.adapter, 'reply_message_chunk_for_target', None)) + ) + reply.native = ( + self.source is not None or target_streaming + ) and await self.adapter.is_stream_output_supported() if reply.native: - await self.adapter.create_message_card(reply.delivery_id, self.source) + if self.source is not None: + await self.adapter.create_message_card(reply.delivery_id, self.source) + else: + target_type, target_id = self.target.get('target_type'), self.target.get('target_id') + if not target_type or not target_id: + raise ValueError('This event has no reply target') + await self.adapter.create_message_card_for_target( + reply.delivery_id, str(target_type), str(target_id) + ) if reply.started and reply.native: await self._update_native(key, reply, final=request.operation != 'update') elif request.operation == 'finish' and reply.text and not self.mock: @@ -131,13 +146,18 @@ class ReplyStreamSession: msg_sequence=reply.sequence, is_final=final, ) - await self.adapter.reply_message_chunk( - message_source=self.source, - bot_message=chunk, - message=self._message(reply.text), - quote_origin=False, - is_final=final, - ) + if self.source is not None: + await self.adapter.reply_message_chunk( + message_source=self.source, + bot_message=chunk, + message=self._message(reply.text), + quote_origin=False, + is_final=final, + ) + else: + await self.adapter.reply_message_chunk_for_target( + message_id=reply.delivery_id, content=reply.text, is_final=final + ) async def close(self): """Close visible streams on cancellation; never send a buffered partial reply.""" diff --git a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py index 6564510c0..12ad43e6b 100644 --- a/src/langbot/pkg/platform/adapters/dingtalk/adapter.py +++ b/src/langbot/pkg/platform/adapters/dingtalk/adapter.py @@ -263,6 +263,25 @@ class DingTalkAdapter(DingTalkAPIMixin, abstract_platform_adapter.AbstractPlatfo self.card_instance_id_dict[message_id] = (card_instance, card_instance_id) return True + async def create_message_card_for_target(self, message_id: str, target_type: str, target_id: str) -> bool: + card_instance, card_instance_id = await self.bot.create_and_card_for_target(target_type, target_id) + self.card_instance_id_dict[message_id] = (card_instance, card_instance_id) + return True + + async def reply_message_chunk_for_target(self, *, message_id: str, content: str, is_final: bool) -> None: + now = time.monotonic() + last_updates = getattr(self, '_card_last_updates', None) + if last_updates is None: + self._card_last_updates = last_updates = {} + if not is_final and now - last_updates.get(message_id, 0) < 0.35: + return + card_instance, card_instance_id = self.card_instance_id_dict[message_id] + await self.bot.send_card_message(card_instance, card_instance_id, content, is_final) + last_updates[message_id] = now + if is_final: + self.card_instance_id_dict.pop(message_id, None) + last_updates.pop(message_id, None) + async def is_stream_output_supported(self) -> bool: return is_stream_card_configured(self.config) diff --git a/tests/unit_tests/agent/test_reply_stream.py b/tests/unit_tests/agent/test_reply_stream.py index 0f45a690f..5ffba91fc 100644 --- a/tests/unit_tests/agent/test_reply_stream.py +++ b/tests/unit_tests/agent/test_reply_stream.py @@ -120,6 +120,28 @@ async def test_sdk_stream_reuses_adapter_or_sends_one_final_message(native, sour await session.close() +async def test_target_streaming_adapter_updates_card_without_source_message(): + session, adapter, _ = make_session(source=False) + adapter.create_message_card_for_target = AsyncMock(return_value=True) + adapter.reply_message_chunk_for_target = AsyncMock() + key = uuid4() + + await session.apply(request(key, text='hello')) + result = await session.apply(request(key, 'finish', 'hello world')) + + assert result['delivery'] == 'streaming' + adapter.create_message_card_for_target.assert_awaited_once_with( + adapter.reply_message_chunk_for_target.await_args.kwargs['message_id'], 'person', 'user-1' + ) + assert [call.kwargs['content'] for call in adapter.reply_message_chunk_for_target.await_args_list] == [ + 'hello', + 'hello world', + ] + assert [call.kwargs['is_final'] for call in adapter.reply_message_chunk_for_target.await_args_list] == [False, True] + adapter.send_message.assert_not_awaited() + await session.close() + + @pytest.mark.parametrize('native', [True, False]) @pytest.mark.parametrize('error', [RuntimeError, asyncio.CancelledError]) async def test_exception_finalizes_visible_card_without_sending_buffered_partial(native, error): diff --git a/tests/unit_tests/platform/test_dingtalk_api.py b/tests/unit_tests/platform/test_dingtalk_api.py index 73ead0fd6..0ab1ad764 100644 --- a/tests/unit_tests/platform/test_dingtalk_api.py +++ b/tests/unit_tests/platform/test_dingtalk_api.py @@ -46,6 +46,20 @@ def test_text_card_keeps_brand_when_content_updates(): assert update_text_card_messages('surface-id', 'done')[0]['updateDataModel']['path'] == '/content' +@pytest.mark.asyncio +async def test_create_card_for_target_uses_robot_a2ui_delivery(): + client = _client(None) + client.create_and_deliver_ai_card = AsyncMock(return_value={}) + + card, track_id = await client.create_and_card_for_target('person', 'user-1') + + assert card == {'surface_id': f'langbot-{track_id}'} + request = client.create_and_deliver_ai_card.await_args.kwargs + assert request['open_space_id'] == 'dtv1.card//im_robot.user-1' + assert request['is_group'] is False + assert request['a2ui_messages'][2]['updateComponents']['components'][0]['children'][0] == 'langbot-brand' + + @pytest.mark.asyncio async def test_create_ai_card_uses_group_robot_delivery_model(): post = AsyncMock( diff --git a/tests/unit_tests/platform/test_dingtalk_eba_adapter.py b/tests/unit_tests/platform/test_dingtalk_eba_adapter.py index 916311f6a..3009e70d9 100644 --- a/tests/unit_tests/platform/test_dingtalk_eba_adapter.py +++ b/tests/unit_tests/platform/test_dingtalk_eba_adapter.py @@ -51,6 +51,7 @@ class DummyDingTalkClient(DingTalkClient): self.get_audio_url = AsyncMock(return_value='data:audio/ogg;base64,AAAA') self.download_image = AsyncMock(return_value='data:image/png;base64,BBBB') self.create_and_card = AsyncMock(return_value=('card', 'card-id')) + self.create_and_card_for_target = AsyncMock(return_value=({'surface_id': 'surface-id'}, 'card-id')) self.send_card_message = AsyncMock() self.create_and_deliver_ai_card = AsyncMock(return_value={}) self.update_ai_card = AsyncMock(return_value={}) @@ -318,6 +319,20 @@ async def test_dingtalk_send_reply_and_platform_api_use_underlying_client(): assert file_url == {'url': 'https://example.test/file'} +@pytest.mark.asyncio +async def test_dingtalk_target_reply_stream_updates_one_card(): + adapter = make_adapter() + + assert await adapter.create_message_card_for_target('reply-id', 'person', 'user-1') is True + await adapter.reply_message_chunk_for_target(message_id='reply-id', content='hello', is_final=False) + await adapter.reply_message_chunk_for_target(message_id='reply-id', content='hello world', is_final=True) + + adapter.bot.create_and_card_for_target.assert_awaited_once_with('person', 'user-1') + assert [call.args[2] for call in adapter.bot.send_card_message.await_args_list] == ['hello', 'hello world'] + assert [call.args[3] for call in adapter.bot.send_card_message.await_args_list] == [False, True] + assert 'reply-id' not in adapter.card_instance_id_dict + + @pytest.mark.asyncio async def test_dingtalk_interaction_delivery_and_callback_event(): adapter = make_adapter()