在事件溯源系统中集成智能体:实时反欺诈架构实践 AI Engineer 2026-07-30

事件溯源系统与智能体集成的价值

在过去数年中,许多企业在微服务和事件驱动架构上投入了大量资源。通常,人们认为 AI 智能体(AI Agents:能够自主规划、调用工具并执行复杂任务的 AI 系统)仅适用于聊天机器人或编程助手。然而,AI 智能体真正的商业价值在于将它们无缝融入现有的业务工作流中。以实时反欺诈(Real-time Fraud Detection:在交易发生的瞬间对其进行欺诈性评估的系统)为例,传统的系统往往依赖于静态的规则引擎或传统的机器学习模型。虽然这些系统能高效处理大部分显而易见的正常或欺诈交易,但面对新型欺诈手段时,规则的维护成本极高,且在阈值边缘的“灰色地带”交易上缺乏足够的上下文信息,容易导致误判。

将 AI 智能体引入现有的事件驱动系统,并非要推翻重建,而是将其作为系统能力的一种自然延伸。当传统的规则或 ML 引擎(一阶系统)无法对交易做出明确的判断时,系统会将交易事件路由给由 AI 智能体组成的二阶系统,利用其推理能力和实时动态上下文进行深度研判。

Original English

So, we are going to learn how to integrate AI agents in your existing system. So, this system it could be an even so system or it could be an even driven system or it could be any architecture that your business has invested over last few years, right? Because I always believe that these AI agents are more not just for the chatbots or the coding assistance, right? So, the real value that you could bring out of these AI agents is like when you start to apply these into your business workflows. And that is what we are going to learn. And the problem space that we are going to work on today is the real-time fraud detection.

双层研判:从规则引擎到智能体协作

在实际的业务场景中,例如购买一台价值 3500 美元的笔记本电脑,传统反欺诈系统可能会因为交易金额触发静态阈值或缺乏实时交易习惯上下文而直接拒绝交易,且客服人员也无法解释具体原因。为了解决这种高误报率并提升交易决策的精准度,系统采用了双层架构。一阶系统(Tier 1 System:处理常规决策的规则引擎或传统 ML 模型)负责拦截或批准大部分边界清晰的交易。而对于落入灰色地带的疑难交易,则流转至二阶系统(Tier 2 System:基于智能体的高级决策层)。

在二阶系统中,采用 扇出模式(Fan-out Pattern:将单个事件并发分发给多个处理器的设计模式)将事件同时分发给两个独立的智能体进行分析:

  • 风险分析智能体(Risk Analyzer Agent):专注于分析历史欺诈记录、设备信任度等安全指标。
  • 行为分析智能体(Behavior Analyzer Agent):通过特定插件评估交易者的历史消费行为模式。

最后,由第三个智能体作为裁决智能体(Verdict Agent)对两者的结论进行综合评估,并输出最终的裁决事件,从而极大降低了误报率。

Original English

So, the domain that we are going to talk about is the real-time fraud detection as I mentioned before. So, we had this rule-based engine like uh 5 years before. And this rule-based engine was perfectly fine like it was working perfectly fine for few of the cases. And but but the problem with this rule-based engine is like the maintainability because the fraudsters are trying to get intruded into a system like by a lot of different ways and you just need to keep on updating these static rules day by day and it it it's going to be really difficult for you to manage. And that's when like we started to tie up with a third-party provider like who helped us to develop this ML model. So we we have this ML base approach like where we shared with them transaction history or different features with them like based on that they trained the ML model and we were able to get a risk code based on that which with which like we were able to block or approve the transaction. But the problem with either of these approaches like either like we we were able to handle most of the transaction because it would fall below a certain threshold then we would approve the transaction and if it goes beyond a certain threshold we would be blocking those transaction. But majority of the transaction like few of the transaction like goes under the gray zone area and this is the area where it is really uncertain for those systems to really come to a conclusion whether it is an fraudulent transaction or a legitimate transaction. So what we are trying to do is like we we had built a system where we had both these tier one system which has this rule based or the traditional ML model and then we also had a tier two system which is agentic AI approach like most of the cases would be handled really well by these existing system we already had because our thought process is not to exclude the systems that we already had. We we are just trying to handle few of the areas like that is the gray zone areas with the help of agentic AI processing.

So, under this orchestration layer like we have different sub orchestrator. So, we segregated the tier one. Earlier like we just had the tier one. Now, we had the tier one layer, which is handling the um rule-based engine or the ML-based engine. And we also had now the tier two layer, which is going to be agent AK processing. So, for this agent AK processing like we used fan out pattern. So, we kind of use multiple agents within this layer, and we are trying to use a fan out pattern like um once the event has been reached out to the tier two layer, we will be fanning out this event to two different agents. One is the risk analyzer agent, the other one is the behavior analyzer agent. And once these agents like come to a conclusion based on the different tools that it has, it will finally send the response to the verdict. And this verdict could be a metric. It could be just an if condition inside your application layer, or it could be an another region. Because what we seen is like if we are using just the metrics, it is again going to back to the same criteria like where we had this rule-based mechanism. So, there are many false positive cases that we are that we faced. So, so we we in turn like we are trying to use a third agent in this verdict layer, which analyzes both the agents' responses and come to a conclusion, which is then going to be emitted as an event back to this message broker.

