diff --git a/handlers/knowledge.py b/handlers/knowledge.py index cb14227..24afa18 100644 --- a/handlers/knowledge.py +++ b/handlers/knowledge.py @@ -1,6 +1,7 @@ """知识库问答 handler""" import asyncio import logging +import re from telegram import Update from telegram.ext import CommandHandler, ContextTypes, filters @@ -10,39 +11,94 @@ import config logger = logging.getLogger("spam_guard") -def split_answer(text: str, max_len: int = 3500, max_parts: int = 3) -> list[str]: - """Telegram 单条消息长度有限，长回答按段发送，不再硬截断。""" +def looks_like_markdown(text: str) -> bool: + return bool(re.search(r"(^|\n)(#{1,3}\s|[-*]\s|\d+\.\s|```)|\*\*[^*]+\*\*|\*[^*]+\*|`[^`]+`", text)) + + +def optimize_plain_answer(text: str) -> str: + """非 Markdown 回答也做一点可读性整理。""" + text = text.strip() + if looks_like_markdown(text): + return text + lines = [line.strip() for line in text.splitlines() if line.strip()] + if len(lines) <= 1: + return f"*回答*\n{text}" + return "*回答*\n" + "\n".join(f"- {line}" if not line.startswith(("-", "•", "✅", "⚠️", "📚")) else line for line in lines) + + +def telegram_markdown(text: str) -> str: + """把常见 Markdown 调整为 Telegram legacy Markdown 更容易接受的形式。""" + text = re.sub(r"\*\*([^*]+)\*\*", r"*\1*", text) + text = re.sub(r"^#{1,3}\s+(.+)$", r"*\1*", text, flags=re.MULTILINE) + return text + + +def compact_answer(text: str, max_len: int = 3800) -> str: + """尽量压到一条 Telegram 消息；超长时保留开头和关键尾部。""" + text = optimize_plain_answer(text) + if len(text) <= max_len: + return text + tail_markers = ["**文档**", "📚", "相关文档", "文档"] + tail = "" + for marker in tail_markers: + idx = text.rfind(marker) + if idx > max_len // 2: + tail = text[idx:].strip() + break + head_len = max_len - len(tail) - 80 if tail else max_len - 40 + head = text[:max(1000, head_len)].rstrip() + return head + "\n\n...（内容较长已压缩，可追问具体步骤）" + (("\n\n" + tail) if tail else "") + + +def split_answer(text: str, max_len: int = 3900, max_parts: int = 2) -> list[str]: + """优先单条；实在超长最多拆 2 条。""" if len(text) <= max_len: return [text] - parts = [] rest = text while rest and len(parts) < max_parts: - if len(rest) <= max_len: - parts.append(rest) - break - cut = rest.rfind("\n\n", 0, max_len) - if cut < max_len // 2: - cut = rest.rfind("\n", 0, max_len) - if cut < max_len // 2: - cut = max_len + cut = min(len(rest), max_len) + if cut < len(rest): + better = rest.rfind("\n\n", 0, cut) + if better > max_len // 2: + cut = better parts.append(rest[:cut].strip()) rest = rest[cut:].strip() - if rest and parts: - parts[-1] += "\n\n...（回答较长，后续内容已省略；可以追问具体部分）" + parts[-1] += "\n\n...（后续内容已省略，可继续追问）" return parts -async def safe_send(update: Update, context: ContextTypes.DEFAULT_TYPE, text: str): - """优先回复原消息；若原消息已被自动删除，退回普通 send_message。""" +async def safe_send(update: Update, context: ContextTypes.DEFAULT_TYPE, text: str, markdown: bool | None = None): + """优先回复原消息；Markdown 失败或原消息被删时自动降级。""" chat_id = update.effective_chat.id + use_markdown = looks_like_markdown(text) if markdown is None else markdown + send_text = telegram_markdown(text) if use_markdown else text + + async def _send_plain(): + try: + return await update.message.reply_text(text) + except Exception: + return await context.bot.send_message(chat_id=chat_id, text=text) + + if use_markdown: + try: + return await update.message.reply_text(send_text, parse_mode="Markdown", disable_web_page_preview=True) + except Exception as e: + if "Message to be replied not found" in str(e): + try: + return await context.bot.send_message(chat_id=chat_id, text=send_text, parse_mode="Markdown", disable_web_page_preview=True) + except Exception: + return await context.bot.send_message(chat_id=chat_id, text=text) + logger.warning(f"Markdown 发送失败，降级普通文本: {e}") + return await _send_plain() + try: - return await update.message.reply_text(text) + return await update.message.reply_text(text, disable_web_page_preview=True) except Exception as e: if "Message to be replied not found" not in str(e): logger.warning(f"reply_text 失败，改用 send_message: {e}") - return await context.bot.send_message(chat_id=chat_id, text=text) + return await context.bot.send_message(chat_id=chat_id, text=text, disable_web_page_preview=True) async def cmd_ask(update: Update, context: ContextTypes.DEFAULT_TYPE): @@ -65,12 +121,12 @@ async def cmd_ask(update: Update, context: ContextTypes.DEFAULT_TYPE): try: from services.knowledge_base import ask_knowledge_base - answer = await ask_knowledge_base(query) + answer = compact_answer(await ask_knowledge_base(query)) replies = [] parts = split_answer(answer) for i, part in enumerate(parts, 1): - prefix = f"📚 回答 {i}/{len(parts)}\n\n" if len(parts) > 1 else "" - reply = await safe_send(update, context, prefix + part) + prefix = f"*回答 {i}/{len(parts)}*\n\n" if len(parts) > 1 else "" + reply = await safe_send(update, context, prefix + part, markdown=looks_like_markdown(prefix + part)) replies.append(reply) for reply in replies: asyncio.create_task(auto_delete(reply, seconds=180)) diff --git a/services/knowledge_base.py b/services/knowledge_base.py index ae17951..8ed6027 100644 --- a/services/knowledge_base.py +++ b/services/knowledge_base.py @@ -14,7 +14,7 @@ from knowledge import ( logger = logging.getLogger("spam_guard") -KNOWLEDGE_PROMPT = """你是收割机（go-harvest）文档助手。请根据知识库内容，给用户一个完整、可执行、细节充分的中文回答。 +KNOWLEDGE_PROMPT = """你是收割机（go-harvest）文档助手。请根据知识库内容，给用户一个准确、紧凑、可执行的中文回答。 相关文档内容： {context_text} @@ -28,19 +28,18 @@ KNOWLEDGE_PROMPT = """你是收割机（go-harvest）文档助手。请根据知 4. 对 Go Harvest 安装配置，若资料中出现 `newptools/go-harvest:latest`、`EMAIL`、`TOKEN`、`GO_WEB_PORT`、`POSTGRES_DB` 等，请保持这些原名，不要改成其他名字。 5. 不要把常识当成文档内容。可以说“文档没有说明”，不能猜。 -回答质量要求： -1. 不要过度简略。优先给出完整操作步骤、配置位置、示例、注意事项和排错建议。 -2. 如果问题是“怎么做/如何配置/怎么安装/报错怎么办”，请按以下结构回答： - ✅ 简短结论 - 🧭 操作步骤 - ⚙️ 配置示例/关键参数（仅限资料中出现的内容） - ⚠️ 注意事项 - 🧪 验证方式/排错方向 - 📚 相关文档 -3. 如果知识库信息不足，要明确说“文档里没有直接说明”，再基于已有内容给出可确认的部分。 -4. 回答长度建议 800~2500 字；复杂问题可以更长，但避免废话。 -5. 使用清晰排版，可以使用 emoji、小标题、编号、代码块。 -6. 最后列出相关文档链接，最多 5 个。""" +回答格式要求： +1. 尽量控制在 Telegram 单条消息内，优先 1200 字以内，最多不要超过 3000 字。 +2. 使用 Telegram 兼容 Markdown 排版：标题用 *粗体*，列表用 - 或 1.，命令/变量用反引号，配置示例用代码块。 +3. 如果问题是“怎么做/如何配置/怎么安装/报错怎么办”，请使用紧凑结构： + *结论* + *步骤* + *关键配置/命令*（仅限资料中出现的内容） + *注意* + *文档* +4. 不要展开无关背景，不要重复用户问题，不要写长篇解释。 +5. 如果知识库信息不足，要明确说“文档里没有直接说明”，只列可确认部分。 +6. 最后列出相关文档链接，最多 3 个。""" def build_context(docs: list[dict], max_chars: int = 12000) -> str: