基于Bedrock Agentcore 实现智能成本分析与告警系统

from bedrock_agentcore.runtime import BedrockAgentCoreApp

# 初始化AgentCore Runtime App
app = BedrockAgentCoreApp()

# 注册Agent到Runtime
@app.agent()
def invoke_agent(prompt: str) -> str:
    return cost_agent.chat(prompt)

在deploy.py 中，完成镜像打包和agent推送

# AgentCore Runtime创建
def create_agent_runtime():
    agentcore_client = boto3.client('bedrock-agentcore')
    
    response = agentcore_client.create_agent_runtime(
        agentRuntimeName=runtime_name,
        containerImage=container_image_uri,
        executionRoleArn=execution_role_arn,
        # 其他配置参数
    )
    
    return response['agentRuntimeArn']

def detect_cost_anomalies(days: int = 7) -> str:
    """
    检测AWS成本异常，使用机器学习算法识别异常模式
    """
    try:
        ce_client = boto3.client("ce")
        
        # 计算时间范围
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days)
        
        # 获取异常检测结果
        response = ce_client.get_anomalies(
            DateInterval={
                'StartDate': start_date.strftime('%Y-%m-%d'),
                'EndDate': end_date.strftime('%Y-%m-%d')
            },
            MaxResults=50
        )
        
        anomalies = response.get('Anomalies', [])
        
        if not anomalies:
            return f"过去{days}天未检测到成本异常"
        
        # 分析异常详情
        analysis = []
        total_impact = 0
        
        for anomaly in anomalies:
            impact = float(anomaly.get('Impact', {}).get('MaxImpact', 0))
            total_impact += impact
            
            # 获取异常根因分析
            root_causes = []
            for cause in anomaly.get('RootCauses', []):
                service = cause.get('Service', 'Unknown')
                region = cause.get('Region', 'Unknown')
                usage_type = cause.get('UsageType', 'Unknown')
                root_causes.append(f"{service} ({region}) - {usage_type}")
            
            analysis.append({
                'date': anomaly.get('AnomalyStartDate'),
                'impact': impact,
                'score': anomaly.get('AnomalyScore', {}).get('MaxScore', 0),
                'root_causes': root_causes
            })
        
        # 生成分析报告
        report = f"检测到 {len(anomalies)} 个成本异常\n"
        report += f"总影响金额: ${total_impact:.2f}\n\n"
        
        for i, item in enumerate(analysis[:3], 1):
            report += f"异常 {i}:\n"
            report += f" 日期: {item['date']}\n"
            report += f" 影响: ${item['impact']:.2f}\n"
            report += f" 异常分数: {item['score']:.1f}\n"
            report += f" 根因: {', '.join(item['root_causes'][:2])}\n\n"
        
        return report
        
    except Exception as e:
        return f"异常检测失败: {str(e)}"

def get_multi_account_costs(days: int = 30) -> str:
# 获取所有关联账户
        try:
            accounts_response = org_client.list_accounts()
            all_accounts = {
                acc['Id']: acc['Name'] 
                for acc in accounts_response['Accounts']
            }
        except Exception:
            # 如果不是组织主账户，使用当前账户
            sts_client = boto3.client("sts")
            current_account = sts_client.get_caller_identity()['Account']
            all_accounts = {current_account: "Current Account"}
        
        # 按账户分组获取成本数据
        response = ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.strftime('%Y-%m-%d'),
                'End': end_date.strftime('%Y-%m-%d')
            },
            Granularity='MONTHLY',
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'LINKED_ACCOUNT'}
            ]
        )
      # 按成本排序
        sorted_accounts = sorted(
            account_costs.items(), 
            key=lambda x: x[1], 
            reverse=True
        )
        report = f"多账户成本分析 (过去{days}天)\n"
        report += f"总成本: ${total_cost:.2f}\n"
        report += f"账户数量: {len(account_costs)}\n\n"
        
    return report

其他工具功能说明

def lambda_handler(event, context):
    """Lambda处理函数 - 每日成本异常检查"""
    
    agent_runtime_arn = os.environ['AGENT_RUNTIME_ARN']
    sns_topic_arn = os.environ['SNS_TOPIC_ARN']
    try:
        query = "检查过去7天的成本异常，如果发现异常请详细说明"
        
        response = agentcore_client.invoke_agent_runtime(
            agentRuntimeArn=agent_runtime_arn,
            qualifier='DEFAULT',
            payload=json.dumps({"prompt": query})
        )
 # 异常检测
        if detect_anomaly_keywords(agent_response):
            send_anomaly_alert(sns_client, sns_topic_arn, agent_response)
            return create_response(200, '发现成本异常，已发送告警', True, agent_response)
        else:
            return create_response(200, '成本检查正常，无异常', False, agent_response)
            
def detect_anomaly_keywords(agent_response):
    """基于关键词和上下文的异常检测"""
    anomaly_keywords = [
        '异常', '异常检测', '成本激增', '超出预期', 
        '显著增长', '预算超支', '成本飙升', '异常波动'
    ]
    
    # 关键词匹配
    has_keywords = any(keyword in agent_response for keyword in anomaly_keywords)
    
    # 上下文分析（简化版）
    negative_indicators = ['正常', '未检测到', '无异常', '稳定']
    has_negative = any(indicator in agent_response for indicator in negative_indicators)
    
    return has_keywords and not has_negative

cost_optimization_agent.py          # 主代理文件
├── tools/
│   ├── cost_explorer_tools.py      # Cost Explorer API工具
│   ├── budget_tools.py             # Budgets API工具
│   └── multi_account_tools.py      # 多账户成本分析工具
├── test_local.py                   # 本地基础功能测试
├── test_agentcore_runtime.py       # runtime功能测试
└── deploy.py                       # AgentCore部署脚本