-
Notifications
You must be signed in to change notification settings - Fork 1
feat: 重构 SDK Client 为 SQS 异步架构 #11
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.
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
Merged
Changes from all commits
Commits
Show all changes
6 commits
Select commit
Hold shift + click to select a range
f67d856
feat: 重构 SDK Client 为 SQS 异步架构
BukeLy 8b2dc5e
merge: 解决与 main 分支的 README.md 冲突
BukeLy ee12d84
fix: 添加 ConsumerFunction 的 SQS 权限策略并增加 HTTP 状态码检查
BukeLy e869aee
fix: 修复code review的3个高优先级问题
BukeLy d18e3bd
docs: 添加架构调整的CHANGELOG
BukeLy e6e64cb
fix: 增强异常处理和Issue#12修复
BukeLy File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Some comments aren't visible on the classic Files Changed page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| # Changelog | ||
|
|
||
| ## [Unreleased] | ||
|
|
||
| ### Changed | ||
| - **架构调整：从同步到异步处理模式** | ||
| - 重构SDK Client为SQS异步架构 | ||
| - Producer Lambda（SdkClientFunction）：接收Telegram webhook，立即返回200，消息写入SQS队列 | ||
| - Consumer Lambda（ConsumerFunction）：异步处理SQS消息，调用Agent Server，返回结果给Telegram | ||
| - 解决Telegram 30秒webhook超时问题，支持长运行任务 | ||
|
|
||
| ### Added | ||
| - SQS任务队列（TaskQueue）和死信队列（DLQueue） | ||
| - SNS告警主题（AlarmTopic）用于CloudWatch通知 | ||
| - CloudWatch告警和自定义指标监控 | ||
| - DynamoDB会话表用于多轮对话状态管理 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| """Lambda handler for SQS Consumer. | ||
|
|
||
| Processes messages from SQS queue, calls Agent Server, sends response to Telegram. | ||
| """ | ||
| import asyncio | ||
| import json | ||
| from typing import Any | ||
|
|
||
| import httpx | ||
| from telegram import Bot, Update | ||
| from telegram.constants import ParseMode, ChatAction | ||
| from telegram.helpers import escape_markdown | ||
| from telegram.error import BadRequest | ||
|
|
||
| from config import Config | ||
|
|
||
|
|
||
| def lambda_handler(event: dict, context: Any) -> dict: | ||
| """SQS Consumer Lambda entry point.""" | ||
| for record in event['Records']: | ||
| try: | ||
| message_data = json.loads(record['body']) | ||
| except json.JSONDecodeError as e: | ||
| # Invalid message format - log and skip | ||
| import logging | ||
| logger = logging.getLogger() | ||
| logger.error(f"Failed to parse SQS message: {e}") | ||
| continue | ||
|
|
||
| try: | ||
| asyncio.run(process_message(message_data)) | ||
| except Exception as e: | ||
| # Log and let SQS retry on failure | ||
| import logging | ||
| logger = logging.getLogger() | ||
| logger.exception(f"Failed to process message: {e}") | ||
| raise # Re-raise to fail the batch item | ||
|
|
||
| return {'statusCode': 200} | ||
|
|
||
|
|
||
| async def process_message(message_data: dict) -> None: | ||
| """Process single message from SQS queue.""" | ||
| import logging | ||
| logger = logging.getLogger() | ||
|
|
||
| config = Config.from_env() | ||
| bot = Bot(config.telegram_token) | ||
|
|
||
| # Reconstruct Update object from stored data | ||
| update = Update.de_json(message_data['telegram_update'], bot) | ||
| message = update.message or update.edited_message | ||
|
Comment on lines
+50
to
+52
|
||
|
|
||
| if not message: | ||
| logger.warning("Received update with no message or edited_message") | ||
| return | ||
|
|
||
| # Send typing indicator | ||
| await bot.send_chat_action( | ||
| chat_id=message.chat_id, | ||
| action=ChatAction.TYPING, | ||
| message_thread_id=message.message_thread_id, | ||
| ) | ||
|
|
||
| # Initialize result with default error response | ||
| # This ensures result is always defined, even if Agent Server call fails | ||
| result = { | ||
| 'response': '', | ||
| 'is_error': True, | ||
| 'error_message': 'Failed to get response from Agent Server' | ||
| } | ||
|
|
||
| # Call Agent Server | ||
| try: | ||
| async with httpx.AsyncClient(timeout=600.0) as client: | ||
| response = await client.post( | ||
| config.agent_server_url, | ||
| headers={ | ||
| 'Authorization': f'Bearer {config.auth_token}', | ||
| 'Content-Type': 'application/json', | ||
| }, | ||
| json={ | ||
| 'user_message': message.text, | ||
| 'chat_id': str(message.chat_id), | ||
| 'thread_id': str(message.message_thread_id) if message.message_thread_id else None, | ||
| }, | ||
| ) | ||
BukeLy marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| response.raise_for_status() | ||
| result = response.json() | ||
|
|
||
| except httpx.TimeoutException: | ||
| logger.warning(f"Agent Server timeout for chat_id={message.chat_id}") | ||
| await bot.send_message( | ||
| chat_id=message.chat_id, | ||
| text="Request timed out.", | ||
| message_thread_id=message.message_thread_id, | ||
| ) | ||
| raise # Re-raise to trigger SQS retry for transient errors | ||
|
|
||
| except Exception as e: | ||
| logger.exception(f"Agent Server error for chat_id={message.chat_id}") | ||
| error_text = f"Error: {str(e)[:200]}" | ||
| try: | ||
| await bot.send_message( | ||
| chat_id=message.chat_id, | ||
| text=error_text, | ||
| message_thread_id=message.message_thread_id, | ||
| ) | ||
| except Exception as send_error: | ||
| logger.error(f"Failed to send error message to Telegram: {send_error}") | ||
| # Don't re-raise - error message already sent to user, retrying would cause duplicate messages | ||
|
|
||
| # Format response (result is guaranteed to be defined now) | ||
| if result.get('is_error'): | ||
| text = f"Agent error: {result.get('error_message', 'Unknown')}" | ||
| else: | ||
| text = result.get('response') or 'No response' | ||
|
|
||
| if len(text) > 4000: | ||
| text = text[:4000] + "\n\n... (truncated)" | ||
|
|
||
| # Send response to Telegram | ||
| try: | ||
| await bot.send_message( | ||
| chat_id=message.chat_id, | ||
| text=text, | ||
| parse_mode=ParseMode.MARKDOWN_V2, | ||
| message_thread_id=message.message_thread_id, | ||
| reply_to_message_id=message.message_id, | ||
| ) | ||
| except BadRequest as e: | ||
| if "parse entities" in str(e).lower(): | ||
| safe_text = escape_markdown(text, version=2) | ||
| await bot.send_message( | ||
| chat_id=message.chat_id, | ||
| text=safe_text, | ||
| parse_mode=ParseMode.MARKDOWN_V2, | ||
| message_thread_id=message.message_thread_id, | ||
| reply_to_message_id=message.message_id, | ||
| ) | ||
| else: | ||
| raise | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.