当前位置：首页 > news >正文

crewAI 部署形态：本地、Docker、K8s 与 Serverless 化实践

news 2026/5/12 8:43:25

crewAI 部署形态：本地、Docker、K8s 与 Serverless 化实践

本文基于 crewAI v1.11.0，全面覆盖从本地开发到企业级部署的完整实践方案。

一、部署复杂度的阶梯

crewAI 项目的部署需求随规模增长呈阶梯式上升：

Stage 1：本地开发 └── crewai run / python main.py └── 单机，调试为主 Stage 2：容器化 └── Docker 打包，可在任何环境一致运行 └── 解决"在我机器上能跑"问题 Stage 3：容器编排 └── Kubernetes 部署 └── 高可用、自动扩缩容、资源管理 Stage 4：Serverless └── AWS Lambda / Cloud Functions └── 按需运行，零维护，成本最优

不同业务场景选择不同的部署形态，没有"最好"，只有"最合适"。

二、本地开发模式

2.1 标准运行方式

# 进入项目目录cdmy_research_crew# 确保环境变量已配置cat.env# OPENAI_API_KEY=sk-...# 运行 Crewcrewai run# 带参数运行（在 main.py 中读取命令行参数）python src/my_research_crew/main.py--topic"具身智能"

2.2 开发模式热重载

# 使用 watchdog 监听文件变化，自动重启pipinstallwatchdog# watchmedo auto-restartwatchmedo auto-restart\--patterns="*.py;*.yaml"\--recursive\-- python src/my_research_crew/main.py

2.3 调试模式

# main.py 中添加调试模式支持importargparsedefmain():parser=argparse.ArgumentParser()parser.add_argument("--topic",default="AI技术趋势")parser.add_argument("--debug",action="store_true",help="启用详细调试日志")parser.add_argument("--dry-run",action="store_true",help="验证配置但不实际执行")args=parser.parse_args()ifargs.debug:os.environ["CREWAI_LOG_LEVEL"]="DEBUG"ifargs.dry_run:print("配置验证通过，不执行实际任务")crew=MyResearchCrew().crew()print(f"Agent 数量:{len(crew.agents)}")print(f"Task 数量:{len(crew.tasks)}")returnresult=MyResearchCrew().crew().kickoff(inputs={"topic":args.topic})print(result.raw)

三、Docker 容器化

3.1 Dockerfile（多阶段构建）

# Dockerfile # 多阶段构建：减小最终镜像体积 # ─── 构建阶段 ─────────────────────────────────── FROM python:3.11-slim AS builder WORKDIR /build # 安装依赖（使用 uv 加速） COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv COPY pyproject.toml uv.lock* ./ RUN uv pip install --system --no-cache 'crewai[tools]' && \ uv pip install --system --no-cache -r pyproject.toml # ─── 运行阶段 ─────────────────────────────────── FROM python:3.11-slim AS runtime # 安全：使用非 root 用户 RUN useradd --create-home --shell /bin/bash crewai WORKDIR /app # 从构建阶段复制已安装的包 COPY --from=builder /usr/local/lib/python3.11/site-packages/ /usr/local/lib/python3.11/site-packages/ COPY --from=builder /usr/local/bin/ /usr/local/bin/ # 复制应用代码 COPY --chown=crewai:crewai . . USER crewai # 配置 ENV PYTHONUNBUFFERED=1 ENV PYTHONDONTWRITEBYTECODE=1 # 健康检查 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD python -c "import crewai; print('ok')" || exit 1 # 默认入口 CMD ["python", "src/my_research_crew/main.py"]

3.2 构建与运行

# 构建镜像dockerbuild-tmy-research-crew:latest.dockerbuild-tmy-research-crew:v1.0.0.# 本地运行（通过环境变量注入密钥）dockerrun\-eOPENAI_API_KEY=$OPENAI_API_KEY\-eSERPER_API_KEY=$SERPER_API_KEY\-v$(pwd)/outputs:/app/outputs\my-research-crew:latest# 使用 .env 文件注入环境变量dockerrun --env-file .env my-research-crew:latest

3.3 docker-compose 开发环境

# docker-compose.ymlversion:'3.8'services:crew:build:.env_file:.envvolumes:-./outputs:/app/outputs# 输出文件持久化-./knowledge:/app/knowledge# 知识库持久化depends_on:-chromaenvironment:-CHROMA_HOST=chroma-CHROMA_PORT=8000chroma:image:chromadb/chroma:latestports:-"8000:8000"volumes:-chroma_data:/chroma/chromalangfuse:image:langfuse/langfuse:latestports:-"3000:3000"environment:-DATABASE_URL=postgresql://langfuse:langfuse@postgres:5432/langfuse-NEXTAUTH_SECRET=mysecretdepends_on:-postgrespostgres:image:postgres:15environment:POSTGRES_USER:langfusePOSTGRES_PASSWORD:langfusePOSTGRES_DB:langfusevolumes:chroma_data:

四、Kubernetes 部署

4.1 Job 部署（一次性任务）

# k8s/crew-job.yamlapiVersion:batch/v1kind:Jobmetadata:name:research-crew-joblabels:app:research-crewspec:ttlSecondsAfterFinished:3600# 完成后1小时自动清理template:spec:restartPolicy:Never# Job 失败不重试containers:-name:crewimage:my-research-crew:v1.0.0resources:requests:memory:"512Mi"cpu:"250m"limits:memory:"2Gi"# crewAI 内存使用可能较高cpu:"1000m"env:-name:TOPICvalue:"AI在制造业的应用"envFrom:-secretRef:name:crew-secrets# 从 Kubernetes Secret 注入 API Key

4.2 CronJob 部署（定时任务）

# k8s/crew-cronjob.yamlapiVersion:batch/v1kind:CronJobmetadata:name:daily-research-crewspec:schedule:"0 9 * * 1-5"# 工作日每天上午9点concurrencyPolicy:Forbid# 禁止并发执行failedJobsHistoryLimit:3successfulJobsHistoryLimit:5jobTemplate:spec:template:spec:containers:-name:crewimage:my-research-crew:v1.0.0envFrom:-secretRef:name:crew-secrets

4.3 Secret 管理

# 创建 Kubernetes Secret（不要把密钥写在 YAML 里！）kubectl create secret generic crew-secrets\--from-literal=OPENAI_API_KEY=$OPENAI_API_KEY\--from-literal=SERPER_API_KEY=$SERPER_API_KEY\--from-literal=ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY# 验证kubectl get secret crew-secrets

4.4 水平扩缩容（并行处理）

# k8s/crew-deployment.yaml（用于高并发场景）apiVersion:apps/v1kind:Deploymentmetadata:name:crew-workerspec:replicas:3# 3个并行工作实例selector:matchLabels:app:crew-workertemplate:spec:containers:-name:crewimage:my-research-crew:v1.0.0command:["python","src/my_research_crew/worker.py"]# 队列消费模式envFrom:-secretRef:name:crew-secrets---# HorizontalPodAutoscaler：基于 CPU 自动扩缩容apiVersion:autoscaling/v2kind:HorizontalPodAutoscalermetadata:name:crew-worker-hpaspec:scaleTargetRef:apiVersion:apps/v1kind:Deploymentname:crew-workerminReplicas:1maxReplicas:10metrics:-type:Resourceresource:name:cputarget:type:UtilizationaverageUtilization:70

五、Serverless 部署

5.1 AWS Lambda 适配

crewAI 的主要挑战是执行时间可能超过 Lambda 的默认 15 分钟上限。解决方案是结合 crewAI Flows 的断点续传功能：

# lambda_handler.pyimportjsonimportboto3frommy_research_crew.crewimportMyResearchCrew# 使用 SQS 解决超时问题sqs=boto3.client('sqs')RESULT_QUEUE_URL=os.environ['RESULT_QUEUE_URL']defhandler(event,context):"""Lambda 入口函数"""try:topic=event.get('topic','AI趋势')# 配置较短的 max_iter 以控制执行时间crew=MyResearchCrew().crew()# 为每个 Agent 设置较低的迭代上限foragentincrew.agents:agent.max_iter=8result=crew.kickoff(inputs={'topic':topic})# 将结果发送到 SQSsqs.send_message(QueueUrl=RESULT_QUEUE_URL,MessageBody=json.dumps({'topic':topic,'result':result.raw,'token_usage':{'total':result.token_usage.total_tokensifresult.token_usageelse0}}))return{'statusCode':200,'body':json.dumps({'status':'submitted','message':'任务已提交，结果将发送到队列'})}exceptExceptionase:return{'statusCode':500,'body':json.dumps({'error':str(e)})}

5.2 冷启动优化

# 在模块级别初始化（Lambda 容器复用时避免重复初始化）importosfromcrewaiimportCrew,Agent,Taskfromcrewai_toolsimportSerperDevTool# 全局初始化（只在容器初次启动时执行）_crew=None_search_tool=SerperDevTool()# 工具只初始化一次defget_crew()->Crew:global_crewif_crewisNone:researcher=Agent(role="研究员",tools=[_search_tool],goal="...",backstory="...")task=Task(description="...",agent=researcher)_crew=Crew(agents=[researcher],tasks=[task])return_crewdefhandler(event,context):crew=get_crew()# 复用已初始化的 Crewresult=crew.kickoff(inputs={'topic':event['topic']})return{'result':result.raw}

六、部署选型决策矩阵

部署形态	成本	维护复杂度	扩展性	适用场景
本地运行	最低	最简单	无	开发/测试
Docker 单机	低	低	有限	小规模生产
Docker Compose	低	中	中	中小规模
Kubernetes Job	中	高	高	生产批处理
K8s CronJob	中	高	高	定时任务
Lambda	按需（可极低）	低	极高	低频/突发任务