背景与问题
审计工作需要对海量财务凭证、合同文本、业务日志进行合规性检查。传统方法依赖预设规则和人工抽样，存在三个明显痛点：
- 规则覆盖有限：舞弊手段不断变化，固定规则难以识别新型风险。
- 响应周期长：从问题出现到审计发现往往间隔数月，无法及时止损。
- 解释成本高：即使发现问题，也需要人工整理证据链与说明，耗时耗力。
随着大语言模型（LLM）的发展，审计智能化开始从“规则驱动”走向“智能体驱动”。借助AI辅助审计预警智能体平台，可以将大模型的语义理解、推理能力与实时数据流结合，形成持续监控、预警、解释的闭环。
方案设计
本平台采用事件驱动架构，以阿里云核心产品为基础：
- 数据接入层：通过 DataWorks 或 API 网关收集业务系统的交易流水、合同文本和审批日志。
- 存储与计算层：使用阿里云云数据库 RDS 存储结构化审计数据，使用对象存储 OSS 保存非结构化文档，函数计算 FC 承载无服务器计算逻辑。
- 智能预警层：调用阿里云百炼（Bailian）的大模型服务，对文本进行风险识别、实体抽取和解释生成；同时运行自定义规则引擎，对大额异常交易进行阈值报警。
- 预警输出层：将预警结果写入消息队列，通过钉钉机器人或邮件通知审计人员，并在审计工作台展示风险等级与推理依据。
整体流程：数据变化触发函数计算 → 调用百炼模型分析 → 结合规则引擎综合判定 → 生成预警工单。
实现步骤
1. 准备阿里云资源
开通阿里云百炼服务，在控制台创建API-KEY，并获取需要调用的模型名称（例如 qwen-plus）。同时创建函数计算服务与云数据库RDS实例，准备一张 audit_alerts 表用于存储预警记录。
2. 设计风险识别提示词
审计风险识别的关键在于让模型输出结构化结果。可以设计如下提示词模板：
你是一位审计专家。请分析以下业务描述，判断是否存在潜在合规风险。
输出格式：
{"risk_level": "高/中/低/无", "risk_type": "类型", "reason": "简短解释"}
业务描述：{text}
3. 编写函数计算代码
以下 Python 示例演示如何使用阿里云百炼的 OpenAI 兼容接口对一条审计文本进行风险分析，并依据结果写入数据库。为便于运行，使用 requests 库直接调用 HTTP 接口，也可以替换为官方 SDK。
import json
import os
import requests
import pymysql
# 环境变量配置
DASHSCOPE_API_KEY = os.environ["DASHSCOPE_API_KEY"]
MODEL_NAME = os.environ.get("MODEL_NAME", "qwen-plus")
# 百炼兼容OpenAI接口地址
BAILIAN_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
def analyze_risk(text: str):
"""调用百炼大模型，返回结构化风险结果"""
prompt = f"""你是一位审计专家。请分析以下业务描述，判断是否存在潜在合规风险。
输出格式：
{
{"risk_level": "高/中/低/无", "risk_type": "类型", "reason": "简短解释"}}
业务描述：{text}
"""
payload = {
"model": MODEL_NAME,
"messages": [{
"role": "user", "content": prompt}],
"temperature": 0.2 # 降低随机性，提高稳定性
}
headers = {
"Authorization": f"Bearer {DASHSCOPE_API_KEY}",
"Content-Type": "application/json"
}
resp = requests.post(BAILIAN_URL, headers=headers, data=json.dumps(payload, ensure_ascii=False))
resp.raise_for_status()
content = resp.json()["choices"][0]["message"]["content"]
# 清理可能的markdown代码块标记
content = content.replace("```json", "").replace("```", "").strip()
return json.loads(content)
def save_alert(biz_id: str, text: str, result: dict):
"""将预警结果写入RDS数据库"""
conn = pymysql.connect(
host=os.environ["RDS_HOST"],
user=os.environ["RDS_USER"],
password=os.environ["RDS_PASSWORD"],
database=os.environ["RDS_DB"],
charset="utf8mb4"
)
try:
with conn.cursor() as cursor:
sql = """
INSERT INTO audit_alerts (biz_id, biz_text, risk_level, risk_type, reason, created_at)
VALUES (%s, %s, %s, %s, %s, NOW())
"""
cursor.execute(sql, (biz_id, text, result["risk_level"], result["risk_type"], result["reason"]))
conn.commit()
finally:
conn.close()
def handler(event, context):
"""函数计算入口函数"""
# 假设event中传入业务ID和业务描述文本
event_data = json.loads(event)
biz_id = event_data["biz_id"]
text = event_data["text"]
try:
result = analyze_risk(text)
# 可在此处增加规则引擎：若金额超过阈值，强制提升风险等级
if event_data.get("amount", 0) > 1000000 and result["risk_level"] == "低":
result["risk_level"] = "中"
result["reason"] += "；且涉及大额交易"
save_alert(biz_id, text, result)
return {
"statusCode": 200,
"body": json.dumps({
"message": "预警分析完成", "result": result}, ensure_ascii=False)
}
except Exception as e:
return {
"statusCode": 500,
"body": json.dumps({
"error": str(e)})
}
4. 配置触发器与告警
将上述代码部署到函数计算后，为函数创建 HTTP 触发器或消息队列触发器。当业务系统推送新数据时，函数自动执行。预警结果写入数据库后，可以再创建一个定时查询函数，将未处理的“高”风险记录通过钉钉机器人推送。
5. 关键步骤说明
- 提示词工程：要求模型输出JSON格式，便于程序解析；设置 temperature=0.2 以保证结果相对稳定。
- 规则与AI结合：大模型对语义敏感，但对精确数值不敏感。因此将“金额超过百万”这类硬性条件放在规则引擎中，实现优势互补。
- 异常处理：模型调用可能失败或返回非JSON内容，应增加重试与解析容错机制。
注意事项
- 数据安全：审计数据敏感，调用百炼服务前应进行脱敏处理，或使用阿里云专有网络与 PrivateLink 避免公网传输。
- 成本控制：大模型调用按 Token 计费，建议对长文本先做截断或摘要，并设置函数并发上限，防止突发流量导致费用飙升。
- 模型幻觉：大模型可能生成看似合理但无依据的判断，需要将模型输出标记为“辅助建议”，最终结论由人工复核。建议将原始证据和模型理由一并存储，便于追溯。
- 监控与容错：为函数计算配置日志服务，对模型调用失败率、响应时长设置告警。可配置兜底规则，在模型不可用时自动降级到纯规则预警。
总结
本文提出的AI辅助审计预警智能体平台，利用阿里云百炼的强大语义理解能力，结合函数计算的无服务器弹性与云数据库的可靠存储，构建了一个可快速落地的审计预警闭环。通过“AI+规则”的双重校验，既提升了风险识别的广度与准度，又保留了审计所需的可解释性。未来可以进一步引入知识库（如审计准则、历史案例）增强模型的专业性，并利用阿里云机器学习平台对预警结果进行持续反馈训练，逐步实现审计预警的全面智能化。
