Dockerfile 示例

提供一个适合容器化部署的最小 FastAPI 应用：包含健康检查、版本信息与一个 基于规则的 Agent 问答接口（不依赖外部 LLM）。同时提供 generate_dockerfile 生成可用的 Dockerfile 文本，演示镜像构建最佳实践。 对应教程：第12卷《部署》第1章 Docker。 运行方式:: uvicorn examples.volume-12-deployment.dockerfile_example:app --host 0.0.0.0 --port 8000 构建镜像（在项目根目录）:: docker build -t ai-agent-app .

文件路径: examples/volume-12-deployment\dockerfile_example.py

---

源代码

"""容器化部署示例。

提供一个适合容器化部署的最小 FastAPI 应用：包含健康检查、版本信息与一个
基于规则的 Agent 问答接口（不依赖外部 LLM）。同时提供 ``generate_dockerfile``
生成可用的 Dockerfile 文本，演示镜像构建最佳实践。

对应教程：第12卷《部署》第1章 Docker。

运行方式::

    uvicorn examples.volume-12-deployment.dockerfile_example:app --host 0.0.0.0 --port 8000

构建镜像（在项目根目录）::

    docker build -t ai-agent-app .
"""

from __future__ import annotations

import os

from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator


class AskRequest(BaseModel):
    """问答请求体。

    Attributes:
        question: 用户问题。
    """

    question: str = Field(min_length=1, description="用户问题，不能为空")

    @field_validator("question", mode="before")
    @classmethod
    def _strip_question(cls, v: object) -> object:
        return v.strip() if isinstance(v, str) else v


class AskResponse(BaseModel):
    """问答响应体。

    Attributes:
        answer: Agent 回答。
        question: 原始问题（便于日志追踪）。
        elapsed_ms: 处理耗时（毫秒）。
    """

    answer: str
    question: str
    elapsed_ms: float


def rule_based_answer(question: str) -> str:
    """基于规则的占位 Agent（教学用，避免真实 API 调用）。

    Args:
        question: 用户问题。

    Returns:
        回答字符串。
    """
    q = question.strip()
    if not q:
        return "请输入有效问题。"
    if "你好" in q or "hello" in q.lower():
        return "你好！我是部署在容器中的示例 Agent。"
    if "天气" in q:
        return "（示例）今天晴，25°C。"
    if q.endswith("?") or q.endswith("？"):
        return f"这是一个好问题：{q}"
    return f"已收到你的问题：{q}"


def generate_dockerfile(
    app_module: str = "dockerfile_example:app",
    python_version: str = "3.11",
    port: int = 8000,
) -> str:
    """生成一个多阶段 Dockerfile 文本。

    Args:
        app_module: uvicorn 启动时指向的 ``模块:app``。
        python_version: 基础镜像 Python 版本。
        port: 容器暴露端口。

    Returns:
        Dockerfile 文本。
    """
    return f"""# ===== 构建阶段 =====
FROM python:{python_version}-slim AS builder

WORKDIR /build

# 仅复制依赖文件，利用 Docker 层缓存
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# ===== 运行阶段 =====
FROM python:{python_version}-slim AS runtime

# 安装非 root 用户
RUN useradd --create-home appuser
WORKDIR /app

# 复制依赖
COPY --from=builder /root/.local /home/appuser/.local
ENV PATH=/home/appuser/.local/bin:$PATH

# 复制应用代码
COPY . /app

# 设置环境变量
ENV PYTHONUNBUFFERED=1 \\
    PYTHONDONTWRITEBYTECODE=1 \\
    PORT={port}

USER appuser

EXPOSE {port}

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --retries=3 \\
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:{port}/health')" || exit 1

CMD ["uvicorn", "{app_module}", "--host", "0.0.0.0", "--port", "{port}"]
"""


app = FastAPI(title="容器化 Agent 示例", version="1.0.0")


@app.get("/health")
def health() -> dict[str, str]:
    """健康检查接口。

    Returns:
        状态字典，``status`` 为 ``ok`` 表示服务正常。
    """
    return {"status": "ok"}


@app.get("/version")
def version() -> dict[str, str]:
    """返回服务版本信息。

    Returns:
        包含 ``app``、``version``、``env`` 的字典。
    """
    return {
        "app": app.title,
        "version": app.version,
        "env": os.environ.get("APP_ENV", "development"),
    }


@app.post("/agent/ask", response_model=AskResponse)
def ask(req: AskRequest) -> AskResponse:
    """Agent 问答接口（基于规则）。

    Args:
        req: 问答请求体。

    Returns:
        问答响应体。
    """
    import time

    start = time.perf_counter()
    answer = rule_based_answer(req.question)
    elapsed_ms = (time.perf_counter() - start) * 1000
    return AskResponse(answer=answer, question=req.question, elapsed_ms=round(elapsed_ms, 3))


def main() -> None:
    """本地启动服务（或打印 Dockerfile）。"""
    import sys

    if "--dockerfile" in sys.argv:
        print(generate_dockerfile())
        return
    import uvicorn

    port = int(os.environ.get("PORT", "8000"))
    uvicorn.run(app, host="0.0.0.0", port=port)


if __name__ == "__main__":
    main()