AI Agent Harness实时风控规则配置：构建智能风险防御系统

---

一、 引言 (Introduction)

1.1 钩子 (The Hook)

你是否曾经在凌晨3点被手机上的可疑交易提醒惊醒？或者在使用新应用时，突然发现自己的账户被异常操作？在当今数字化世界中，金融欺诈、身份盗用和网络攻击已经成为常态。据统计，2023年全球因网络犯罪造成的经济损失超过8万亿美元，这一数字令人震惊。更令人担忧的是，随着AI技术的快速发展，攻击者的手段也变得更加复杂和难以预测。

在这场攻防之战中，传统的规则引擎往往显得力不从心。它们要么过于僵化，无法应对新型攻击；要么响应太慢，等到检测到问题时，损失已经造成。那么，有没有一种方法能够将人工智能的智能性与规则引擎的确定性结合起来，构建一个既聪明又可靠的实时风控系统呢？

答案是肯定的——这就是我们今天要探讨的AI Agent Harness实时风控规则配置。

1.2 定义问题/阐述背景 (The “Why”)

在深入探讨技术细节之前，让我们先明确几个关键概念：

AI Agent：简单来说，AI Agent是一种能够感知环境、做出决策并采取行动的智能体。它结合了大语言模型(LLM)的理解能力、规划能力和工具使用能力，能够自主完成复杂任务。

Harness：在技术语境中，Harness通常指的是一个框架或平台，用于管理、协调和监控各种组件的运行。AI Agent Harness就是专门为AI Agent设计的这样一个平台。

实时风控：指的是在业务事件发生的瞬间，对其进行风险评估和决策的过程。这要求系统必须具备极低的延迟（通常在毫秒级别）和极高的准确性。

传统风控系统面临的主要挑战包括：

1. 规则更新滞后：新型欺诈手段层出不穷，而传统规则的制定和更新往往需要数天甚至数周时间。
2. 误判率高：过于严格的规则会影响用户体验，过于宽松的规则又会放过风险。
3. 无法处理复杂模式：许多现代攻击采用多步骤、多维度的复杂模式，传统规则难以识别。
4. 扩展性差：随着业务增长，数据量和交易频率呈指数级上升，传统系统往往难以承受。

AI Agent Harness实时风控规则配置正是为了解决这些问题而诞生的。它通过将AI的智能分析能力与规则引擎的确定性执行能力相结合，构建了一个既能快速适应新威胁，又能保持高可靠性的风控系统。

1.3 亮明观点/文章目标 (The “What” & “How”)

在这篇文章中，我们将带你从零开始，全面了解AI Agent Harness实时风控规则配置的理论基础和实践方法。具体来说，你将学到：

1. 核心概念：深入理解AI Agent、实时风控以及它们如何结合在一起。
2. 架构设计：了解AI Agent Harness风控系统的整体架构和关键组件。
3. 规则配置方法：掌握如何在这个框架中配置和管理风控规则。
4. 实战演练：通过一个完整的案例，学习如何构建一个简单但功能完整的AI Agent风控系统。
5. 最佳实践：了解在实际应用中需要注意的问题和优化策略。

我们将采用理论与实践相结合的方式，不仅解释"是什么"和"为什么"，更重要的是告诉你"怎么做"。无论你是风控专家、AI工程师还是系统架构师，相信这篇文章都能给你带来有价值的启发。

---

二、 基础知识/背景铺垫 (Foundational Concepts)

2.1 核心概念定义

在开始深入探讨AI Agent Harness实时风控规则配置之前，让我们先建立一些必要的概念基础。这些概念将贯穿全文，是我们理解后续内容的关键。

2.1.1 人工智能代理 (AI Agent)

AI Agent是近年来人工智能领域最热门的概念之一。从根本上说，AI Agent是一种能够自主感知环境、做出决策并采取行动的智能系统。它不仅仅是一个被动的工具，而是一个具有主动性和目标导向性的实体。

一个完整的AI Agent通常包含以下几个核心组件：

1. 感知模块：负责收集和理解环境信息。这可能包括文本输入、传感器数据、用户行为等。
2. 推理引擎：基于感知到的信息和内部知识，进行推理和决策。
3. 行动模块：将决策转化为具体的行动，如调用API、发送通知、修改数据等。
4. 记忆系统：存储历史信息、经验和知识，用于支持未来的决策。
5. 目标管理：设定和优化Agent的目标，确保其行动符合预期。

在风控场景中，AI Agent的作用尤为重要。它可以实时分析交易数据，识别潜在风险，并自动采取相应的措施，如拒绝交易、要求额外验证或通知人工审核。

2.1.2 实时风控系统

实时风控系统是一种专门设计用于在业务事件发生的瞬间进行风险评估和决策的系统。与传统的批处理风控系统不同，实时风控系统对延迟有极其严格的要求——通常需要在几十毫秒甚至几毫秒内完成从数据接收、分析到决策的全过程。

