Integrando FastAPI con Gradio para endpoints API funcionales

app.py

@@ -8,6 +8,9 @@ import io
 import base64
 import os
 from huggingface_hub import login
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 
 # Configurar autenticación con Hugging Face
 HF_TOKEN = os.getenv("HF_TOKEN")
@@ -20,6 +23,28 @@ if HF_TOKEN:
 else:
     print("⚠️ No se encontró HF_TOKEN - modelos gated no estarán disponibles")
 
+# Clases para los endpoints API
+class TextRequest(BaseModel):
+    prompt: str
+    model_name: str
+    max_length: int = 100
+
+class ImageRequest(BaseModel):
+    prompt: str
+    model_name: str
+    num_inference_steps: int = 20
+
+class VideoRequest(BaseModel):
+    prompt: str
+    model_name: str
+    num_frames: int = 16
+    num_inference_steps: int = 20
+
+class ChatRequest(BaseModel):
+    message: str
+    history: list
+    model_name: str
+
 # Configuración de modelos libres
 MODELS = {
     "text": {
@@ -558,128 +583,79 @@ with gr.Blocks(title="Modelos Libres de IA", theme=gr.themes.Soft()) as demo:
 
 # Configuración para Hugging Face Spaces
 if __name__ == "__main__":
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False
+    # Crear la aplicación FastAPI
+    app = FastAPI(title="NTIA Space API")
+    
+    # Configurar CORS
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
     )
+    
+    # Endpoints API
+    @app.get("/models")
+    async def api_models():
+        try:
+            return {
+                "text": [{"id": k, "label": v} for k, v in MODELS["text"].items()],
+                "image": [{"id": k, "label": v} for k, v in MODELS["image"].items()],
+                "video": [{"id": k, "label": v} for k, v in MODELS["video"].items()],
+                "chat": [{"id": k, "label": v} for k, v in MODELS["chat"].items()]
+            }
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
 
-# Endpoints API para integración con chat NTIA
-import gradio_client
-from fastapi import FastAPI, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
-import base64
-from io import BytesIO
-
-app = FastAPI(title="NTIA Space API")
-
-# Configurar CORS
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-class TextRequest(BaseModel):
-    prompt: str
-    model_name: str
-    max_length: int = 100
-
-class ImageRequest(BaseModel):
-    prompt: str
-    model_name: str
-    num_inference_steps: int = 20
-
-class VideoRequest(BaseModel):
-    prompt: str
-    model_name: str
-    num_frames: int = 16
-    num_inference_steps: int = 20
-
-class ChatRequest(BaseModel):
-    message: str
-    history: list
-    model_name: str
-
-@app.post("/generate_text")
-async def api_generate_text(request: TextRequest):
-    try:
-        result = generate_text(request.prompt, request.model_name, request.max_length)
-        return {"response": result}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-@app.post("/generate_image")
-async def api_generate_image(request: ImageRequest):
-    try:
-        result = generate_image(request.prompt, request.model_name, request.num_inference_steps)
-        
-        # Convertir imagen a base64
-        if isinstance(result, str) and result.startswith("Error"):
-            raise HTTPException(status_code=500, detail=result)
-        
-        # Convertir PIL Image a base64
-        buffer = BytesIO()
-        result.save(buffer, format="PNG")
-        img_str = base64.b64encode(buffer.getvalue()).decode()
-        
-        return {"image": f"data:image/png;base64,{img_str}"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-@app.post("/generate_video")
-async def api_generate_video(request: VideoRequest):
-    try:
-        result = generate_video(request.prompt, request.model_name, request.num_frames, request.num_inference_steps)
-        
-        if isinstance(result, str) and result.startswith("Error"):
-            raise HTTPException(status_code=500, detail=result)
-        
-        # Convertir frames a video (simplificado)
-        # En producción, usarías ffmpeg para crear un video real
-        return {"video": "video_data_placeholder", "frames": len(result) if isinstance(result, list) else 0}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+    @app.post("/generate_text")
+    async def api_generate_text(request: TextRequest):
+        try:
+            result = generate_text(request.prompt, request.model_name, request.max_length)
+            return {"response": result}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
 
-@app.post("/chat")
-async def api_chat(request: ChatRequest):
-    try:
-        result = chat_with_model(request.message, request.history, request.model_name)
-        return {"response": result[-1]["content"] if result else "No response"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+    @app.post("/generate_image")
+    async def api_generate_image(request: ImageRequest):
+        try:
+            result = generate_image(request.prompt, request.model_name, request.num_inference_steps)
+            
+            if isinstance(result, str) and result.startswith("Error"):
+                raise HTTPException(status_code=500, detail=result)
+            
+            # Convertir PIL Image a base64
+            buffer = BytesIO()
+            result.save(buffer, format="PNG")
+            img_str = base64.b64encode(buffer.getvalue()).decode()
+            
+            return {"image": f"data:image/png;base64,{img_str}"}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
 
-@app.get("/models")
-async def api_models():
-    try:
-        # Devolver la estructura MODELS pero solo los nombres y labels
-        return {
-            "text": [
-                {"id": k, "label": v} for k, v in MODELS["text"].items()
-            ] if isinstance(MODELS["text"], dict) else [
-                {"id": k, "label": v} for k, v in MODELS["text"].items()
-            ],
-            "image": [
-                {"id": k, "label": v} for k, v in MODELS["image"].items()
-            ] if isinstance(MODELS["image"], dict) else [
-                {"id": k, "label": v} for k, v in MODELS["image"].items()
-            ],
-            "video": [
-                {"id": k, "label": v} for k, v in MODELS["video"].items()
-            ] if isinstance(MODELS["video"], dict) else [
-                {"id": k, "label": v} for k, v in MODELS["video"].items()
-            ],
-            "chat": [
-                {"id": k, "label": v} for k, v in MODELS["chat"].items()
-            ] if isinstance(MODELS["chat"], dict) else [
-                {"id": k, "label": v} for k, v in MODELS["chat"].items()
-            ]
-        }
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+    @app.post("/generate_video")
+    async def api_generate_video(request: VideoRequest):
+        try:
+            result = generate_video(request.prompt, request.model_name, request.num_frames, request.num_inference_steps)
+            
+            if isinstance(result, str) and result.startswith("Error"):
+                raise HTTPException(status_code=500, detail=result)
+            
+            return {"video": "video_data_placeholder", "frames": len(result) if isinstance(result, list) else 0}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
 
-# Montar la API en el mismo servidor que Gradio
-demo.app = app 
+    @app.post("/chat")
+    async def api_chat(request: ChatRequest):
+        try:
+            result = chat_with_model(request.message, request.history, request.model_name)
+            return {"response": result[-1]["content"] if result else "No response"}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
+    
+    # Montar Gradio en FastAPI
+    app = gr.mount_gradio_app(app, demo, path="/")
+    
+    # Lanzar con uvicorn
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)