Telethon实现Telegram账号ChatGPT机器人的输入状态显示问题

问题描述 #

我编写了一段将ChatGPT集成到Telegram普通账号的Python代码，希望通过async with client.action(event.chat_id, 'typing'): time.sleep(delay)模拟“正在输入”状态，但代码无报错却无法显示该状态，请求协助解决。

代码如下：

# Import necessary libraries
import telethon
from telethon import TelegramClient, events
from tinydb import TinyDB, Query
from telethon.tl.types import SendMessageTypingAction
import random
import time
import openai

# Set your OpenAI API key
openai.api_key = "key"

# Initialize the Telegram client with your API ID and hash
api_id = "id"  # Replace with your API ID
api_hash = "hash"  # Replace with your API hash
client = TelegramClient('bot', api_id, api_hash)

# Initialize a TinyDB database to store user conversations
db = TinyDB('db.json')
Telegram = Query()

# Read the chat bot prompt from a file
with open('path to prompt', 'r') as f:
    chat_bot_prompt = f.read()

# Register an event handler for incoming messages
@client.on(events.NewMessage)
async def handle_new_message(event):
    before_time = time.time()
    # Get the username and message from the event
    username = event.message.sender_id
    message = event.message.text

    # Check if the user is new or has an existing conversation
    if not (db.search(Telegram.username == username)):
        # If the user is new, start a new conversation
        conversation : list[str] = []
        conversation.append(message)
        
        # Generate a response using the GPT-3.5 Turbo model
        response = openai.ChatCompletion.create(
            model='gpt-3.5-turbo',
            messages=[{'role': 'system', 'content': chat_bot_prompt}, {'role': 'user', 'content': message}],
            temperature=0
        )
        reply = response['choices'][0]['message']['content']
        conversation.append(reply)
        
        # Store the conversation in the database
        db.insert({'username': username, 'conversation': conversation})
    else:
        # If the user has an existing conversation, continue it
        user = db.search(Telegram.username == username)
        conversation = user[0]['conversation']
        conversation.append(message)
        messages : list = [{'role': 'system', 'content': chat_bot_prompt}]
        
        # Prepare the conversation history for GPT-3.5 Turbo
        for i in range(len(conversation)):
            if (i % 2 == 0):
                messages.append({'role': 'user', 'content': conversation[i]})
            else:
                messages.append({'role': 'assistant', 'content': conversation[i]})
        
        # Generate a response using the GPT-3.5 Turbo model
        response = openai.ChatCompletion.create(
            model='gpt-3.5-turbo',
            messages=messages,
            temperature=0
        )
        reply = response['choices'][0]['message']['content']
        conversation.append(reply)
        
        # Update the conversation in the database
        db.update({'conversation': conversation}, Telegram.username == username)
    
    print(f"Received a message from {username}: {message}, Replied with: {reply}")

    # Calculate the delay based on the length of the reply
    delay = len(reply) / 6 - (time.time() -  before_time)

    if (delay < 0):
        delay = 0

    # Simulate typing by setting the 'typing' action
    async with client.action(event.chat_id, 'typing'):
        time.sleep(delay)

    # Send the reply with the calculated delay
    await event.respond(reply)

# Start the Telegram client
client.start()

# Run the client until Ctrl+C is pressed
client.run_until_disconnected()

解决方案 #

问题根源 #

1. 同步阻塞破坏异步逻辑：time.sleep(delay)是同步函数，在异步事件循环中调用会直接阻塞整个程序，导致Telegram客户端无法与服务器通信维持“正在输入”的状态心跳。
2. 输入状态时机错误：原代码在生成完ChatGPT回复后才触发输入状态，此时用户已经等待了回复生成的时间，输入状态失去实际交互意义。

修复步骤 #

1. 导入asyncio模块，用异步的asyncio.sleep()替代同步的time.sleep()，避免阻塞事件循环。
2. 调整输入状态的触发时机：将输入状态上下文包裹住整个ChatGPT回复生成过程，让输入状态覆盖真实的思考时长。
3. 优化延迟处理：仅在回复生成时间短于预期输入时长时，才补充异步延迟。

修改后的代码 #

# Import necessary libraries
import telethon
from telethon import TelegramClient, events
from tinydb import TinyDB, Query
import asyncio  # 新增异步模块
import random
import time
import openai

# Set your OpenAI API key
openai.api_key = "key"

# Initialize the Telegram client with your API ID and hash
api_id = "id"  # Replace with your API ID
api_hash = "hash"  # Replace with your API hash
client = TelegramClient('bot', api_id, api_hash)

# Initialize a TinyDB database to store user conversations
db = TinyDB('db.json')
Telegram = Query()

# Read the chat bot prompt from a file
with open('path to prompt', 'r') as f:
    chat_bot_prompt = f.read()

# Register an event handler for incoming messages
@client.on(events.NewMessage)
async def handle_new_message(event):
    # Get the username and message from the event
    username = event.message.sender_id
    message = event.message.text
    reply = ""
    conversation = []

    # 启动输入状态，覆盖整个回复生成过程
    async with client.action(event.chat_id, 'typing'):
        before_time = time.time()
        
        # Check if the user is new or has an existing conversation
        if not (db.search(Telegram.username == username)):
            # If the user is new, start a new conversation
            conversation.append(message)
            
            # Generate a response using the GPT-3.5 Turbo model
            response = openai.ChatCompletion.create(
                model='gpt-3.5-turbo',
                messages=[{'role': 'system', 'content': chat_bot_prompt}, {'role': 'user', 'content': message}],
                temperature=0
            )
            reply = response['choices'][0]['message']['content']
            conversation.append(reply)
            
            # Store the conversation in the database
            db.insert({'username': username, 'conversation': conversation})
        else:
            # If the user has an existing conversation, continue it
            user = db.search(Telegram.username == username)
            conversation = user[0]['conversation']
            conversation.append(message)
            messages : list = [{'role': 'system', 'content': chat_bot_prompt}]
            
            # Prepare the conversation history for GPT-3.5 Turbo
            for i in range(len(conversation)):
                if (i % 2 == 0):
                    messages.append({'role': 'user', 'content': conversation[i]})
                else:
                    messages.append({'role': 'assistant', 'content': conversation[i]})
            
            # Generate a response using the GPT-3.5 Turbo model
            response = openai.ChatCompletion.create(
                model='gpt-3.5-turbo',
                messages=messages,
                temperature=0
            )
            reply = response['choices'][0]['message']['content']
            conversation.append(reply)
            
            # Update the conversation in the database
            db.update({'conversation': conversation}, Telegram.username == username)
        
        # Calculate the delay based on the length of the reply
        delay = len(reply) / 6 - (time.time() - before_time)
        if delay > 0:
            await asyncio.sleep(delay)  # 使用异步sleep替代同步sleep
    
    print(f"Received a message from {username}: {message}, Replied with: {reply}")
    # Send the reply
    await event.respond(reply)

# Start the Telegram client
client.start()

# Run the client until Ctrl+C is pressed
client.run_until_disconnected()

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