From 4b6931d2945a4dee78abeefc33e7fd8bd8c61f59 Mon Sep 17 00:00:00 2001 From: fdc310 <2213070223@qq.com> Date: Sat, 11 Jul 2026 22:40:08 +0800 Subject: [PATCH] feat: Enhance error handling and caching for Dify and WeCom interactions --- .../libs/dify_service_api/v1/client.py | 5 +- src/langbot/libs/wecom_ai_bot_api/api.py | 9 +- .../libs/wecom_ai_bot_api/ws_client.py | 13 +-- src/langbot/pkg/platform/sources/telegram.py | 44 +++++-- src/langbot/pkg/provider/runners/difysvapi.py | 6 +- .../platform/test_telegram_adapter.py | 20 ++++ .../platform/test_wecombot_template_card.py | 28 ++++- .../provider/runners/test_difysvapi_runner.py | 108 +++++++++++++++++- 8 files changed, 202 insertions(+), 31 deletions(-) diff --git a/src/langbot/libs/dify_service_api/v1/client.py b/src/langbot/libs/dify_service_api/v1/client.py index a608a0dbb..45a52415a 100644 --- a/src/langbot/libs/dify_service_api/v1/client.py +++ b/src/langbot/libs/dify_service_api/v1/client.py @@ -156,9 +156,10 @@ class AsyncDifyServiceClient: headers={'Authorization': f'Bearer {self.api_key}'}, params={'user': user}, ) as r: + if r.status_code != 200: + body = (await r.aread()).decode(errors='replace') + raise DifyAPIError(f'{r.status_code} {body}') async for chunk in r.aiter_lines(): - if r.status_code != 200: - raise DifyAPIError(f'{r.status_code} {chunk}') if chunk.strip() == '': continue if chunk.startswith('data:'): diff --git a/src/langbot/libs/wecom_ai_bot_api/api.py b/src/langbot/libs/wecom_ai_bot_api/api.py index c274f3792..4e2c6a28a 100644 --- a/src/langbot/libs/wecom_ai_bot_api/api.py +++ b/src/langbot/libs/wecom_ai_bot_api/api.py @@ -2042,19 +2042,18 @@ class WecomBotClient: previous_content = self._stream_last_content.get(msg_id, '') if previous_content and content.startswith(previous_content): - delta_content = content[len(previous_content) :] next_content = content elif previous_content and not content: - delta_content = '' next_content = previous_content else: - delta_content = content next_content = previous_content + content if previous_content else content - if not is_final and not delta_content: + if not is_final and next_content == previous_content: return True - chunk = StreamChunk(content=delta_content, is_final=is_final) + # Follow-up responses replace the displayed stream body in WeCom. + # Publish the complete snapshot so earlier chunks remain visible. + chunk = StreamChunk(content=next_content, is_final=is_final) await self.stream_sessions.publish(stream_id, chunk) self._stream_last_content[msg_id] = next_content if is_final: diff --git a/src/langbot/libs/wecom_ai_bot_api/ws_client.py b/src/langbot/libs/wecom_ai_bot_api/ws_client.py index 3b0a3972c..32c11bd50 100644 --- a/src/langbot/libs/wecom_ai_bot_api/ws_client.py +++ b/src/langbot/libs/wecom_ai_bot_api/ws_client.py @@ -453,20 +453,17 @@ class WecomBotWsClient: try: previous_content = self._stream_last_content.get(msg_id, '') if previous_content and content.startswith(previous_content): - delta_content = content[len(previous_content) :] next_content = content elif previous_content and not content: - delta_content = '' next_content = previous_content else: - delta_content = content next_content = previous_content + content if previous_content else content # Skip sending if content hasn't changed (e.g. during tool call argument streaming) - if not is_final and not delta_content: + if not is_final and next_content == previous_content: return True - # Skip empty/whitespace-only chunks — the runner injects a + # Skip empty/whitespace-only snapshots — the runner injects a # zero-width space ('​') as a pass-through when workflow_paused # fires without any preceding LLM output. WeCom renders that # as an empty bubble that sits before the form card; skip it. @@ -476,7 +473,7 @@ class WecomBotWsClient: if not is_final: import re as _re - if not _re.sub(r'[\s​‌‍]', '', delta_content): + if not _re.sub(r'[\s​‌‍]', '', next_content): return True # Generate feedback_id for final chunk @@ -489,7 +486,9 @@ class WecomBotWsClient: if session_info: self._feedback_sessions[feedback_id] = session_info - await self.reply_stream(req_id, stream_id, delta_content, finish=is_final, feedback_id=feedback_id) + # WeCom replaces the displayed stream content on each refresh, so + # every frame must contain the complete snapshot, not only a delta. + await self.reply_stream(req_id, stream_id, next_content, finish=is_final, feedback_id=feedback_id) self._stream_last_content[msg_id] = next_content if is_final: self._stream_ids.pop(msg_id, None) diff --git a/src/langbot/pkg/platform/sources/telegram.py b/src/langbot/pkg/platform/sources/telegram.py index 62d5ee49c..295c3b274 100644 --- a/src/langbot/pkg/platform/sources/telegram.py +++ b/src/langbot/pkg/platform/sources/telegram.py @@ -10,6 +10,8 @@ import typing import traceback import json import base64 +import time +import uuid import pydantic from langbot.pkg.utils import httpclient @@ -260,7 +262,38 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None], ] = {} - _form_action_titles: typing.Dict[str, str] = {} # callback_data -> display title + _FORM_ACTION_CACHE_TTL = 30 * 60 + # callback_data -> (display title, expiration monotonic time, form group id) + _form_action_titles: typing.Dict[str, tuple[str, float, str]] = {} + + def _prune_form_action_titles(self, now: float | None = None) -> None: + now = time.monotonic() if now is None else now + expired = [key for key, (_, expires_at, _) in self._form_action_titles.items() if expires_at <= now] + for key in expired: + self._form_action_titles.pop(key, None) + + def _cache_form_action_titles(self, mappings: dict[str, str], now: float | None = None) -> None: + now = time.monotonic() if now is None else now + self._prune_form_action_titles(now) + group_id = uuid.uuid4().hex + expires_at = now + self._FORM_ACTION_CACHE_TTL + self._form_action_titles.update( + {callback_data: (title, expires_at, group_id) for callback_data, title in mappings.items()} + ) + + def _take_form_action_title(self, callback_data: str, now: float | None = None) -> str | None: + """Consume a callback and invalidate every button from the same form.""" + self._prune_form_action_titles(now) + entry = self._form_action_titles.get(callback_data) + if entry is None: + return None + title, _, group_id = entry + group_keys = [ + key for key, (_, _, cached_group_id) in self._form_action_titles.items() if cached_group_id == group_id + ] + for key in group_keys: + self._form_action_titles.pop(key, None) + return title def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): async def telegram_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): @@ -297,11 +330,11 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): w_suffix = data.get('w', '') session_key = data.get('session_key') or data.get('s', '') callback_action = _telegram_form_action_from_callback(data) - if callback_action is None or query.data not in self._form_action_titles: + action_title = self._take_form_action_title(query.data) if callback_action is not None else None + if callback_action is None or action_title is None: await self.logger.warning(f'Invalid or stale Telegram form callback: {query.data!r}') return # Show selected action feedback by editing the original message - action_title = self._form_action_titles[query.data] try: original_text = query.message.text or '' selected_text = f'{original_text}\n\n✅ {action_title}' @@ -309,9 +342,6 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): except Exception: # If edit fails (e.g. message too long), just pass pass - finally: - # Clean up the stored title - self._form_action_titles.pop(query.data, None) if session_key.startswith('group_') or session_key.startswith('g:'): launcher_type = provider_session.LauncherTypes.GROUP @@ -702,7 +732,7 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): 'text': '\n\n'.join(text_lines), } elif keyboard: - self._form_action_titles.update(pending_title_mappings) + self._cache_form_action_titles(pending_title_mappings) reply_markup = InlineKeyboardMarkup(keyboard) args = { 'chat_id': chat_id, diff --git a/src/langbot/pkg/provider/runners/difysvapi.py b/src/langbot/pkg/provider/runners/difysvapi.py index 7226bf9fe..1807294d0 100644 --- a/src/langbot/pkg/provider/runners/difysvapi.py +++ b/src/langbot/pkg/provider/runners/difysvapi.py @@ -1235,9 +1235,7 @@ class DifyServiceAPIRunner(runner.RequestRunner): if not saw_event: raise errors.DifyAPIError('Dify API did not return any workflow resume events') - if pending_content: - content, _ = self._process_thinking_content(pending_content) - yield provider_message.Message(role='assistant', content=content) + raise errors.DifyAPIError('Dify workflow resume stream ended before a terminal event') def _resolve_pending_form(self, session_key: PendingFormKey, form_action: dict) -> dict | None: """Locate the pending form this action targets. @@ -2001,6 +1999,8 @@ class DifyServiceAPIRunner(runner.RequestRunner): if is_final: return + raise errors.DifyAPIError('Dify workflow resume stream ended before a terminal event') + async def _workflow_messages_chunk( self, query: pipeline_query.Query ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]: diff --git a/tests/unit_tests/platform/test_telegram_adapter.py b/tests/unit_tests/platform/test_telegram_adapter.py index f3308f16e..c45a5c6dc 100644 --- a/tests/unit_tests/platform/test_telegram_adapter.py +++ b/tests/unit_tests/platform/test_telegram_adapter.py @@ -53,6 +53,26 @@ def test_telegram_invalid_select_callback_is_rejected(): assert _telegram_form_action_from_callback({'f': 1, 'x': 'invalid'}) is None +def test_telegram_form_callback_cache_consumes_the_whole_form_group(): + adapter = TelegramAdapter.model_construct() + adapter._form_action_titles = {} + adapter._cache_form_action_titles({'callback-a': 'A', 'callback-b': 'B'}, now=100.0) + + assert adapter._take_form_action_title('callback-a', now=101.0) == 'A' + assert adapter._take_form_action_title('callback-a', now=101.0) is None + assert adapter._take_form_action_title('callback-b', now=101.0) is None + assert adapter._form_action_titles == {} + + +def test_telegram_form_callback_cache_prunes_expired_entries(): + adapter = TelegramAdapter.model_construct() + adapter._form_action_titles = {} + adapter._cache_form_action_titles({'callback-a': 'A'}, now=100.0) + + assert adapter._take_form_action_title('callback-a', now=100.0 + adapter._FORM_ACTION_CACHE_TTL) is None + assert adapter._form_action_titles == {} + + @pytest.mark.asyncio async def test_telegram_select_field_sends_two_column_inline_keyboard(): bot = MagicMock() diff --git a/tests/unit_tests/platform/test_wecombot_template_card.py b/tests/unit_tests/platform/test_wecombot_template_card.py index 7d5c1de8e..c7ebfbee3 100644 --- a/tests/unit_tests/platform/test_wecombot_template_card.py +++ b/tests/unit_tests/platform/test_wecombot_template_card.py @@ -9,6 +9,7 @@ logger_module.EventLogger = object sys.modules.setdefault('langbot.pkg.platform.logger', logger_module) from langbot.libs.wecom_ai_bot_api.api import ( # noqa: E402 + WecomBotClient, build_button_interaction_payload, build_human_input_template_card_payload, build_button_interaction_update_card, @@ -254,7 +255,7 @@ async def test_ws_push_form_pause_sends_text_prompt_without_empty_card(): @pytest.mark.asyncio -async def test_ws_stream_sends_delta_chunks_to_wecom(): +async def test_ws_stream_sends_cumulative_snapshots_to_wecom(): client = WecomBotWsClient('bot-id', 'secret', object()) client._stream_ids['msg-1'] = 'req-1|stream-1' client._stream_sessions['msg-1'] = {} @@ -272,8 +273,29 @@ async def test_ws_stream_sends_delta_chunks_to_wecom(): assert sent == [ ('req-1', 'stream-1', '你', False), - ('req-1', 'stream-1', '好', False), - ('req-1', 'stream-1', '', True), + ('req-1', 'stream-1', '你好', False), + ('req-1', 'stream-1', '你好', True), + ] + + +@pytest.mark.asyncio +async def test_webhook_stream_queues_cumulative_snapshots_for_followups(): + client = WecomBotClient('', '', '', object(), unified_mode=True) + session, _ = client.stream_sessions.create_or_get({'msgid': 'msg-1', 'chatid': '', 'from': {'userid': 'user-1'}}) + + assert await client.push_stream_chunk('msg-1', '你', is_final=False) + assert await client.push_stream_chunk('msg-1', '你好', is_final=False) + assert await client.push_stream_chunk('msg-1', '你好', is_final=True) + + chunks = [ + await client.stream_sessions.consume(session.stream_id), + await client.stream_sessions.consume(session.stream_id), + await client.stream_sessions.consume(session.stream_id), + ] + assert [(chunk.content, chunk.is_final) for chunk in chunks] == [ + ('你', False), + ('你好', False), + ('你好', True), ] diff --git a/tests/unit_tests/provider/runners/test_difysvapi_runner.py b/tests/unit_tests/provider/runners/test_difysvapi_runner.py index 2efff4cd3..38bf1d2a5 100644 --- a/tests/unit_tests/provider/runners/test_difysvapi_runner.py +++ b/tests/unit_tests/provider/runners/test_difysvapi_runner.py @@ -11,6 +11,62 @@ from unittest.mock import AsyncMock, MagicMock import langbot_plugin.api.entities.builtin.platform.message as platform_message +class TestDifyWorkflowSubmitClient: + @pytest.mark.asyncio + async def test_rejects_empty_error_response_before_iterating_sse(self, monkeypatch): + from langbot.libs.dify_service_api.v1 import client, errors + + class FakeResponse: + status_code = 503 + + async def aread(self): + return b'' + + async def aiter_lines(self): + raise AssertionError('error responses must not enter the SSE loop') + yield + + class FakeStreamContext: + async def __aenter__(self): + return FakeResponse() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + class FakeClient: + def __init__(self, **kwargs): + del kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback): + return False + + async def post(self, *args, **kwargs): + del args, kwargs + response = MagicMock() + response.status_code = 200 + return response + + def stream(self, *args, **kwargs): + del args, kwargs + return FakeStreamContext() + + monkeypatch.setattr(client.httpx, 'AsyncClient', FakeClient) + dify_client = client.AsyncDifyServiceClient('test-key', 'https://dify.example/v1') + + with pytest.raises(errors.DifyAPIError, match='503'): + await anext( + dify_client.workflow_submit( + form_token='token-1', + workflow_run_id='run-1', + inputs={}, + user='person_user-1', + ) + ) + + class TestDifyExtractTextOutput: """Tests for _extract_dify_text_output method.""" @@ -834,7 +890,7 @@ class TestDifyHumanInputForms: assert difysvapi._get_pending_form_by_token(session_key, 'token-1') is None @pytest.mark.asyncio - async def test_failed_blocking_resume_keeps_pending_form_for_retry(self): + async def test_incomplete_blocking_resume_keeps_pending_form_for_retry(self): from langbot.libs.dify_service_api.v1.errors import DifyAPIError from langbot.pkg.provider.runners import difysvapi @@ -843,8 +899,7 @@ class TestDifyHumanInputForms: async def workflow_submit(**kwargs): del kwargs - raise DifyAPIError('temporary failure') - yield + yield {'event': 'message', 'answer': 'partial answer'} runner.dify_client.workflow_submit = workflow_submit query = MagicMock() @@ -873,12 +928,57 @@ class TestDifyHumanInputForms: } } - with pytest.raises(DifyAPIError, match='temporary failure'): + with pytest.raises(DifyAPIError, match='before a terminal event'): _ = [message async for message in runner._workflow_messages(query)] assert difysvapi._get_pending_form_by_token(session_key, 'token-1') is not None difysvapi._PENDING_FORMS.clear() + @pytest.mark.asyncio + async def test_incomplete_streaming_resume_keeps_pending_form_for_retry(self): + from langbot.libs.dify_service_api.v1.errors import DifyAPIError + from langbot.pkg.provider.runners import difysvapi + + runner = self._create_runner() + query = MagicMock() + query.session.launcher_type.value = 'person' + query.session.launcher_id = 'user-1' + query.bot_uuid = 'bot-1' + query.pipeline_uuid = 'pipeline-1' + query.variables = { + '_dify_form_action': { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'action_id': 'yes', + 'user': 'person_user-1', + 'inputs': {}, + } + } + session_key = difysvapi._session_key_from_query(query) + difysvapi._PENDING_FORMS.clear() + difysvapi._set_pending_form( + session_key, + { + 'form_token': 'token-1', + 'workflow_run_id': 'run-1', + 'actions': [{'id': 'yes', 'title': 'Yes'}], + 'inputs': {}, + 'user': 'person_user-1', + }, + ) + + async def workflow_submit(**kwargs): + del kwargs + yield {'event': 'text_chunk', 'data': {'text': 'partial answer'}} + + runner.dify_client.workflow_submit = workflow_submit + + with pytest.raises(DifyAPIError, match='before a terminal event'): + _ = [message async for message in runner._workflow_messages_chunk(query)] + + assert difysvapi._get_pending_form_by_token(session_key, 'token-1') is not None + difysvapi._PENDING_FORMS.clear() + class TestDifyCumulativeStreaming: def _create_runner(self, *, remove_think: bool = False):