armistace/steward_mirror
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: improve Telegram conversation heuristics (#14)

Browse Source 
Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
Daniel 2026-07-26 21:39:59 +10:00 committed by GitHub
parent 62e4b5e7fb
commit 293591e48a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 258 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

208
steward/bot/telegram.py
View File

@ -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(

57
tests/test_bot.py
View File

@ -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."""