LLM Gateway 架构：控制面、数据面与生产边界

{
  "request_id": "req_01",
  "tenant_id": "tenant_red",
  "config_version": "2026-08-09.3",
  "virtual_model": "support-chat",
  "backend_id": "provider_a_chat",
  "attempt": 1,
  "status": "completed",
  "input_units": 1840,
  "output_units": 276,
  "usage_source": "provider_response",
  "price_catalog_version": "catalog_42",
  "estimated_cost": "0.000000",
  "currency": "USD"
}

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


def require(condition: bool, message: str, errors: list[str]) -> None:
    if not condition:
        errors.append(message)


def string_set(value: Any, field: str, errors: list[str]) -> set[str]:
    if not isinstance(value, list) or any(
        not isinstance(item, str) or not item for item in value
    ):
        errors.append(f"{field} must be a list of non-empty strings")
        return set()
    return set(value)


def validate(policy: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    require(
        isinstance(policy.get("config_version"), str)
        and bool(policy["config_version"]),
        "config_version must be a non-empty string",
        errors,
    )

    backend_rows = policy.get("backends")
    route_rows = policy.get("routes")
    tenant_rows = policy.get("tenants")
    require(isinstance(backend_rows, list), "backends must be a list", errors)
    require(isinstance(route_rows, list), "routes must be a list", errors)
    require(isinstance(tenant_rows, list), "tenants must be a list", errors)
    if not all(isinstance(rows, list) for rows in (
        backend_rows, route_rows, tenant_rows
    )):
        return errors

    backends: dict[str, set[str]] = {}
    for index, row in enumerate(backend_rows):
        if not isinstance(row, dict):
            errors.append(f"backends[{index}] must be an object")
            continue
        backend_id = row.get("id")
        if not isinstance(backend_id, str) or not backend_id:
            errors.append(f"backends[{index}].id must be non-empty")
            continue
        require(
            backend_id not in backends,
            f"duplicate backend id: {backend_id}",
            errors,
        )
        backends[backend_id] = string_set(
            row.get("capabilities"),
            f"backend {backend_id} capabilities",
            errors,
        )
        require(
            isinstance(row.get("secret_ref"), str)
            and bool(row["secret_ref"]),
            f"backend {backend_id} must use secret_ref",
            errors,
        )

    aliases: set[str] = set()
    for index, route in enumerate(route_rows):
        if not isinstance(route, dict):
            errors.append(f"routes[{index}] must be an object")
            continue
        alias = route.get("alias")
        if not isinstance(alias, str) or not alias:
            errors.append(f"routes[{index}].alias must be non-empty")
            continue
        require(alias not in aliases, f"duplicate route alias: {alias}", errors)
        aliases.add(alias)
        required = string_set(
            route.get("required_capabilities"),
            f"route {alias} required_capabilities",
            errors,
        )
        candidates = string_set(
            route.get("candidates"), f"route {alias} candidates", errors
        )
        require(bool(candidates), f"route {alias} has no candidates", errors)
        for backend_id in candidates:
            require(
                backend_id in backends,
                f"route {alias} references unknown backend {backend_id}",
                errors,
            )
            if backend_id in backends:
                missing = required - backends[backend_id]
                require(
                    not missing,
                    f"route {alias} backend {backend_id} lacks "
                    f"{sorted(missing)}",
                    errors,
                )

        retry = route.get("retry")
        if not isinstance(retry, dict):
            errors.append(f"route {alias} retry must be an object")
        else:
            attempts = retry.get("max_attempts")
            require(
                isinstance(attempts, int) and not isinstance(attempts, bool)
                and 1 <= attempts <= 3,
                f"route {alias} max_attempts must be between 1 and 3",
                errors,
            )
            require(
                retry.get("after_downstream_started") is False,
                f"route {alias} must not retry after streaming starts",
                errors,
            )
            ratio = retry.get("budget_ratio")
            require(
                isinstance(ratio, (int, float))
                and not isinstance(ratio, bool)
                and 0 <= ratio <= 1,
                f"route {alias} retry budget_ratio must be in [0, 1]",
                errors,
            )

    for index, tenant in enumerate(tenant_rows):
        if not isinstance(tenant, dict):
            errors.append(f"tenants[{index}] must be an object")
            continue
        tenant_id = tenant.get("id")
        require(
            isinstance(tenant_id, str) and bool(tenant_id),
            f"tenants[{index}].id must be non-empty",
            errors,
        )
        require(
            isinstance(tenant.get("cache_namespace"), str)
            and bool(tenant["cache_namespace"]),
            f"tenant {tenant_id!r} needs an isolated cache_namespace",
            errors,
        )
        require(
            tenant.get("on_budget_exhausted") in {"reject", "manual_approval"},
            f"tenant {tenant_id!r} must reject or require approval "
            "when budget is exhausted",
            errors,
        )

    telemetry = policy.get("telemetry")
    require(isinstance(telemetry, dict), "telemetry must be an object", errors)
    if isinstance(telemetry, dict):
        require(
            telemetry.get("log_prompt_body") is False,
            "telemetry.log_prompt_body must default to false",
            errors,
        )
    return errors


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("policy", type=Path)
    args = parser.parse_args()
    try:
        value = json.loads(args.policy.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        parser.error(str(error))
    if not isinstance(value, dict):
        parser.error("policy root must be an object")
    errors = validate(value)
    if errors:
        for error in errors:
            print(f"ERROR: {error}")
        raise SystemExit(1)
    print(f"valid policy: {args.policy}")


if __name__ == "__main__":
    main()