# 第09章：部署指南 ## 📌 本章目标 - 了解 ADK Agent 的多种部署方式 - 掌握使用 Docker 容器化部署 - 学习部署到 Google Cloud Run - 了解部署到 Vertex AI Agent Engine - 掌握 API Server 的使用方法 --- ## 9.1 部署方式概览 ``` ┌──────────────────────────────────────────────────────┐ │ ADK 部署选项 │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌────────────┐ │ │ │ 本地运行 │ │ Cloud Run │ │ Vertex AI │ │ │ │ (开发测试) │ │ (容器部署) │ │ Agent Eng. │ │ │ └──────────────┘ └──────────────┘ └────────────┘ │ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ API Server │ │ 自定义部署 │ │ │ │ (FastAPI) │ │ (Docker) │ │ │ └──────────────┘ └──────────────┘ │ └──────────────────────────────────────────────────────┘ ``` | 方式 | 适用场景 | 复杂度 | 成本 | |------|----------|--------|------| | **本地运行** | 开发、测试 | 低 | 免费 | | **API Server** | 集成到现有应用 | 低 | 取决于服务器 | | **Cloud Run** | 生产环境、自动扩缩 | 中 | 按使用付费 | | **Vertex AI** | 企业级、大规模 | 高 | 按使用付费 | | **Docker** | 自托管、私有化 | 中 | 取决于服务器 | --- ## 9.2 API Server 部署 ### 9.2.1 启动 API Server ```bash # 启动 FastAPI 服务器 adk api_server --port 8080 # 指定 Agent 目录 adk api_server --agent-dir ./my_agent --port 8080 # 启动后，API Server 提供以下端点： # POST /apps/{app_name}/users/{user_id}/sessions/{session_id}:run # GET /apps/{app_name}/users/{user_id}/sessions # POST /apps/{app_name}/users/{user_id}/sessions ``` ### 9.2.2 调用 API Server ```python """ 调用 ADK API Server 通过 HTTP 请求与 Agent 交互 """ import requests # 导入 HTTP 请求库 import json # 导入 JSON 模块 # ======================================== # 配置 API Server 地址 # ======================================== API_BASE_URL = "http://localhost:8080" # API Server 地址 APP_NAME = "my_agent" # 应用名称 USER_ID = "user_001" # 用户 ID SESSION_ID = "session_001" # 会话 ID # ======================================== # 创建会话 # ======================================== def create_session(): """创建新的会话""" url = f"{API_BASE_URL}/apps/{APP_NAME}/users/{USER_ID}/sessions" # API URL response = requests.post( # 发送 POST 请求 url, # 请求 URL json={ # 请求体 "session_id": SESSION_ID, # 会话 ID }, ) session = response.json() # 解析 JSON 响应 print(f"会话创建成功: {session}") # 打印结果 return session # 返回会话信息 # ======================================== # 发送消息给 Agent # ======================================== def send_message(message: str): """发送消息并获取 Agent 响应""" url = ( # 构建 URL f"{API_BASE_URL}/apps/{APP_NAME}/" f"users/{USER_ID}/sessions/{SESSION_ID}:run" ) payload = { # 请求体 "user_id": USER_ID, # 用户 ID "session_id": SESSION_ID, # 会话 ID "new_message": { # 新消息 "role": "user", # 角色 "parts": [{"text": message}], # 消息内容 }, } response = requests.post( # 发送 POST 请求 url, # URL json=payload, # 请求体 stream=True, # 流式响应 ) # 处理流式响应 for line in response.iter_lines(): # 逐行读取 if line: # 如果有内容 data = json.loads(line) # 解析 JSON print(f"事件: {data}") # 打印事件 # ======================================== # 使用示例 # ======================================== if __name__ == "__main__": create_session() # 创建会话 send_message("你好！") # 发送消息 ``` --- ## 9.3 Docker 容器化部署 ### 9.3.1 创建 Dockerfile ```dockerfile # ======================================== # ADK Agent Docker 部署文件 # ======================================== # 使用 Python 3.11 作为基础镜像 FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 设置环境变量 # PYTHONUNBUFFERED: 确保 Python 输出直接显示到控制台 ENV PYTHONUNBUFFERED=1 # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制 Agent 代码 COPY my_agent/ ./my_agent/ # 复制启动脚本 COPY start.sh . # 赋予启动脚本执行权限 RUN chmod +x start.sh # 暴露端口 EXPOSE 8080 # 启动 API Server CMD ["./start.sh"] ``` ### 9.3.2 创建 requirements.txt ```text # ADK Agent 依赖文件 google-adk>=1.0.0 # 如果需要使用其他模型 litellm>=1.0.0 ``` ### 9.3.3 创建启动脚本 ```bash #!/bin/bash # ADK Agent 启动脚本 # 设置 API Key（从环境变量读取） export GOOGLE_API_KEY="${GOOGLE_API_KEY}" # 启动 API Server adk api_server \ --agent-dir ./my_agent \ --port 8080 \ --host 0.0.0.0 ``` ### 9.3.4 构建和运行 Docker 镜像 ```bash # 构建 Docker 镜像 docker build -t my-adk-agent . # 运行 Docker 容器 docker run -d \ --name my-agent \ -p 8080:8080 \ -e GOOGLE_API_KEY="your_api_key" \ my-adk-agent # 查看日志 docker logs -f my-agent # 测试 API curl -X POST http://localhost:8080/apps/my_agent/users/user_001/sessions \ -H "Content-Type: application/json" \ -d '{"session_id": "test_session"}' ``` --- ## 9.4 部署到 Google Cloud Run ### 9.4.1 使用 ADK CLI 部署 ```bash # 使用 adk deploy 命令部署到 Cloud Run adk deploy my_agent --platform cloud-run # 部署过程： # 1. 自动构建 Docker 镜像 # 2. 推送到 Google Container Registry # 3. 部署到 Cloud Run # 4. 配置环境变量和密钥 ``` ### 9.4.2 手动部署到 Cloud Run ```bash # 第一步：配置 gcloud CLI gcloud auth login # 登录 Google Cloud gcloud config set project YOUR_PROJECT_ID # 设置项目 ID # 第二步：构建并推送镜像 gcloud builds submit --tag gcr.io/YOUR_PROJECT_ID/my-agent # 第三步：部署到 Cloud Run gcloud run deploy my-agent \ --image gcr.io/YOUR_PROJECT_ID/my-agent \ --platform managed \ --region us-central1 \ --allow-unauthenticated \ --set-env-vars "GOOGLE_API_KEY=your_api_key" # 第四步：获取服务 URL gcloud run services describe my-agent \ --region us-central1 \ --format 'value(status.url)' ``` --- ## 9.5 部署到 Vertex AI Agent Engine ### 9.5.1 使用 ADK CLI 部署 ```bash # 部署到 Vertex AI Agent Engine adk deploy my_agent --platform vertex-ai # Vertex AI Agent Engine 提供： # - 自动扩缩容 # - 内置监控和日志 # - 企业级安全 # - 与 Google Cloud 生态集成 ``` ### 9.5.2 部署配置 ```python """ Vertex AI Agent Engine 部署配置 """ # 在 agent.py 中添加部署元数据 from google.adk.agents import Agent # 导入 Agent 类 root_agent = Agent( name="my_agent", # Agent 名称 model="gemini-2.0-flash", # 模型 instruction="你是一个助手。", # 指令 # Vertex AI 部署相关配置 # output_key="response", # 输出键 # before_model_callback=..., # 回调函数 ) ``` --- ## 9.6 生产环境最佳实践 ### 9.6.1 安全配置 ```python """ 生产环境安全配置 """ # ======================================== # 1. API Key 管理 # ======================================== # ❌ 不要硬编码 API Key # BAD: api_key = "AIzaSy..." # ✅ 使用环境变量 import os # 导入 os 模块 api_key = os.environ.get("GOOGLE_API_KEY") # 从环境变量读取 # ✅ 使用 Secret Manager（Google Cloud） # from google.cloud import secretmanager # client = secretmanager.SecretManagerServiceClient() # secret = client.access_secret_version(name="projects/.../secrets/api-key/versions/latest") # ======================================== # 2. 输入验证 # ======================================== async def safe_before_model_callback(cb_ctx, inv_ctx): """安全回调：验证输入""" # 获取用户输入 user_input = inv_ctx.session.state.get("temp:user_input", "") # 读取输入 # 检查输入长度 if len(user_input) > 10000: # 如果输入过长 raise ValueError("输入过长，请缩短您的消息。") # 抛出异常 # 检查敏感内容（示例） sensitive_patterns = [ # 敏感内容模式 "密码", "credit card", "ssn", # 敏感词列表 ] for pattern in sensitive_patterns: # 遍历模式 if pattern.lower() in user_input.lower(): # 如果匹配 print(f"[安全] 检测到敏感内容: {pattern}") # 记录日志 # ======================================== # 3. 速率限制 # ======================================== # 在 API Server 前面配置速率限制 # 可以使用 Nginx、Cloud Armor 等工具 ``` ### 9.6.2 监控和日志 ```python """ 生产环境监控配置 """ import logging # 导入日志模块 import time # 导入时间模块 # 配置日志 logging.basicConfig( # 配置日志格式 level=logging.INFO, # 日志级别 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', # 格式 ) logger = logging.getLogger("adk_agent") # 创建日志记录器 async def monitoring_callback(cb_ctx, inv_ctx): """监控回调：记录性能指标""" start_time = time.time() # 记录开始时间 # 记录请求信息 logger.info( # 记录信息日志 f"Agent: {inv_ctx.agent.name}, " # Agent 名称 f"User: {inv_ctx.session.user_id}" # 用户 ID ) # 在 after 回调中计算耗时 elapsed = time.time() - start_time # 计算耗时 logger.info(f"处理耗时: {elapsed:.3f}s") # 记录耗时 # 如果耗时过长，记录警告 if elapsed > 5.0: # 如果超过5秒 logger.warning(f"请求处理耗时过长: {elapsed:.3f}s") # 记录警告 ``` --- ## 📌 本章小结 - ADK 支持多种部署方式：本地、API Server、Cloud Run、Vertex AI - API Server 基于 FastAPI，提供 REST API 接口 - Docker 容器化是推荐的生产部署方式 - Cloud Run 提供自动扩缩容和按使用付费 - Vertex AI Agent Engine 提供企业级功能 - 生产环境需要注意安全、监控和日志 **下一章**：[第10章 - 高级主题](./10-advanced-topics.md) → 学习安全最佳实践、A2A 协议等高级功能。