使用Label Studio监控与评估生产环境中的模型

[labelstudio]
# Your Label Studio instance url
LabelStudioURL =
# Your Label Studio API key
LabelStudioAPI =
# OPTIONAL The ID of the Label Studio project that you want to use the config from.
LabelStudioProjectID = 
# OPTIONAL Your Label Studio Config. Note that you need a space at the beginning of every line!
LabelingConfig =

[data]
# Total number of samples to upload to LabelStudio for review.
total_to_extract = 
# BOOLEAN whether or not to sample your data by date
# If true, take an even sample across all 6 days sampled, getting as close as we can to the total_to_extract number without going over.
sample_by_date = True

[logs]
username =
api_key =
bucket =

[notifications]
# Email from which to send the notification
email_sender = 
#email or emails, comma separated with no spaces, to receive notification emails
email_recipient =
# SMTP server -- provided by your email client
smtp_server = smtp.gmail.com
smtp_port = 465
# the password or app key for your email. If using gmail, generate an app key here: https://myaccount.google.com/apppasswords
email_password =

# Connect to your server here.
# In this example, we use an S3 bucket
session = Session(aws_access_key_id=logs_username, aws_secret_access_key=logs_password)
s3 = session.resource('s3')
bucket = s3.Bucket(logs_bucket)

# for our test, we assume that files have the name "qalogs_MM:DD:YY.txt"
for s3_file in bucket.objects.all():
    key = s3_file.key
    if "qalogs" in key:
        key_date = key.split(".")[0]
        key_date = key_date.split("_")[1]
        key_date = datetime.strptime(key_date, '%m:%d:%y').date()
        print(key_date, end, start)
        if key_date <= end and key_date >= start:
            body = s3_file.get()['Body'].read().decode("utf-8")
   #         # the scrape file method does the file processing.
            all_data.extend(scrape_file(body))

def scrape_file(body):
   # basic file processing template
   # ALL logic will need to be customized based on the format of your logs.
   all_data = []
   if body:
       curr_data = {}
       for line in body.split('\n'):
           if "Timestamp" in line:
               if curr_data:
                   print(curr_data)
                   all_data.append(curr_data)
                   curr_data = {}
               line = line.split(' ')
               date = line[1]
               print(f'date {date}')
               curr_data["date"] = date
           if "User Input" in line:
               line = line.split('User Input: ')
               question = line[1]
               curr_data["question"] = question.strip("\"")
           if "Model Response" in line:
               line = line.split('Model Response: ')
               answer = line[1]
               curr_data["answer"] = answer.strip("\"")
           if "Version" in line:
               line = line.split("Version: ")
               model_version = line[1]
               curr_data["model_version"] = model_version
       if curr_data:
           all_data.append(curr_data)
       return all_data

def get_data_subset(all_data, total_to_extract, sample_by_date):
   """
   Get a sample of all your data. If the total len of data is less than the sample size, return all.
   Else, if sample_by_date, take an even sample across all date ranges as for a total as close as we can get
   to the intended sample number while keeping an even sample across all dates.
   :param all_data: a list of dictionaries containing all the data scraped from the logs
   :param total_to_extract: the goal number of samples to have in total
   :param sample_by_date: boolean, if true, sample the subset by date.
   :return:
   """
   if len(all_data) <= total_to_extract:
       return all_data

   if sample_by_date:
       by_date = {}
       for t in all_data:
           if t["date"] not in by_date.keys():
               by_date[t["date"]] = [t]
           else:
               by_date[t["date"]].append(t)

       per_date = floor(total_to_extract / len(by_date.keys()))
       new_data = []
       for date, data in by_date.items():
           random_sample = sample(data, per_date)
           new_data.extend(random_sample)
       return new_data

   #ToDo: Sample by class?

   else:
       random_sample = sample(all_data, total_to_extract)
       return random_sample

<View>
  <Header value="Question:"/>
  <Text name="question" value="$question"/>
  <TextArea name="answer" toName="question" editable="true" smart="false" maxSubmissions="1"/>
</View>

task_data = {
   "data":
       {"question": task["question"]}
}

{
    "model_version": "sample model version",
    "score": 0.0,
    "result": [
        {
            "id": "ab417d1e-b4ee-4f8f-b4e9-930d35da5e60",
            "from_name": "answer",
            "to_name": "question",
            "type": "textarea",
            "value": {
                "text": [
                    "Lorem Ipsum
                ]
            }
        }
    ]
}

from label_studio_sdk.label_interface.objects import PredictionValue
prediction = PredictionValue(
   model_version=task["model_version"],
   result=[
       {
           "from_name": "answer",
           "to_name": "question",
           "type": "textarea",
           "value": {
               "text": [task["answer"]]
           }
       }
   ]
)