FastAPI+Azure构建生产级机器学习API服务

1. 项目概述：为什么一个轻量级 API 封装，比“直接跑模型”重要十倍

我第一次在生产环境里把训练好的 PyTorch 模型扔进一个 Flask 脚本里，用app.run(host='0.0.0.0', port=5000)启动，然后发微信给测试同事：“好了，接口地址是 http://192.168.1.100:5000/predict，你试试。”——结果他刚发了三轮并发请求，服务器内存就飙到 98%，top里看到 Python 进程占满 CPU，日志里全是Killed process。那不是部署，那是给服务器下战书。

这件事过去三年，我现在带团队做 MLOps 支撑时，第一课永远讲同一件事：模型本身不等于服务，服务的核心价值从来不在“能跑通”，而在于“可控、可测、可扩、可维”。FastAPI 不是又一个 Web 框架的跟风选择，它是为现代机器学习服务量身定制的“操作系统内核”——它原生支持异步 I/O、自动 OpenAPI 文档、类型驱动的请求校验、依赖注入机制，这些不是锦上添花的功能点，而是解决真实痛点的工程杠杆。比如，当你的模型推理需要调用外部向量数据库、触发异步特征计算、或对输入图像做预处理时，同步阻塞式框架（如老版本 Flask）会把整个线程卡死，而 FastAPI 的async def可以让 IO 等待期间释放线程去处理其他请求，实测在中等负载下 QPS 提升 3.2 倍，P95 延迟下降 67%。

Azure 则提供了从底层到顶层的“确定性交付链”：不是让你自己配 Ubuntu、装 Docker、调 Nginx 反向代理、写 systemd 服务脚本、再手动轮换 TLS 证书——而是用 Azure Container Apps 或 Azure App Service，把“模型 + API + 配置 + 扩缩策略 + 日志 + 监控”打包成一个声明式单元。我去年上线一个金融风控评分模型，从本地验证完模型、写好 FastAPI 接口、构建 Docker 镜像，到在 Azure 上完成部署、配置自动扩缩（CPU > 70% 时自动加 1 实例）、接入 Application Insights 做端到端追踪，全程只用了 47 分钟。这背后不是魔法，是 Azure 把基础设施的“不确定性”全部封装掉了，你只需要专注两件事：模型逻辑是否正确，API 接口是否健壮。

所以，当你看到标题里“Deploying ML Models with FastAPI and Azure”，别把它当成一个技术组合的罗列。它是一套经过千次线上事故淬炼出的最小可行服务范式：FastAPI 是你的服务大脑（负责定义契约、校验输入、编排流程），Azure 是你的服务躯干（负责承载、调度、观测、保障）。关键词里的 “Towards AI - Medium” 提示我们，这个实践不是闭门造车的理论推演，而是来自一线工程师在真实数据产品中踩坑、复盘、沉淀下来的路径。它适合三类人：刚训完模型不知如何落地的算法同学；被业务方催着“明天就要上线”的后端/运维同学；以及想把离线实验快速转为线上能力的数据产品负责人。接下来，我会带你从零开始，把一个.pkl模型文件，变成一个带健康检查、输入校验、错误追踪、自动扩缩的生产级 API 服务——每一步都附带我亲手验证过的参数、命令和避坑提示。

2. 整体架构设计与关键决策解析

2.1 为什么选 FastAPI 而非 Flask/Django/Fastify？

这个问题我被问过至少 37 次，答案从来不是“因为新”或“因为快”，而是四个不可替代的工程事实：

第一，类型系统即契约，省掉 80% 的手动校验代码。
假设你的模型需要接收一个包含user_id（字符串）、transaction_amount（浮点数）、is_weekend（布尔值）的 JSON 请求体。在 Flask 中，你得写：

@app.route('/predict', methods=['POST']) def predict(): data = request.get_json() if not isinstance(data.get('user_id'), str): return jsonify({'error': 'user_id must be string'}), 400 if not isinstance(data.get('transaction_amount'), (int, float)): return jsonify({'error': 'amount must be number'}), 400 # ... 还有 is_weekend 校验、缺失字段检查、范围限制……

而在 FastAPI 中，你只需定义 Pydantic 模型：

from pydantic import BaseModel class PredictionRequest(BaseModel): user_id: str transaction_amount: float is_weekend: bool @app.post("/predict") def predict(request: PredictionRequest): # request 已经是完全校验、类型转换、缺失字段报错的干净对象 # 无需任何 if-else 校验逻辑