一个典型的实时风控系统包含以下几个关键要素：

1. 数据接入层：负责实时接收和处理各种来源的数据流。
2. 特征计算层：从原始数据中提取和计算各种风险特征。
3. 风险评估层：基于特征和规则，对事件进行风险评分和分类。
4. 决策执行层：根据风险评估结果，执行相应的决策动作。
5. 监控和反馈层：监控系统性能和决策效果，并通过反馈机制持续优化。

实时风控系统在金融、电商、游戏等多个领域都有广泛应用。例如，在支付场景中，它可以实时判断一笔交易是否存在欺诈风险；在登录场景中，它可以评估用户的登录行为是否正常。

2.1.3 规则引擎

规则引擎是一种将业务规则从应用程序代码中分离出来的组件。它允许业务人员或风险分析师通过配置而非编程的方式来定义和修改规则，从而大大提高了系统的灵活性和响应速度。

规则引擎通常包含以下几个核心部分：

1. 规则库：存储所有定义好的规则。
2. 事实库：存储当前需要评估的数据或事实。
3. 推理引擎：根据规则库中的规则和事实库中的事实，进行推理和决策。
4. 规则管理界面：允许用户创建、修改和管理规则。

在风控场景中，规则引擎的作用至关重要。它可以实现各种风控策略，如：

• 如果用户从陌生地理位置登录，则要求二次验证
• 如果交易金额超过用户历史平均值的5倍，则拒绝交易
• 如果同一IP地址在短时间内尝试多次失败登录，则暂时阻止该IP

然而，传统的规则引擎也有其局限性。它们通常只能处理明确的、预先定义好的规则，对于模糊的、复杂的或新型的风险模式往往无能为力。这正是AI Agent能够发挥作用的地方。

2.2 AI Agent与传统风控的融合

现在我们已经了解了AI Agent和实时风控系统的基本概念，接下来让我们探讨它们是如何结合在一起，形成AI Agent Harness实时风控规则配置系统的。

2.2.1 传统风控的局限性

如前所述，传统的风控系统主要依赖于规则引擎，虽然它们在处理明确的、已知的风险模式时表现良好，但在面对以下情况时往往显得力不从心：

1. 未知风险模式：新型欺诈手段不断出现，传统规则无法及时识别。
2. 复杂关联关系：许多现代攻击涉及多个实体、多个步骤和多种数据类型，传统规则难以捕捉这种复杂的关联关系。
3. 上下文理解：传统规则往往只关注表面特征，而缺乏对上下文的深入理解。
4. 自适应能力：传统规则需要人工更新，无法自动适应环境变化。

这些局限性导致传统风控系统要么误报率过高，影响用户体验；要么漏报率过高，造成实际损失。

2.2.2 AI Agent的优势

相比之下，AI Agent在以下方面具有明显优势：

1. 模式识别能力：通过机器学习和深度学习技术，AI Agent能够识别复杂的、非线性的风险模式。
2. 自适应学习：AI Agent可以从历史数据和反馈中持续学习，自动适应新的风险模式。
3. 上下文理解：借助大语言模型(LLM)，AI Agent能够理解和分析非结构化数据，如文本、图像等，从而获得更全面的上下文信息。
4. 自主决策：AI Agent不仅能识别风险，还能根据情况自主决定采取什么行动。
5. 多Agent协作：多个AI Agent可以协同工作，分别关注不同的风险维度，然后综合判断。

2.2.3 最佳组合：AI+规则

然而，这并不意味着AI Agent可以完全取代传统的规则引擎。实际上，最有效的风控系统往往是两者的结合：

1. 规则处理确定性问题：对于已知的、明确的风险模式，传统规则仍然是最可靠、最高效的选择。
2. AI处理不确定性问题：对于复杂的、未知的或模糊的风险模式，AI Agent能够发挥其优势。
3. AI增强规则：AI可以帮助发现新的规则，或优化现有规则的参数。
4. 规则约束AI：规则可以为AI设定边界和约束，确保其决策符合业务要求和合规标准。

AI Agent Harness实时风控规则配置系统正是基于这种理念设计的。它提供了一个统一的平台，让AI Agent和规则引擎能够协同工作，发挥各自的优势。

2.3 相关技术概览

在构建AI Agent Harness实时风控规则配置系统时，我们会用到多种技术。让我们简要了解一下这些技术：

2.3.1 大语言模型(LLM)

大语言模型是AI Agent的"大脑"，为其提供理解、推理和生成能力。在风控场景中，LLM可以用于：

• 分析和理解非结构化数据，如用户评论、支持工单等
• 生成风险分析报告
• 解释风控决策的原因
• 协助规则的生成和优化

目前常用的LLM包括GPT系列、Claude、Llama等。选择合适的LLM需要考虑性能、成本、隐私等多种因素。

