Cesar Mendivil
8b458a3ddf
- Add Next.js app structure with base configs, linting, and formatting - Implement LiveKit Meet page, types, and utility functions - Add Docker, Compose, and deployment scripts for backend and token server - Provide E2E and smoke test scaffolding with Puppeteer and Playwright helpers - Include CSS modules and global styles for UI - Add postMessage and studio integration utilities - Update package.json with dependencies and scripts for development and testing
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
gemini_agent_server.py
|
|
|
|
Server HTTP mínimo (FastAPI) que expone un endpoint /query para recibir un prompt,
|
|
opcionalmente enviar ese prompt a la API de Gemini (si GEMINI_API_KEY está configurada),
|
|
interpretar la instrucción y ejecutar la acción usando `gemini_log_agent`.
|
|
|
|
- Instalar dependencias:
|
|
python3 -m pip install --user fastapi uvicorn requests
|
|
|
|
- Ejecutar:
|
|
export GEMINI_API_KEY="YOUR_KEY"
|
|
export PLAYWRIGHT_WS="ws://192.168.1.20:3003"
|
|
python3 -m uvicorn packages.broadcast_panel.e2e.gemini_agent_server:app --host 0.0.0.0 --port 5001
|
|
|
|
Nota: el módulo se implementa para usarse en el workspace; si uvicorn no encuentra
|
|
el módulo, ejecuta desde la raíz del repo:
|
|
uvicorn packages.broadcast-panel.e2e.gemini_agent_server:app --reload
|
|
"""
|
|
from __future__ import annotations
|
|
import os
|
|
import json
|
|
import traceback
|
|
from typing import Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
# Import local agent
|
|
from .gemini_log_agent import interpret_prompt, handle_action
|
|
|
|
import requests
|
|
|
|
app = FastAPI(title="Gemini Log Agent Server")
|
|
|
|
GEMINI_API_KEY = os.environ.get('GEMINI_API_KEY')
|
|
PLAYWRIGHT_WS = os.environ.get('PLAYWRIGHT_WS', 'ws://192.168.1.20:3003')
|
|
|
|
class QueryRequest(BaseModel):
|
|
prompt: str
|
|
use_llm: Optional[bool] = False
|
|
lines: Optional[int] = 200
|
|
ws: Optional[str] = None
|
|
|
|
class QueryResponse(BaseModel):
|
|
ok: bool
|
|
mapping: dict
|
|
result: str
|
|
|
|
|
|
@app.get('/health')
|
|
def health():
|
|
return {"status": "ok", "playwright_ws": PLAYWRIGHT_WS, "llm": bool(GEMINI_API_KEY)}
|
|
|
|
|
|
def call_gemini(prompt: str) -> Optional[str]:
|
|
"""Call Google Generative Language (Gemini) REST endpoint using API key if available.
|
|
Returns the generated text or None on failure.
|
|
Note: we use the text-bison endpoint (v1beta2) as a best-effort; if the network fails
|
|
we return None and fallback to local heuristics.
|
|
"""
|
|
if not GEMINI_API_KEY:
|
|
return None
|
|
try:
|
|
url = f"https://generativelanguage.googleapis.com/v1beta2/models/text-bison-001:generateText?key={GEMINI_API_KEY}"
|
|
payload = { "prompt": { "text": prompt }, "temperature": 0.2, "maxOutputTokens": 256 }
|
|
headers = { 'Content-Type': 'application/json' }
|
|
resp = requests.post(url, json=payload, headers=headers, timeout=15)
|
|
if resp.status_code != 200:
|
|
return None
|
|
body = resp.json()
|
|
# Response shape: {'candidates':[{'content':'...'}], ...}
|
|
if 'candidates' in body and len(body['candidates'])>0:
|
|
return body['candidates'][0].get('content')
|
|
# Older shapes may have 'output' or 'result'
|
|
if 'output' in body and isinstance(body['output'], list) and len(body['output'])>0:
|
|
return body['output'][0].get('content')
|
|
if 'result' in body and isinstance(body['result'], dict):
|
|
return body['result'].get('content')
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
@app.post('/query', response_model=QueryResponse)
|
|
def query(req: QueryRequest):
|
|
try:
|
|
prompt = req.prompt
|
|
use_llm = bool(req.use_llm)
|
|
lines = int(req.lines or 200)
|
|
ws = req.ws or PLAYWRIGHT_WS
|
|
|
|
if use_llm and GEMINI_API_KEY:
|
|
gen = call_gemini(prompt)
|
|
if gen:
|
|
# Use the generated text to form a mapping prompt (pass-through)
|
|
interpreted = interpret_prompt(gen)
|
|
else:
|
|
interpreted = interpret_prompt(prompt)
|
|
else:
|
|
interpreted = interpret_prompt(prompt)
|
|
|
|
# Attach ws info if available for run_session
|
|
if interpreted.get('action') == 'run_session':
|
|
interpreted['ws'] = ws
|
|
|
|
result = handle_action(interpreted, lines=lines)
|
|
return QueryResponse(ok=True, mapping=interpreted, result=result)
|
|
except Exception as e:
|
|
traceback.print_exc()
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|