File size: 3,131 Bytes
246d201 |
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 |
import time
from pydantic import BaseModel, Field
class Cost(BaseModel):
model: str
cost: float
timestamp: float = Field(default_factory=time.time)
class ResponseLatency(BaseModel):
"""Metric tracking the round-trip time per completion call."""
model: str
latency: float
response_id: str
class Metrics:
"""Metrics class can record various metrics during running and evaluation.
Currently, we define the following metrics:
accumulated_cost: the total cost (USD $) of the current LLM.
response_latency: the time taken for each LLM completion call.
"""
def __init__(self, model_name: str = 'default') -> None:
self._accumulated_cost: float = 0.0
self._costs: list[Cost] = []
self._response_latencies: list[ResponseLatency] = []
self.model_name = model_name
@property
def accumulated_cost(self) -> float:
return self._accumulated_cost
@accumulated_cost.setter
def accumulated_cost(self, value: float) -> None:
if value < 0:
raise ValueError('Total cost cannot be negative.')
self._accumulated_cost = value
@property
def costs(self) -> list[Cost]:
return self._costs
@property
def response_latencies(self) -> list[ResponseLatency]:
if not hasattr(self, '_response_latencies'):
self._response_latencies = []
return self._response_latencies
@response_latencies.setter
def response_latencies(self, value: list[ResponseLatency]) -> None:
self._response_latencies = value
def add_cost(self, value: float) -> None:
if value < 0:
raise ValueError('Added cost cannot be negative.')
self._accumulated_cost += value
self._costs.append(Cost(cost=value, model=self.model_name))
def add_response_latency(self, value: float, response_id: str) -> None:
self._response_latencies.append(
ResponseLatency(
latency=max(0.0, value), model=self.model_name, response_id=response_id
)
)
def merge(self, other: 'Metrics') -> None:
self._accumulated_cost += other.accumulated_cost
self._costs += other._costs
self._response_latencies += other._response_latencies
def get(self) -> dict:
"""Return the metrics in a dictionary."""
return {
'accumulated_cost': self._accumulated_cost,
'costs': [cost.model_dump() for cost in self._costs],
'response_latencies': [
latency.model_dump() for latency in self._response_latencies
],
}
def reset(self):
self._accumulated_cost = 0.0
self._costs = []
self._response_latencies = []
def log(self):
"""Log the metrics."""
metrics = self.get()
logs = ''
for key, value in metrics.items():
logs += f'{key}: {value}\n'
return logs
def __repr__(self):
return f'Metrics({self.get()}'
|