import logging
import os
from datetime import datetime
from decimal import Decimal
from typing import List
import boto3
from boto3.dynamodb.conditions import Attr, Key
from datasets import Dataset
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
# Create a DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
def _create_arena_table():
dynamodb.create_table(
TableName='oaaic_chatbot_arena',
KeySchema=[
{
'AttributeName': 'arena_battle_id',
'KeyType': 'HASH'
},
],
AttributeDefinitions=[
{
'AttributeName': 'arena_battle_id',
'AttributeType': 'S'
},
{
'AttributeName': 'timestamp',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
},
GlobalSecondaryIndexes=[
{
'IndexName': 'TimestampIndex',
'KeySchema': [
{
'AttributeName': 'arena_battle_id',
'KeyType': 'HASH'
},
{
'AttributeName': 'timestamp',
'KeyType': 'RANGE'
},
],
'Projection': {
'ProjectionType': 'ALL',
},
'ProvisionedThroughput': {
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
}
},
]
)
def _create_elo_scores_table():
dynamodb.create_table(
TableName='elo_scores',
KeySchema=[
{
'AttributeName': 'chatbot_name',
'KeyType': 'HASH' # Partition key
},
],
AttributeDefinitions=[
{
'AttributeName': 'chatbot_name',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5
}
)
def _create_elo_logs_table():
dynamodb.create_table(
TableName='elo_logs',
KeySchema=[
{
'AttributeName': 'arena_battle_id',
'KeyType': 'HASH' # Partition key
},
{
'AttributeName': 'battle_timestamp',
'KeyType': 'RANGE' # Sort key
},
],
AttributeDefinitions=[
{
'AttributeName': 'arena_battle_id',
'AttributeType': 'S'
},
{
'AttributeName': 'battle_timestamp',
'AttributeType': 'S'
},
{
'AttributeName': 'all',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
},
GlobalSecondaryIndexes=[
{
'IndexName': 'AllTimestampIndex',
'KeySchema': [
{
'AttributeName': 'all',
'KeyType': 'HASH' # Partition key for the GSI
},
{
'AttributeName': 'battle_timestamp',
'KeyType': 'RANGE' # Sort key for the GSI
}
],
'Projection': {
'ProjectionType': 'ALL'
},
'ProvisionedThroughput': {
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
},
]
)
def get_unprocessed_battles(last_processed_timestamp):
# Use boto3 to create a DynamoDB resource and reference the table
table = dynamodb.Table('oaaic_chatbot_arena')
# Use a query to retrieve unprocessed battles in temporal order
response = table.scan(
FilterExpression=Attr('timestamp').gt(last_processed_timestamp),
# ScanIndexForward=True
)
return response['Items']
def calculate_elo(rating1, rating2, result, K=32):
# Convert ratings to float
rating1 = float(rating1)
rating2 = float(rating2)
# Calculate the expected outcomes
expected_outcome1 = 1.0 / (1.0 + 10.0 ** ((rating2 - rating1) / 400.0))
expected_outcome2 = 1.0 - expected_outcome1
# Calculate the new Elo ratings
new_rating1 = rating1 + K * (result - expected_outcome1)
new_rating2 = rating2 + K * ((1.0 - result) - expected_outcome2)
return Decimal(new_rating1).quantize(Decimal('0.00')), Decimal(new_rating2).quantize(Decimal('0.00'))
def get_last_processed_timestamp():
table = dynamodb.Table('elo_logs')
# Scan the table sorted by timestamp in descending order
response = table.query(
IndexName='AllTimestampIndex',
KeyConditionExpression=Key('all').eq('ALL'),
ScanIndexForward=False,
Limit=1
)
# If there are no items in the table, return a default timestamp
if not response['Items']:
return '1970-01-01T00:00:00'
# Otherwise, return the timestamp of the latest item
return response['Items'][0]['battle_timestamp']
def log_elo_update(arena_battle_id, battle_timestamp, new_rating1, new_rating2):
# Reference the elo_logs table
table = dynamodb.Table('elo_logs')
# Update the table
table.put_item(
Item={
'arena_battle_id': arena_battle_id,
'battle_timestamp': battle_timestamp, # Use the timestamp of the battle
'log_timestamp': datetime.now().isoformat(), # Also store the timestamp of the log for completeness
'new_rating1': new_rating1,
'new_rating2': new_rating2,
'all': 'ALL',
}
)
def get_elo_score(chatbot_name, elo_scores):
if chatbot_name in elo_scores:
return elo_scores[chatbot_name]
table = dynamodb.Table('elo_scores')
response = table.get_item(Key={'chatbot_name': chatbot_name})
# If there is no item in the table, return a default score
if 'Item' not in response:
return 1500
return response['Item']['elo_score']
def update_elo_score(chatbot_name, new_elo_score):
table = dynamodb.Table('elo_scores')
# This will create a new item if it doesn't exist
table.put_item(
Item={
'chatbot_name': chatbot_name,
'elo_score': Decimal(str(new_elo_score)),
}
)
def get_elo_scores():
table = dynamodb.Table('elo_scores')
response = table.scan()
data = response['Items']
return data
def _backfill_logs():
table = dynamodb.Table('elo_logs')
# Initialize the scan operation
response = table.scan()
for item in response['Items']:
table.update_item(
Key={
'arena_battle_id': item['arena_battle_id'],
'battle_timestamp': item['battle_timestamp']
},
UpdateExpression="SET #all = :value",
ExpressionAttributeNames={
'#all': 'all'
},
ExpressionAttributeValues={
':value': 'ALL'
}
)
def main():
last_processed_timestamp = get_last_processed_timestamp()
battles: List[dict] = get_unprocessed_battles(last_processed_timestamp)
battles = sorted(battles, key=lambda x: x['timestamp'])
elo_scores = {}
for battle in battles:
print(repr(battle))
if battle['label'] in {-1, 0, 1, 2}:
outcome = battle['label']
for chatbot_name in [battle['choice1_name'], battle['choice2_name']]:
if chatbot_name not in elo_scores:
elo_scores[chatbot_name] = get_elo_score(chatbot_name, elo_scores)
# 1: This means that the first player (or team) won the match.
# 0.5: This means that the match ended in a draw.
# 0: This means that the first player (or team) lost the match.
if outcome == 0 or outcome == -1:
elo_result = 0.5
elif outcome == 1:
elo_result = 1
else:
elo_result = 0
new_rating1, new_rating2 = calculate_elo(elo_scores[battle['choice1_name']], elo_scores[battle['choice2_name']], elo_result)
logging.info(f"{battle['choice1_name']}: {elo_scores[battle['choice1_name']]} -> {new_rating1} | {battle['choice2_name']}: {elo_scores[battle['choice2_name']]} -> {new_rating2}")
elo_scores[battle['choice1_name']] = new_rating1
elo_scores[battle['choice2_name']] = new_rating2
log_elo_update(battle['arena_battle_id'], battle['timestamp'], new_rating1, new_rating2)
update_elo_score(battle['choice1_name'], new_rating1)
update_elo_score(battle['choice2_name'], new_rating2)
elo_scores[battle['choice1_name']] = new_rating1
elo_scores[battle['choice2_name']] = new_rating2
elo_scores = get_elo_scores()
for i, j in enumerate(elo_scores):
j["elo_score"] = float(j["elo_score"])
elo_scores[i] = j
print(elo_scores)
if battles:
# Convert the data into a format suitable for Hugging Face Dataset
elo_dataset = Dataset.from_list(elo_scores)
elo_dataset.push_to_hub("openaccess-ai-collective/chatbot-arena-elo-scores", private=False)
if __name__ == "__main__":
main()