基于AI實現阿里云的智能財務管家
1.余額與信控
本次迭代實現了技術路徑與應用價值的雙重進階：
- 數據獲取革新：經嚴謹的技術驗證，阿里云賬號余額與信用控制額度存在數據采集盲區。為此，我們對接阿里云官方 SDK，構建起直連數據源的底層鏈路，確保數據獲取的完整性與實時性。
- 多賬號協同升級：打造智能多賬號數據處理中樞，突破單賬號數據獲取的局限。該功能支持在復雜多賬號架構下，并行抓取各賬號余額與信控數據，顯著提升數據采集效率。
- 智能預測賦能：依托歷史數據分析引擎，系統自動回溯各賬號近三個月的資源使用軌跡，結合動態消費模型，精準測算余額可用時長。
- 全局優化建議：創新性引入數據整合與智能決策模塊，系統自動聚合多賬號財務數據，基于機器學習算法分析各賬號的資源消耗特性，生成信控額度合理劃撥方案。
流程圖
圖片
操作步驟
FC函數計算
異步并發查詢阿里云賬戶財務信息，通過配置文件讀取多個賬戶的憑證，利用阿里云 BSS API 批量查詢賬戶的可用額度、信控額度及近三個月消費記錄，采用線程池和信號量控制并發請求，結合自動重試機制確保網絡穩定性，最終匯總所有賬單數據。具體代碼如下所示：
config.ini
[ALIYUN]
account_info_dic = {"阿里云賬號1":{"uid":"請填寫阿里云賬號uid","ak":"請填寫ak","sk":"請填寫sk"},"阿里云賬號2":{"uid":"請填寫阿里云賬號uid","ak":"請填寫ak","sk":"請填寫sk"}}index.py
import json,configparser,logging,asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List
from datetime import datetime
from dateutil.relativedelta import relativedelta
from tenacity import retry, stop_after_attempt, wait_fixed
from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_credentials.models import Config as CreConfig
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_bssopenapi20171214 import models as bss_open_api_20171214_models
# 日志配置
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 全局常量定義（避免污染類）
current = datetime.now()
months = [(current - relativedelta(mnotallow=i)).strftime("%Y-%m") for i in range(3, 0, -1)]
class Billing:
# 類級配置
config = configparser.ConfigParser()
config.read('config.ini')
account_info_dic = json.loads(config['ALIYUN']['account_info_dic'])
executor = ThreadPoolExecutor(max_workers=100)
semaphore = asyncio.Semaphore(100)
months = months # 將 months 作為類屬性引用外部定義
@staticmethod
def create_client(ak: str, sk: str) -> BssOpenApi20171214Client:
"""創建阿里云 BSS API 客戶端"""
credentials_config = CreConfig(
type='access_key',
access_key_id=ak,
access_key_secret=sk,
)
credential = CredentialClient(credentials_config)
client_config = open_api_models.Config(
credential=credential
)
client_config.endpoint = 'business.aliyuncs.com'
return BssOpenApi20171214Client(client_config)
@staticmethod
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def get_credit_amount(args: List[str]) -> tuple:
"""獲取賬戶信用額度"""
client = args[0]
try:
runtime = util_models.RuntimeOptions()
res = client.query_account_balance_with_options(runtime).body.data
available_amount = res.available_amount
credit_amount = res.credit_amount
return available_amount, credit_amount
except Exception as e:
logger.error("獲取信用額度失敗", exc_info=True)
return (0, 0)
@staticmethod
async def async_get_credit_amount(client: BssOpenApi20171214Client):
"""異步調用 get_credit_amount 方法"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(Billing.executor, Billing.get_credit_amount, [client])
@staticmethod
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def get_history_billing_data(client, account_uid: str, months_list: list) -> str:
"""獲取歷史賬單數據"""
history_amount = ""
for billing_cycle in months_list:
page_num = 1
page_size = 100
pretax_amount = 0
while True:
query_account_bill_request = bss_open_api_20171214_models.QueryAccountBillRequest(
billing_cycle=billing_cycle,
owner_id=account_uid,
granularity='MONTHLY',
page_num=page_num,
page_size=page_size
)
runtime = util_models.RuntimeOptions()
res = client.query_account_bill_with_options(query_account_bill_request, runtime).body.data.items.item
if not res:
break
for re in res:
pretax_amount += re.pretax_amount
page_num += 1
history_amount += f"{billing_cycle}: {round(pretax_amount, 2)}元 "
return history_amount
@staticmethod
async def async_get_history_billing_data(client, account_uid: str, months_list: list):
"""異步調用 get_history_billing_data 方法"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(Billing.executor, Billing.get_history_billing_data, client, account_uid, months_list)
@staticmethod
async def process_account(account_name: str, account_info: dict):
"""處理單個賬戶的信用額度與賬單信息"""
client = Billing.create_client(account_info["ak"], account_info["sk"])
async with Billing.semaphore:
available_amount, credit_amount = await Billing.async_get_credit_amount(client)
history_amount = await Billing.async_get_history_billing_data(client, account_info["uid"], Billing.months)
return account_name, account_info["uid"], available_amount, credit_amount, history_amount
@staticmethod
async def get_all_credit_amount_async():
"""并發獲取所有賬戶信用額度和賬單信息"""
tasks = [
Billing.process_account(account_name, account_info)
for account_name, account_info in Billing.account_info_dic.items()
]
results = await asyncio.gather(*tasks)
message = ""
for account_name, uid, available_amount, credit_amount, history_amount in results:
message += "\n{}（{}）:\n 可用額度為{}元\n 信控額度為{}元\n 前三個月消費金額為 {}".format(
account_name, uid, available_amount, credit_amount, history_amount
)
logger.info(message)
return message
@staticmethod
async def main():
"""主入口方法"""
return await Billing.get_all_credit_amount_async()
def main(event,context):
"""模塊化主函數入口"""
return asyncio.run(Billing.main())百煉
首先由意圖分類節點判斷用戶問題，若涉及余額、信控相關內容，調用上文的函數計算代碼獲取數據，再通過引導大模型節點，結合用戶問題與獲取數據進行答復；若問題與余額、信控無關，因賬單機器人無法回應其他問題，觸發文本轉換節點，告知用戶暫無法回答，建議聯系工程師 。
圖片
效果展示
如下圖所示，賬單機器人會顯示每個阿里云賬戶的當前可用額度、信控額度以及前三個月的消費情況，并據此預測每個賬戶的預計可用時長。基于這些數據，系統會提出建議，例如從某個賬戶劃撥部分可用額度給另一個賬戶，以更合理地分配資源，確保所有賬戶都有足夠的可用額度來滿足需求。
圖片
2.鑒權
在百煉中使用鑒權功能主要出于架構適配和安全管控的雙重考量。由于采用計算巢替代傳統代碼對接百煉與飛書，而計算巢本身缺乏鑒權能力，因此將鑒權邏輯前置到百煉工作流應用中是必要的架構調整。通過飛書機器人將用戶 ID（user_id）傳遞至百煉，在百煉側實現用戶身份驗證與權限校驗，能夠有效控制對賬單數據的訪問權限。這種設計既彌補了計算巢在安全管控上的不足，又利用百煉的工作流能力實現了鑒權與業務邏輯的解耦，確保只有經過授權的用戶才能獲取敏感的賬單數據，同時保持了系統間的集成靈活性與安全性。
飛書應用
在配置前請確保賬單機器人有“獲取用戶 user ID”的權限，否則計算巢會收不到user ID。
圖片
計算巢
進入計算巢連接流，編輯“阿里云百煉”連接器。
圖片
如下圖所示配置“工作流應用自定義參數傳遞”，左邊填寫“userId”，右邊插入變量“Node1.userId”。
圖片
百煉
在“開始”節點添加輸入變量，變量名為“userId”，類型為“String”。在“條件判斷”節點判斷用戶，如果提問的用戶在“添加判斷”節點內，則進行下一步操作，否則會告訴用戶權限不足。
效果展示
如下圖所示，若賬單機器人檢測到用戶未通過鑒權驗證，將自動觸發權限提示機制，明確告知用戶其賬號當前暫未開通該功能的使用權限。
圖片
3.鑒查詢包年包月訂單
分賬賬單數據在計費模式維度存在天然缺失，無法對包年包月與按量付費的計費類型進行有效區分。該類精細化數據需通過阿里云官方 SDK 接口實現深度調取，方可獲取包含計費模式的完整數據集合。
FC函數計算
對接阿里云 BSS OpenAPI SDK，實現了包年包月（預付費）訂單數據的自動化查詢與統計分析，指定時間范圍檢索已支付訂單，自動處理分頁數據并通過 API 限流機制控制調用頻率；將英文產品代碼映射為中文名稱（如 DataWorks、MongoDB 等），解析訂單詳情中的實例 ID 與優惠券抵扣金額，按產品類型和訂單類型（新購、續費等）進行金額匯總，最終以 JSON 格式返回總消費金額、分類統計結果及訂單詳情。具體代碼如下所示：
# -*- coding: utf-8 -*-
# 導入所需的阿里云SDK模塊和其他依賴庫
from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_bssopenapi20171214 import models as bss_open_api_20171214_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_credentials.models import Config as CreConfig
from ratelimit import limits, sleep_and_retry # 用于限制API調用頻率
import logging, json
# 初始化日志記錄器
logger = logging.getLogger()
# 阿里云訪問密鑰（需要替換為真實值）
ak = "請輸入AK"
sk = "請輸入SK"
# 訂單類型映射表，將英文訂單類型轉換為中文描述
order_type_mapping = {'New': '新購','Renew': '續費','Upgrade': '升級','TempUpgrade': '短時升級','Downgrade': '降級','Refund': '退款','Convert': '付費類型轉換','ResizeDisk': '調整磁盤大小','CompensatoryRenew': '補償續費','IncreaseUpgrade': '帶寬升級','Exchange': '換購','ChangeOperatingSystem': '更換操作系統'}
class Sample:
def __init__(self):
pass
@staticmethod
def create_client() -> BssOpenApi20171214Client:
"""創建BSS Open API客戶端"""
credentialsConfig = CreConfig(type='access_key',access_key_id=ak,access_key_secret=sk,)
credential = CredentialClient(credentialsConfig)
config = open_api_models.Config(credential=credential)
config.endpoint = f'business.vpc-proxy.aliyuncs.com'
return BssOpenApi20171214Client(config)
@sleep_and_retry
@limits(calls=10, period=1) # 限制每秒最多調用10次
@staticmethod
def GetOrderDetail(order_id):
"""獲取訂單詳情信息，包括實例ID和優惠券抵扣金額"""
client = Sample.create_client()
get_order_detail_request = bss_open_api_20171214_models.GetOrderDetailRequest(order_id=order_id)
runtime = util_models.RuntimeOptions()
order_list = client.get_order_detail_with_options(get_order_detail_request, runtime).body.data.order_list.order
voucher_amount = 0 # 優惠券抵扣金額初始化
# 遍歷訂單列表，計算優惠券抵扣金額
for order in order_list:
instance_ids = order.instance_ids # 獲取實例ID
voucher = order.extend_infos.get("DeductedByCoupons", "0") # 獲取優惠券抵扣信息
voucher_amount += float(voucher) # 累加優惠券金額
return instance_ids, voucher_amount # 返回實例ID和優惠券金額
@sleep_and_retry
@limits(calls=10, period=1) # 限制API調用頻率
@staticmethod
def main(start_time, end_time):
"""主函數：查詢指定時間范圍內的訂單信息"""
client = Sample.create_client() # 創建API客戶端
page_num = 1 # 當前頁碼
page_size = 100 # 每頁數量
result_dict = {} # 結果字典，存儲統計信息
while True:
# 創建查詢訂單的請求
query_orders_request = bss_open_api_20171214_models.QueryOrdersRequest(
create_time_start=start_time, # 開始時間
create_time_end=end_time, # 結束時間
page_num=page_num, # 當前頁碼
page_size=page_size, # 每頁數量
subscription_type='Subscription' # 查詢預付費訂單
)
runtime = util_models.RuntimeOptions()
# 記錄日志：開始查詢第N頁數據
logger.info("第{}頁訂單數據開始查詢".format(page_num))
# 發送請求并獲取響應
order_list = client.query_orders_with_options(query_orders_request, runtime).body.data.order_list.order
if order_list == []: # 如果沒有更多數據，退出循環
break
else:
page_num += 1 # 否則增加頁碼
order_info = [] # 存儲訂單詳細信息
# 處理每個訂單
for order in order_list:
if order.payment_status == "Paid": # 只處理已支付的訂單
logger.info(str(order.__dict__)) # 記錄訂單原始信息
product_code = order.product_code # 獲取產品代碼
# 對一些特殊產品代碼進行重命名
if product_code == 'dide': product_code = 'DataWorks'
elif product_code == 'dds': product_code = 'MongoDB'
elif product_code in ['kvstore', 'redisa']: product_code = 'Redis'
elif product_code == 'odpsplus': product_code = 'MaxCompute'
# 獲取訂單類型、金額等信息
order_type = order_type_mapping[order.order_type]
pretax_amount = order.pretax_amount
order_id = order.order_id
# 獲取訂單的實例ID和優惠券抵扣金額
instance_ids, voucher_amount = Sample.GetOrderDetail(order_id)
# 計算實際金額（原價+優惠券）
amount = pretax_amount + voucher_amount
# 收集訂單信息
order_info.append({"order_id": order_id,"order_type": order_type,"product_code": product_code,"amount": amount,"instance_ids": instance_ids})
# 匯總統計信息
key = (product_code, order_type)
if key in result_dict: result_dict[key] += amount
else: result_dict[key] = amount
return result_dict, order_info # 返回結果
def main(event, context):
"""處理事件觸發的入口函數"""
data = event.decode('utf-8') # 解碼輸入數據
logger.info("收到的數據：{}".format(data)) # 記錄接收到的數據
# 解析JSON數據中的時間范圍
time = json.loads(data)["time"].split(",")
start_time = time[0] # 開始時間
end_time = time[1] # 結束時間
# 記錄時間范圍
logger.info("開始時間：{}".format(start_time))
logger.info("結束時間：{}".format(end_time))
# 查詢訂單信息
result_dict, order_info = Sample.main(start_time, end_time)
# 計算總金額
amount = 0
for key, value in result_dict.items():
amount += value
# 記錄統計結果
logger.info("總金額：{}".format(str(amount)))
logger.info("訂單匯總：{}".format(str(result_dict)))
logger.info("訂單詳情：{}".format(str(order_info)))
# 構造響應JSON
response_json = json.dumps({"amount": str(amount),"result_dict": str(result_dict),"order_info": str(order_info)}, indent=4)
# 記錄返回結果
logger.info("返回結果：{}".format(response_json))
return response_json # 返回結果百煉
智能體應用
借助百煉智能體應用的語義解析能力，可精準分析用戶提問意圖，從中提取所需查詢的時間范圍，并將其轉換為適配 FC 函數中阿里云 SDK 要求的格式。以 “查詢昨天的賬單” 為例，智能體能夠基于當前日期（2025-06-30）自動推算 “昨天” 的時間區間，生成符合 ISO 8601 標準的 UTC 時間格式字符串“2025-06-29T00:00:00Z,2025-06-30T00:00:00Z”，確保起始與結束時間滿足 SDK 接口的參數規范。
圖片
工作流應用
工作流從接收輸入的賬單查詢時間范圍開始，通過函數計算節點調用上文的函數，利用阿里云 SDK 獲取該時間段內包年包月的賬單數據，之后引導大模型節點根據用戶問題，不僅答復查詢到的總金額，還會對數據按金額從大到小排序并標注續費、新購等賬單類型，同時分析訂單詳情以確定花費最多的主要實例。
圖片
智能體編排應用
通過「智能體編排應用」的流程引擎，將「智能體應用」與「工作流應用」構建為協同處理鏈路：當系統接收到用戶查詢請求時，「意圖分類節點」首先對語義進行解析，自動觸發負責時間參數提取的「智能體應用」；該智能體將自然語言時間轉換為標準化格式后，無縫傳遞至「工作流應用」執行賬單數據獲取邏輯；工作流完成包年包月訂單的檢索、分析與匯總后，最終以自然語言形式向用戶反饋賬單查詢結果。整個鏈路通過節點間的參數傳遞與能力協同，實現從用戶提問到數據響應的全流程自動化閉環。
圖片
效果展示
當用戶查詢 "昨天包年包月訂單詳情" 時，賬單機器人響應內容包含三部分核心信息：①匯總展示昨日包年包月訂單的總消費金額；②按消費金額從高到低排序，依次列出各產品的具體花費（如 ECS、RDS 等），并標注 "新購"" 續費 " 等訂單類型；③深入分析訂單詳情，定位費用占比突出的花費最高的實例。
圖片
4.Help
為降低用戶的使用門檻，賬單機器人特別配置了「help 菜單」引導功能，幫助初用者快速了解有哪些賬單查詢能力，讓用戶從首次接觸即可輕松掌握系統的操作邏輯與功能邊界。
飛書應用
在飛書開發者后臺配置賬單機器人，在菜單配置中添加主菜單配置，名稱為“help”，響應動作為“發送文字消息”。實現點擊“help”菜單后，會給機器人發送help消息。
圖片
百煉
在百煉工作流應用中，意圖分類節點會首先對用戶輸入進行語義解析，判斷是否觸發 "help" 意圖。若識別為 "help" 意圖，系統將自動激活文本轉換節點，向用戶輸出"賬單機器人具備哪些功能"的固定應答內容。
圖片
效果展示
如下圖所示，當用戶點擊賬單機器人的「help 菜單」時，飛書將自動向機器人發送 "help" 指令。此時，賬單機器人會立即響應并觸發功能說明機制，以清晰簡潔的語言向用戶羅列其具備的各項功能，幫助用戶快速了解機器人的服務范圍與使用方法。
圖片
5.總結
通過阿里云 SDK 突破分賬賬單限制，實現多賬號余額與信控數據的精準獲取；完善鑒權機制，未通過驗證時明確提示賬號權限狀態；基于函數計算架構優化包年包月訂單查詢功能，支持按時間范圍檢索、金額排序及實例消費分析；優化賬單機器人的 help 菜單交互邏輯，通過意圖分類自動觸發功能說明。
愿賬單機器人在 AI 能力深化、多維度數據整合與交互體驗升級的持續迭代中不斷進化，以更智能的財務洞察、更高效的流程自動化與更人性化的交互設計，逐步成長為企業云財務管控的全能助手。