2.3.2 流式计算框架

实时风控系统需要处理大量的实时数据流，因此流式计算框架是必不可少的。流式计算框架能够处理无限的数据流，提供低延迟的计算能力。

常用的流式计算框架包括：

• Apache Kafka Streams
• Apache Flink
• Apache Storm
• Spark Streaming

在选择流式计算框架时，需要考虑吞吐量、延迟、状态管理、容错能力等因素。

2.3.3 特征存储

特征存储是一种专门用于存储和管理机器学习特征的系统。在实时风控场景中，特征存储能够：

• 提供低延迟的特征读取能力
• 确保训练和推理时使用的特征一致性
• 支持特征的版本管理和血缘追踪
• 预计算和缓存常用特征

常用的特征存储包括Feast、Tecton、Hopsworks等。

2.3.4 图数据库

许多风险模式涉及实体之间的复杂关联关系，如图(Graph)结构。图数据库专门用于存储和查询这种图结构数据，能够高效地发现实体之间的关联和模式。

在风控场景中，图数据库可以用于：

• 识别欺诈团伙
• 发现资金流转的异常模式
• 分析用户之间的社交关系

常用的图数据库包括Neo4j、JanusGraph、Amazon Neptune等。

2.3.5 规则引擎

如前所述，规则引擎是风控系统的核心组件之一。在AI Agent Harness系统中，我们通常会选择或构建一个灵活、高效、易于集成的规则引擎。

常用的规则引擎包括：

• Drools
• OWL (Oracle Business Rules)
• IBM Operational Decision Manager
• 自研规则引擎

对于AI Agent Harness系统，我们更倾向于选择轻量级、易于与AI组件集成的规则引擎，或者根据需求自研。

---

三、 核心内容/实战演练 (The Core - “How-To”)

在这一部分，我们将深入探讨AI Agent Harness实时风控规则配置系统的核心内容，并通过一个实战案例，带你从零开始构建一个简化但功能完整的系统。

3.1 系统架构设计

首先，让我们来看一下AI Agent Harness实时风控规则配置系统的整体架构。一个完整的系统通常包含以下几个层次：

3.1.1 数据接入层

数据接入层是系统的入口，负责从各种数据源实时接收数据。这些数据源可能包括：

• 交易系统
• 用户行为日志
• 第三方数据服务
• 历史数据库

数据接入层的主要功能包括：

• 数据采集：从各种数据源获取数据
• 数据清洗：处理缺失值、异常值等
• 数据格式转换：将不同格式的数据转换为统一格式
• 数据路由：将数据发送到合适的处理单元

在实现上，我们通常会使用消息队列系统，如Kafka、RabbitMQ等，来实现数据的异步处理和解耦。

3.1.2 特征计算层

特征计算层负责从原始数据中提取和计算各种风险特征。这些特征是后续风险评估的基础。

特征可以分为几类：

• 基础特征：直接从原始数据中提取的特征，如交易金额、交易时间等
• 统计特征：基于历史数据计算的统计特征，如用户过去30天的平均交易金额等
• 聚合特征：将多个基础特征组合而成的特征
• 模型特征：通过机器学习模型计算得到的特征

特征计算层通常会结合实时计算和批量计算：

• 实时计算：使用流式计算框架，实时计算最新的特征
• 批量计算：使用批处理框架，定期计算和更新历史特征

特征存储在这个层次中起着关键作用，它能够提供低延迟的特征读取能力，并确保特征的一致性。

3.1.3 AI Agent层

AI Agent层是系统的核心，包含多个专门负责不同任务的AI Agent。这些Agent可以根据需要动态配置和扩展。

常见的AI Agent类型包括：

• 风险评估Agent：负责评估交易或事件的风险程度
• 模式识别Agent：负责识别已知和未知的风险模式
• 决策Agent：负责根据风险评估结果做出决策
• 规则生成Agent：负责自动发现和生成新的风控规则
• 协作Agent：负责协调多个Agent之间的工作

每个AI Agent都包含以下组件：

• 感知模块：接收和理解输入数据
• 推理模块：基于输入和知识进行推理
• 行动模块：执行具体的动作
• 记忆模块：存储历史信息和经验
• 通信模块：与其他Agent和系统组件通信

3.1.4 规则引擎层

规则引擎层与AI Agent层协同工作，负责处理明确的、已知的风险规则。

规则引擎层的主要功能包括：

• 规则存储：管理和存储所有风控规则
• 规则执行：根据输入数据执行规则
• 规则冲突解决：处理规则之间的冲突
• 规则版本管理：管理规则的版本和变更历史

规则可以通过多种方式定义：

• 决策表：适用于条件和结果都比较明确的规则
• 决策树：适用于有层次结构的规则
• 脚本：适用于复杂的、需要编程逻辑的规则
• 自然语言：借助LLM，可以使用自然语言定义规则

