Grok API Pool 部署指南

# Ubuntu 22.04+ / Python 3.10+
sudo apt update && sudo apt install -y python3.11 python3.11-venv curl jq
python3.11 -m venv /opt/litellm/venv
source /opt/litellm/venv/bin/activate
pip install -U "litellm[proxy]"

# ── 模型路由表：同一个对外名 grok-main 挂多个 Key ──
model_list:
  - model_name: grok-main          # 对外暴露名，客户端请求用这个
    litellm_params:
      model: grok/grok-3           # 真实上游模型（xAI 命名，看官方 docs）
      api_key: os.environ/GROK_API_KEY_1
      rpm: 60                      # 这个 Key 每分钟上限，加权轮询依据
    model_info:
      tier: primary
  - model_name: grok-main
    litellm_params:
      model: grok/grok-3
      api_key: os.environ/GROK_API_KEY_2
      rpm: 60
    model_info:
      tier: primary
  - model_name: grok-main
    litellm_params:
      model: grok/grok-3
      api_key: os.environ/GROK_API_KEY_3
      rpm: 30
    model_info:
      tier: primary
  - model_name: grok-fallback      # 兜底模型：主模型全挂时用
    litellm_params:
      model: grok/grok-3-mini
      api_key: os.environ/GROK_API_KEY_1

# ── Router 核心：轮询 + 熔断 + 重试 + fallback ──
router_settings:
  routing_strategy: simple-shuffle # 轮询分派（或 least-busy / weighted）
  enable_pre_call_checks: true     # 发送前先查配额，减少无效请求
  cooldown_time: 60                # 熔断后冷却秒数（M 分钟恢复）
  allowed_fails: 3                 # 连续失败 N 次触发熔断
  num_retries: 2                   # 失败自动换 Key 重试次数
  request_timeout: 60              # 单请求超时（秒）
  fallbacks: [{"grok-main": ["grok-fallback"]}]   # 全部 Key 挂晒→兜底模型

litellm_settings:
  drop_params: true
  set_verbose: true
  telemetry: false                 # 关闭遥测，隐私友好

# 三个独立账号的 Grok API Key（不要用同一个账号开的多个 Key！原因见第 7 节）
GROK_API_KEY_1=YOUR_GROK_API_KEY_1
GROK_API_KEY_2=YOUR_GROK_API_KEY_2
GROK_API_KEY_3=YOUR_GROK_API_KEY_3

# 保护 Pool 入口的 Master Key（客户端请求时要带）
LITELLM_MASTER_KEY=YOUR_MASTER_KEY

[Unit]
Description=LiteLLM Proxy (Grok API Pool)
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/litellm
EnvironmentFile=/opt/litellm/.env
ExecStart=/opt/litellm/venv/bin/litellm --config /opt/litellm/config.yaml --port 4000
Restart=always
RestartSec=5
NoNewPrivileges=true

[Install]
WantedBy=multi-user.target

sudo chown -R www-data:www-data /opt/litellm
sudo systemctl daemon-reload
sudo systemctl enable --now litellm
sudo systemctl status litellm --no-pager

# 1) 健康检查
curl -s http://127.0.0.1:4000/health/liveliness

# 2) 连续发 5 次请求，看日志确认 Key 有轮换
for i in 1 2 3 4 5; do
  curl -s http://127.0.0.1:4000/v1/chat/completions \
    -H "Authorization: Bearer YOUR_MASTER_KEY" \
    -H "Content-Type: application/json" \
    -d '{"model":"grok-main","messages":[{"role":"user","content":"hi"}],"max_tokens":10}'
  echo
done

# 3) 看日志：留意每个请求用了哪个 Key（provider/API Key 尾几位）
journalctl -u litellm -f --no-pager

# Ubuntu / Debian 官方源
sudo apt install -y openresty   # 或者按 openresty.org 加官方源

# 检查 Lua 模块
openresty -V 2>&1 | grep -o 'lua-nginx-module'

# Grok API Pool —— Nginx + Lua 轻量网关
# 部署前替换: YOUR_GROK_API_KEY_1..3 / YOUR_POOL_ENTRY_KEY / your-domain.com

lua_shared_dict grok_pool 1m;   # 跨 worker 共享熔断状态

init_by_lua_block {
    -- Key 池（占位符，部署前替换）
    KEYS = { "YOUR_GROK_API_KEY_1", "YOUR_GROK_API_KEY_2", "YOUR_GROK_API_KEY_3" }
    RR = 0              -- round-robin 游标
    MAX_FAILS = 3       -- 连续失败熔断阈值
    COOLDOWN_SEC = 60   -- 冷却秒数
}

