chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(CG5): add get_service_totals to BillingAgent + complete /billing/by-service in app.py

Browse Source
This commit is contained in:
chrischristiansen-glitch 2026-06-09 23:45:08 +02:00
parent 97f0ead1fb
commit fd6aa1f2aa
2 changed files with 120 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
agents/core-logic/app.py
View File

@ -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([])

161
ml/billing_agent.py
View File

@ -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())