3.1.5 决策执行层

决策执行层负责根据AI Agent和规则引擎的决策结果，执行具体的行动。

常见的决策行动包括：

• 通过：允许交易或事件继续进行
• 拒绝：阻止交易或事件
• 审查：将交易或事件标记为需要人工审查
• 验证：要求用户进行额外的验证，如验证码、人脸识别等
• 通知：发送通知给相关人员或系统
• 限制：对用户或账户实施临时限制

决策执行层需要确保决策能够被可靠地执行，并且能够处理各种异常情况。

3.1.6 监控反馈层

监控反馈层负责监控系统的运行状态和决策效果，并通过反馈机制持续优化系统。

主要功能包括：

• 系统监控：监控系统的性能、可用性等指标
• 决策监控：监控决策的准确率、误报率、漏报率等指标
• 反馈收集：收集人工审核结果、用户反馈等信息
• 模型更新：基于反馈数据，更新和优化AI模型和规则
• 报表生成：生成各种报表，供业务人员和管理人员查看

3.2 核心概念与关系

在深入实现之前，让我们更清晰地定义一些核心概念以及它们之间的关系。

3.2.1 核心概念

1. 事件(Event)：系统处理的基本单位，如一次交易、一次登录尝试等。
2. 特征(Feature)：描述事件的属性，如交易金额、用户历史行为等。
3. 规则(Rule)：定义风险判断条件和相应行动的逻辑。
4. Agent：具有感知、推理和行动能力的智能实体。
5. 决策(Decision)：对事件的风险判断结果，如通过、拒绝、审查等。
6. 行动(Action)：基于决策执行的具体操作。
7. 反馈(Feedback)：关于决策质量的信息，用于优化系统。

3.2.2 概念关系

让我们用ER图来表示这些概念之间的关系：

这个ER图展示了各个核心概念之间的基本关系：

• 一个事件由多个特征描述
• 一个事件产生一个决策
• 决策受规则影响，由Agent做出
• 决策触发一个或多个行动
• 事件产生反馈，反馈用于优化规则和训练Agent

接下来，让我们看一下这些概念在系统中的交互流程：

这个时序图展示了一个典型事件的处理流程：

1. 客户端或业务系统发送事件到数据接入层
2. 数据接入层转发事件到特征计算层
3. 特征计算层计算相关特征
4. 特征计算层将事件和特征同时发送给规则引擎层和AI Agent层
5. 规则引擎层执行规则，AI Agent层进行推理
6. 两者都将决策发送给决策执行层
7. 决策执行层综合所有决策，做出最终决策
8. 决策执行层将决策结果返回给客户端
9. 同时，决策执行层将决策记录到监控反馈层
10. 监控反馈层收集反馈，用于更新规则和优化Agent

3.3 数学模型与算法

在AI Agent Harness实时风控规则配置系统中，我们会用到多种数学模型和算法。让我们来了解一些核心的模型和算法。

3.3.1 风险评分模型

风险评分是风控系统的核心功能之一。我们通常会使用一个综合的风险评分模型，将多个风险指标组合成一个单一的风险分数。

一个常用的风险评分模型是加权求和模型：

$$S=\sum _{i=1}^n{w}_i\cdot f_i(x)$$

其中：

• $S$ 是最终的风险评分
• $w_i$ 是第 $i$ 个特征的权重
• $f_i(x)$ 是第 $i$ 个特征的评分函数

在实际应用中，我们可能会使用更复杂的模型，如逻辑回归、随机森林、神经网络等。例如，逻辑回归模型：

$$P(y=1|x)=\frac{1}{1+e^{-(w_0+w_1{x}_1+...+w_n{x}_n)}}$$

其中：

• $P(y=1|x)$ 是给定特征 $x$ 时事件为正例（风险事件）的概率
• $w_0,w_1,...,w_n$ 是模型参数
• $x_1,x_2,...,x_n$ 是特征值

3.3.2 异常检测算法

异常检测是识别未知风险模式的关键技术。常用的异常检测算法包括：

1. 孤立森林(Isolation Forest)：通过随机选择特征和分割值来隔离异常点。
2. 一类支持向量机(One-Class SVM)：学习正常数据的边界，识别边界外的异常点。
3. 局部异常因子(Local Outlier Factor, LOF)：通过比较点与其邻居的密度来识别异常点。

让我们以孤立森林为例，简要介绍其算法原理。孤立森林的基本思想是：异常点通常更容易被孤立。算法通过多次随机选择特征和分割值，构建多个孤立树。一个点的异常分数基于它在这些树中的平均路径长度：

$$s(x,n)=2^{-\frac{E(h(x))}{c(n)}}$$

其中：

• $s(x,n)$ 是点 $x$ 的异常分数
• $E(h(x))$ 是点 $x$ 在所有树中的平均路径长度
• $c(n)$ 是给定 $n$ 个样本时的平均路径长度，用于归一化