upstream grok_backend {
    server api.x.ai:443;
    keepalive 16;       -- 连接复用，降低握手开销
}

server {
    listen 443 ssl;
    server_name your-domain.com;

    ssl_certificate     /etc/letsencrypt/live/your-domain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/your-domain.com/privkey.pem;

    location /v1/ {
        rewrite_by_lua_block {
            -- 1) 校验调用方 Key（保护 Pool 入口）
            if ngx.var.http_authorization ~= "Bearer YOUR_POOL_ENTRY_KEY" then
                return ngx.exit(ngx.HTTP_UNAUTHORIZED)
            end

            -- 2) 轮询选 Key，跳过冷却中的；全部冷却则取第一个硬试
            local d = ngx.shared.grok_pool
            local now = os.time()
            local chosen = KEYS[1]
            ngx.ctx.key_idx = 1
            for i = 1, #KEYS do
                local idx = (RR % #KEYS) + 1
                RR = RR + 1
                local cooldown = d:get("cd:" .. idx) or 0
                if cooldown <= now then
                    chosen = KEYS[idx]
                    ngx.ctx.key_idx = idx
                    break
                end
            end

            -- 3) 注入上游要用的 Key（覆盖调用方传的）
            ngx.req.set_header("Authorization", "Bearer " .. chosen)
        }

        proxy_pass https://grok_backend/v1/;
        proxy_set_header Host api.x.ai;
        proxy_ssl_name api.x.ai;
        proxy_ssl_server_name on;
        proxy_buffering off;        -- 流式响应必须关缓冲
        proxy_read_timeout 300s;
        proxy_connect_timeout 10s;

        header_filter_by_lua_block {
            -- 4) 429/401/503 → 记失败；连续 MAX_FAILS 次 → 冷却 COOLDOWN_SEC
            local status = ngx.status
            if status == 429 or status == 401 or status == 503 then
                local idx = ngx.ctx.key_idx
                if idx then
                    local d = ngx.shared.grok_pool
                    local fails = (d:get("f:" .. idx) or 0) + 1
                    if fails >= MAX_FAILS then
                        d:set("cd:" .. idx, os.time() + COOLDOWN_SEC)
                        d:set("f:" .. idx, 0)
                    else
                        d:set("f:" .. idx, fails)
                    end
                end
            end
        }
    }
}

# 语法检查 + 重载
openresty -t
sudo systemctl reload openresty

# 测试：带入口 Key 请求
curl -s https://your-domain.com/v1/chat/completions \
  -H "Authorization: Bearer YOUR_POOL_ENTRY_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"grok-3","messages":[{"role":"user","content":"hi"}],"max_tokens":10}'

# gateway.py —— Grok API Pool 网关（FastAPI + httpx 异步）
# 启动: uvicorn gateway:app --host 0.0.0.0 --port 8000
# 部署前替换: 环境变量 GROK_KEYS 填真实 Key，逗号分隔
import asyncio
import os
import time

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse

# ── 配置（全部走环境变量，不要硬编码）──
API_KEYS = os.getenv("GROK_KEYS", "").split(",")     # 多个 Key 逗号分隔
GROK_BASE = os.getenv("GROK_BASE", "https://api.x.ai/v1")
MODEL = os.getenv("GROK_MODEL", "grok-3")
MAX_FAILS = int(os.getenv("MAX_FAILS", "3"))         # 熔断阈值
COOLDOWN = int(os.getenv("COOLDOWN_SEC", "60"))      # 冷却秒数
TIMEOUT = float(os.getenv("TIMEOUT_SEC", "60"))
ENTRY_KEY = os.getenv("POOL_ENTRY_KEY", "YOUR_POOL_ENTRY_KEY")

# ── Key 池状态：失败次数 + 冷却截止时间 ──
state = {k: {"fails": 0, "cooldown_until": 0.0} for k in API_KEYS}
lock = asyncio.Lock()
rr = 0                                    # round-robin 游标

app = FastAPI(title="Grok API Pool Gateway")


def _healthy(key):
    return state[key]["cooldown_until"] < time.time()


async def _pick_key():
    # 轮询 + 跳过冷却中的 Key；全部冷却就全部重试（最坏情况）
    global rr
    async with lock:
        now = time.time()
        keys = [k for k in API_KEYS if _healthy(k)]
        if not keys:
            keys = API_KEYS
        k = keys[rr % len(keys)]
        rr += 1
        return k


