253 lines
9.9 KiB
Python
253 lines
9.9 KiB
Python
# agents/core-logic/app.py
|
|
# OSVauco-NMTMD-GCOS — FastAPI HTTP entrypoint for Cloud Run
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Gjor ml-pakken tilgjengelig uansett cwd
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
|
|
|
|
try:
|
|
from google.adk.runners import Runner
|
|
from google.adk.sessions import InMemorySessionService
|
|
from google.genai.types import Content, Part
|
|
from agent import root_agent
|
|
from agents.recommendations_engine import get_recommendations
|
|
except ImportError as e:
|
|
logger.error(f"Failed to import ADK dependencies: {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")
|
|
BQ_BILLING_DATASET = os.environ.get("BQ_BILLING_DATASET", "billing_data")
|
|
|
|
# CG6: in-memory budget store (TODO Fase-C: migrate to Firestore per-user)
|
|
_budget_store: dict = {} # key: user_email | "default"
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
logger.info(f"OSVauco agent '{APP_NAME}' starting up")
|
|
yield
|
|
logger.info(f"OSVauco agent '{APP_NAME}' shutting down")
|
|
|
|
|
|
app = FastAPI(
|
|
title="OSVauco GCP Agent",
|
|
description="ADK-based multi-agent orchestrator on Cloud Run",
|
|
version="1.0.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
class RunRequest(BaseModel):
|
|
user_id: str
|
|
session_id: str
|
|
message: str
|
|
|
|
|
|
class RunResponse(BaseModel):
|
|
user_id: str
|
|
session_id: str
|
|
response: str
|
|
|
|
|
|
class BudgetRequest(BaseModel):
|
|
budget: float
|
|
user: str = "default"
|
|
|
|
|
|
async def _ensure_session(user_id: str, session_id: str):
|
|
try:
|
|
session = await session_service.get_session(
|
|
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
|
|
)
|
|
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)])
|
|
final_response = ""
|
|
async for event in runner.run_async(
|
|
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)
|
|
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():
|
|
try:
|
|
from google.cloud import bigquery
|
|
client = bigquery.Client(project=PROJECT_ID)
|
|
query = f"""
|
|
SELECT
|
|
agent_name,
|
|
model_name,
|
|
SUM(input_tokens) AS total_input_tokens,
|
|
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
|
|
"""
|
|
rows = [
|
|
{
|
|
"agent_name": row.agent_name,
|
|
"model_name": row.model_name,
|
|
"total_input_tokens": row.total_input_tokens,
|
|
"total_output_tokens": row.total_output_tokens,
|
|
"total_tokens": row.total_tokens,
|
|
"total_cost_usd": round(float(row.total_cost_usd), 6),
|
|
}
|
|
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):
|
|
try:
|
|
recommendations = get_recommendations(budget)
|
|
return JSONResponse(recommendations)
|
|
except Exception as e:
|
|
logger.error(f"[/billing/recommendations] Failed: {e}", exc_info=True)
|
|
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):
|
|
try:
|
|
from ml.billing_agent import BillingAgent
|
|
agent = BillingAgent()
|
|
return JSONResponse(agent.get_service_totals(days))
|
|
except Exception as e:
|
|
logger.error(f"[/billing/by-service] Failed: {e}", exc_info=True)
|
|
return JSONResponse([]) # fallback: tom liste, dashbordet krasjer ikke
|
|
|
|
|
|
# ── billing: anomalier (CG6) ─────────────────────────────────────────────────
|
|
@app.get("/billing/anomalies")
|
|
async def billing_anomalies():
|
|
"""
|
|
CG6 — Anomalideteksjon: sammenligner dagens kostnad mot 7-dagers snitt per tjeneste.
|
|
Returnerer tjenester der dagens kostnad > 2x snittet.
|
|
Fallback: tom liste ved feil slik at dashbordet ikke krasjer.
|
|
"""
|
|
try:
|
|
from ml.billing_agent import BillingAgent
|
|
agent = BillingAgent()
|
|
return JSONResponse(agent.get_anomalies())
|
|
except Exception as e:
|
|
logger.error(f"[/billing/anomalies] Failed: {e}", exc_info=True)
|
|
return JSONResponse({"anomalies": []})
|
|
|
|
|
|
# ── billing: historikk for bar-chart (CG6) ───────────────────────────────────
|
|
@app.get("/billing/history")
|
|
async def billing_history(days: int = 30):
|
|
"""
|
|
CG6 — Daglig MTD-historikk for bar-chart i dashbordet.
|
|
Returnerer liste: [{date, mtd}] sortert ASC.
|
|
"""
|
|
try:
|
|
from ml.billing_agent import BillingAgent
|
|
agent = BillingAgent()
|
|
return JSONResponse({"history": agent.get_daily_history(days)})
|
|
except Exception as e:
|
|
logger.error(f"[/billing/history] Failed: {e}", exc_info=True)
|
|
return JSONResponse({"history": []})
|
|
|
|
|
|
# ── billing: summary ─────────────────────────────────────────────────────────
|
|
@app.get("/billing/summary")
|
|
async def billing_summary():
|
|
try:
|
|
from ml.billing_agent import BillingAgent
|
|
agent = BillingAgent()
|
|
return JSONResponse(agent.get_summary())
|
|
except Exception as e:
|
|
logger.error(f"[/billing/summary] Failed: {e}", exc_info=True)
|
|
return JSONResponse({"summary": []})
|
|
|
|
|
|
# ── billing: live / forecast ─────────────────────────────────────────────────
|
|
@app.get("/billing/live")
|
|
async def billing_live():
|
|
try:
|
|
from ml.billing_agent import BillingAgent
|
|
agent = BillingAgent()
|
|
return JSONResponse(agent.get_forecast())
|
|
except Exception as e:
|
|
logger.error(f"[/billing/live] Failed: {e}", exc_info=True)
|
|
return JSONResponse({})
|
|
|
|
|
|
# ── billing: budsjett GET + POST (CG6) ───────────────────────────────────────
|
|
# TODO Fase-C: erstatt _budget_store med Firestore-dokument per bruker
|
|
@app.get("/billing/budget")
|
|
async def get_budget(user: str = "default"):
|
|
"""
|
|
CG6 — Hent lagret budsjett for bruker.
|
|
In-memory; nullstilles ved redeploy (Firestore i Fase C).
|
|
"""
|
|
budget = _budget_store.get(user, 500.0)
|
|
return JSONResponse({"budget": budget, "user": user})
|
|
|
|
|
|
@app.post("/billing/budget")
|
|
async def set_budget(req: BudgetRequest):
|
|
"""
|
|
CG6 — Lagre budsjett for bruker.
|
|
In-memory; nullstilles ved redeploy (Firestore i Fase C).
|
|
"""
|
|
if req.budget <= 0:
|
|
raise HTTPException(status_code=400, detail="Budget must be > 0")
|
|
_budget_store[req.user] = req.budget
|
|
logger.info(f"Budget updated: {req.user} -> {req.budget}")
|
|
return JSONResponse({"budget": req.budget, "user": req.user})
|