前言

本文主要描述Event-Driven开发中的ReAct模式，并且使用一个demo，彻底搞懂怎么在实际工作中使用Event-Driven模式

话不多说，我们开始

代码结构

代码地址

.
├── agent.py       # EventDrivenAgent 主逻辑，负责接收事件、调用 LLM、执行工具
├── event_bus.py   # 极简事件总线，负责发布事件和消费事件
├── main.py        # 程序入口，启动事件循环，并模拟 Prometheus 告警
└── tools.py       # 工具注册中心，定义查询 Pod、查询日志、重启服务三个工具

代码解析

• event_bus.py，这里用的是 Python 标准库里的 queue.Queue，主要就是模拟从消息队列读取消息

• main.py，模拟从消息队列中取出警告

  event = {
      "alert_name": "HighErrorRate",
      "service": "order-service"
  }
  
  publish_event(event)
  

  只要队列里来了新事件，它就取出来，然后交给 agent.handle_event(event) 处理

• tools.py，定义了三个工具

  • 查 Pod 状态：

  def query_pod_status(service):
      print(f"[Tool] 查询 Pod 状态: {service}")
      return f"{service} 有 2 个 Pod CrashLoopBackOff"
  

  • 查服务日志：

  def query_logs(service):
      print(f"[Tool] 查询日志: {service}")
      return "数据库连接超时 timeout"
  

  • 重启服务：

  def restart_service(service):
      print(f"[Tool] 重启服务: {service}")
      return f"{service} 已成功重启"
  

  同时把工具注册到 _TOOL_REGISTRY 里 ，生成一份 function calling 需要的工具 schema

• agent.py，这是与llm打交道的部分

  • 调用工具，拿到对应的数据发给llm，llm思考之后返回结果
  • 在根据结果判断进行下一步

执行链路

Event-Driven 的特点

以上代码分析之后，其实Event-Driven本质就是：事件驱动+ReAct

Event-Driven 最适合的场景

• 告警处理，这是最适合的应用场景了，通过告警回调触发

• 事件驱动的相关事件

  • Pod CrashLoopBackOff
  • 磁盘使用率超过 90%
  • 证书即将过期
  • 任务执行失败
  • 服务发布完成

而通常一个成熟的 aiops agent 可能是这样：

Prometheus 告警事件触发 Agent           # Event-Driven
  ↓
Agent 先生成排查计划                    # Plan-and-Execute
  ↓
执行过程中根据日志和指标不断调整判断    # ReAct
  ↓
需要恢复动作时走审批或白名单            # Guardrail
  ↓
输出结论并写入工单 / 通知群             # Workflow

总结

Event-Driven 最适合告警处理、自动化运维、业务事件、CI/CD、工单流转这类场景。它可以和 ReAct、Plan-and-Execute 组合使用：Event-Driven 负责触发，ReAct 负责边查边判断，Plan-and-Execute 负责长任务规划

联系我

• 联系我，做深入的交流

至此，本文结束

在下才疏学浅，有撒汤漏水的，请各位不吝赐教...