实测下来，一个中等复杂度的预测接口，FastAPI 能帮你省掉 120+ 行脆弱的手动校验代码。更重要的是，这个PredictionRequest类型会自动生成 OpenAPI Schema，前端、测试工具、Postman 都能直接导入，契约从文档变成了可执行的代码。

第二，异步原生支持，直击模型服务的 IO 瓶颈。
很多模型推理本身是 CPU 密集型（如 ResNet 图像分类），但实际生产中，90% 的延迟并不来自模型计算，而是来自 IO：加载大模型权重（GB 级）、读取特征存储（Redis/MongoDB）、调用外部风控规则引擎、写入审计日志。FastAPI 的async def允许你在等待 IO 时释放事件循环：

@app.post("/predict") async def predict(request: PredictionRequest): # 步骤1：异步加载特征（假设特征服务提供 async API） features = await fetch_user_features_async(request.user_id) # 步骤2：同步执行模型推理（CPU-bound，需用 run_in_executor 避免阻塞） result = await loop.run_in_executor(None, model.predict, features) # 步骤3：异步写入日志 await log_prediction_async(request, result) return {"score": result}

对比 Flask 的同步模型，同样的 100 并发请求，FastAPI 的平均响应时间从 1.2s 降到 0.38s，且不会因 IO 等待导致连接池耗尽。

第三，依赖注入机制，让测试和替换变得像换电池一样简单。
模型服务最怕“硬编码依赖”。比如，你把model = joblib.load('model.pkl')写死在路由函数里，测试时想 mock 模型？得 patch 全局变量。而 FastAPI 的依赖注入让你把模型加载抽象成一个可插拔组件：

from fastapi import Depends # 定义依赖：模型加载器 async def get_model(): # 这里可以是 joblib.load, torch.load, 或从 Azure Blob Storage 动态拉取 return load_model_from_cache() @app.post("/predict") def predict(request: PredictionRequest, model = Depends(get_model)): return model.predict(request)

上线时，get_model()从本地磁盘加载；A/B 测试时，你可以注入一个返回 mock 结果的get_mock_model()；灰度发布时，甚至可以注入一个根据request.user_id % 100决定走新旧模型的路由依赖。这种解耦，是服务长期可维护性的基石。

第四，开箱即用的生产就绪特性。

• /docs和/redoc自动生成交互式 API 文档，无需额外配置 Swagger UI；
• /health健康检查端点（需手动加，但一行代码搞定）；
• 请求/响应日志、异常捕获中间件、CORS 配置全部内置；
• 与 Uvicorn（ASGI 服务器）深度集成，启动命令简洁：uvicorn main:app --reload --host 0.0.0.0 --port 8000。

提示：Django 太重（ORM、Admin、模板系统对纯 API 是累赘），Flask 太轻（缺类型、缺异步、缺依赖注入），Node.js 的 Fastify 虽快但 Python 生态（PyTorch/TensorFlow/scikit-learn）的成熟度和算法同学的熟悉度，决定了 Python 栈仍是 ML 服务的首选。FastAPI 是那个“刚刚好”的平衡点。

2.2 为什么选 Azure 而非 AWS/GCP/自建 K8s？

选择云平台，核心看三个维度：交付速度、运维负担、生态契合度。我们逐一对比：

具体到本次项目，我们选择Azure Container Apps而非 App Service，原因很务实：

• App Service是 PaaS，适合传统 Web 应用，但对容器化 ML 服务支持较弱（如无法指定 GPU 节点、GPU 实例价格高且不灵活）；
• Container Apps是 Serverless 容器平台，它把 Kubernetes 的强大能力（自动扩缩、流量切分、Dapr 集成）封装成极简的 YAML 配置，同时保留容器的灵活性。你不需要懂 K8s 的 Pod、Service、Ingress，只需告诉它：“我的镜像是myregistry.azurecr.io/ml-api:v1.2，CPU 限 1 核，内存限 2GB，HTTP 端口 8000，健康检查路径/health”。

更关键的是，Container Apps 的自动扩缩（Scale Rules）直接支持基于 Prometheus 指标（如 CPU 使用率、HTTP 请求延迟）或基于 Kafka/Event Hubs 的事件驱动扩缩。我们曾用它实现“交易高峰时段自动扩容至 10 实例，凌晨自动缩容至 1 实例”，月度计算成本比固定 4 实例方案降低 42%。

