From 68802353ef670e0a5926260229b579045bfd24e3 Mon Sep 17 00:00:00 2001 From: chrischristiansen-glitch Date: Mon, 15 Jun 2026 16:15:51 +0200 Subject: [PATCH] =?UTF-8?q?feat(ML-3a):=20POST=20/emma=20+=20EMMA=5FBACKEN?= =?UTF-8?q?D=20flagg=20(vertex=20n=C3=A5=20=E2=86=92=20local=20VM=20ved=20?= =?UTF-8?q?trigger)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 138 ++++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 119 insertions(+), 19 deletions(-) diff --git a/main.py b/main.py index bba3103..9f891c4 100644 --- a/main.py +++ b/main.py @@ -17,6 +17,7 @@ CG4-credits: GET /billing/credits — kreditt-saldo, burn rate og tom-dato. Krever GOOGLE_CREDIT_TOTAL_USD i Cloud Run env for runway-beregning. TERMINAL: POST /terminal/exec — whitelisted kommandoer: health, billing, build, logs, help. ✅ VM: GET /vm/ssh-key — henter public key fra Compute Engine metadata. ✅ +ML-3a: POST /emma — intern agent (Emma Vauger). EMMA_BACKEND=vertex (nå) | local (VM-trigger). ✅ """ import os @@ -64,6 +65,21 @@ PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5") # Guard+-tiers som får tilgang til SMS-varsler SMS_ALLOWED_TIERS = {"guard", "shield", "enterprise"} +# ── EMMA: backend-flagg ─────────────────────────────────────────────────────── +# EMMA_BACKEND=vertex → Vertex AI (nå, bruker eksisterende kreditter) +# EMMA_BACKEND=local → lokal Gemma 4 på g2-standard-4 Spot VM (trigger: første kunde) +# EMMA_LOCAL_URL → http://:8000/generate (kun brukt når EMMA_BACKEND=local) +EMMA_BACKEND = os.environ.get("EMMA_BACKEND", "vertex") +EMMA_LOCAL_URL = os.environ.get("EMMA_LOCAL_URL", "http://localhost:8000/generate") +EMMA_MODEL = os.environ.get("EMMA_MODEL", "google/gemma-3-12b-it") # Vertex model ID + +# Last Emma sitt system-prompt fra world.md ved oppstart +_WORLD_MD_PATH = pathlib.Path(__file__).parent / "docs" / "gemma" / "world.md" +try: + _EMMA_SYSTEM_PROMPT = _WORLD_MD_PATH.read_text(encoding="utf-8") +except FileNotFoundError: + _EMMA_SYSTEM_PROMPT = "Du er Emma Vauger, intern AI-assistent for Vauco AS. Vær analytisk og direkte." + app = FastAPI( title="OSVauco OPAX Agent", description="Agent for OSVauco-OPAX platform.", @@ -206,14 +222,6 @@ class TerminalExecRequest(BaseModel): @app.post("/terminal/exec") async def terminal_exec(req: TerminalExecRequest): - """ - Whitelisted terminal-kommandoer: - health → GET /health internt - billing → GET /billing/summary - build → GET /opax/build-status - logs → siste 20 linjer fra Cloud Logging - help → returner kommandoliste - """ cmd = req.cmd.strip().lower().split()[0] if req.cmd.strip() else "" if cmd not in TERMINAL_WHITELIST: return JSONResponse( @@ -301,10 +309,6 @@ async def terminal_exec(req: TerminalExecRequest): # ── VM: GET /vm/ssh-key ──────────────────────────────────────────────────────── @app.get("/vm/ssh-key") async def vm_ssh_key(instance: str = "osvauco-dev-vm"): - """ - Henter public SSH-nøkkel fra Compute Engine instance metadata. - Bruker google.auth ADC — krever compute.instances.get på service account. - """ try: creds, project = google.auth.default() creds.refresh(google.auth.transport.requests.Request()) @@ -340,18 +344,116 @@ async def vm_ssh_key(instance: str = "osvauco-dev-vm"): raise HTTPException(status_code=500, detail=f"vm/ssh-key feilet: {e}") -# ── OQ-29: GET /notify/channels — konfigurerte delivery-kanaler ─────────────── +# ── OQ-29: GET /notify/channels ─────────────────────────────────────────────── @app.get("/notify/channels") async def get_notify_channels(): return { "channels": [ - {"type": "email", "configured": bool(os.getenv("SENDGRID_API_KEY")), "target": os.getenv("NOTIFY_EMAIL_TO", None)}, - {"type": "sms", "configured": bool(os.getenv("TWILIO_ACCOUNT_SID")), "target": os.getenv("TWILIO_FROM_NUMBER", None)}, - {"type": "webhook", "configured": bool(os.getenv("NOTIFY_WEBHOOK_URL")), "url": os.getenv("NOTIFY_WEBHOOK_URL", None)}, + {"type": "email", "configured": bool(os.getenv("SENDGRID_API_KEY")), "target": os.getenv("NOTIFY_EMAIL_TO", None)}, + {"type": "sms", "configured": bool(os.getenv("TWILIO_ACCOUNT_SID")), "target": os.getenv("TWILIO_FROM_NUMBER", None)}, + {"type": "webhook", "configured": bool(os.getenv("NOTIFY_WEBHOOK_URL")), "url": os.getenv("NOTIFY_WEBHOOK_URL", None)}, ] } +# ── EMMA: ML-3a intern agent ────────────────────────────────────────────────── +class EmmaRequest(BaseModel): + message: str + session_id: str = "emma-default" + history: Optional[List[Dict[str, str]]] = Field( + default=None, + description="Valgfri samtalehistorikk: [{role: user|model, content: str}]" + ) + + +class EmmaChatService: + """ + Håndterer Emma-kall mot enten: + - Vertex AI (EMMA_BACKEND=vertex) — nå, bruker Vertex-kreditter + - Lokal Gemma 4 VM (EMMA_BACKEND=local) — når trigger er nådd + Bytt backend: sett EMMA_BACKEND=local + EMMA_LOCAL_URL=http://:8000/generate + """ + + async def chat(self, message: str, history: Optional[List[Dict]] = None) -> str: + if EMMA_BACKEND == "local": + return await self._local(message, history) + return await self._vertex(message, history) + + async def _vertex(self, message: str, history: Optional[List[Dict]] = None) -> str: + try: + import vertexai + from vertexai.generative_models import GenerativeModel, Content, Part + + vertexai.init(project=PROJECT_ID, location="us-central1") + model = GenerativeModel( + EMMA_MODEL, + system_instruction=_EMMA_SYSTEM_PROMPT, + ) + + chat_history = [] + for turn in (history or []): + role = turn.get("role", "user") + content = turn.get("content", "") + chat_history.append(Content(role=role, parts=[Part.from_text(content)])) + + chat = model.start_chat(history=chat_history) + response = chat.send_message(message) + return response.text + except Exception as e: + raise HTTPException(status_code=500, detail=f"Emma (Vertex) feilet: {e}") + + async def _local(self, message: str, history: Optional[List[Dict]] = None) -> str: + """ + Kaller lokal Gemma 4 på VM via HTTP. + Forventet format: POST {EMMA_LOCAL_URL} med {prompt, system_prompt} + """ + turns = "" + for turn in (history or []): + role = "User" if turn.get("role") == "user" else "Emma" + turns += f"{role}: {turn.get('content', '')}\n" + full_prompt = f"{turns}User: {message}\nEmma:" + + try: + async with httpx.AsyncClient(timeout=60.0) as client: + resp = await client.post( + EMMA_LOCAL_URL, + json={"prompt": full_prompt, "system_prompt": _EMMA_SYSTEM_PROMPT, "max_tokens": 1024}, + ) + resp.raise_for_status() + data = resp.json() + return data.get("response") or data.get("text") or data.get("generated_text", "") + except httpx.TimeoutException: + raise HTTPException(status_code=504, detail="Emma (lokal VM) svarte ikke innen 60s") + except Exception as e: + raise HTTPException(status_code=500, detail=f"Emma (lokal) feilet: {e}") + + +_emma = EmmaChatService() + + +async def _emma_chat(request: Request, req: EmmaRequest): + """ + POST /emma — intern agent, kun for Chris (IAP-beskyttet). + Backend: EMMA_BACKEND env-var (vertex | local). + """ + start = time.monotonic() + response_text = await _emma.chat(req.message, req.history) + duration = round(time.monotonic() - start, 3) + store = get_store() + store.push("emma", "last_duration_s", duration) + store.push("emma", "last_backend", EMMA_BACKEND) + store.push("emma", "last_success", True) + return JSONResponse({ + "response": response_text, + "backend": EMMA_BACKEND, + "model": EMMA_MODEL, + "duration_s": duration, + "session_id": req.session_id, + }) + +app.add_api_route("/emma", endpoint=require_auth(_emma_chat), methods=["POST"]) + + # ── MODELLER ────────────────────────────────────────────────────────────────── class RunRequest(BaseModel): message: str @@ -756,13 +858,11 @@ async def authenticated_billing_anomalies(request: Request): app.add_api_route("/billing/anomalies", endpoint=require_auth(authenticated_billing_anomalies), methods=["GET"]) -# ── CG4-credits: GET /billing/credits ──────────────────────────────────────── async def authenticated_billing_credits(request: Request, days: int = 90): try: return JSONResponse(BillingAgent().get_credits_status(days=days)) except Exception as exc: return JSONResponse(status_code=500, content={"error": str(exc)}) - app.add_api_route("/billing/credits", endpoint=require_auth(authenticated_billing_credits), methods=["GET"]) @@ -886,7 +986,7 @@ def run_dag(req: DagRequest): store = get_store() log_dag_execution(dag_id=req.session_id, agent_results=results, total_duration_s=total_dur) store.push(AGENT_ID, "last_dag_total_duration_s", total_dur) - store.push(AGENT_ID, "last_dag_agent_count", len(agent_ids)) + store.push(AGENT_ID, "last_dag_agent_count", len(tasks)) store.push(AGENT_ID, "last_dag_success_count", sum(1 for r in results if r["success"])) return { "total_duration_s": total_dur,