import os import subprocess from pathlib import Path import ...
Prompt
import os import subprocess from pathlib import Path import modal import docker MODEL_NAME = "MiniMaxAI/MiniMax-H3" HF_TOKEN = os.environ.get("HF_TOKEN", "") GPU_TYPE = "B200" N_GPUS = 1 TIMEOUT_SECONDS = 3600 STARTUP_TIMEOUT_SECONDS = 3300 app = modal.App("minimax-h3-deployment") h3_image = ( modal.Image.from_registry("nvidia/cuda:12.4.0-devel-ubuntu22.04") .apt_install( "git", "curl", "build-essential", "libssl-dev", "ca-certificates", "docker.io", ) .env( { "HF_HOME": "/huggingface", "HF_HUB_CACHE": "/huggingface/hub", "HF_TOKEN": HF_TOKEN, } ) .run_commands( "python -m pip install --upgrade pip", "python -m pip install torch==2.2.0 --index-url https://download.pytorch.org/whl/cu121", "python -m pip install transformers==4.40.0 accelerate==0.27.0", f"huggingface-cli download {MODEL_NAME} --local-dir /models", ) .pip_install( "fastapi[standard]", "uvicorn[standard]", "huggingface_hub", "python-multipart", "docker", ) ) hf_volume = modal.Volume.from_name("huggingface-cache", create_if_missing=True) @app.function( image=h3_image, gpu=f"{GPU_TYPE}:{N_GPUS}", volumes={"/huggingface": hf_volume}, timeout=STARTUP_TIMEOUT_SECONDS, ) def start_ghost_server(): import sys import time pid = os.fork() if pid == 0: os.setsid() os.umask(0) os.chdir('/') sys.stdout.flush() sys.stderr.flush() process_env = os.environ.copy() if HF_TOKEN and not process_env.get("HF_TOKEN"): process_env["HF_TOKEN"] = HF_TOKEN Path("/huggingface/hub").mkdir(parents=True, exist_ok=True) Path("/models").mkdir(parents=True, exist_ok=True) command = [ "uvicorn", "minimax_h3_server:app", "--host", "127.0.0.1", "--port", "8000", ] with open('/dev/null', 'w') as devnull: subprocess.Popen( command, env=process_env, stdout=devnull, stderr=devnull, start_new_session=True, ) sys.exit(0) else: time.sleep(2) return "Ghost MiniMax-H3 server initiated." @app.function( image=h3_image, gpu=f"{GPU_TYPE}:{N_GPUS}", volumes={"/huggingface": hf_volume}, timeout=STARTUP_TIMEOUT_SECONDS, ) def start_ghost_container(): client = docker.from_env() try: client.images.pull("nvidia/cuda:12.4.0-devel-ubuntu22.04") except: pass container = client.containers.run( "nvidia/cuda:12.4.0-devel-ubuntu22.04", command=[ "uvicorn", "minimax_h3_server:app", "--host", "0.0.0.0", "--port", "8000", ], detach=True, ports={'8000/tcp': 8000}, gpus=N_GPUS, environment={ "HF_TOKEN": HF_TOKEN, }, volumes={ "/huggingface": {"bind": "/huggingface", "mode": "rw"}, }, ) return f"Ghost container started! ID: {container.id}" @app.function( image=h3_image, gpu=f"{GPU_TYPE}:{N_GPUS}", volumes={"/huggingface": hf_volume}, timeout=TIMEOUT_SECONDS, ) @modal.asgi_app() def query_ghost_server(): import httpx from fastapi import FastAPI, HTTPException from starlette.requests import Request from starlette.responses import StreamingResponse from starlette.background import BackgroundTask VLLM_BASE_URL = "http://127.0.0.1:8000" fastapi_app = FastAPI() @fastapi_app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"]) async def proxy(request: Request, path: str): client = httpx.AsyncClient(base_url=VLLM_BASE_URL, timeout=30.0) try: target_url = httpx.URL( path=request.url.path, query=request.url.query.encode("utf-8"), ) request_headers = [ (name, value) for name, value in request.headers.raw if name.lower() not in { b"connection", b"keep-alive", b"proxy-authenticate", b"proxy-authorization", b"te", b"trailer", b"transfer-encoding", b"upgrade", b"host", } ] proxy_request = client.build_request( method=request.method, url=target_url, headers=request_headers, content=await request.body(), ) response = await client.send(proxy_request, stream=True) response_headers = { name.decode("latin-1"): value.decode("latin-1") for name, value in response.headers.raw if name.lower() not in { b"connection", b"keep-alive", b"proxy-authenticate", b"proxy-authorization", b"te", b"trailer", b"transfer-encoding", b"upgrade", } } return StreamingResponse( response.aiter_raw(), status_code=response.status_code, headers=response_headers, background=BackgroundTask(response.aclose), ) except httpx.RequestError as error: raise HTTPException( status_code=502, detail=f"Ghost MiniMax-H3 server request failed: {error}", ) finally: await client.aclose() return fastapi_app @app.function( image=h3_image, gpu=f"{GPU_TYPE}:{N_GPUS}", volumes={"/huggingface": hf_volume}, timeout=TIMEOUT_SECONDS, ) @modal.asgi_app() def minimax_h3_api(): import torch from fastapi import FastAPI, HTTPException from starlette.requests import Request from transformers import AutoModelForCausalLM, AutoProcessor from huggingface_hub import login if HF_TOKEN: login(token=HF_TOKEN) device = "cuda" if torch.cuda.is_available() else "cpu" model = AutoModelForCausalLM.from_pretrained( "/models", torch_dtype=torch.bfloat16, trust_remote_code=True, ).to(device) processor = AutoProcessor.from_pretrained("/models") fastapi_app = FastAPI() @fastapi_app.post("/generate") async def generate(request: Request): try: data = await request.json() prompt = data.get("prompt", "") if not prompt: raise HTTPException(status_code=400, detail="Prompt is required") inputs = processor(text=prompt, return_tensors="pt").to(device) outputs = model.generate(**inputs, max_new_tokens=256) generated_text = processor.decode(outputs[0], skip_special_tokens=True) return {"generated_text": generated_text} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @fastapi_app.post("/generate_video") async def generate_video(request: Request): try: data = await request.json() prompt = data.get("prompt", "") if not prompt: raise HTTPException(status_code=400, detail="Prompt is required") inputs = processor(text=prompt, return_tensors="pt").to(device) outputs = model.generate(**inputs, max_new_tokens=256, do_sample=True) video_data = processor.decode_video(outputs[0]) return {"video_data": video_data} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) return fastapi_app Send back the complete code with all the fixes. Fix each of the listed errors one by one, making sure to actually correct them so that there are 0 errors remaining. Keep the original imports, since the files exist. Write out every single character; do not abbreviate anything. Fix every error. There must be exactly one file. Do not write anything else; just output the complete code, and it must not contain any comments. Never, under any circumstances, use simplified, substitute, dummy, simulated, or fake code. Write the entire file as complete, unabridged, production-ready code in a single code block. It must be 100% error-free, a complete, error-free file, and must be submitted as a downloadable file. These requirements are mandatory and must be strictly adhered to. If no list of errors is provided, you must find all the errors and fix them. If there were comments in the original code, delete them. And most importantly: YOU MUST NEVER SIMPLIFY!