注意：Azure 不是“必须选”，而是“当前阶段最优解”。如果你的团队已深度绑定 AWS（如大量使用 Lambda、Step Functions），那用 ECS Fargate + ALB 也是成熟路径。但对从零开始、追求快速验证、且团队熟悉 Python 的场景，Azure Container Apps 的学习曲线最平缓，试错成本最低。

2.3 整体架构图：从本地代码到云端服务的完整链路

整个部署流程不是“写完代码 → 上传 → 完事”，而是一个闭环的工程流水线。下图描述了我们实际采用的架构（文字版，避免 Mermaid）：

1. 开发侧（Local）：

   • 代码仓库：GitHub/GitLab，含main.py（FastAPI 入口）、model_loader.py（模型加载逻辑）、requirements.txt、Dockerfile；
   • 本地测试：用uvicorn main:app --reload启动，Postman 测试/predict和/health；
   • 模型管理：模型文件（.pkl/.pt）不放入 Git，而是上传至 Azure Blob Storage 或 Azure Machine Learning Model Registry。

2. 构建侧（CI Pipeline）：

   • 触发：Git Push 到main分支；
   • 步骤：安装依赖 → 运行单元测试（测试模型加载、预测逻辑）→ 构建 Docker 镜像 → 推送镜像至 Azure Container Registry（ACR）；
   • 关键配置：Dockerfile 必须使用多阶段构建（multi-stage build），基础镜像选tiangolo/uvicorn-gunicorn-fastapi:python3.9，它已预装 Uvicorn、Gunicorn、FastAPI，体积仅 320MB，比从python:3.9-slim自建小 60%。

3. 部署侧（CD Pipeline / Azure Portal）：

   • 创建 Container App 环境（Environment）：这是 Container Apps 的命名空间，包含网络、日志、监控配置；
   • 创建 Container App：指定 ACR 镜像、端口、环境变量（如MODEL_STORAGE_URL）、健康检查路径；
   • 配置扩缩规则：例如，“当 CPU 平均使用率 > 70% 持续 2 分钟，增加 1 个实例，最多 10 个”；
   • 配置 HTTPS：Azure 自动签发并续期 Let's Encrypt 证书，无需手动操作。

4. 运行侧（Production）：

   • 流量入口：https://ml-api-prod.australiaeast.azurecontainerapps.io；
   • 监控：Application Insights 自动采集 HTTP 请求、依赖调用（如 Redis）、异常、性能计数器；
   • 日志：所有print()和logger.info()输出自动流入 Log Analytics，支持 KQL 查询（如requests | where timestamp > ago(1h) | summarize avg(duration), count() by resultCode）；
   • 运维：通过 Azure CLI 一键重启、查看日志、调整实例数（az containerapp update --name ml-api --resource-group rg-ml --min-replicas 1 --max-replicas 5）。

这个架构的价值在于：每个环节都可独立验证、可回滚、可监控。如果部署后发现预测错误，你可以立刻：

• 查看 Application Insights 的失败请求详情（哪个字段校验失败？）；
• 在 Log Analytics 中搜索ERROR日志定位模型加载异常；
• 用az containerapp revision list查看历史版本，用az containerapp revision activate一键回滚到上一版。

3. 核心细节解析与实操要点

3.1 FastAPI 服务代码：不只是“写个 predict 函数”

一个能上生产的 FastAPI 服务，远不止@app.post("/predict")这一行。它必须包含健壮性、可观测性、可维护性的骨架。以下是我们的标准模板（已脱敏，可直接复用）：

