从少数人试用到工程平台：Airbnb 扩大前沿模型访问的启示

mkdir model-gateway && cd model-gateway
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn httpx
export UPSTREAM_API_KEY='replace-me'
export UPSTREAM_URL='https://api.example.com/v1/chat/completions'

import os
import time
import uuid
from typing import Literal

import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Engineering Model Gateway")

UPSTREAM_URL = os.getenv(
    "UPSTREAM_URL",
    "https://api.example.com/v1/chat/completions",
)
UPSTREAM_API_KEY = os.environ["UPSTREAM_API_KEY"]

MODEL_MAP = {
    "bug-analysis": "gpt-6-astra",
    "system-design": "frontier-default",
    "test-generation": "frontier-default",
}

TEAM_POLICIES = {
    "payments": {"bug-analysis", "test-generation"},
    "platform": {"bug-analysis", "system-design", "test-generation"},
}

class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str = Field(min_length=1, max_length=20_000)

class GatewayRequest(BaseModel):
    task: Literal["bug-analysis", "system-design", "test-generation"]
    messages: list[Message] = Field(min_length=1, max_length=30)

@app.post("/v1/engineering-assistant")
async def engineering_assistant(
    request: GatewayRequest,
    x_team: str = Header(...),
):
    allowed_tasks = TEAM_POLICIES.get(x_team, set())
    if request.task not in allowed_tasks:
        raise HTTPException(status_code=403, detail="Task is not allowed for this team")

    request_id = str(uuid.uuid4())
    model = MODEL_MAP[request.task]
    started = time.perf_counter()

    payload = {
        "model": model,
        "messages": [message.model_dump() for message in request.messages],
        "temperature": 0.2,
    }

    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            UPSTREAM_URL,
            headers={
                "Authorization": f"Bearer {UPSTREAM_API_KEY}",
                "Content-Type": "application/json",
                "X-Request-ID": request_id,
            },
            json=payload,
        )

    latency_ms = round((time.perf_counter() - started) * 1000)
    print({
        "request_id": request_id,
        "team": x_team,
        "task": request.task,
        "model": model,
        "status": response.status_code,
        "latency_ms": latency_ms,
    })

    if response.is_error:
        raise HTTPException(status_code=502, detail="Upstream model request failed")

    return {
        "request_id": request_id,
        "model": model,
        "result": response.json(),
    }

uvicorn app:app --host 127.0.0.1 --port 8000

curl http://127.0.0.1:8000/v1/engineering-assistant \
  -H 'Content-Type: application/json' \
  -H 'X-Team: platform' \
  -d '{
    "task": "system-design",
    "messages": [
      {
        "role": "system",
        "content": "Act as a reviewer. List assumptions, failure modes, and open questions. Do not invent metrics."
      },
      {
        "role": "user",
        "content": "Review a plan to move an image-processing job from a request handler to an asynchronous queue."
      }
    ]
  }'