如何在Telegram Bot中流式输出Gemini（Google生成式AI）的响应并避免Markdown解析错误？

你遇到的问题我之前做类似Bot时也碰到过——核心原因就是Telegram的Markdown解析器对格式完整性要求极其严格：当你流式更新消息时，如果当前累积的文本里存在未闭合的Markdown实体（比如只有开头的*却没结尾的*、未闭合的代码块标记```，或者半截的链接[xxx），Telegram服务器直接就会返回解析错误。

下面给你几个可行的解决方案，从简单到进阶都有，你可以根据需求选：

方案1：流式更新用纯文本，最终版本再应用Markdown #

这是最省心的实现方式，既能保留流式更新的实时感，又能彻底避免中间的解析错误：

• 流式返回chunk的过程中，编辑消息时不启用任何格式解析（显式设为ParseMode.NONE），以纯文本展示累积内容；
• 等所有chunk接收完成后，最后一次编辑消息时再开启ParseMode.MARKDOWN，展示完整的带格式内容。

修改你的handle_prompt函数即可：

@dp.message()
async def handle_prompt(message: types.Message):
    async with AsyncSession(engine) as db:
        user, __ = await get_or_create_telegram_user(message, db)
        prompt = message.text

        temp_message_text = await _("timer_1", user.language_id, db)
        temp_message = await message.answer(temp_message_text)

        accumulated_text = ""

        try:
            async for chunk in generate_response(prompt):
                accumulated_text += chunk
                # 流式阶段用纯文本，跳过格式解析
                await temp_message.edit_text(
                    accumulated_text,
                    disable_web_page_preview=True,
                    parse_mode=ParseMode.NONE
                )
            # 所有内容加载完成后，最后一次编辑应用Markdown格式
            await temp_message.edit_text(
                accumulated_text,
                disable_web_page_preview=True,
                parse_mode=ParseMode.MARKDOWN
            )
        except Exception as e:
            print(f"Error in handle_prompt: {e}")
            await temp_message.edit_text("An error occurred during processing.")
        finally:
            await refresh_last_interaction_date(user.chat_id, db)

方案2：实时修复未闭合的Markdown实体（进阶） #

如果你希望流式更新的过程中也能展示合法的Markdown格式，可以在每次累积文本后，检查并修复未闭合的格式标记。我自己写过一个简单的修复函数，你可以参考：

def fix_incomplete_markdown(text: str) -> str:
    # 处理斜体/粗体的*标记：奇数个则转义最后一个
    star_count = text.count("*")
    if star_count % 2 != 0:
        last_star_idx = text.rfind("*")
        if last_star_idx != -1:
            text = text[:last_star_idx] + r"\*" + text[last_star_idx+1:]
    
    # 处理行内代码的`标记：同样转义未闭合的最后一个
    backtick_count = text.count("`")
    if backtick_count % 2 != 0:
        last_backtick_idx = text.rfind("`")
        if last_backtick_idx != -1:
            text = text[:last_backtick_idx] + r"\`" + text[last_backtick_idx+1:]
    
    # 处理代码块的```标记：未闭合则临时补全
    code_block_count = text.count("```")
    if code_block_count % 2 != 0:
        text += "```"
    
    # 链接、下划线等其他标记可以按需扩展修复逻辑
    return text

然后在流式更新时调用这个函数：

async for chunk in generate_response(prompt):
    accumulated_text += chunk
    # 修复后再编辑消息，注意用MARKDOWN_V2匹配转义规则
    safe_text = fix_incomplete_markdown(accumulated_text)
    await temp_message.edit_text(
        safe_text,
        disable_web_page_preview=True,
        parse_mode=ParseMode.MARKDOWN_V2
    )

方案3：切换为HTML格式解析 #

Telegram的HTML解析容错性比Markdown高一些，部分未闭合的标签不会直接报错。你可以尝试让Gemini返回HTML格式的内容，再用ParseMode.HTML解析：

先修改generate_response函数，指定输出格式：

async def generate_response(prompt: str, **kwargs) -> AsyncGenerator[str, None]:
    try:
        # 让Gemini生成HTML格式的响应
        response = sync_to_async_iterable(model.generate_content(
            prompt, 
            stream=True,
            generation_config={"response_mime_type": "text/html"},
            **kwargs
        ))
        async for content in response:
            yield content.text
    except Exception as e:
        print(f"Error in generate_response: {e}")
        yield "An error occurred during processing."

然后在编辑消息时切换解析模式：

await temp_message.edit_text(
    accumulated_text,
    disable_web_page_preview=True,
    parse_mode=ParseMode.HTML
)

额外小技巧 #

如果担心还是会有偶发的解析错误，可以在编辑消息时捕获TelegramBadRequest，自动回退到纯文本模式：

try:
    await temp_message.edit_text(safe_text, parse_mode=ParseMode.MARKDOWN_V2)
except aiogram.exceptions.TelegramBadRequest:
    # 解析失败时用纯文本兜底
    await temp_message.edit_text(accumulated_text, parse_mode=ParseMode.NONE)

备注：内容来源于stack exchange，提问作者urban freestyler