From 293591e48af2f0c194f8661c76aa7cc342fe47d6 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sun, 26 Jul 2026 21:39:59 +1000 Subject: [PATCH] feat: improve Telegram conversation heuristics (#14) Co-authored-by: openhands --- steward/bot/telegram.py | 208 ++++++++++++++++++++++++++++++++++++++-- tests/test_bot.py | 57 +++++++++++ 2 files changed, 258 insertions(+), 7 deletions(-) diff --git a/steward/bot/telegram.py b/steward/bot/telegram.py index 0565dff..6d2e837 100644 --- a/steward/bot/telegram.py +++ b/steward/bot/telegram.py @@ -1,7 +1,9 @@ """Telegram bot interface for Steward.""" import logging +import re from collections import defaultdict +from dataclasses import dataclass, field from typing import Any from telegram import Update @@ -55,6 +57,59 @@ _KB_CONTEXT_HEADER = ( "but do not repeat their contents unless directly asked." ) +_CONVERSATION_SYSTEM_APPENDIX = """ +Telegram conversation guidance: +- Reply like a thoughtful software engineer in a chat, not like a one-shot FAQ bot. +- Ask a brief follow-up question when the requested outcome, constraints, or preferred + option is unclear. +- You may send multiple short Telegram messages when that makes the conversation easier to follow. +- To send multiple messages, separate each message with a line containing only [MESSAGE]. +- When the user needs to choose from 2-10 concise options, use Telegram's native poll format: +[POLL] +question: The decision to make? +- First option +- Second option +[/POLL] +- Use polls only for discrete choices. Use a normal clarifying question for open-ended input. +""".strip() + +_MESSAGE_MARKER_RE = re.compile(r"^\s*\[MESSAGE\]\s*$", re.IGNORECASE) +_POLL_START_RE = re.compile(r"^\s*\[POLL\]\s*$", re.IGNORECASE) +_POLL_END_RE = re.compile(r"^\s*\[/POLL\]\s*$", re.IGNORECASE) +_OPTION_RE = re.compile(r"^\s*(?:[-*•]|\d+[.)]|option\s*:?)\s*(.+?)\s*$", re.IGNORECASE) +_QUESTION_RE = re.compile(r"^\s*question\s*:\s*(.+?)\s*$", re.IGNORECASE) +_MAX_POLL_OPTIONS = 10 +_MAX_POLL_QUESTION_LENGTH = 300 +_MAX_POLL_OPTION_LENGTH = 100 + + +@dataclass(frozen=True) +class PollRequest: + """A Telegram poll requested by the assistant response.""" + + question: str + options: list[str] + + +TelegramResponseAction = str | PollRequest + + +@dataclass(frozen=True) +class TelegramResponsePlan: + """Telegram actions derived from a single assistant response.""" + + actions: list[TelegramResponseAction] = field(default_factory=list) + + @property + def messages(self) -> list[str]: + """Text messages in this response plan.""" + return [action for action in self.actions if isinstance(action, str)] + + @property + def polls(self) -> list[PollRequest]: + """Polls in this response plan.""" + return [action for action in self.actions if isinstance(action, PollRequest)] + def _thread_key(update: Update) -> tuple[int, int] | None: """Return the (chat_id, thread_id) key if the message is part of a thread, else None.""" @@ -105,6 +160,128 @@ def _is_chat_enabled(chat: Any, settings: Settings) -> bool: return allowed +def _telegram_system_prompt(settings: Settings) -> str: + """Return the conversational Telegram system prompt for normal chat turns.""" + base_prompt = settings.openai_system_prompt.strip() + if not base_prompt: + return _CONVERSATION_SYSTEM_APPENDIX + return f"{base_prompt}\n\n{_CONVERSATION_SYSTEM_APPENDIX}" + + +def _append_message_action(actions: list[TelegramResponseAction], lines: list[str]) -> None: + message = "\n".join(lines).strip() + if message: + actions.append(message) + lines.clear() + + +def _compact_telegram_text(text: str, max_length: int) -> str: + return re.sub(r"\s+", " ", text).strip()[:max_length] + + +def _parse_poll(lines: list[str]) -> PollRequest | None: + question = "" + options: list[str] = [] + + for line in lines: + stripped = line.strip() + if not stripped: + continue + + question_match = _QUESTION_RE.match(stripped) + if question_match and not question: + question = question_match.group(1) + continue + + option_match = _OPTION_RE.match(stripped) + if option_match: + options.append(option_match.group(1)) + continue + + if not question: + question = stripped + + question = _compact_telegram_text(question, _MAX_POLL_QUESTION_LENGTH) + unique_options: list[str] = [] + for option in options: + compact_option = _compact_telegram_text(option, _MAX_POLL_OPTION_LENGTH) + if compact_option and compact_option not in unique_options: + unique_options.append(compact_option) + + unique_options = unique_options[:_MAX_POLL_OPTIONS] + if not question or len(unique_options) < 2: + return None + + return PollRequest(question=question, options=unique_options) + + +def _parse_telegram_response(reply: str) -> TelegramResponsePlan: + """Parse assistant response directives into ordered Telegram actions.""" + actions: list[TelegramResponseAction] = [] + message_lines: list[str] = [] + poll_lines: list[str] = [] + in_poll = False + + for line in reply.splitlines(): + if in_poll: + if _POLL_END_RE.match(line): + poll = _parse_poll(poll_lines) + if poll is None: + message_lines.extend(["[POLL]", *poll_lines, "[/POLL]"]) + else: + actions.append(poll) + poll_lines.clear() + in_poll = False + else: + poll_lines.append(line) + continue + + if _MESSAGE_MARKER_RE.match(line): + _append_message_action(actions, message_lines) + continue + + if _POLL_START_RE.match(line): + _append_message_action(actions, message_lines) + in_poll = True + continue + + message_lines.append(line) + + if in_poll: + message_lines.extend(["[POLL]", *poll_lines]) + _append_message_action(actions, message_lines) + + if not actions and reply.strip(): + actions.append(reply.strip()) + + return TelegramResponsePlan(actions=actions) + + +def _format_response_for_history(plan: TelegramResponsePlan) -> str: + """Render Telegram actions back into natural assistant history text.""" + history_parts: list[str] = [] + for action in plan.actions: + if isinstance(action, str): + history_parts.append(action) + else: + options = "; ".join(action.options) + history_parts.append(f"Poll: {action.question} ({options})") + return "\n\n".join(history_parts) + + +async def _send_telegram_response(update: Update, plan: TelegramResponsePlan) -> None: + """Send all Telegram actions for a parsed assistant response.""" + for action in plan.actions: + if isinstance(action, str): + await _send_long(update, action) + else: + await update.message.reply_poll( # type: ignore[union-attr] + question=action.question, + options=action.options, + is_anonymous=False, + ) + + async def _send_long(update: Update, text: str) -> None: """Send text, splitting across messages if it exceeds Telegram's 4096-char limit.""" limit = 4096 @@ -419,31 +596,48 @@ async def message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> if not text: return + system_prompt = _telegram_system_prompt(settings) key = _thread_key(update) if key is not None: # Thread message: unbounded history history: list[dict[str, Any]] = _thread_history[key] call_history = _with_kb_context(history, store.search(text)) if tool_client is not None: - reply = await llm.chat_with_tools(text, tool_client, history=call_history) + reply = await llm.chat_with_tools( + text, + tool_client, + history=call_history, + system_prompt=system_prompt, + ) else: - reply = await llm.chat(text, history=call_history) + reply = await llm.chat(text, history=call_history, system_prompt=system_prompt) + response_plan = _parse_telegram_response(reply) history.append({"role": "user", "content": text}) - history.append({"role": "assistant", "content": reply}) + history.append( + {"role": "assistant", "content": _format_response_for_history(response_plan)} + ) else: # Non-thread message: capped history per user user_history: list[dict[str, Any]] = _history[user.id] call_history = _with_kb_context(user_history, store.search(text)) if tool_client is not None: - reply = await llm.chat_with_tools(text, tool_client, history=call_history) + reply = await llm.chat_with_tools( + text, + tool_client, + history=call_history, + system_prompt=system_prompt, + ) else: - reply = await llm.chat(text, history=call_history) + reply = await llm.chat(text, history=call_history, system_prompt=system_prompt) + response_plan = _parse_telegram_response(reply) user_history.append({"role": "user", "content": text}) - user_history.append({"role": "assistant", "content": reply}) + user_history.append( + {"role": "assistant", "content": _format_response_for_history(response_plan)} + ) if len(user_history) > _MAX_HISTORY * 2: _history[user.id] = user_history[-(_MAX_HISTORY * 2) :] - await _send_long(update, reply) + await _send_telegram_response(update, response_plan) def build_application( diff --git a/tests/test_bot.py b/tests/test_bot.py index b081289..cac9efb 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -44,6 +44,7 @@ def _make_update( message = MagicMock(spec=Message) message.text = text message.reply_text = AsyncMock() + message.reply_poll = AsyncMock() message.message_thread_id = thread_id chat = MagicMock(spec=Chat) @@ -185,6 +186,62 @@ async def test_message_handler_calls_llm_and_replies(): assert _history[77][1]["role"] == "assistant" +@pytest.mark.asyncio +async def test_message_handler_sends_multiple_reply_messages(): + settings = _make_settings() + mock_llm = MagicMock(spec=LLMClient) + mock_llm.chat = AsyncMock( + return_value="[MESSAGE]\nFirst thought.\n[MESSAGE]\nWhat outcome do you want?" + ) + + update = _make_update(user_id=78, text="Can you help with this vague thing?") + ctx = _make_context(settings, llm=mock_llm) + + _history[78].clear() + await message_handler(update, ctx) + + assert update.message.reply_text.await_count == 2 + assert update.message.reply_text.await_args_list[0].args[0] == "First thought." + assert update.message.reply_text.await_args_list[1].args[0] == "What outcome do you want?" + update.message.reply_poll.assert_not_awaited() + assert _history[78][1]["content"] == "First thought.\n\nWhat outcome do you want?" + + +@pytest.mark.asyncio +async def test_message_handler_sends_native_poll_from_directive(): + settings = _make_settings() + mock_llm = MagicMock(spec=LLMClient) + mock_llm.chat = AsyncMock( + return_value=( + "[MESSAGE]\nI can turn that into a vote.\n" + "[POLL]\n" + "question: Which implementation should we choose?\n" + "- Minimal change\n" + "- Full redesign\n" + "[/POLL]\n" + "[MESSAGE]\nI'll use the winning option." + ) + ) + + update = _make_update(user_id=79, text="Should we do a minimal change or redesign?") + ctx = _make_context(settings, llm=mock_llm) + + _history[79].clear() + await message_handler(update, ctx) + + assert update.message.reply_text.await_count == 2 + update.message.reply_poll.assert_awaited_once_with( + question="Which implementation should we choose?", + options=["Minimal change", "Full redesign"], + is_anonymous=False, + ) + assert _history[79][1]["content"] == ( + "I can turn that into a vote.\n\n" + "Poll: Which implementation should we choose? (Minimal change; Full redesign)\n\n" + "I'll use the winning option." + ) + + @pytest.mark.asyncio async def test_message_handler_non_thread_trims_history(): """Non-thread history is trimmed to _MAX_HISTORY turns."""