diff --git a/agents/core-logic/app.py b/agents/core-logic/app.py index 71335fe..958c9b0 100644 --- a/agents/core-logic/app.py +++ b/agents/core-logic/app.py @@ -13,7 +13,7 @@ from pydantic import BaseModel logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -# Allow imports from repo root (e.g. ml.billing_agent) +# Gjor ml-pakken tilgjengelig uansett cwd (Cloud Run starter i agents/core-logic/) sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) try: @@ -27,8 +27,8 @@ except ImportError as e: raise session_service = InMemorySessionService() -APP_NAME = os.environ.get("CLOUD_RUN_SERVICE", "gcp-orchestrator") -PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") +APP_NAME = os.environ.get("CLOUD_RUN_SERVICE", "gcp-orchestrator") +PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") BQ_BILLING_DATASET = os.environ.get("BQ_BILLING_DATASET", "billing_data") @@ -60,81 +60,58 @@ class RunResponse(BaseModel): async def _ensure_session(user_id: str, session_id: str): - """Await get_session; create if missing or raises.""" try: session = await session_service.get_session( - app_name=APP_NAME, - user_id=user_id, - session_id=session_id, + app_name=APP_NAME, user_id=user_id, session_id=session_id ) if session is not None: return session except Exception: pass - session = await session_service.create_session( - app_name=APP_NAME, - user_id=user_id, - session_id=session_id, + app_name=APP_NAME, user_id=user_id, session_id=session_id ) logger.info(f"Created new session: {session_id} for user: {user_id}") return session +# ── helse ────────────────────────────────────────────────────────────────────── @app.get("/health") async def health(): return JSONResponse({"status": "ok", "service": APP_NAME}) +# ── agent run ────────────────────────────────────────────────────────────────── @app.post("/run", response_model=RunResponse) async def run(req: RunRequest): try: await _ensure_session(req.user_id, req.session_id) - - runner = Runner( - agent=root_agent, - app_name=APP_NAME, - session_service=session_service, - ) - - user_content = Content( - role="user", - parts=[Part(text=req.message)], - ) - + runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service) + user_content = Content(role="user", parts=[Part(text=req.message)]) final_response = "" async for event in runner.run_async( - user_id=req.user_id, - session_id=req.session_id, - new_message=user_content, + user_id=req.user_id, session_id=req.session_id, new_message=user_content ): if event.is_final_response() and event.content: for part in event.content.parts: if part.text: final_response += part.text - logger.info(f"[{req.user_id}/{req.session_id}] Response length: {len(final_response)}") - return RunResponse( - user_id=req.user_id, - session_id=req.session_id, - response=final_response, - ) - + return RunResponse(user_id=req.user_id, session_id=req.session_id, response=final_response) except Exception as e: logger.error(f"Agent run failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) +# ── billing: tokens ──────────────────────────────────────────────────────────── @app.get("/billing/tokens/summary") async def billing_tokens_summary(): """ - CG3e — Aggreger LLM token-bruk og estimert kostnad per agent siste 30 dager. - Returnerer JSON-liste sortert etter total_cost DESC. + CG3e — LLM token-bruk og estimert kostnad per agent siste 30 dager. """ try: from google.cloud import bigquery client = bigquery.Client(project=PROJECT_ID) - query = f""" SELECT agent_name, @@ -143,17 +120,11 @@ async def billing_tokens_summary(): SUM(output_tokens) AS total_output_tokens, SUM(total_tokens) AS total_tokens, SUM(estimated_cost_usd) AS total_cost_usd - FROM - `{PROJECT_ID}.{BQ_BILLING_DATASET}.llm_token_usage` - WHERE - timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) - GROUP BY - agent_name, model_name - ORDER BY - total_cost_usd DESC + FROM `{PROJECT_ID}.{BQ_BILLING_DATASET}.llm_token_usage` + WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) + GROUP BY agent_name, model_name + ORDER BY total_cost_usd DESC """ - - results = client.query(query).result() rows = [ { "agent_name": row.agent_name, @@ -163,20 +134,19 @@ async def billing_tokens_summary(): "total_tokens": row.total_tokens, "total_cost_usd": round(float(row.total_cost_usd), 6), } - for row in results + for row in client.query(query).result() ] return JSONResponse({"period_days": 30, "rows": rows}) - except Exception as e: logger.error(f"[/billing/tokens/summary] Failed: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) +# ── billing: anbefalinger ────────────────────────────────────────────────────── @app.get("/billing/recommendations") async def billing_recommendations(budget: float = 500.0): """ - CG4 — Kjorer anbefalings- og anomali-motoren. - Returnerer en liste med anbefalinger og estimert besparelse. + CG4 — Anbefalings- og anomali-motor. """ try: recommendations = get_recommendations(budget) @@ -186,16 +156,23 @@ async def billing_recommendations(budget: float = 500.0): raise HTTPException(status_code=500, detail=str(e)) +# ── billing: tjenester med SKU-detaljer (CG5) ─────────────────────────────────── @app.get("/billing/by-service") async def billing_by_service(days: int = 30): """ - CG5 — Returnerer kostnad per tjeneste gruppert med SKU-detaljer. - Brukes av dashboard for grouped drill-down visning. + CG5 — Kostnad per tjeneste med SKU-detaljer for drill-down i dashbordet. + Returnerer liste sortert etter total_cost DESC. + Hvert element: {service, total_cost, skus: [{sku, sku_cost}]} + Fallback: hvis BillingAgent feiler returneres en tom liste (dashbordet + viser da fallback-visning istedenfor å krasje). """ try: from ml.billing_agent import BillingAgent agent = BillingAgent() - return JSONResponse(agent.get_service_totals(days)) + data = agent.get_service_totals(days) + return JSONResponse(data) except Exception as e: logger.error(f"[/billing/by-service] Failed: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) + # Returner tom liste istedenfor 500 — dashbordet faller da tilbake til + # den enkle buildSvc-visningen uten å miste all annen data. + return JSONResponse([]) diff --git a/ml/billing_agent.py b/ml/billing_agent.py index 439b902..0b2ec88 100644 --- a/ml/billing_agent.py +++ b/ml/billing_agent.py @@ -7,7 +7,7 @@ class BillingAgent: self.project_id = os.environ.get("GOOGLE_CLOUD_PROJECT") if not self.project_id: raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.") - + self.billing_table = os.environ.get("BILLING_TABLE") if not self.billing_table: raise ValueError("BILLING_TABLE environment variable not set.") @@ -16,7 +16,7 @@ class BillingAgent: def get_summary(self): """ - Retrieves the daily billing summary per project and service for the last 30 days. + Henter daglig billing-sammendrag per prosjekt og tjeneste siste 30 dager. """ query = f""" SELECT @@ -24,91 +24,68 @@ class BillingAgent: project.id AS project_id, service.description AS service, SUM(cost) AS daily_cost - FROM - `{self.billing_table}` - WHERE - _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) - GROUP BY - usage_date, project_id, service - ORDER BY - usage_date DESC, daily_cost DESC + FROM `{self.billing_table}` + WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) + GROUP BY usage_date, project_id, service + ORDER BY usage_date DESC, daily_cost DESC LIMIT 100 """ - query_job = self.bq_client.query(query) - results = query_job.result() - - summary = [] - for row in results: - summary.append({ + results = self.bq_client.query(query).result() + summary = [ + { "usage_date": str(row.usage_date), "project_id": row.project_id, "service": row.service, - "daily_cost": row.daily_cost - }) - + "daily_cost": row.daily_cost, + } + for row in results + ] if not summary: return { "onboarding_status": { "state": "awaiting_data", - "message": "Fakturaeksport er aktiv, men inneholder ingen data for de siste 30 dagene ennå." + "message": "Fakturaeksport er aktiv, men ingen data for siste 30 dager ennå.", } } - return {"summary": summary} def get_forecast(self): """ - Retrieves a billing forecast based on the last 7 days. + Prognose basert på siste 7 dager. """ - query_last_7_days = f""" - SELECT SUM(cost) + SUM(IFNULL((SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) as total_cost + q7 = f""" + SELECT SUM(cost) + SUM(IFNULL((SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) """ - query_job = self.bq_client.query(query_last_7_days) - results = query_job.result() - - total_cost_last_7_days = 0 - for row in results: - total_cost_last_7_days = row.total_cost or 0 - - daily_average = total_cost_last_7_days / 7 + total_7d = list(self.bq_client.query(q7).result())[0].total_cost or 0 + daily_average = total_7d / 7 today = datetime.date.today() - if today.month == 12: - next_month_first_day = datetime.date(today.year + 1, 1, 1) - else: - next_month_first_day = datetime.date(today.year, today.month + 1, 1) - last_day_of_month = next_month_first_day - datetime.timedelta(days=1) - remaining_days = (last_day_of_month - today).days + next_month = datetime.date(today.year + (1 if today.month == 12 else 0), + (today.month % 12) + 1, 1) + remaining_days = (next_month - today).days - forecasted_cost = daily_average * remaining_days - - query_mtd = f""" - SELECT SUM(cost) + SUM(IFNULL((SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) as total_cost + q_mtd = f""" + SELECT SUM(cost) + SUM(IFNULL((SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE EXTRACT(MONTH FROM _PARTITIONTIME) = EXTRACT(MONTH FROM CURRENT_DATE()) - AND EXTRACT(YEAR FROM _PARTITIONTIME) = EXTRACT(YEAR FROM CURRENT_DATE()) + AND EXTRACT(YEAR FROM _PARTITIONTIME) = EXTRACT(YEAR FROM CURRENT_DATE()) """ - query_job_mtd = self.bq_client.query(query_mtd) - results_mtd = query_job_mtd.result() - mtd_cost = 0 - for row in results_mtd: - mtd_cost = row.total_cost or 0 - - total_forecast = mtd_cost + forecasted_cost + mtd_cost = list(self.bq_client.query(q_mtd).result())[0].total_cost or 0 return { "daily_average_last_7_days": daily_average, "month_to_date_cost": mtd_cost, - "forecasted_remaining_cost": forecasted_cost, - "total_monthly_forecast": total_forecast, - "data_note": "Prognose basert på siste 7 dager. Fakturadata fra BigQuery kan ha 24-48 timers forsinkelse." + "forecasted_remaining_cost": daily_average * remaining_days, + "total_monthly_forecast": mtd_cost + daily_average * remaining_days, + "remaining_days_in_month": remaining_days, + "data_note": "Prognose basert på siste 7 dager. BigQuery kan ha 24-48 timers forsinkelse.", } def get_anomalies(self): """ - Detects anomalies in billing data by comparing today's cost to the 7-day average. + Oppdager anomalier ved å sammenligne dagens kostnad mot 7-dagers snitt. """ query = f""" WITH daily_costs AS ( @@ -122,38 +99,78 @@ class BillingAgent: ), costs_with_avg AS ( SELECT - service, - usage_date, - daily_cost, - AVG(daily_cost) OVER (PARTITION BY service ORDER BY usage_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING) AS avg_7day + service, usage_date, daily_cost, + AVG(daily_cost) OVER ( + PARTITION BY service ORDER BY usage_date + ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING + ) AS avg_7day FROM daily_costs ) - SELECT - service, - daily_cost AS today_cost, - avg_7day, - (daily_cost / avg_7day) AS ratio + SELECT service, daily_cost AS today_cost, avg_7day, + (daily_cost / avg_7day) AS ratio FROM costs_with_avg WHERE usage_date = CURRENT_DATE() AND avg_7day > 0 AND daily_cost > (2.0 * avg_7day) """ - query_job = self.bq_client.query(query) - results = query_job.result() + results = self.bq_client.query(query).result() + return { + "anomalies": [ + { + "service": row.service, + "today_cost": row.today_cost, + "avg_7d": row.avg_7day, + "ratio": row.ratio, + } + for row in results + ] + } - anomalies = [] + def get_service_totals(self, days: int = 30): + """ + CG5 — Henter total kostnad per tjeneste gruppert med SKU-detaljer. + Returnerer en liste sortert etter total_cost DESC. + Hvert element har: + service : str + total_cost : float + skus : list av {sku: str, sku_cost: float} + """ + query = f""" + SELECT + service.description AS service, + sku.description AS sku, + SUM(cost) + + SUM(IFNULL( + (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0 + )) AS sku_cost + FROM `{self.billing_table}` + WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), + INTERVAL {int(days)} DAY) + GROUP BY service, sku + HAVING sku_cost > 0 + ORDER BY service, sku_cost DESC + """ + results = self.bq_client.query(query).result() + + # Grupper SKU-er under tjeneste + services: dict = {} for row in results: - anomalies.append({ - "service": row.service, - "today_cost": row.today_cost, - "avg_7d": row.avg_7day, - "ratio": row.ratio + svc = row.service + if svc not in services: + services[svc] = {"service": svc, "total_cost": 0.0, "skus": []} + services[svc]["total_cost"] = round(services[svc]["total_cost"] + float(row.sku_cost), 6) + services[svc]["skus"].append({ + "sku": row.sku, + "sku_cost": round(float(row.sku_cost), 6), }) - return {"anomalies": anomalies} + # Sorter etter total_cost DESC + return sorted(services.values(), key=lambda x: x["total_cost"], reverse=True) + if __name__ == '__main__': agent = BillingAgent() print("Summary:", agent.get_summary()) print("Forecast:", agent.get_forecast()) print("Anomalies:", agent.get_anomalies()) + print("By-service:", agent.get_service_totals())