事件驱动与领域上下文的解耦设计

在现代微服务设计中,系统通常被划分为多个限界上下文(Bounded Context:领域驱动设计中定义模型适用边界的显式空间),例如交易上下文(Transaction Context)、账户上下文(Accounts Context)、设备上下文(Device Context)和支付上下文(Payment Context)。这些上下文之间的数据默认是相互隔离的,以保证高内聚和低耦合。

为了在分布式环境下协同完成复杂的反欺诈判定,系统引入了Saga编排器(Saga Orchestrator:负责协调多个微服务之间事务一致性与流程流转的中心组件)。所有的领域事件(如 TransactionCreated)和集成事件(如 PaymentApprovedTransactionRejected)都通过消息代理进行异步分发。同时,交易上下文使用 事件溯源(Event Sourcing:不保存对象的当前状态,而是将所有状态改变作为一系列事件持久化在只追加存储中的模式)作为核心存储模式,将每一次状态变更作为不可变的事实追加到 Cosmos DB 这样的事件存储中。

Original English

But then our architecture, our domain is really complicated. So, this is our different bounded contexts that we have internally in our domain. So, the transaction context holds all the details about your transactions. Like, it knows about the merchants. It knows about the amounts that that you transact. And everything related to a transaction would be residing on this particular context. And it doesn't have any information about the customer it is handling, or it doesn't have any information about the payments or the device details. That is what the bounded context means, right? Because if you are from the DDD background or software engineering, like you would know that these are different bounded contexts like with like you wouldn't share the data among themselves. Like, you need to do an asynchronous way of communication. And like all sort of things like that you would do in a microservice communication would necessarily be done here as well. So, then we do have accounts context. All the details about the accounts like the KYC complaining, whether the user is KYC compliant, or anything about the customer. Like, if you want to know about the customer, this would be the right context that we need to reach out to. And there is a device context. Like, we had stored all the device fingerprints, browser fingerprints, the OS that they are using over these device contexts. Like, that would be really helpful like for detecting these kinds of real-time frauds. And finally, like we had payment context. So, any sort of chargebacks or anything related to the payments will be residing on this particular context. So, now the problem that we have with this kind of an architecture is like we we really we really don't have a way or means to share these data across different bounded contexts, right? So, that's where like we started to introduce an orchestrator layer. So, earlier like we had this orchestrator layer, but now we also have an agentic AI inside this orchestrator. So, what this orchestrator does is like similar to a saga orchestration. So, all the communication would go through this layer and then it would be communicated to other services like who are interested in those events. And that is how like we define this orchestration layer. And we also have an asynchronous way of communicating with within other different contexts like through a message broker.

If I zoom in a little bit on the transaction context, you could see what what are the details that these transactions are holding. So, essentially like what happens is like when you have particular domain, you will be having different events that are emitted from that system. So, those are really called as domain events. And within our transaction context like we do have transaction created domain event, which is the start of entire transaction. And we do have lot of integration events. So, these are the events that comes from other contexts like payment contexts, device contexts, or account contexts. So, events like transaction rejected, payment approved, payment rejected would come from other systems to our transaction contexts. And these entire informations are stored in our NoSQL database. We are currently using Cosmos DB as our event store. And we are following event sourcing as our methodology to store the events. So, what happens is like whenever user initiate a command, so that goes into our event store as an event as a business fact. So, we are not mutating the state, but instead like we are appending all the events as when it arrives.

利用 CDC 构建语义投影与智能体运行机制

由于事件存储(Event Store)只适合按追加方式写入事件,不便于进行直接的复杂只读查询。因此,系统利用 变更数据捕获(Change Data Capture (CDC):监听数据库数据变化并将其近乎实时地推送到其他存储或系统中的机制,在 NoSQL 中通常表现为 Change Feed)机制,将事件流实时同步至不同的只读模型(Read Models:专为读取操作进行过结构优化的去规范化数据视图)。

这些数据在投影层被进一步整合,构建起一个统一的语义层(Semantic Layer:为 AI 智能体提供统一、结构化且富含业务上下文的数据读取接口的媒介)。智能体运行在这个语义层之上,通过内置的工具接口获取实时 massaged 后的数据:

  1. 交易上下文投影:提供交易次数反规范化统计、平均消费额度及近期交易流水。
  2. 设备上下文投影:提供设备信任评分、异地登录历史与 IP 段变更。
  3. 账户上下文投影:提供 KYC 合规状态与注册时长等信息。