异常分数的范围是(0, 1)，越接近1表示越可能是异常点。

3.3.3 规则冲突解决算法

当多个规则同时触发时，我们需要一种机制来解决冲突。常用的规则冲突解决策略包括：

1. 优先级策略：为每个规则分配优先级，高优先级规则优先执行。
2. 特异性策略：更具体的规则优先于更一般的规则。
3. 最近性策略：最近使用或修改的规则优先。
4. 投票策略：多个规则投票决定最终结果。

让我们定义一个形式化的规则冲突解决算法。假设我们有一组触发的规则 R={r1,r2,...,rk}R = \{r_1, r_2, ..., r_k\}R={r1​,r2​,...,rk​}，每个规则 $r_i$ 有优先级 $p_i$、特异性 $s_i$、最近性 $t_i$，以及决策结果 $d_i$。我们可以计算每个规则的综合权重：

$$w_i=\alpha \cdot p_i+\beta \cdot s_i+\gamma \cdot t_i$$

其中 $\alpha ,\beta ,\gamma$ 是权重系数，满足 $\alpha +\beta +\gamma =1$。

然后，我们可以根据综合权重对决策结果进行加权投票：

$$D=\arg \underset{d}{\max }\sum _{i:d_i=d}{w}_i$$

其中 $D$ 是最终决策。

3.4 实战演练：构建一个简化的AI Agent风控系统

现在让我们通过一个实战案例，来学习如何构建一个简化但功能完整的AI Agent风控系统。我们将使用Python作为主要编程语言，并结合一些常用的开源库。

3.4.1 项目介绍

我们将构建一个用于支付交易的风控系统，它具有以下功能：

1. 接收和处理交易事件
2. 计算风险特征
3. 使用规则引擎进行初步风险判断
4. 使用AI Agent进行深入风险分析
5. 综合规则和AI的结果，做出最终决策
6. 提供规则配置界面
7. 收集反馈，持续优化系统

为了简化实现，我们将重点关注核心功能，而不是构建一个生产级别的系统。

3.4.2 环境安装

首先，让我们准备开发环境。我们需要安装以下Python库：

# 创建虚拟环境
python -m venv ai_agent_risk_control
source ai_agent_risk_control/bin/activate  # Linux/Mac
# 或
ai_agent_risk_control\Scripts\activate  # Windows

# 安装必要的库
pip install fastapi uvicorn  # Web框架
pip install pandas numpy  # 数据处理
pip install scikit-learn  # 机器学习
pip install pydantic  # 数据验证
pip install python-dotenv  # 环境变量管理
pip install openai  # OpenAI API（用于AI Agent）
pip install networkx  # 图处理
pip install matplotlib  # 可视化

3.4.3 系统设计

让我们先来设计系统的核心组件和接口。

3.4.3.1 数据模型

首先，我们定义一些核心的数据模型：

# models.py
from pydantic import BaseModel, Field
from typing import List, Dict, Optional, Any
from enum import Enum
from datetime import datetime

class TransactionStatus(str, Enum):
    PENDING = "pending"
    APPROVED = "approved"
    DECLINED = "declined"
    REVIEW = "review"

class DecisionSource(str, Enum):
    RULE = "rule"
    AI_AGENT = "ai_agent"
    HYBRID = "hybrid"

class Transaction(BaseModel):
    transaction_id: str = Field(..., description="交易唯一标识")
    user_id: str = Field(..., description="用户唯一标识")
    amount: float = Field(..., description="交易金额")
    currency: str = Field(default="USD", description="货币类型")
    merchant_id: str = Field(..., description="商户唯一标识")
    timestamp: datetime = Field(default_factory=datetime.now, description="交易时间戳")
    device_id: Optional[str] = Field(None, description="设备标识")
    ip_address: Optional[str] = Field(None, description="IP地址")
    location: Optional[str] = Field(None, description="地理位置")
    additional_data: Optional[Dict[str, Any]] = Field(default_factory=dict, description="额外数据")

class RiskFeatures(BaseModel):
    transaction_id: str
    amount_to_average_ratio: float = Field(..., description="交易金额与用户平均交易金额的比率")
    is_new_merchant: bool = Field(..., description="是否是新商户")
    is_new_device: bool = Field(..., description="是否是新设备")
    is_high_risk_location: bool = Field(..., description="是否是高风险地区")
    velocity_hour: int = Field(..., description="过去1小时的交易次数")
    velocity_day: int = Field(..., description="过去24小时的交易次数")
    failed_attempts_last_24h: int = Field(..., description="过去24小时的失败尝试次数")
    additional_features: Optional[Dict[str, Any]] = Field(default_factory=dict)

