From f549fef84c94a8614751ef0b148a947438b505fb Mon Sep 17 00:00:00 2001 From: Pratyush Shukla Date: Sat, 23 Nov 2024 15:12:36 +0530 Subject: [PATCH] oops --- agentops/session.py | 75 --------------------------------------------- 1 file changed, 75 deletions(-) diff --git a/agentops/session.py b/agentops/session.py index 8c68e8a0..4fa50a8a 100644 --- a/agentops/session.py +++ b/agentops/session.py @@ -642,80 +642,5 @@ def get_analytics(self) -> Optional[Dict[str, Any]]: "Cost": formatted_cost, } - @staticmethod - def _format_duration(start_time, end_time): - start = datetime.fromisoformat(start_time.replace("Z", "+00:00")) - end = datetime.fromisoformat(end_time.replace("Z", "+00:00")) - duration = end - start - - hours, remainder = divmod(duration.total_seconds(), 3600) - minutes, seconds = divmod(remainder, 60) - - parts = [] - if hours > 0: - parts.append(f"{int(hours)}h") - if minutes > 0: - parts.append(f"{int(minutes)}m") - parts.append(f"{seconds:.1f}s") - - return " ".join(parts) - - def _get_response(self) -> Optional[Response]: - with self._lock: - payload = {"session": self.__dict__} - try: - response = HttpClient.post( - f"{self.config.endpoint}/v2/update_session", - json.dumps(filter_unjsonable(payload)).encode("utf-8"), - jwt=self.jwt, - ) - except ApiServerException as e: - logger.error(f"Could not fetch response from server - {e}") - return None - - logger.debug(response.body) - return response - - def _get_token_cost(self, response: Response) -> Decimal: - token_cost = response.body.get("token_cost", "unknown") - if token_cost == "unknown" or token_cost is None: - return Decimal(0) - return Decimal(token_cost) - - @staticmethod - def _format_token_cost(token_cost_d): - return ( - "{:.2f}".format(token_cost_d) - if token_cost_d == 0 - else "{:.6f}".format(token_cost_d.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)) - ) - - def get_analytics(self) -> Optional[dict[str, Union[Decimal, str]]]: - if not self.end_timestamp: - self.end_timestamp = get_ISO_time() - - formatted_duration = self._format_duration(self.init_timestamp, self.end_timestamp) - - response = self._get_response() - if response is None: - return None - - self.token_cost = self._get_token_cost(response) - formatted_cost = self._format_token_cost(self.token_cost) - - self.session_url = response.body.get( - "session_url", - f"https://app.agentops.ai/drilldown?session_id={self.session_id}", - ) - - return { - "LLM calls": self.event_counts["llms"], - "Tool calls": self.event_counts["tools"], - "Actions": self.event_counts["actions"], - "Errors": self.event_counts["errors"], - "Duration": formatted_duration, - "Cost": formatted_cost, - } - active_sessions: List[Session] = []