# main.py from fastapi import FastAPI, HTTPException, Depends, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any import logging import time import asyncio from contextlib import asynccontextmanager # 配置日志（关键！所有日志必须结构化） logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) logger = logging.getLogger(__name__) # 模型加载依赖（异步，避免阻塞启动） async def get_model(): logger.info("Loading model from storage...") # 实际代码：从 Azure Blob 或本地挂载卷加载 # model = load_model_from_azure_blob("models/risk-v2.pt") # return model return MockModel() # 占位符，下文详解 # 健康检查模型 class HealthResponse(BaseModel): status: str = "ok" timestamp: float = Field(default_factory=time.time) uptime_seconds: float = 0.0 # 预测请求模型（严格定义输入契约） class PredictionRequest(BaseModel): user_id: str = Field(..., min_length=5, max_length=32, description="用户唯一标识") transaction_amount: float = Field(..., gt=0.0, le=1000000.0, description="交易金额，单位：元") is_weekend: bool = Field(default=False, description="是否为周末") # 可扩展字段，如 device_fingerprint, ip_address 等 metadata: Optional[Dict[str, Any]] = Field(default=None, description="附加元数据") # 预测响应模型（定义输出契约） class PredictionResponse(BaseModel): score: float = Field(..., ge=0.0, le=1.0, description="风险评分，0~1") risk_level: str = Field(..., description="风险等级：low/medium/high") explanation: str = Field(..., description="评分依据简述") request_id: str = Field(..., description="本次请求唯一ID") # 全局应用实例 app = FastAPI( title="Risk Scoring API", description="实时风控评分服务", version="1.2.0", docs_url="/docs", # Swagger UI redoc_url="/redoc", # ReDoc UI ) # CORS 配置（生产环境务必限制 origin） app.add_middleware( CORSMiddleware, allow_origins=["https://your-frontend.com"], # 替换为实际前端域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 自定义异常处理器（统一错误格式） @app.exception_handler(HTTPException) async def http_exception_handler(request, exc): logger.error(f"HTTP Exception: {exc.status_code} - {exc.detail}") return JSONResponse( status_code=exc.status_code, content={"error": {"code": exc.status_code, "message": exc.detail}}, ) # 健康检查端点（必须！K8s/Container Apps 依赖此判断实例是否就绪） @app.get("/health", response_model=HealthResponse) async def health_check(): return HealthResponse( status="ok", uptime_seconds=time.time() - app.state.start_time ) # 主预测端点 @app.post("/predict", response_model=PredictionResponse) async def predict( request: PredictionRequest, model = Depends(get_model) # 依赖注入模型 ): start_time = time.time() try: # 记录请求日志（结构化，含 trace_id） logger.info( "Prediction request started", extra={ "user_id": request.user_id, "amount": request.transaction_amount, "is_weekend": request.is_weekend, "trace_id": request.metadata.get("trace_id", "N/A") if request.metadata else "N/A" } ) # 模型预测（此处应为 CPU 密集型，用 run_in_executor 避免阻塞） loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, model.predict, request.user_id, request.transaction_amount, request.is_weekend ) # 计算耗时并记录 duration_ms = (time.time() - start_time) * 1000 logger.info( "Prediction completed successfully", extra={ "score": result["score"], "risk_level": result["risk_level"], "duration_ms": round(duration_ms, 2), "user_id": request.user_id } ) return PredictionResponse(**result) except ValueError as e: logger.error(f"Validation error in prediction: {e}", exc_info=True) raise HTTPException(status_code=400, detail=f"Input validation failed: {str(e)}") except Exception as e: logger.error(f"Unexpected error during prediction: {e}", exc_info=True) raise HTTPException(status_code=500, detail="Internal server error") # 应用生命周期管理（启动时记录时间） @app.on_event("startup") async def startup_event(): app.state.start_time = time.time() logger.info("Application startup complete") # 模拟模型类（实际替换为你的模型） class MockModel: def predict(self, user_id: str, amount: float, is_weekend: bool) -> Dict: # 模拟耗时计算（真实模型会更长） time.sleep(0.05) score = min(0.95, 0.1 + (amount / 10000) + (0.2 if is_weekend else 0.0)) risk_level = "high" if score > 0.7 else "medium" if score > 0.3 else "low" return { "score": round(score, 4), "risk_level": risk_level, "explanation": f"Base on amount ({amount}) and weekend flag ({is_weekend})", "request_id": f"req-{int(time.time())}-{user_id[:4]}" }

关键细节说明：

• 结构化日志：logger.info(..., extra={...})是核心。Azure Application Insights 会自动解析extra字典中的字段，生成可查询的结构化日志。不要用logger.info(f"user_id={user_id}, amount={amount}")，那只是字符串，无法做聚合分析。
• 健康检查/health：必须返回200 OK且响应体包含有意义的状态（如uptime_seconds）。Container Apps 会每 10 秒调用一次，连续 3 次失败则标记实例为不健康并重启。
• run_in_executor的必要性：FastAPI 的事件循环是单线程的。如果model.predict()是纯 Python 计算（如 sklearn 的predict），它会阻塞整个事件循环，导致其他请求排队。run_in_executor把它扔进线程池执行，释放事件循环。对于 PyTorch/TensorFlow，它们内部已做异步优化，但保险起见仍建议包裹。
• 异常处理器：统一将HTTPException转为标准 JSON 错误格式，前端无需处理多种错误结构。
• Field的约束：min_length,gt,ge等不是装饰，而是 Pydantic 的运行时校验。它会在请求解析阶段就拦截非法输入，避免进入业务逻辑。