class RuleCondition(BaseModel):
    feature: str = Field(..., description="特征名称")
    operator: str = Field(..., description="比较运算符")
    value: Any = Field(..., description="比较值")

class RuleAction(str, Enum):
    APPROVE = "approve"
    DECLINE = "decline"
    REVIEW = "review"
    ESCALATE = "escalate"

class RiskRule(BaseModel):
    rule_id: str = Field(..., description="规则唯一标识")
    name: str = Field(..., description="规则名称")
    description: Optional[str] = Field(None, description="规则描述")
    conditions: List[RuleCondition] = Field(..., description="规则条件列表")
    action: RuleAction = Field(..., description="规则触发后的动作")
    priority: int = Field(default=0, description="规则优先级，数字越大优先级越高")
    is_active: bool = Field(default=True, description="规则是否激活")
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

class RuleDecision(BaseModel):
    rule_id: str
    rule_name: str
    triggered: bool
    action: Optional[RuleAction] = None
    reasoning: Optional[str] = None

class AgentDecision(BaseModel):
    agent_id: str
    risk_score: float = Field(..., ge=0, le=1, description="风险评分，0为最低风险，1为最高风险")
    recommended_action: RuleAction
    confidence: float = Field(..., ge=0, le=1, description="决策置信度")
    reasoning: Optional[str] = None
    risk_factors: Optional[List[str]] = None

class FinalDecision(BaseModel):
    transaction_id: str
    status: TransactionStatus
    source: DecisionSource
    rule_decisions: List[RuleDecision]
    agent_decisions: List[AgentDecision]
    final_reasoning: str
    timestamp: datetime = Field(default_factory=datetime.now)

class Feedback(BaseModel):
    transaction_id: str
    correct_decision: TransactionStatus
    comments: Optional[str] = None
    provided_at: datetime = Field(default_factory=datetime.now)

3.4.3.2 系统接口

接下来，我们定义系统的主要接口：

# api.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import JSONResponse
from typing import List, Dict, Any
import datetime

from models import (
    Transaction, RiskFeatures, RiskRule, RuleAction,
    RuleDecision, AgentDecision, FinalDecision, Feedback,
    TransactionStatus, DecisionSource
)

app = FastAPI(title="AI Agent Harness Risk Control API", version="0.1.0")

# 内存存储（在生产环境中应使用数据库）
transactions_db: Dict[str, Transaction] = {}
rules_db: Dict[str, RiskRule] = {}
decisions_db: Dict[str, FinalDecision] = {}
feedbacks_db: Dict[str, Feedback] = {}

# 简单的用户历史数据（在生产环境中应从数据库或特征存储获取）
user_history: Dict[str, Dict[str, Any]] = {
    # 示例数据
    "user_123": {
        "average_amount": 100.0,
        "known_merchants": ["merchant_456", "merchant_789"],
        "known_devices": ["device_abc"],
        "transaction_count_1h": 0,
        "transaction_count_24h": 2,
        "failed_attempts_24h": 0,
        "last_transaction_time": datetime.datetime.now() - datetime.timedelta(hours=2)
    }
}

# 初始化一些示例规则
def initialize_sample_rules():
    rule1 = RiskRule(
        rule_id="rule_001",
        name="大额交易规则",
        description="当交易金额超过用户平均金额的5倍时，标记为需要审查",
        conditions=[
            {"feature": "amount_to_average_ratio", "operator": ">", "value": 5.0}
        ],
        action=RuleAction.REVIEW,
        priority=10
    )
    
    rule2 = RiskRule(
        rule_id="rule_002",
        name="新设备+高金额规则",
        description="当使用新设备且交易金额超过用户平均金额的3倍时，拒绝交易",
        conditions=[
            {"feature": "is_new_device", "operator": "==", "value": True},
            {"feature": "amount_to_average_ratio", "operator": ">", "value": 3.0}
        ],
        action=RuleAction.DECLINE,
        priority=20
    )
    
    rule3 = RiskRule(
        rule_id="rule_003",
        name="高频交易规则",
        description="当用户在过去1小时内交易超过10次时，标记为需要审查",
        conditions=[
            {"feature": "velocity_hour", "operator": ">", "value": 10}
        ],
        action=RuleAction.REVIEW,
        priority=15
    )
    
    rules_db[rule1.rule_id] = rule1
    rules_db[rule2.rule_id] = rule2
    rules_db[rule3.rule_id] = rule3

initialize_sample_rules()

# API路由
@app.post("/transactions", response_model=FinalDecision)
async def process_transaction(transaction: Transaction):
    """处理交易并返回风控决策"""
    # 存储交易
    transactions_db[transaction.transaction_id] = transaction
    
    # 计算特征
    features = await calculate_risk_features(transaction)
    
    # 执行规则引擎
    rule_decisions = await execute_rules(features)
    
    # 执行AI Agent
    agent_decisions = await execute_ai_agents(transaction, features)
    
    # 综合决策
    final_decision = await make_final_decision(
        transaction, rule_decisions, agent_decisions
    )
    
    # 存储决策
    decisions_db[final_decision.transaction_id] = final_decision
    
    # 更新用户历史（简化版）
    update_user_history(transaction)
    
    return final_decision

