Spaces:
Sleeping
Sleeping
File size: 3,663 Bytes
7db0ae4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
#### What this does ####
# On success + failure, log events to aispend.io
import datetime
import traceback
import dotenv
import os
import requests
dotenv.load_dotenv() # Loading env variables using dotenv
# convert to {completion: xx, tokens: xx}
def parse_usage(usage):
return {
"completion": usage["completion_tokens"] if "completion_tokens" in usage else 0,
"prompt": usage["prompt_tokens"] if "prompt_tokens" in usage else 0,
}
def parse_messages(input):
if input is None:
return None
def clean_message(message):
# if is strin, return as is
if isinstance(message, str):
return message
if "message" in message:
return clean_message(message["message"])
text = message["content"]
if text == None:
text = message.get("function_call", None)
return {
"role": message["role"],
"text": text,
}
if isinstance(input, list):
if len(input) == 1:
return clean_message(input[0])
else:
return [clean_message(msg) for msg in input]
else:
return clean_message(input)
class LLMonitorLogger:
# Class variables or attributes
def __init__(self):
# Instance variables
self.api_url = os.getenv("LLMONITOR_API_URL") or "https://app.llmonitor.com"
self.app_id = os.getenv("LLMONITOR_APP_ID")
def log_event(
self,
type,
event,
run_id,
model,
print_verbose,
input=None,
user_id=None,
response_obj=None,
start_time=datetime.datetime.now(),
end_time=datetime.datetime.now(),
error=None,
):
# Method definition
try:
print_verbose(f"LLMonitor Logging - Logging request for model {model}")
if response_obj:
usage = (
parse_usage(response_obj["usage"])
if "usage" in response_obj
else None
)
output = response_obj["choices"] if "choices" in response_obj else None
else:
usage = None
output = None
if error:
error_obj = {"stack": error}
else:
error_obj = None
data = [
{
"type": type,
"name": model,
"runId": run_id,
"app": self.app_id,
"event": "start",
"timestamp": start_time.isoformat(),
"userId": user_id,
"input": parse_messages(input),
},
{
"type": type,
"runId": run_id,
"app": self.app_id,
"event": event,
"error": error_obj,
"timestamp": end_time.isoformat(),
"userId": user_id,
"output": parse_messages(output),
"tokensUsage": usage,
},
]
print_verbose(f"LLMonitor Logging - final data object: {data}")
response = requests.post(
self.api_url + "/api/report",
headers={"Content-Type": "application/json"},
json={"events": data},
)
print_verbose(f"LLMonitor Logging - response: {response}")
except:
# traceback.print_exc()
print_verbose(f"LLMonitor Logging Error - {traceback.format_exc()}")
pass
|