随着数字化转型成为企业发展的必答题，财务管理智能化已成为提升企业竞争力的关键。上一篇《基于 AI 实现阿里云的智能财务管家》中，我们以 AI 技术为笔，阿里云为纸，初步勾勒出智能财务管家的轮廓。如今，我们再度启程，为这一创新成果增添新的功能模块，旨在让财务工作效率与精准度再上一个台阶。
01
余额与信控
相较于上一篇文章所阐述的功能体系，本次迭代实现了技术路径与应用价值的双重进阶：
- 数据获取革新：经严谨的技术验证，阿里云账号余额与信用控制额度存在数据采集盲区。为此，我们对接阿里云官方 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(months=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 += "
{}（{}）:
可用额度为{}元
信控额度为{}元
前三个月消费金额为 {}".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())百炼
第一由意图分类节点判断用户问题，若涉及余额、信控相关内容，调用上文的函数计算代码获取数据，再通过引导大模型节点，结合用户问题与获取数据进行答复；若问题与余额、信控无关，因账单机器人无法回应其他问题，触发文本转换节点，告知用户暂无法回答，提议联系工程师 。
效果展示
如下图所示，账单机器人会显示每个阿里云账户的当前可用额度、信控额度以及前三个月的消费情况，并据此预测每个账户的预计可用时长。基于这些数据，系统会提出提议，例如从某个账户划拨部分可用额度给另一个账户，以更合理地分配资源，确保所有账户都有足够的可用额度来满足需求。
02
鉴 权
在百炼中使用鉴权功能主要出于架构适配和安全管控的双重考量。由于采用计算巢替代传统代码对接百炼与飞书，而计算巢本身缺乏鉴权能力，因此将鉴权逻辑前置到百炼工作流应用中是必要的架构调整。通过飞书机器人将用户 ID（user_id）传递至百炼，在百炼侧实现用户身份验证与权限校验，能够有效控制对账单数据的访问权限。这种设计既弥补了计算巢在安全管控上的不足，又利用百炼的工作流能力实现了鉴权与业务逻辑的解耦，确保只有经过授权的用户才能获取敏感的账单数据，同时保持了系统间的集成灵活性与安全性。
飞书应用
在配置前请确保账单机器人有“获取用户 user ID”的权限，否则计算巢会收不到user ID。
计算巢
进入计算巢连接流，编辑“阿里云百炼”连接器。
如下图所示配置“工作流应用自定义参数传递”，左边填写“userId”，右边插入变量“Node1.userId”。
百炼
在“开始”节点添加输入变量，变量名为“userId”，类型为“String”。在“条件判断”节点判断用户，如果提问的用户在“添加判断”节点内，则进行下一步操作，否则会告知用户权限不足。
效果展示
如下图所示，若账单机器人检测到用户未通过鉴权验证，将自动触发权限提示机制，明确告知用户其账号当前暂未开通该功能的使用权限。
03
鉴查询包年包月订单
分账账单数据在计费模式维度存在天然缺失，无法对包年包月与按量付费的计费类型进行有效区分。该类精细化数据需通过阿里云官方 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 等），并标注 “新购”” 续费 ” 等订单类型；③深入分析订单详情，定位费用占比突出的花费最高的实例。
04
Help
为降低用户的使用门槛，账单机器人特别配置了「help 菜单」引导功能，协助初用者快速了解有哪些账单查询能力，让用户从首次接触即可轻松掌握系统的操作逻辑与功能边界。
飞书应用
在飞书开发者后台配置账单机器人，在菜单配置中添加主菜单配置，名称为“help”，响应动作为“发送文字消息”。实现点击“help”菜单后，会给机器人发送help消息。
百炼
在百炼工作流应用中，意图分类节点会第一对用户输入进行语义解析，判断是否触发 “help” 意图。若识别为 “help” 意图，系统将自动激活文本转换节点，向用户输出”账单机器人具备哪些功能”的固定应答内容。
效果展示
如下图所示，当用户点击账单机器人的「help 菜单」时，飞书将自动向机器人发送 “help” 指令。此时，账单机器人会立即响应并触发功能说明机制，以清晰简洁的语言向用户罗列其具备的各项功能，协助用户快速了解机器人的服务范围与使用方法。
05
总 结
相较于上篇文章，此次优化在多方面实现功能升级：通过阿里云 SDK 突破分账账单限制，实现多账号余额与信控数据的精准获取；完善鉴权机制，未通过验证时明确提示账号权限状态；基于函数计算架构优化包年包月订单查询功能，支持按时间范围检索、金额排序及实例消费分析；优化账单机器人的 help 菜单交互逻辑，通过意图分类自动触发功能说明。
愿账单机器人在 AI 能力深化、多维度数据整合与交互体验升级的持续迭代中不断进化，以更智能的财务洞察、更高效的流程自动化与更人性化的交互设计，逐步成长为企业云财务管控的全能助手。
如有相关问题，请在文章后面给小编留言，小编安排作者第一时间和您联系，为您答疑解惑。
暂无评论内容