@app.get("/rules", response_model=List[RiskRule])
async def get_rules():
    """获取所有规则"""
    return list(rules_db.values())

@app.post("/rules", response_model=RiskRule)
async def create_rule(rule: RiskRule):
    """创建新规则"""
    if rule.rule_id in rules_db:
        raise HTTPException(status_code=400, detail="Rule with this ID already exists")
    
    rules_db[rule.rule_id] = rule
    return rule

@app.put("/rules/{rule_id}", response_model=RiskRule)
async def update_rule(rule_id: str, rule: RiskRule):
    """更新规则"""
    if rule_id not in rules_db:
        raise HTTPException(status_code=404, detail="Rule not found")
    
    rule.rule_id = rule_id
    rule.updated_at = datetime.datetime.now()
    rules_db[rule_id] = rule
    return rule

@app.delete("/rules/{rule_id}")
async def delete_rule(rule_id: str):
    """删除规则"""
    if rule_id not in rules_db:
        raise HTTPException(status_code=404, detail="Rule not found")
    
    del rules_db[rule_id]
    return {"message": "Rule deleted successfully"}

@app.post("/feedback", response_model=Feedback)
async def submit_feedback(feedback: Feedback):
    """提交反馈"""
    if feedback.transaction_id not in transactions_db:
        raise HTTPException(status_code=404, detail="Transaction not found")
    
    feedbacks_db[feedback.transaction_id] = feedback
    
    # 在实际系统中，这里会触发模型重新训练或规则更新
    await process_feedback(feedback)
    
    return feedback

@app.get("/decisions/{transaction_id}", response_model=FinalDecision)
async def get_decision(transaction_id: str):
    """获取交易决策"""
    if transaction_id not in decisions_db:
        raise HTTPException(status_code=404, detail="Decision not found")
    
    return decisions_db[transaction_id]

# 核心功能实现
async def calculate_risk_features(transaction: Transaction) -> RiskFeatures:
    """计算风险特征"""
    # 获取用户历史数据
    user_data = user_history.get(transaction.user_id, {
        "average_amount": transaction.amount,  # 新用户默认使用当前交易金额
        "known_merchants": [],
        "known_devices": [],
        "transaction_count_1h": 0,
        "transaction_count_24h": 0,
        "failed_attempts_24h": 0
    })
    
    # 计算特征
    amount_to_average_ratio = transaction.amount / user_data.get("average_amount", 1.0)
    is_new_merchant = transaction.merchant_id not in user_data.get("known_merchants", [])
    is_new_device = transaction.device_id not in user_data.get("known_devices", []) if transaction.device_id else True
    
    # 简化版的地理位置风险判断
    is_high_risk_location = transaction.location in ["高风险地区A", "高风险地区B"] if transaction.location else False
    
    velocity_hour = user_data.get("transaction_count_1h", 0)
    velocity_day = user_data.get("transaction_count_24h", 0)
    failed_attempts_last_24h = user_data.get("failed_attempts_24h", 0)
    
    return RiskFeatures(
        transaction_id=transaction.transaction_id,
        amount_to_average_ratio=amount_to_average_ratio,
        is_new_merchant=is_new_merchant,
        is_new_device=is_new_device,
        is_high_risk_location=is_high_risk_location,
        velocity_hour=velocity_hour,
        velocity_day=velocity_day,
        failed_attempts_last_24h=failed_attempts_last_24h
    )

async def execute_rules(features: RiskFeatures) -> List[RuleDecision]:
    """执行规则引擎"""
    decisions = []
    
    # 将特征转换为字典，方便访问
    features_dict = features.dict()
    
    # 按优先级排序规则
    sorted_rules = sorted(
        [rule for rule in rules_db.values() if rule.is_active],
        key=lambda r: -r.priority
    )
    
    for rule in sorted_rules:
        # 检查规则条件
        triggered = True
        for condition in rule.conditions:
            feature_value = features_dict.get(condition.feature)
            if not evaluate_condition(feature_value, condition.operator, condition.value):
                triggered = False
                break
        
        decision = RuleDecision(
            rule_id=rule.rule_id,
            rule_name=rule.name,
            triggered=triggered,
            action=rule.action if triggered else None,
            reasoning=f"规则 '{rule.name}' 被触发" if triggered else f"规则 '{rule.name}' 未触发"
        )
        decisions.append(decision)
    
    return decisions

