如何在Telethon中持续显示1-2分钟的输入状态？

Telethon的client.action上下文管理器本应在上下文存续期间自动维持输入状态，但如果你的慢操作（比如OpenAI调用）阻塞了事件循环，就会导致无法自动续期，只能显示5秒。以下是两种可行的解决方法：

方法1：避免阻塞事件循环（推荐） #

如果你的OpenAI调用是同步函数，用asyncio.to_thread将其包装到后台线程运行，让事件循环能正常处理Telethon的动作续期：

import asyncio
from telethon import TelegramClient

async def handle_event(event):
    async with client.action(event.chat_id, 'typing'):
        # 将同步慢函数放到线程中执行
        result = await asyncio.to_thread(your_slow_openai_call, *args, **kwargs)
    await event.reply(result)

如果已经使用OpenAI的异步SDK（如openai.AsyncOpenAI），直接await异步调用即可：

async def handle_event(event):
    async with client.action(event.chat_id, 'typing'):
        # 直接await异步慢操作
        result = await your_async_openai_call(*args, **kwargs)
    await event.reply(result)

方法2：手动定时维持输入状态（备用） #

如果方法1无效，可以创建后台任务，每隔4秒（短于Telegram的5秒超时）发送一次输入状态，直到慢操作完成：

import asyncio
from telethon import TelegramClient

async def keep_typing(chat_id, stop_signal):
    while not stop_signal.is_set():
        await client.send_chat_action(chat_id, 'typing')
        await asyncio.sleep(4)

async def handle_event(event):
    stop_signal = asyncio.Event()
    # 启动后台维持输入状态的任务
    typing_task = asyncio.create_task(keep_typing(event.chat_id, stop_signal))
    
    try:
        # 执行慢操作
        result = await your_slow_function(*args, **kwargs)
    finally:
        # 操作完成后停止后台任务
        stop_signal.set()
        await typing_task
    
    await event.reply(result)

注意事项 #

• 禁止在异步函数中直接调用同步阻塞代码，否则会卡住事件循环，导致Telethon无法自动续期输入状态。
• 手动定时发送状态时，间隔不要过短，避免触发Telegram的频率限制。

内容的提问来源于stack exchange，提问作者rvnikita