def _on_fail(key, reset=False):
    # 失败计数；连续 MAX_FAILS 次 → 熔断冷却
    s = state[key]
    if reset:
        s["fails"] = 0
        s["cooldown_until"] = 0.0
    else:
        s["fails"] += 1
        if s["fails"] >= MAX_FAILS:
            s["cooldown_until"] = time.time() + COOLDOWN
            s["fails"] = 0


@app.get("/v1/models")
async def models():
    return {"object": "list", "data": [{"id": MODEL, "object": "model"}]}


@app.post("/v1/chat/completions")
async def chat(request: Request):
    # 入口鉴权：调用方要带自己的 Key
    if request.headers.get("Authorization") != "Bearer " + ENTRY_KEY:
        return JSONResponse({"error": {"message": "unauthorized"}}, status_code=401)

    body = await request.json()
    headers = {"Content-Type": "application/json"}

    # 流式请求：单次直通（重试会破坏 SSE 顺序，不做）
    if body.get("stream"):
        async def gen():
            async with httpx.AsyncClient(timeout=TIMEOUT) as client:
                async with client.stream(
                    "POST", f"{GROK_BASE}/chat/completions",
                    json=body, headers=headers,
                ) as resp:
                    async for chunk in resp.aiter_bytes():
                        yield chunk

        key = await _pick_key()
        headers["Authorization"] = f"Bearer {key}"
        return StreamingResponse(gen(), media_type="text/event-stream")

    # 非流式：失败换 Key 重试，直到全部 Key 试完
    for _ in range(len(API_KEYS) + 1):
        key = await _pick_key()
        headers["Authorization"] = f"Bearer {key}"
        try:
            async with httpx.AsyncClient(timeout=TIMEOUT) as client:
                resp = await client.post(
                    f"{GROK_BASE}/chat/completions", json=body, headers=headers
                )
            if resp.status_code in (429, 401, 503):
                _on_fail(key)          # 限流/凭证错/上游挂 → 换 Key
                continue
            if resp.status_code >= 500:
                _on_fail(key)
                continue
            _on_fail(key, reset=True)  # 成功 → 重置计数
            return JSONResponse(content=resp.json(), status_code=resp.status_code)
        except httpx.TimeoutException:
            _on_fail(key)
        except Exception:
            _on_fail(key)

    return JSONResponse(
        {"error": {"message": "all keys unavailable"}}, status_code=503
    )

fastapi
uvicorn[standard]
httpx

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY gateway.py .
EXPOSE 8000
CMD ["uvicorn", "gateway:app", "--host", "0.0.0.0", "--port", "8000"]

services:
  grok-gateway:
    build: .
    ports:
      - "127.0.0.1:8000:8000"          # 只绑本机，外面经反向代理入
    environment:
      GROK_KEYS: "YOUR_GROK_API_KEY_1,YOUR_GROK_API_KEY_2,YOUR_GROK_API_KEY_3"
      GROK_BASE: "https://api.x.ai/v1"
      GROK_MODEL: "grok-3"
      POOL_ENTRY_KEY: "YOUR_POOL_ENTRY_KEY"
      MAX_FAILS: "3"
      COOLDOWN_SEC: "60"
    restart: unless-stopped

docker compose up -d --build
# 测试
curl -s http://127.0.0.1:8000/v1/chat/completions \
  -H "Authorization: Bearer YOUR_POOL_ENTRY_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"grok-3","messages":[{"role":"user","content":"hi"}],"max_tokens":10}'

server {
    listen 80;
    server_name your-domain.com;
    # 强制跳 HTTPS（certbot 会帮你写好，这里示意）
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    server_name your-domain.com;
    ssl_certificate     /etc/letsencrypt/live/your-domain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/your-domain.com/privkey.pem;

    # LiteLLM 自带 Master Key 鉴权，这里再加一层 IP 白名单（可选）
    allow <YOUR_SERVER_IP>/32;
    deny all;

    location / {
        proxy_pass http://127.0.0.1:4000;      # 指去 LiteLLM
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_buffering off;                    # 流式响应
        proxy_read_timeout 300s;
    }
}

