如何使上下文链接到电报机器人中的每个用户?

Question

我需要不同的用户有不同的上下文，而不是一个共同的。所以每个用户都有不同的答案。以下是我的代码，其中所有用户都具有相同的上下文:

import openai
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor
import json
from datetime import datetime

file = open('config.json', 'r')
config = json.load(file)
openai.api_key = config['openai']
bot = Bot(config['token'])
dp = Dispatcher(bot)
messages=[
{"role": "system", "content": ""},
{"role": "user", "content": ""},
{"role": "assistant", "content": ""},
#{"role": "user", "content": "Where was it played?"}
]

def update(messages, role, content):
messages.append({"role": role, "content": content})
return messages

#personal messages
@dp.message_handler(content_types=["text"], chat_type=["private"])
async def echo(message : types.Message):
update(messages, "user", message.text)
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
temperature = 1,
#max_tokens=4096
presence_penalty = 1
)
await message.answer(response['choices'][0]['message']['content'])
# saving a message to a log file
user_id = message.from_user.id
user_name = message.from_user.username if message.from_user.username else message.from_user.first_name
message_text = message.text
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_data = {
"timestamp": timestamp,
"user_id": user_id,
"user_name": user_name,
"message": message_text
}
with open("updates.log", "a", encoding="utf-8") as f:
f.write(json.dumps(log_data, ensure_ascii=False) + "n")
print(log_data)
executor.start_polling(dp, skip_updates=True)

我尝试使用SQLite来保存数据库中的上下文。但这已经够糟糕了，我得到的最大的误解是:(此代码不工作!)

import openai
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor
import json
from datetime import datetime
import sqlite3
file = open('config.json', 'r')
config = json.load(file)
openai.api_key = config['openai']
bot = Bot(config['token'])
dp = Dispatcher(bot)

conn = sqlite3.connect('database.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
context TEXT
)
''')
conn.close()

def get_context(user_id):
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
cursor.execute('SELECT context FROM users WHERE user_id = ?', (user_id,))
row = cursor.fetchone()
conn.close()
if row:
return json.loads(row[0])
else:
return []
def save_context(user_id, context):
conn = sqlite3.connect('database.db')
cursor = conn.cursor()
cursor.execute('INSERT OR REPLACE INTO users (user_id, context) VALUES (?, ?)', (user_id, json.dumps(context)))
conn.commit()
conn.close()

@dp.message_handler(content_types=["text"], chat_type=["private"])
async def echo(message : types.Message):
user_id = message.from_user.id
context = get_context(user_id)
messages = [{"role": "system", "content": ""}] + context + [{"role": "user", "content": message.text}]
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=1,
# max_tokens=4096
presence_penalty=1
)
await message.answer(response.choices[0].text)
context = response.choices[0].context
save_context(user_id, context)

if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

Answer 1

我不确定您想如何处理这个问题，但是为了为每个用户提供不同的上下文，您可以分别存储每个用户的对话历史记录和上下文。你提到了sqlite，所以这就是我使用的(contexts.db):

import openai
import sqlite3
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils import executor
import json
from datetime import datetime

# Initialize database
conn = sqlite3.connect('contexts.db')
c = conn.cursor()
# Create table to store user contexts
c.execute('''CREATE TABLE IF NOT EXISTS contexts
(user_id text PRIMARY KEY, context text)''')
# Load OpenAI API key and Telegram Bot token from config.json
file = open('config.json', 'r')
config = json.load(file)
openai.api_key = config['openai']
bot = Bot(config['token'])
dp = Dispatcher(bot)

# Get context for user from database
def get_context(user_id):
c.execute("SELECT context FROM contexts WHERE user_id=?", (user_id,))
row = c.fetchone()
if row:
return json.loads(row[0])
else:
return []

# Save context for user to database
def save_context(user_id, context):
c.execute("REPLACE INTO contexts (user_id, context) VALUES (?, ?)",
(user_id, json.dumps(context)))
conn.commit()

# Update conversation history and context
def update(context, messages, role, content):
messages.append({"role": role, "content": content})
context["messages"] = messages
return context

# Handler for personal messages
@dp.message_handler(content_types=["text"], chat_type=["private"])
async def echo(message: types.Message):
# Get user ID and context
user_id = message.from_user.id
context = get_context(user_id)
# Update conversation history and context
messages = context.get("messages", [])
messages = update(context, messages, "user", message.text)
# Get response from OpenAI API
response = openai.Completion.create(
engine="davinci",
prompt="n".join([m["content"] for m in messages]),
temperature=0.5,
max_tokens=1024,
top_p=1,
frequency_penalty=0,
presence_penalty=0,
stop=["n"]
)
# Update conversation history and context with response
messages = update(context, messages, "assistant",
response.choices[0].text.strip())
# Save context to database
save_context(user_id, context)
# Send response to user
await message.answer(response.choices[0].text.strip())
# Log message
user_name = message.from_user.username if message.from_user.username else message.from_user.first_name
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_data = {
"timestamp": timestamp,
"user_id": user_id,
"user_name": user_name,
"message": message.text,
"response": response.choices[0].text.strip()
}
with open("updates.log", "a", encoding="utf-8") as f:
f.write(json.dumps(log_data, ensure_ascii=False) + "n")

executor.start_polling(dp, skip_updates=True)