实操心得：我在一个电商推荐服务中，曾忽略Field(..., min_length=1)，导致前端传空字符串""的user_id，模型加载时抛出KeyError，错误日志里只有KeyError: ''，花了 2 小时才定位。加上min_length=1后，错误直接变成422 Unprocessable Entity，明确提示"user_id field required"。防御性编程的第一道防线，永远在输入校验层。

3.2 Dockerfile 编写：小体积、快启动、稳运行

Dockerfile 不是“能跑就行”，它直接影响部署速度、内存占用、安全扫描结果。我们采用多阶段构建，最终镜像仅含运行时必需项：

# 构建阶段 FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9 AS builder # 设置工作目录 WORKDIR /app # 复制依赖文件（先于代码，利用 Docker 缓存） COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制源码 COPY . . # 运行时阶段 FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9 # 创建非 root 用户（安全最佳实践） RUN addgroup -g 1001 -f appgroup && adduser -S appuser -u 1001 # 复制构建阶段安装的依赖和代码 COPY --from=builder /app /app COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages # 切换到非 root 用户 USER appuser # 暴露端口 EXPOSE 8000 # 启动命令（覆盖基础镜像默认命令，确保使用 Gunicorn + Uvicorn） CMD exec gunicorn --bind :8000 --workers 2 --worker-class uvicorn.workers.UvicornWorker --timeout 120 --max-requests 1000 --max-requests-jitter 100 --access-logfile '-' --error-logfile '-' main:app

requirements.txt示例（精简、锁定版本）：

fastapi==0.104.1 uvicorn[standard]==0.23.2 pydantic==2.4.2 numpy==1.24.4 scikit-learn==1.3.0 joblib==1.3.2 azure-storage-blob==12.18.3 opentelemetry-api==1.22.0 opentelemetry-sdk==1.22.0 opentelemetry-instrumentation-fastapi==0.42b0

关键点解析：

• 基础镜像选择：tiangolo/uvicorn-gunicorn-fastapi是 FastAPI 官方维护的生产就绪镜像，已预装 Gunicorn（进程管理）、Uvicorn（ASGI 服务器）、FastAPI，且针对 Alpine Linux 优化，体积小、漏洞少。不要用python:3.9-slim从头装，那会多出 200MB 且需手动调优。
• 多阶段构建：--from=builder只复制最终需要的文件和包，不包含构建时的临时文件、缓存、dev 依赖（如 pytest），镜像体积从 1.2GB 降到 320MB。
• 非 root 用户：adduser -S appuser创建系统用户，USER appuser切换，避免容器以 root 运行带来的安全风险（如 CVE-2022-29154）。Azure Container Apps 强制要求非 root 用户。
• Gunicorn 配置：--workers 2是经验公式：2 * (CPU 核数) + 1，对于 1 核容器，2 个 worker 最佳；--timeout 120防止长请求拖垮服务；--max-requests 1000强制 worker 重启，避免内存泄漏累积。
• OpenTelemetry 集成：opentelemetry-instrumentation-fastapi会自动为所有 FastAPI 路由添加分布式追踪，无需修改业务代码，Application Insights 即可看到端到端调用链。

注意事项：模型文件（.pkl/.pt）绝不COPY 进 Docker 镜像！原因有三：

1. 镜像体积暴增（GB 级），每次模型更新都要重建、推送 GB 级镜像，CI/CD 流水线卡死；
2. 模型版本与代码版本强耦合，无法独立灰度（如代码 v1.2 部署，但想先用模型 v1.1 测试）；
3. 安全风险：模型文件可能含敏感训练数据。
   正确做法：模型存 Azure Blob Storage，服务启动时按需下载（首次访问慢，但后续缓存），或挂载 Azure Files 作为共享卷。

3.3 Azure Container Apps 部署：YAML 配置详解

Azure Container Apps 的核心是声明式 YAML。我们不用 Portal 点点点，而是用containerapp.yaml文件，确保环境一致、可复现、可版本控制：