/var/log/litellm/*.log {
    daily
    rotate 14
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}

scrape_configs:
  - job_name: litellm
    metrics_path: /metrics
    static_configs:
      - targets: ["127.0.0.1:4000"]

#!/usr/bin/env bash
# 用 curl 查 Prometheus，剩低 <20% 就推告警（占位符自行替换）
THRESHOLD=20
QUOTA=$(curl -s "http://127.0.0.1:9090/api/v1/query" \
  --data-urlencode 'query=litellm_remaining_quota' | jq -r '.data.result[0].value[1]' 2>/dev/null)
[ -z "$QUOTA" ] && exit 0
if [ "$QUOTA" -lt "$THRESHOLD" ]; then
  curl -s "https://api.telegram.org/botYOUR_BOT_TOKEN/sendMessage" \
    -d "chat_id=YOUR_CHAT_ID" \
    -d "text=⚠️ Grok Key 剩余配额得 $QUOTA%"
fi

/opt/litellm/
├── config.yaml          # 路由配置（多 Key / 熔断 / fallback）
├── .env                 # 密钥（chmod 600，不入 Git）
├── venv/                # Python 虚拟环境
└── (Docker 方案则用 compose 管理)

/etc/systemd/system/litellm.service   # systemd 单元
/etc/logrotate.d/litellm              # 日志轮替
/etc/nginx/conf.d/grok-entry.conf     # 对外入口（HTTPS + 反代）

#!/bin/bash
# 由 Hermes cron 呼叫，預設每批 5 個
exec ~/grok-pool/cron-batch.sh "${@:-5}"

(crontab -l 2>/dev/null; echo ""; echo "# Grok 帳戶自動註冊批次"; \
echo "0 4 * * * /home/YOUR_USERNAME/.hermes/scripts/grok-pool-batch.sh 5"; \
echo "0 16 * * * /home/YOUR_USERNAME/.hermes/scripts/grok-pool-batch.sh 5") | crontab -

cd /home/YOUR_USERNAME/grok-pool && ./cron-batch.sh 5

ls -t ~/grok-pool/logs/batch_*.log | head -1 | xargs tail -30

python3 /home/YOUR_USERNAME/grok-pool/reapply-picker-patch.py

systemd-run --user --on-active=90s --unit=grok-restart \
bash /home/YOUR_USERNAME/grok-pool/restart-gateway.sh

# 1) 登录拿 token
curl -X POST http://127.0.0.1:8001/api/admin/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"YOUR_ADMIN_PASSWORD"}'

# 2) GET 全量配置（拿到 revision）
# 3) 修改以下字段后再 PUT {"config": {...}, "revision": "0"}

routing:
  maxAttempts: 10        # 默认 999 —— 降级时 retry 风暴会打爆上游（429 雪崩）
  capacityWait: 1.5s     # 默认 500ms —— 账号全忙时排队代替硬撞
accounts:
  autoCleanReauthEnabled: true   # 自动清理需要重新登录的死账号

curl -X POST https://your-domain/v1/images/generations \
  -H "Authorization: Bearer YOUR_GROK_API_KEY" -H "Content-Type: application/json" -d '{
    "model": "Console/grok-imagine-image",   ← 必须 Console/ 前缀
    "prompt": "Tokyo alley, film look, 8k",
    "size": "1024x1536",                     ← 2:3 直向; 1:1=1024x1024
    "n": 1
  }'

curl -X POST https://your-domain/v1/videos/generations \
  -H "Authorization: Bearer YOUR_GROK_API_KEY" -H "Content-Type: application/json" -d '{
    "model": "Console/grok-imagine-video",
    "prompt": "woman walks then jumps, slow motion",
    "image": {"url": "data:image/jpeg;base64,"},   ← 必须 HTTPS 或 data URL
    "aspect_ratio": "2:3"                                  ← 关键! 否则固定 16:9 会压扁直向图
  }'
# 返回 {"request_id":"video_xxx"} → 轮询状态
curl "https://your-domain/v1/videos/video_xxx" -H "Authorization: Bearer YOUR_GROK_API_KEY"
# {"status":"done","video":{"url":".../v1/videos/video_xxx/content"}} → 下载
curl "https://your-domain/v1/videos/video_xxx/content" -H "Authorization: Bearer YOUR_GROK_API_KEY" -o out.mp4

用法:
  python3 grok_gen.py image "prompt" [-ar 2:3|16:9|1:1|9:16] [-o out.png]
  python3 grok_gen.py video "動作描述" [-i input.png] [-ar 2:3] [-o out.mp4]

核心邏輯（關鍵三行）:
  body = {"model": "Console/grok-imagine-image", "prompt": prompt, "size": sizes.get(ar)}
  body = {"model": "Console/grok-imagine-video", "prompt": prompt,
          "image": {"url": "data:...;base64,..."}, "aspect_ratio": ar}
  # video 下载: urllib Request 带 Authorization header → /videos/{id}/content