今天第 25 天 大致上 ai 相關的討論差不多了,在現代應用當中,通常要自動化處理這件事
所以整合 CI/CI 去完成部署是相對重要的事情,今天探討的主題是把我們開發完成的 AI 應用
進行 CI/CD pipeline,當然不熟悉 CI/CD 的部分,我這裡會簡單介紹這是什麼和怎麼實現等
我們這裡以 AWS 全家桶內部的東西來進行完整的整合
CI (Continuous Integration,持續整合)
CD (Continuous Delivery/Deployment,持續交付/部署)
AWS CodePipeline
完整的CI/CD編排服務,整合其他AWS服務
AWS CodeBuild
全託管的建置服務,支援多種程式語言和框架
AWS CodeDeploy
自動化部署服務,支援EC2、Lambda、ECS等
Github Action + AWS
使用 GitHub Actions 搭配AWS服務,適合已使用 GitHub 的團隊
用AWS CodePipeline + CodeBuild + CodeDeploy 完成 CI/CD pipeline
架構
Github Repository -> CodePipeline(編排) ->
CodeBuild(建置與測試) ->
CodeDeploy(部署到 Lambda/ECS) ->
Production Environment
這裡建立一個 ai 專案
my-ai-app/
├── src/
│ ├── lambda/
│ │ └── inference.py
│ └── api/
│ └── app.py
├── tests/
│ ├── test_inference.py
│ └── test_api.py
├── infrastructure/
│ └── cloudformation.yaml
├── buildspec.yml # CodeBuild配置
├── appspec.yml # CodeDeploy配置
├── requirements.txt
└── Dockerfile
buidsepc.yml
version: 0.2
phases:
pre_build:
commands:
- echo Logging in to Amazon ECR...
- aws ecr get-login-password --region $AWS_DEFAULT_REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com
- REPOSITORY_URI=$AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com/my-ai-app
- COMMIT_HASH=$(echo $CODEBUILD_RESOLVED_SOURCE_VERSION | cut -c 1-7)
- IMAGE_TAG=${COMMIT_HASH:=latest}
build:
commands:
- echo Build started on `date`
- echo Running tests...
- pip install -r requirements.txt
- python -m pytest tests/ -v
- echo Building the Docker image...
- docker build -t $REPOSITORY_URI:latest .
- docker tag $REPOSITORY_URI:latest $REPOSITORY_URI:$IMAGE_TAG
post_build:
commands:
- echo Build completed on `date`
- echo Pushing the Docker images...
- docker push $REPOSITORY_URI:latest
- docker push $REPOSITORY_URI:$IMAGE_TAG
- echo Writing image definitions file...
- printf '[{"name":"my-ai-app","imageUri":"%s"}]' $REPOSITORY_URI:$IMAGE_TAG > imagedefinitions.json
artifacts:
files:
- imagedefinitions.json
- appspec.yml
- taskdef.json
cache:
paths:
- '/root/.cache/pip/**/*'
import boto3
import json
iam = boto3.client('iam')
# CodeBuild服務角色
codebuild_policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"ecr:GetAuthorizationToken",
"ecr:BatchCheckLayerAvailability",
"ecr:GetDownloadUrlForLayer",
"ecr:PutImage",
"ecr:InitiateLayerUpload",
"ecr:UploadLayerPart",
"ecr:CompleteLayerUpload"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"s3:GetObject",
"s3:PutObject"
],
"Resource": "arn:aws:s3:::my-pipeline-bucket/*"
},
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel"
],
"Resource": "*"
}
]
}
# 建立CodeBuild角色
assume_role_policy = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "codebuild.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
response = iam.create_role(
RoleName='CodeBuildServiceRole',
AssumeRolePolicyDocument=json.dumps(assume_role_policy),
Description='Role for CodeBuild to build AI applications'
)
iam.put_role_policy(
RoleName='CodeBuildServiceRole',
PolicyName='CodeBuildPolicy',
PolicyDocument=json.dumps(codebuild_policy)
)
print(f"CodeBuild角色ARN: {response['Role']['Arn']}")
import boto3
codebuild = boto3.client('codebuild')
response = codebuild.create_project(
name='my-ai-app-build',
description='Build project for AI application',
source={
'type': 'GITHUB',
'location': 'https://github.com/your-username/my-ai-app.git',
'buildspec': 'buildspec.yml'
},
artifacts={
'type': 'S3',
'location': 'my-pipeline-bucket',
'path': 'builds/',
'name': 'build-output.zip',
'packaging': 'ZIP'
},
environment={
'type': 'LINUX_CONTAINER',
'image': 'aws/codebuild/standard:7.0',
'computeType': 'BUILD_GENERAL1_SMALL',
'privilegedMode': True, # 需要建置Docker映像
'environmentVariables': [
{
'name': 'AWS_DEFAULT_REGION',
'value': 'us-east-1',
'type': 'PLAINTEXT'
},
{
'name': 'AWS_ACCOUNT_ID',
'value': '123456789012',
'type': 'PLAINTEXT'
}
]
},
serviceRole='arn:aws:iam::123456789012:role/CodeBuildServiceRole',
cache={
'type': 'S3',
'location': 'my-pipeline-bucket/cache'
}
)
print(f"CodeBuild專案已建立: {response['project']['name']}")
import boto3
codepipeline = boto3.client('codepipeline')
response = codepipeline.create_pipeline(
pipeline={
'name': 'my-ai-app-pipeline',
'roleArn': 'arn:aws:iam::123456789012:role/CodePipelineServiceRole',
'artifactStore': {
'type': 'S3',
'location': 'my-pipeline-bucket'
},
'stages': [
{
'name': 'Source',
'actions': [
{
'name': 'SourceAction',
'actionTypeId': {
'category': 'Source',
'owner': 'ThirdParty',
'provider': 'GitHub',
'version': '1'
},
'configuration': {
'Owner': 'your-username',
'Repo': 'my-ai-app',
'Branch': 'main',
'OAuthToken': '{{resolve:secretsmanager:github-token}}'
},
'outputArtifacts': [
{
'name': 'SourceOutput'
}
]
}
]
},
{
'name': 'Build',
'actions': [
{
'name': 'BuildAction',
'actionTypeId': {
'category': 'Build',
'owner': 'AWS',
'provider': 'CodeBuild',
'version': '1'
},
'configuration': {
'ProjectName': 'my-ai-app-build'
},
'inputArtifacts': [
{
'name': 'SourceOutput'
}
],
'outputArtifacts': [
{
'name': 'BuildOutput'
}
]
}
]
},
{
'name': 'Deploy',
'actions': [
{
'name': 'DeployAction',
'actionTypeId': {
'category': 'Deploy',
'owner': 'AWS',
'provider': 'ECS',
'version': '1'
},
'configuration': {
'ClusterName': 'my-ai-app-cluster',
'ServiceName': 'my-ai-app-service',
'FileName': 'imagedefinitions.json'
},
'inputArtifacts': [
{
'name': 'BuildOutput'
}
]
}
]
}
]
}
)
print(f"Pipeline已建立: {response['pipeline']['name']}")
# buildspec.yml 中的測試階段
version: 0.2
phases:
pre_build:
commands:
- echo Installing dependencies...
- pip install -r requirements.txt
- pip install pytest pytest-cov boto3
build:
commands:
- echo Running unit tests...
- python -m pytest tests/unit/ -v --cov=src --cov-report=xml
- echo Running integration tests...
- python -m pytest tests/integration/ -v
- echo Testing Bedrock integration...
- python tests/test_bedrock_integration.py
- echo Building Docker image...
- docker build -t $REPOSITORY_URI:latest .
- echo Running container tests...
- docker run --rm $REPOSITORY_URI:latest python -m pytest tests/container/ -v
reports:
coverage-report:
files:
- 'coverage.xml'
file-format: 'COBERTURAXML'
test-report:
files:
- 'test-results.xml'
file-format: 'JUNITXML'
tests/test_bedrock_integratoin.py
import boto3
import json
import os
def test_bedrock_connection():
"""測試Bedrock連線"""
bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1')
try:
response = bedrock_runtime.invoke_model(
modelId='anthropic.claude-3-haiku-20240307-v1:0',
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 100,
"messages": [
{
"role": "user",
"content": "Hello, this is a test."
}
]
})
)
result = json.loads(response['body'].read())
assert 'content' in result
print("✓ Bedrock連線測試通過")
return True
except Exception as e:
print(f"✗ Bedrock連線測試失敗: {str(e)}")
return False
def test_model_inference():
"""測試模型推論"""
from src.lambda.inference import lambda_handler
event = {
"body": json.dumps({
"prompt": "What is AWS?",
"max_tokens": 100
})
}
response = lambda_handler(event, None)
assert response['statusCode'] == 200
body = json.loads(response['body'])
assert 'result' in body
print("✓ 模型推論測試通過")
if __name__ == "__main__":
test_bedrock_connection()
test_model_inference()
version: 0.0
Resources:
- TargetService:
Type: AWS::ECS::Service
Properties:
TaskDefinition: <TASK_DEFINITION>
LoadBalancerInfo:
ContainerName: "my-ai-app"
ContainerPort: 8080
PlatformVersion: "LATEST"
NetworkConfiguration:
AwsvpcConfiguration:
Subnets:
- "subnet-12345678"
- "subnet-87654321"
SecurityGroups:
- "sg-12345678"
AssignPublicIp: "ENABLED"
Hooks:
- BeforeInstall: "LambdaFunctionToValidateBeforeInstall"
- AfterInstall: "LambdaFunctionToValidateAfterInstall"
- AfterAllowTestTraffic: "LambdaFunctionToValidateAfterTestTrafficStarts"
- BeforeAllowTraffic: "LambdaFunctionToValidateBeforeAllowingProductionTraffic"
- AfterAllowTraffic: "LambdaFunctionToValidateAfterAllowingProductionTraffic"
設定 Pipeline 執行與通知 - AWS SNS
import boto3
sns = boto3.client('sns')
events = boto3.client('events')
# 建立SNS主題
topic_response = sns.create_topic(Name='pipeline-notifications')
topic_arn = topic_response['TopicArn']
# 訂閱email通知
sns.subscribe(
TopicArn=topic_arn,
Protocol='email',
Endpoint='your-email@example.com'
)
# 建立EventBridge規則
rule_response = events.put_rule(
Name='pipeline-state-change',
EventPattern=json.dumps({
"source": ["aws.codepipeline"],
"detail-type": ["CodePipeline Pipeline Execution State Change"],
"detail": {
"pipeline": ["my-ai-app-pipeline"],
"state": ["FAILED", "SUCCEEDED"]
}
}),
State='ENABLED',
Description='Notify on pipeline state changes'
)
# 將SNS設為目標
events.put_targets(
Rule='pipeline-state-change',
Targets=[
{
'Id': '1',
'Arn': topic_arn,
'InputTransformer': {
'InputPathsMap': {
'pipeline': '$.detail.pipeline',
'state': '$.detail.state',
'execution': '$.detail.execution-id'
},
'InputTemplate': '"Pipeline <pipeline> execution <execution> has <state>"'
}
}
]
)
print("通知設定完成")