在智能体具体运行机制上,通常包含大语言模型/小语言模型(LLM/SLM)、工具和内存(Memory)。因为反欺诈判定具有严苛的 服务等级协议(Service Level Agreement (SLA):服务提供商与用户之间关于服务质量、响应时间等达成的协议,此场景下要求在 500 毫秒内完成响应),系统采用内存级短期记忆(In-memory Short-term Memory)来跟踪推理链条,避免使用高延迟的长周期存储,并通过合理的指标或退出机制打破推理循环,彻底杜绝智能体在非确定性推理中出现死循环的风险。

graph TD
    subgraph 限界上下文 (Bounded Contexts)
        TC[交易上下文]
        AC[账户上下文]
        DC[设备上下文]
        PC[支付上下文]
    end

    TC -- 产生事件 --> EventStore[(Cosmos DB Event Store)]
    EventStore -- CDC / Change Feed --> ReadModels[只读模型投影层]
    ReadModels --> SL[语义层 Materialized View]

    subgraph Saga 编排与智能体层 (Orchestrator & Agents)
        SO[Saga 编排器]
        T1[Tier 1: 规则与机器学习引擎]
        T2[Tier 2: 智能体分析层]
        
        RA[风险分析智能体]
        BA[行为分析智能体]
        VA[裁决智能体]
        
        T2 --> RA & BA
        RA & BA --> VA
    end

    SL -- 提供工具上下文 --> RA & BA
    SO --> T1
    T1 -- 灰色地带交易 --> T2
    VA -- 输出最终判定事件 --> SO
    SO -- 异步事件广播 --> MessageBroker[消息代理 Message Broker]
    MessageBroker --> PC
Original English

And what it really helped us to do is like once it arrives into the event store, so we have a kind of a mechanism called CDC or in no sequel word this is called as a change feed with which like whenever there is a change or update happens over a table, so you will be getting notified and those changes could be propagated into different read models. So, because in a real scenario like you won't be able to rely entirely upon the event store for the query operation for the read operation. What you would do is like internal like you will be having different read models which are optimized for the read operations. So, what we really had is like we had multiple read models. So, one for timelines, one for customer information, and one for the fraud in the indicators, the risk view. So, these are the different read model later layers that we had within our transaction context. And essentially like you can't say other teams to follow the same patterns because even sourcing is not the one that other teams are also following. So, what we did is like we also had this asynchronous way of communication by emitting those events into the message broker and those events in turn will be processed by a worker process and then it will be reaching out to our projection layer. So, the idea is like we need to gather all the data from all these different contexts and to have or build a semantic layer or you could call it as a materialized view which you could further use within your agent.

Now, coming to the agents. So, we all know like agents are comprised of these three main components, the language models, the tools, and the memories, right? So, this language model it is not necessary to be a large language model. It could be a SLM. It could be an open-source model. And the tools are by which like you could interact with the external APIs or the meta methods that you define inside your application layer. And you also need to have a memory layer. For this particular use case like we are using an short memory because you can't really rely on the long-term memory because you need to adhere to the SLA that you provided to the customers because for the transaction to be processed like it should be sub 500 milliseconds. And you we are currently using in-memory for this. And essentially like what happens is like whenever there is an query comes into our language model, so it has this reasoning capability, the thinking capability with which like it tries to decompose the task into multiple chunks. And each of these tasks in turn will go into the language model, do certain processing with the help of the tools, and it will see like if the end goal is reached. And if it is not reached like it will try to go on in this loop. So, this is what essentially happens inside the agentic framework. But you should really know like when to stop this loop because this depends this might be varying. This might be differing for different use cases. For our use case like we do have a metrics with beyond which like if we go like we could break out of this loop. And this could be varying for different use cases. So, you should be really careful on avoiding this infinite loop.

Now, as we seen earlier, like how does these agents gather contexts? So, from the projection layer. So, these orchestration is going to consume these datas from the projection layers and how it is going to do is by the help of tools that it already has. So, the transaction context, it is going to denormalize some of the counts, averages, average amounts that we had, the recent transactions, all gets into this semantic layer, and the device context is going to send all the details about the device trust score, location histories, and all other related information about the locations, IP addresses to this semantic layer. And the account context is going to send the statuses of the accounts, the KYC status, account age, whether the customer has been with us for the past few years. So, based on that, like it will try to validate the customer based on those informations. And there is also the payment context. From this, like we get to know about the recent payments that we gathered over these contexts into the semantic layer. And as I said earlier, so these tools get access to these projection layer, and which in turn will be used by these agents to come to a conclusion.

📌 文中提及的人物和组织

人物: Divakar Kumar

公司/组织: FlyersSoft

关键字: event-sourcing ai-agents saga-orchestration fraud-detection change-data-capture