From 41c79ac5c43ec08de4c0a6364d460864b60b27ae Mon Sep 17 00:00:00 2001 From: YooooEX Date: Tue, 28 Apr 2026 11:21:09 +0800 Subject: [PATCH 1/2] feat(tg): stream Telegram responses by turn with summaries --- frontends/tgapp.py | 241 ++++++++++++++++++++++++++++++++++++++------- 1 file changed, 204 insertions(+), 37 deletions(-) diff --git a/frontends/tgapp.py b/frontends/tgapp.py index 3d8fb30c..e4f2e657 100644 --- a/frontends/tgapp.py +++ b/frontends/tgapp.py @@ -42,16 +42,27 @@ _ASK_CANCEL_LABEL = "none of these above" _ASK_CANCEL_PROMPT = "已取消选择，请直接发送下一步操作。" _ask_menu_events = Q.Queue() _ask_menu_store = {} +_QUOTE_OPEN_TAG = "<_quote_>" +_QUOTE_CLOSE_TAG = "" +_QUOTE_TOKEN_PATTERN = re.escape(_QUOTE_OPEN_TAG) + r"([\s\S]*?)" + re.escape(_QUOTE_CLOSE_TAG) _MD_TOKEN_RE = re.compile( - r"(`{3,})([A-Za-z0-9_+-]*)\n([\s\S]*?)\1" - r"|\[([^\]]+)\]\(([^)\n]+)\)" - r"|`([^`\n]+)`" - r"|\*\*([^\n]+?)\*\*" - r"|__([^\n]+?)__" - r"|~~([^\n]+?)~~" - r"|(?\s*(.*?)\s*", re.DOTALL) +_TURN_SUMMARY_SEARCH_STRIP_RE = re.compile(r"`{3,}[\s\S]*?`{3,}|[\s\S]*?", re.DOTALL) def _make_draft_id(): return random.randint(1, 2**31 - 1) @@ -60,6 +71,50 @@ def _visible_segments(text): text = (text or "").strip() return split_text(text, _STREAM_SEGMENT_LIMIT) if text else [] +def _line_complete(line): + return (line or "").endswith(("\n", "\r")) + +def _turn_marker_number(line): + match = _TURN_MARKER_RE.fullmatch((line or "").strip()) + return int(match.group(1)) if match else None + +def _maybe_partial_turn_marker(line): + text = (line or "").strip().lstrip("*") + if not text: + return False + marker_head = "LLM Running (Turn " + return marker_head.startswith(text) or text.startswith(marker_head) + +def _maybe_partial_code_fence(line): + return bool(re.match(r"^\s*`{1,}[^`\r\n]*$", line or "")) + +def _extract_turn_summary(raw_text): + search_text = _TURN_SUMMARY_SEARCH_STRIP_RE.sub("", raw_text or "") + match = _TURN_SUMMARY_RE.search(search_text) + if not match: + return "" + summary = re.sub(r"\s+", " ", match.group(1)).strip() + if len(summary) > _TURN_SUMMARY_LIMIT: + summary = summary[:_TURN_SUMMARY_LIMIT - 3].rstrip() + "..." + return summary + +def _quote_tag(text): + safe_text = (text or "").strip().replace(_QUOTE_OPEN_TAG, "").replace(_QUOTE_CLOSE_TAG, "") + return f"{_QUOTE_OPEN_TAG}{safe_text}{_QUOTE_CLOSE_TAG}" + +def _inject_turn_summary(body, summary): + if not (body or "").strip() or not (summary or "").strip(): + return body + lines = (body or "").splitlines() + if not lines or _turn_marker_number(lines[0]) is None: + return body + title = lines[0].strip() + rest = "\n".join(lines[1:]).strip() + summary_line = _quote_tag(summary) + if rest: + return f"{title}\n\n{summary_line}\n\n{rest}" + return f"{title}\n\n{summary_line}" + def _resolve_files(paths): files, seen = [], set() for fpath in paths: @@ -77,6 +132,28 @@ def _render_file_markers(text): return os.path.basename(match.group(1)) return re.sub(r"\[FILE:([^\]]+)\]", repl, text or "").strip() +def _files_from_text(text): + cleaned = clean_reply(text) if (text or "").strip() else "" + return _resolve_files(extract_files(cleaned)) + +async def _send_files(root_msg, files): + for fpath in files: + if fpath.lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".webp")): + try: + with open(fpath, "rb") as fp: + await root_msg.reply_photo(fp) + except Exception: + pass + else: + try: + with open(fpath, "rb") as fp: + await root_msg.reply_document(fp) + except Exception: + pass + +async def _send_files_from_text(root_msg, text): + await _send_files(root_msg, _files_from_text(text)) + def _escape_pre(text): return escape_markdown(text or "", version=2, entity_type="pre") @@ -86,6 +163,10 @@ def _escape_code(text): def _escape_link_target(text): return escape_markdown(text or "", version=2, entity_type="text_link") +def _quote_to_markdown_v2(text): + lines = (text or "").strip().splitlines() or [""] + return "\n".join(f"> {escape_markdown(line, version=2)}" for line in lines) + def _to_markdown_v2(text): if not text: return "" @@ -98,19 +179,21 @@ def _to_markdown_v2(text): header = f"```{lang}\n" if lang else "```\n" parts.append(f"{header}{code}\n```") elif match.group(4) is not None: - label = escape_markdown(match.group(4), version=2) - target = _escape_link_target(match.group(5)) + parts.append(_quote_to_markdown_v2(match.group(4))) + elif match.group(5) is not None: + label = escape_markdown(match.group(5), version=2) + target = _escape_link_target(match.group(6)) parts.append(f"[{label}]({target})") - elif match.group(6) is not None: - parts.append(f"`{_escape_code(match.group(6))}`") elif match.group(7) is not None: - parts.append(f"*{escape_markdown(match.group(7), version=2)}*") + parts.append(f"`{_escape_code(match.group(7))}`") elif match.group(8) is not None: parts.append(f"*{escape_markdown(match.group(8), version=2)}*") elif match.group(9) is not None: - parts.append(f"~{escape_markdown(match.group(9), version=2)}~") + parts.append(f"*{escape_markdown(match.group(9), version=2)}*") elif match.group(10) is not None: - parts.append(f"_{escape_markdown(match.group(10), version=2)}_") + parts.append(f"~{escape_markdown(match.group(10), version=2)}~") + elif match.group(11) is not None: + parts.append(f"_{escape_markdown(match.group(11), version=2)}_") pos = match.end() parts.append(escape_markdown(text[pos:], version=2)) return "".join(parts) @@ -251,7 +334,7 @@ class _TelegramStreamSession: def __init__(self, root_msg): self.root_msg = root_msg self.private_chat = getattr(getattr(root_msg, "chat", None), "type", "") == ChatType.PRIVATE - self.can_use_draft = True # update tg client! + self.can_use_draft = self.private_chat # update tg client! self.draft_id = _make_draft_id() self.live_msg = None self.raw_text = "" @@ -291,9 +374,10 @@ class _TelegramStreamSession: self.active_display = "" async def _refresh(self, done, send_files): + summary = _extract_turn_summary(self.raw_text) cleaned = clean_reply(self.raw_text) if self.raw_text.strip() else "" - self.files = _resolve_files(extract_files(cleaned)) - body = _render_file_markers(cleaned) + self.files = _files_from_text(cleaned) + body = _inject_turn_summary(_render_file_markers(cleaned), summary) if done and not body and self.files: body = "已生成附件" elif done and not body: @@ -334,17 +418,7 @@ class _TelegramStreamSession: self.draft_id = _make_draft_id() async def _send_files(self): - for fpath in self.files: - if fpath.lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".webp")): - try: - with open(fpath, "rb") as fp: - await self.root_msg.reply_photo(fp) - except Exception: pass - else: - try: - with open(fpath, "rb") as fp: - await self.root_msg.reply_document(fp) - except Exception: pass + await _send_files(self.root_msg, self.files) async def _send_draft(self, text): try: @@ -388,9 +462,98 @@ class _TelegramStreamSession: self.live_msg = await self._edit_text(self.live_msg, text) +class _TelegramTurnStreamCoordinator: + def __init__(self, root_msg): + self.root_msg = root_msg + self.session = None + self.pending_line = "" + self.code_fence_len = 0 + self.last_turn = 0 + + async def prime(self): + await self._ensure_session() + + async def add_chunk(self, chunk): + if not chunk: + return + text = self.pending_line + chunk + self.pending_line = "" + for line in text.splitlines(keepends=True): + if _line_complete(line): + await self._process_line(line) + elif _maybe_partial_turn_marker(line) or _maybe_partial_code_fence(line): + self.pending_line = line + else: + await self._process_line(line) + + async def finalize(self, done_text="", send_files=True): + await self._flush_pending_line() + if self.session is None: + if done_text: + await self._add_to_current(done_text) + elif not self.session.raw_text.strip() and done_text: + await self.session.finalize(done_text, send_files=False) + if send_files: + await _send_files_from_text(self.root_msg, done_text) + return + if self.session is not None: + await self.session.finalize(send_files=False) + if send_files: + await _send_files_from_text(self.root_msg, done_text) + + async def finish_with_notice(self, notice): + await self._flush_pending_line() + await self._ensure_session() + await self.session.finish_with_notice(notice) + + async def _ensure_session(self): + if self.session is None: + self.session = _TelegramStreamSession(self.root_msg) + await self.session.prime() + + async def _start_turn(self, marker): + if self.session is not None and self.session.raw_text.strip(): + await self.session.finalize(send_files=False) + self.session = None + await self._ensure_session() + await self.session.add_chunk(marker) + + async def _add_to_current(self, text): + if not text: + return + await self._ensure_session() + await self.session.add_chunk(text) + + async def _process_line(self, line): + turn_no = _turn_marker_number(line) + if self.code_fence_len == 0 and turn_no == self.last_turn + 1: + self.last_turn = turn_no + await self._start_turn(line) + return + await self._add_to_current(line) + self._update_code_fence(line) + + async def _flush_pending_line(self): + if not self.pending_line: + return + line = self.pending_line + self.pending_line = "" + await self._add_to_current(line) + + def _update_code_fence(self, line): + match = _CODE_FENCE_RE.match(line or "") + if not match: + return + fence_len = len(match.group(1)) + if self.code_fence_len: + if fence_len >= self.code_fence_len: + self.code_fence_len = 0 + return + self.code_fence_len = fence_len + async def _stream(dq, msg): - session = _TelegramStreamSession(msg) - await session.prime() + stream = _TelegramTurnStreamCoordinator(msg) + await stream.prime() try: while True: try: first = await asyncio.to_thread(dq.get, True, _QUEUE_WAIT_SECONDS) @@ -399,21 +562,25 @@ async def _stream(dq, msg): try: while True: items.append(dq.get_nowait()) except Q.Empty: pass - done_item = next((item for item in items if "done" in item), None) + done_item = None + for item in items: + chunk = item.get("next", "") + if chunk: + await stream.add_chunk(chunk) + if "done" in item: + done_item = item + break if done_item is not None: - await session.finalize(done_item.get("done", "")) + await stream.finalize(done_item.get("done", "")) event = _drain_latest_ask_user_event() if event: await _send_ask_user_menu(msg, event) break - chunk = "".join(item.get("next", "") for item in items if item.get("next")) - if chunk: - await session.add_chunk(chunk) except asyncio.CancelledError: - await session.finish_with_notice("⏹️ 已停止") + await stream.finish_with_notice("⏹️ 已停止") except Exception as exc: print(f"[TG stream error] {type(exc).__name__}: {exc}", flush=True) - await session.finish_with_notice(f"❌ 输出失败: {exc}") + await stream.finish_with_notice(f"❌ 输出失败: {exc}") def _normalized_command(text): parts = (text or "").strip().split(None, 1) From 5e1a788af1cd6e369044d4b68ddad1a8dc9ab86f Mon Sep 17 00:00:00 2001 From: YooooEX Date: Tue, 28 Apr 2026 18:34:36 +0800 Subject: [PATCH 2/2] fix(tgapp): handle Telegram stream rate limits --- frontends/tgapp.py | 202 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 183 insertions(+), 19 deletions(-) diff --git a/frontends/tgapp.py b/frontends/tgapp.py index e4f2e657..eae5d7f8 100644 --- a/frontends/tgapp.py +++ b/frontends/tgapp.py @@ -5,6 +5,7 @@ from agentmain import GeneraticAgent try: from telegram import BotCommand, InlineKeyboardButton, InlineKeyboardMarkup from telegram.constants import ChatType, MessageLimit, ParseMode + from telegram.error import RetryAfter from telegram.ext import ApplicationBuilder, CallbackQueryHandler, MessageHandler, filters, ContextTypes from telegram.helpers import escape_markdown from telegram.request import HTTPXRequest @@ -34,6 +35,9 @@ ALLOWED = set(mykeys.get('tg_allowed_users', [])) _DRAFT_HINT = "thinking..." _STREAM_SUFFIX = " ⏳" _STREAM_SEGMENT_LIMIT = max(1200, MessageLimit.MAX_TEXT_LENGTH - 256) +_STREAM_UPDATE_INTERVAL_SECONDS = 2.0 +_STREAM_MIN_UPDATE_CHARS = 400 +_RETRY_AFTER_MARGIN_SECONDS = 1.0 _QUEUE_WAIT_SECONDS = 1 _ASK_USER_HOOK_KEY = "telegram_ask_user_menu" _ASK_CALLBACK_PREFIX = "ask:" @@ -69,7 +73,41 @@ def _make_draft_id(): def _visible_segments(text): text = (text or "").strip() - return split_text(text, _STREAM_SEGMENT_LIMIT) if text else [] + if not text: + return [] + segments = [] + for part in split_text(text, _STREAM_SEGMENT_LIMIT): + segments.extend(_markdown_safe_segments(part)) + return segments + +def _markdown_safe_segments(text, limit=None): + limit = limit or MessageLimit.MAX_TEXT_LENGTH + text = (text or "").strip() + if not text: + return [] + if len(_to_markdown_v2(text)) <= limit: + return [text] + parts = [] + remaining = text + while remaining: + if len(_to_markdown_v2(remaining)) <= limit: + parts.append(remaining) + break + low, high, best = 1, len(remaining), 1 + while low <= high: + mid = (low + high) // 2 + if len(_to_markdown_v2(remaining[:mid].rstrip() or remaining[:mid])) <= limit: + best = mid + low = mid + 1 + else: + high = mid - 1 + cut = remaining.rfind("\n", 0, best) + if cut < max(1, best * 0.6): + cut = best + chunk = remaining[:cut].rstrip() or remaining[:best] + parts.append(chunk) + remaining = remaining[len(chunk):].lstrip() + return parts def _line_complete(line): return (line or "").endswith(("\n", "\r")) @@ -341,12 +379,77 @@ class _TelegramStreamSession: self.files = [] self.sent_segments = 0 self.active_display = "" + self.pending_display = "" + self.retry_until = 0.0 + self.last_update_at = 0.0 + self.last_update_raw_len = 0 + + def _now(self): + return time.monotonic() + + def _retry_after_seconds(self, exc): + retry_after = getattr(exc, "_retry_after", None) + if retry_after is None: + retry_after = getattr(exc, "retry_after", 0) or 0 + if hasattr(retry_after, "total_seconds"): + retry_after = retry_after.total_seconds() + try: + return max(0.0, float(retry_after)) + except (TypeError, ValueError): + return 0.0 + + def _set_retry_after(self, exc): + wait_seconds = self._retry_after_seconds(exc) + _RETRY_AFTER_MARGIN_SECONDS + self.retry_until = max(self.retry_until, self._now() + wait_seconds) + + def _is_retrying(self): + return self._now() < self.retry_until + + async def _wait_for_retry(self): + remaining = self.retry_until - self._now() + if remaining > 0: + await asyncio.sleep(remaining) + + def _should_stream_update(self, display): + if display == self.active_display: + return False + if self.last_update_at <= 0: + return True + elapsed = self._now() - self.last_update_at + raw_delta = len(self.raw_text) - self.last_update_raw_len + return elapsed >= _STREAM_UPDATE_INTERVAL_SECONDS or raw_delta >= _STREAM_MIN_UPDATE_CHARS + + def _mark_stream_update(self, display): + self.active_display = display + self.pending_display = "" + self.last_update_at = self._now() + self.last_update_raw_len = len(self.raw_text) + + def _stream_display(self, text): + base = (text or _DRAFT_HINT).strip() or _DRAFT_HINT + safe_parts = _markdown_safe_segments(base) + base = safe_parts[-1] if safe_parts else _DRAFT_HINT + if base == _DRAFT_HINT: + return base + display = base + _STREAM_SUFFIX + if len(_to_markdown_v2(display)) <= MessageLimit.MAX_TEXT_LENGTH: + return display + return base async def prime(self): - if self.can_use_draft and await self._send_draft(_DRAFT_HINT): + if self.can_use_draft: + draft_result = await self._send_draft(_DRAFT_HINT) + if draft_result is True: + self.active_display = _DRAFT_HINT + return + if draft_result is None: + self.active_display = _DRAFT_HINT + return + try: + await self._upsert_live_message(_DRAFT_HINT, wait_retry=False) + except RetryAfter: self.active_display = _DRAFT_HINT return - await self._upsert_live_message(_DRAFT_HINT) self.active_display = _DRAFT_HINT async def add_chunk(self, chunk): @@ -395,16 +498,24 @@ class _TelegramStreamSession: await self._stream_active(active_text) async def _stream_active(self, text): - display = (text or _DRAFT_HINT).strip() or _DRAFT_HINT - if display != _DRAFT_HINT: - display = display + _STREAM_SUFFIX + display = self._stream_display(text) if display == self.active_display: return - if self.can_use_draft and await self._send_draft(display): - self.active_display = display + self.pending_display = display + if self._is_retrying() or not self._should_stream_update(display): + return + try: + if self.can_use_draft: + draft_result = await self._send_draft(display) + if draft_result is True: + self._mark_stream_update(display) + return + if draft_result is None: + return + await self._upsert_live_message(display, wait_retry=False) + self._mark_stream_update(display) + except RetryAfter: return - await self._upsert_live_message(display) - self.active_display = display async def _finalize_segment(self, text): final_text = (text or "").strip() or "..." @@ -428,6 +539,9 @@ class _TelegramStreamSession: parse_mode=ParseMode.MARKDOWN_V2, ) return True + except RetryAfter as exc: + self._set_retry_after(exc) + return None except Exception as exc: if _is_not_modified_error(exc): return True @@ -436,30 +550,71 @@ class _TelegramStreamSession: self.draft_id = _make_draft_id() return False - async def _reply_text(self, text): + async def _retry_call(self, func, *args): + while True: + await self._wait_for_retry() + try: + return await func(*args) + except RetryAfter as exc: + self._set_retry_after(exc) + + async def _reply_text_once(self, text): markdown = _to_markdown_v2(text) try: return await self.root_msg.reply_text(markdown, parse_mode=ParseMode.MARKDOWN_V2) + except RetryAfter as exc: + self._set_retry_after(exc) + raise except Exception as exc: if _is_not_modified_error(exc): return None - return await self.root_msg.reply_text(text) + try: + return await self.root_msg.reply_text(text) + except RetryAfter as retry_exc: + self._set_retry_after(retry_exc) + raise - async def _edit_text(self, msg, text): + async def _reply_text(self, text, wait_retry=True): + last_msg = None + for segment in _markdown_safe_segments(text) or ["..."]: + if wait_retry: + last_msg = await self._retry_call(self._reply_text_once, segment) + else: + last_msg = await self._reply_text_once(segment) + return last_msg + + async def _edit_text_once(self, msg, text): markdown = _to_markdown_v2(text) try: updated = await msg.edit_text(markdown, parse_mode=ParseMode.MARKDOWN_V2) + except RetryAfter as exc: + self._set_retry_after(exc) + raise except Exception as exc: if _is_not_modified_error(exc): return msg - updated = await msg.edit_text(text) + try: + updated = await msg.edit_text(text) + except RetryAfter as retry_exc: + self._set_retry_after(retry_exc) + raise return updated if hasattr(updated, "edit_text") else msg - async def _upsert_live_message(self, text): - if self.live_msg is None: - self.live_msg = await self._reply_text(text) + async def _edit_text(self, msg, text, wait_retry=True): + segments = _markdown_safe_segments(text) or ["..."] + if wait_retry: + updated = await self._retry_call(self._edit_text_once, msg, segments[0]) else: - self.live_msg = await self._edit_text(self.live_msg, text) + updated = await self._edit_text_once(msg, segments[0]) + for segment in segments[1:]: + updated = await self._reply_text(segment, wait_retry=wait_retry) + return updated if hasattr(updated, "edit_text") else msg + + async def _upsert_live_message(self, text, wait_retry=True): + if self.live_msg is None: + self.live_msg = await self._reply_text(text, wait_retry=wait_retry) + else: + self.live_msg = await self._edit_text(self.live_msg, text, wait_retry=wait_retry) class _TelegramTurnStreamCoordinator: @@ -578,9 +733,18 @@ async def _stream(dq, msg): break except asyncio.CancelledError: await stream.finish_with_notice("⏹️ 已停止") + except RetryAfter as exc: + print(f"[TG stream retry_after] {type(exc).__name__}: {exc}", flush=True) + if stream.session is not None: + stream.session._set_retry_after(exc) except Exception as exc: print(f"[TG stream error] {type(exc).__name__}: {exc}", flush=True) - await stream.finish_with_notice(f"❌ 输出失败: {exc}") + if stream.session is not None and stream.session._is_retrying(): + return + try: + await stream.finish_with_notice(f"❌ 输出失败: {exc}") + except RetryAfter as retry_exc: + print(f"[TG stream error notice retry_after] {type(retry_exc).__name__}: {retry_exc}", flush=True) def _normalized_command(text): parts = (text or "").strip().split(None, 1)