def evaluate_condition(feature_value, operator, condition_value):
    """评估单个条件"""
    if operator == "==":
        return feature_value == condition_value
    elif operator == "!=":
        return feature_value != condition_value
    elif operator == ">":
        return feature_value > condition_value
    elif operator == "<":
        return feature_value < condition_value
    elif operator == ">=":
        return feature_value >= condition_value
    elif operator == "<=":
        return feature_value <= condition_value
    elif operator == "in":
        return feature_value in condition_value
    else:
        return False

async def execute_ai_agents(transaction: Transaction, features: RiskFeatures) -> List[AgentDecision]:
    """执行AI Agent（简化版）"""
    decisions = []
    
    # 在实际系统中，这里会调用真实的AI模型或LLM
    # 这里我们使用一个简化的基于规则的模拟
    
    # 计算风险评分
    risk_score = 0.0
    risk_factors = []
    
    # 基于特征计算风险分数
    if features.amount_to_average_ratio > 5:
        risk_score += 0.3
        risk_factors.append("交易金额显著高于用户平均水平")
    elif features.amount_to_average_ratio > 3:
        risk_score += 0.15
    
    if features.is_new_device:
        risk_score += 0.2
        risk_factors.append("使用新设备")
    
    if features.is_new_merchant:
        risk_score += 0.1
        risk_factors.append("与新商户交易")
    
    if features.is_high_risk_location:
        risk_score += 0.25
        risk_factors.append("来自高风险地区")
    
    if features.velocity_hour > 5:
        risk_score += 0.1
        risk_factors.append("交易频率异常高")
    
    # 确保风险评分在0-1范围内
    risk_score = min(risk_score, 1.0)
    
    # 根据风险评分决定推荐动作
    if risk_score > 0.7:
        recommended_action = RuleAction.DECLINE
        confidence = 0.8
    elif risk_score > 0.4:
        recommended_action = RuleAction.REVIEW
        confidence = 0.7
    else:
        recommended_action = RuleAction.APPROVE
        confidence = 0.9
    
    decision = AgentDecision(
        agent_id="risk_assessment_agent_001",
        risk_score=risk_score,
        recommended_action=recommended_action,
        confidence=confidence,
        reasoning=f"基于风险特征分析，该交易的风险评分为{risk_score:.2f}。" + 
                  (f"主要风险因素包括：{', '.join(risk_factors)}。" if risk_factors else ""),
        risk_factors=risk_factors
    )
    
    decisions.append(decision)
    return decisions

async def make_final_decision(
    transaction: Transaction,
    rule_decisions: List[RuleDecision],
    agent_decisions: List[AgentDecision]
) -> FinalDecision:
    """做出最终决策"""
    # 收集所有触发的规则动作
    triggered_actions = [dec.action for dec in rule_decisions if dec.triggered]
    
    # 收集AI Agent的推荐动作
    agent_actions = [dec.recommended_action for dec in agent_decisions]
    agent_risk_scores = [dec.risk_score for dec in agent_decisions]
    
    # 决策逻辑（简化版）
    final_status = TransactionStatus.APPROVED
    decision_source = DecisionSource.HYBRID
    reasoning_parts = []
    
    # 优先考虑规则决策
    if RuleAction.DECLINE in triggered_actions:
        final_status = TransactionStatus.DECLINED
        decision_source = DecisionSource.RULE
        reasoning_parts.append("交易被规则引擎拒绝。")
    elif RuleAction.REVIEW in triggered_actions:
        final_status = TransactionStatus.REVIEW
        decision_source = DecisionSource.RULE
        reasoning_parts.append("交易被规则引擎标记为需要审查。")
    # 如果没有规则触发，使用AI Agent决策
    elif agent_actions:
        avg_risk_score = sum(agent_risk_scores) / len(agent_risk_scores)
        if avg_risk_score > 0.7:
            final_status = TransactionStatus.DECLINED
            reasoning_parts.append(f"基于AI Agent分析，风险评分{avg_risk_score:.2f}过高，交易被拒绝。")
        elif avg_risk_score > 0.4:
            final_status = TransactionStatus.REVIEW
            reasoning_parts.append(f"基于AI Agent分析，风险评分{avg_risk_score:.2f}中等，交易需要审查。")
        else:
            reasoning_parts.append(f"基于AI Agent分析，风险评分{avg_risk_score:.2f}较低，交易通过。")
        decision_source = DecisionSource.AI_AGENT
    
    # 收集详细的决策原因
    for dec in rule_decisions:
        if dec.triggered:
            reasoning_parts.append(f"规则 '{dec.rule_name}' 触发: {dec.reasoning}")
    
    for dec in agent_decisions:
        reasoning_parts.append(f"AI Agent '{dec.agent_id}': {dec.reasoning}")
    
    return FinalDecision(
        transaction_id=transaction.transaction_id,
        status=final_status,
        source=decision_source,
        rule_decisions=rule_decisions,
        agent_decisions=agent_decisions,
        final_reasoning=" ".join(reasoning_parts)
    )