AI回答采集系统需要定时向多个大模型API发送问题并存储回答。本地原型验证可行后，迁移到云上需解决定时调度、并发控制、数据持久化和可观测性问题。本文以阿里云ECS和OSS为例，分享从本地脚本到容器化部署的完整过程。读者需具备Python、Docker和阿里云基础操作知识，并已开通ECS、OSS、日志服务等产品（免费额度可试用）。
业务目标与云上约束
采集系统需要定时向多个大模型API发送问题，收集回答并存储。本地原型通常单线程运行，数据存本地文件。上云后需满足：
定时调度：每天固定时间执行采集任务。
并发控制：多个API请求并发，但需控制速率避免限流。
数据持久化：采集结果可靠存储，支持回溯。
可观测：任务状态、失败原因、耗时等可监控。
成本可控：按需使用资源，避免闲置浪费。
整体架构
flowchart LR
A[定时触发器] --> B[任务调度器]
B --> C[采集Worker集群]
C --> D[大模型API]
D --> C
C --> E[OSS存储]
C --> F[日志服务]
E --> G[数据湖/分析]
定时触发器：使用ECS Cron触发。
任务调度器：管理任务队列和并发。
采集Worker：无状态容器，执行HTTP请求。
OSS：存储原始回答JSON。
日志服务：记录任务日志和指标。
环境与资源准备
云资源清单
资源 规格 用途
ECS ecs.t6-c1m1.large (2C2G) 运行任务调度器和Worker
OSS 标准存储 存储采集结果
日志服务 按量付费 日志采集与监控
容器镜像服务 免费 存储Docker镜像
本地环境
Python 3.10+
Docker 20.10+
阿里云CLI配置完成
权限准备
RAM用户需具备ECS、OSS、日志服务的读写权限。
创建AccessKey并配置到环境变量：
export ALIBABA_CLOUD_ACCESS_KEY_ID=""
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=""
方案对比与选择
方案 优点 缺点
单ECS + Cron 简单，成本低 单点故障，扩展性差
函数计算 自动伸缩，免运维 冷启动，执行时长限制
ECS + Docker Compose 可控，易迁移 需手动管理
容器服务ACK 高可用，弹性强 运维复杂，成本高
本文选择ECS + Docker Compose方案，平衡可控性与成本。
核心实现
- 容器化采集脚本
将本地采集脚本打包为Docker镜像。关键Dockerfile：
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "collector.py"]
collector.py核心逻辑：
import os
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from aliyun_oss import OSSClient
API_LIST = [
{"name": "model_a", "url": "https://api.example.com/v1/chat", "api_key": os.getenv("API_KEY_A")},
{"name": "model_b", "url": "https://api.example2.com/v1/completions", "api_key": os.getenv("API_KEY_B")},
]
QUESTIONS = [
"什么是生成式AI？",
"如何选择大模型？",
]
def fetch_answer(api, question):
headers = {"Authorization": f"Bearer {api['api_key']}"}
payload = {"model": "default", "messages": [{"role": "user", "content": question}]}
try:
resp = requests.post(api["url"], json=payload, headers=headers, timeout=30)
resp.raise_for_status()
return {"api": api["name"], "question": question, "answer": resp.json(), "status": "success"}
except Exception as e:
return {"api": api["name"], "question": question, "error": str(e), "status": "failed"}
def main():
tasks = []
with ThreadPoolExecutor(max_workers=5) as executor:
for api in API_LIST:
for q in QUESTIONS:
tasks.append(executor.submit(fetch_answer, api, q))
results = [t.result() for t in as_completed(tasks)]
oss = OSSClient(os.getenv("OSS_BUCKET"), os.getenv("OSS_ENDPOINT"))
timestamp = int(time.time())
key = f"results/{timestamp}.json"
oss.put_object(key, json.dumps(results, ensure_ascii=False))
print(f"Uploaded to oss://{oss.bucket}/{key}")
if name == "main":
main()
说明：
使用ThreadPoolExecutor控制并发数，避免触发API限流。
每个请求设置超时，捕获异常并记录。
结果上传到OSS，按时间戳分片存储。
- 构建与推送镜像
docker build -t collector:latest .
docker tag collector:latest registry.cn-hangzhou.aliyuncs.com//collector:latest
docker push registry.cn-hangzhou.aliyuncs.com//collector:latest
- ECS部署
在ECS上安装Docker，编写docker-compose.yml：
version: '3'
services:
collector:
image: registry.cn-hangzhou.aliyuncs.com//collector:latest
environment:
- API_KEY_A=${API_KEY_A}
- API_KEY_B=${API_KEY_B}
- OSS_BUCKET=${OSS_BUCKET}
- OSS_ENDPOINT=${OSS_ENDPOINT}
- ALIBABA_CLOUD_ACCESS_KEY_ID=${ALIBABA_CLOUD_ACCESS_KEY_ID}
- ALIBABA_CLOUD_ACCESS_KEY_SECRET=${ALIBABA_CLOUD_ACCESS_KEY_SECRET}
restart: unless-stopped
创建.env文件填入环境变量，然后启动：
docker-compose up -d
- 定时任务
使用Crontab每天凌晨2点执行：
0 2 * cd /home/ecs-user/collector && docker-compose run --rm collector
测试与监控
验证采集结果
登录OSS控制台，查看results/目录下是否有当天JSON文件。下载并检查内容：
[
{
"api": "model_a",
"question": "什么是生成式AI？",
"answer": {"choices": [{"message": {"content": "生成式AI是指..."}}]},
"status": "success"
}
]
日志监控
采集脚本输出到stdout，通过Docker日志查看：
docker-compose logs collector
配置阿里云日志服务采集ECS上的Docker日志，设置告警：当出现"status": "failed"时触发通知。
成本与安全分析
成本估算
ECS：ecs.t6-c1m1.large 按量约0.1元/小时，每天运行1小时，月成本约3元。
OSS：存储少量JSON，月成本忽略。
日志服务：按量付费，月成本约1元。
合计：约5元/月。
具体费用以官方控制台为准。
安全注意事项
API Key和AccessKey通过环境变量注入，不写在代码或镜像中。
RAM用户权限最小化，仅授予OSS写入和日志服务写入权限。
ECS安全组仅允许出站HTTP/HTTPS，禁止入站。
OSS Bucket设置为私有，通过RAM授权访问。
踩坑与适用边界
常见问题
API限流：多个API同时请求可能导致限流。解决：在fetch_answer中加入指数退避重试。
OSS上传失败：检查Endpoint是否与Bucket地域一致，AccessKey是否有权限。
容器时区：默认UTC，需在Dockerfile设置时区：ENV TZ=Asia/Shanghai。
适用场景
每日采集量小于1000次请求。
对实时性要求不高，可容忍延迟。
团队规模小，运维能力有限。
不适用场景
高并发实时采集（需使用ACK+消息队列）。
需要分布式任务调度（建议使用Celery+Redis）。
总结
本文从本地原型到云上生产部署，完整介绍了AI回答采集系统的容器化、部署和监控过程。通过ECS+Docker Compose方案，以极低成本实现了稳定、可观测的采集系统。后续可扩展为：增加任务队列支持更多API、集成数据湖分析、配置自动伸缩应对突发流量。
可复用清单
[ ] 容器化采集脚本
[ ] 环境变量管理
[ ] 定时任务配置
[ ] 日志监控告警
[ ] 成本预算与资源释放策略