# containerapp.yaml # 生成命令：az containerapp create -n ml-api -g rg-ml -f containerapp.yaml name: ml-api type: Microsoft.Web/containerApps location: australiaeast properties: managedEnvironmentId: "/subscriptions/xxx-xxx-xxx/resourceGroups/rg-ml-env/providers/Microsoft.App/managedEnvironments/ml-env" configuration: activeRevisionsMode: Single ingress: external: true allowInsecure: false targetPort: 8000 traffic: - revisionName: ml-api--00001 weight: 100 customDomains: - ml-api-prod.australiaeast.azurecontainerapps.io registries: - server: myregistry.azurecr.io username: myregistry passwordSecretRef: acr-password template: revisionSuffix: 00001 containers: - name: ml-api image: myregistry.azurecr.io/ml-api:v1.2.0 env: - name: MODEL_STORAGE_URL value: "https://mystorage.blob.core.windows.net/models/risk-v2.pt" - name: AZURE_STORAGE_CONNECTION_STRING secretRef: storage-connection-string resources: cpu: 1.0 memory: 2.0Gi scale: minReplicas: 1 maxReplicas: 10 rules: - http: metadata: concurrentRequests: "100" name: http-concurrency - custom: type: cpu metadata: threshold: "70" averageValue: "70" name: cpu-utilization secrets: - name: acr-password value: "your-acr-password-here" # 实际应存 Key Vault - name: storage-connection-string value: "DefaultEndpointsProtocol=https;AccountName=mystorage;AccountKey=xxx;EndpointSuffix=core.windows.net"

配置项逐条解读：

• managedEnvironmentId：指向已创建的 Container Apps 环境（Environment），它是网络、日志、监控的父级资源，一个环境可托管多个 Container App。
• ingress.external: true：启用公网访问，Azure 自动分配 DNS 名称（ml-api-prod.australiaeast.azurecontainerapps.io）并配置 HTTPS。
• targetPort: 8000：必须与 Dockerfile 中EXPOSE和 FastAPI 启动端口一致。
• registries：指定私有镜像仓库（ACR），passwordSecretRef引用密钥，避免明文密码。
• env：环境变量，MODEL_STORAGE_URL告诉服务去哪里下载模型，AZURE_STORAGE_CONNECTION_STRING用于访问 Blob Storage。
• resources：为容器设置 CPU 和内存限制。注意：Container Apps 的cpu单位是“vCPU”，memory单位是 GiB。1.0 vCPU ≈ 1 个物理核心，2.0Gi 内存是硬限制，超限会被 OOM Kill。
• scale.rules：定义扩缩策略。这里配置了两个规则：
  • http-concurrency：当每秒并发请求数 > 100，触发扩缩；
  • cpu-utilization：当 CPU 平均使用率 > 70%，触发扩缩。
    Container Apps 会同时监控这两个指标，任一满足即扩缩。
• secrets：密钥存储，value字段应替换为 Azure Key Vault 中的密钥引用（如@Microsoft.KeyVault(SecretUri=https://myvault.vault.azure.net/secrets/acr-password/xxx)），本文为简化展示明文。

实操心得：我曾因targetPort写错（写成8080而 FastAPI 启动在8000），导致所有请求返回502 Bad Gateway，排查了 40 分钟才意识到是端口不匹配。Container Apps 的 ingress 配置是“黑盒”，错误不会在部署时报错，只会在运行时静默失败。解决方案：部署后立即用curl -I https://your-app-url/health检查状态码，再用az containerapp logs show -n ml-api -g rg-ml查看容器日志，确认 Uvicorn 是否成功监听8000端口。

4. 实操过程与核心环节实现

4.1 从零开始：本地开发与测试全流程

我们以一个真实的风控模型为例，走一遍从本地编码到 Azure 部署的完整链路。假设你已有一个训练好的risk_model_v2.pkl（scikit-learn 模型），目标是暴露/predict接口。

步骤 1：初始化项目结构

mkdir ml-api && cd ml-api # 创建虚拟环境（推荐，隔离依赖） python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 初始化 Git git init echo "venv/" > .gitignore echo "__pycache__/" >> .gitignore echo "*.pyc" >> .gitignore

步骤 2：编写核心代码
创建main.py（内容见 3.1 节），model_loader.py：

# model_loader.py import joblib import os from azure.storage.blob import BlobServiceClient from io import BytesIO def load_model_from_azure_blob(blob_url: str) -> object: """ 从 Azure Blob Storage 加载模型 blob_url 示例: https://mystorage.blob.core.windows.net/models/risk-v2.pkl """ # 解析 URL 获取 account_name, container_name, blob_name from urllib.parse import urlparse parsed = urlparse(blob_url) account_name = parsed.netloc.split('.')[0] # mystorage container_name = parsed.path.split('/')[1] # models blob_name = '/'.join(parsed.path.split('/')[2:]) # risk-v2.pkl # 从环境变量获取连接字符串 connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING") if not connection_string: raise EnvironmentError("AZURE_STORAGE_CONNECTION_STRING not set