本文解决AI回答采集系统从本地原型迁移到云上生产环境时遇到的部署、配置、性能和成本问题。系统基于Python和FastAPI构建，使用阿里云ECS、RDS MySQL、OSS和日志服务。最终实现一个可稳定运行、支持并发采集、数据可追溯、成本可控的云上服务。适合有Python开发经验、需要将AI应用部署到云上的开发者。前提是已开通阿里云账号并完成实名认证，了解ECS、RDS、OSS的基本概念。本文不涉及模型训练和调优。
整体方案
系统架构如下：
flowchart LR
A[采集任务调度] --> B[FastAPI应用]
B --> C[阿里云百炼模型服务]
C --> D[结果解析与清洗]
D --> E[(RDS MySQL)]
D --> F[(OSS存储)]
B --> G[日志服务]
采集任务由Celery调度，FastAPI提供API接口，调用阿里云百炼模型服务获取AI回答，解析后存入RDS，原始数据存OSS，日志采集到日志服务。
环境与资源准备
操作系统：Ubuntu 22.04 LTS
Python 3.10+
FastAPI、Celery、Redis、SQLAlchemy
阿里云ECS（2核4GB）、RDS MySQL（1核1GB）、OSS、日志服务
地域：华东1（杭州）
在ECS上安装Python依赖：
pip install fastapi uvicorn celery redis sqlalchemy pymysql oss2
账号与权限配置
创建RAM用户，授予以下权限：
ECS相关：无特殊权限，使用AccessKey调用API
RDS：连接数据库
OSS：读写Bucket
日志服务：写入日志
将AccessKey配置为环境变量：
export ALIBABA_CLOUD_ACCESS_KEY_ID=""
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=""
不要将密钥写入代码仓库。
核心实现
模型调用
使用阿里云百炼的DashScope SDK调用通义千问模型：
import dashscope
from dashscope import Generation
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
def get_ai_answer(question):
response = Generation.call(
model="qwen-turbo",
prompt=question,
max_tokens=500
)
if response.status_code == 200:
return response.output.text
else:
raise Exception(f"API error: {response.code}")
数据存储
使用SQLAlchemy定义模型，将采集结果存入RDS：
class AnswerRecord(Base):
tablename = "answers"
id = Column(Integer, primary_key=True)
question = Column(String(500))
answer = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
原始回答JSON存入OSS：
import oss2
auth = oss2.Auth(os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"))
bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "my-answer-bucket")
def save_raw(question, response):
key = f"raw/{datetime.now():%Y%m%d}/{uuid4().hex}.json"
bucket.put_object(key, json.dumps(response))
异步任务
使用Celery处理采集任务，避免阻塞API：
from celery import Celery
app = Celery("tasks", broker="redis://localhost:6379/0")
@app.task
def collect_answer(question):
try:
answer = get_ai_answer(question)
save_raw(question, answer)
# 解析并存入数据库
except Exception as e:
log_error(e)
部署到ECS
使用systemd管理FastAPI服务：
[Unit]
Description=FastAPI App
After=network.target
[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/ai-collector
ExecStart=/usr/bin/uvicorn main:app --host 0.0.0.0 --port 8000
Restart=always
[Install]
WantedBy=multi-user.target
启动服务：
sudo systemctl start fastapi
sudo systemctl enable fastapi
验证服务启动
检查服务状态：
sudo systemctl status fastapi
正常情况下应显示 active (running)。
查看应用日志：
sudo journalctl -u fastapi -f
应看到类似 Uvicorn running on http://0.0.0.0:8000 的日志。
调用健康检查接口（如果实现了）：
curl http://localhost:8000/health
应返回 {"status":"ok"}。
验证采集任务
手动触发一个测试任务：
from tasks import collect_answer
result = collect_answer.delay("什么是RAG？")
print(result.id)
查看任务状态：
celery -A tasks status
检查数据库和OSS：
在RDS中查询 answers 表，应出现对应记录。
在OSS Bucket中查看 raw/ 目录，应有对应JSON文件。
性能优化与监控
数据库连接池配置
使用SQLAlchemy连接池，调整 pool_size 和 max_overflow 参数：
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://user:pass@host:3306/dbname",
pool_size=10,
max_overflow=20,
pool_pre_ping=True
)
pool_size 设置连接池大小，max_overflow 设置最大溢出连接数，pool_pre_ping 在每次连接前检查连接是否有效。
模型调用超时与重试
为模型调用设置超时和重试，避免因限流导致任务失败：
import time
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_ai_answer_with_retry(question):
return get_ai_answer(question)
日志与监控告警
使用阿里云日志服务收集应用日志，并设置告警：
在日志服务中创建Project和Logstore。
在ECS上安装Logtail并配置采集。
设置告警规则，例如当错误日志出现时触发告警。
成本控制
ECS按量付费，测试完成后释放。
RDS选择最小规格，使用按量付费。
OSS存储费用根据实际存储量计费，具体费用请参考官方定价。
模型调用按Token计费，控制请求频率和Token数量。
常见问题与避坑
API Key未生效
现象：调用模型时返回401或InvalidApiKey。
排查：检查环境变量是否设置正确：
echo $DASHSCOPE_API_KEY
解决：重新设置环境变量，并重启服务。
数据库连接失败
现象：应用启动时报数据库连接错误。
排查：检查RDS白名单是否包含ECS公网IP。
解决：在RDS控制台添加ECS公网IP到白名单。
任务重复执行
现象：同一任务被多次执行。
解决：在Celery任务中设置幂等性，使用唯一ID去重。
日志丢失
现象：日志服务中看不到日志。
排查：检查Logtail配置和网络连通性。
解决：确认日志服务Project和Logstore配置正确。
总结
本文介绍了AI回答采集系统上云的完整流程，从环境准备、代码实现到部署监控。通过合理规划资源和成本控制，可以实现稳定、可维护的云上系统。关键点包括：使用RAM最小权限、配置环境变量、使用异步任务、设置监控告警。希望本文能帮助开发者顺利完成上云迁移。
