在网络安全领域，做过 SOC（安全运营中心）或者自动化攻防的大伙儿应该都有体会：让一个大模型包揽所有安全运维工作，纯属“强大模型所难”。
你让它去分析 SQL 注入日志，它表现不错；但如果你让它一边分析日志、一边跑 Threat Intelligence（威胁情报）对比、一边还要自动生成防火墙封禁策略，单智能体（Single Agent）的上下文（Context）瞬间就爆了，逻辑也容易错乱。
这就是为什么今年 AI 安全领域都在提 Multi-Agent（多智能体协同）。
今天这篇文章，我们就抛开那些晦涩的概念，用最接地气的方式，聊聊在网络安全场景下，多智能体到底有哪些架构？它们之间又是怎么传递信息的？文章最后，我还会带大家用 Python 写一个轻量级的安全多智能体路由系统。
一、 为什么网络安全需要 Multi-Agent？
过去我们用 LangChain 或 AutoGPT 搞安全助手，基本上是“一个大模型 + 一堆 Tools（如 Nmap、Sqlmap、微步 API）”。
这种模式处理简单任务没问题，一旦碰到复杂攻击溯源就崩了。
多智能体的核心思想，说白了就是“专人做专事”：
- Agent A（日志分析员）： 只看 Syslog / Web 访问日志。
- Agent B（威胁情报员）： 专门去查 IP/Hash 的黑灰产情报。
- Agent C（应急处置员）： 负责调用防火墙 API 执行拦截。
那么问题来了：这几个“安全专家”之间，到底该怎么配合？
二、 盘点多智能体的三大主流架构
在实际落地中，目前行业里主流的构建方式主要有以下几种：
1. 网状自由流转模式（Network Pattern）
这种模式下，各个 Agent 之间是平级的。
工作逻辑： Agent A 处理完日志，觉得有威胁，自由决定把消息发给 Agent B；Agent B 查完情报，再自由决定发给 Agent C。
安全人的吐槽： 千万慎用！ 这种模式太灵活了。在网络安全场景下，极其容易造成死循环。比如：日志 Agent 丢给拦截 Agent，拦截 Agent 发现规则冲突又丢给分析 Agent……你的 API Tokens 瞬间就会被烧光，甚至把业务系统搞崩。
2. 主从路由模式（Supervisor / Router Pattern）【最推荐】
这是目前安全大模型落地的“王炸架构”。
[ 用户 / 安全事件 ] │ ▼ ┌──────────────────┐ │ 路由智能体 Router │ └─────────┬────────┘ │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ SQLi Agent│ │ DDoS Agent│ │Webshell Ag│ └───────────┘ └───────────┘ └───────────┘
工作逻辑： 前端设立一个“安全指挥官”（Router Agent）。当一条威胁告警进来时，指挥官先判断攻击类型，再精确路由给具体的专家 Agent。
优势： 逻辑清晰、可控性强、不会死循环。
3. 分层金字塔模式（Hierarchical Pattern）
如果你做的是大型集团的 SOC 自动化系统，这种架构最合适。
工作 logic： 顶层是“总安全指挥官”，下一级按业务线划分为“云安全指挥官”、“终端安全指挥官”，最底层的具体 Agent 再去执行端口扫描、流量分析等微观任务。
三、 多智能体协同的核心：状态共享（State）
搞懂了架构，很多同学最关心的问题是：这些 Agent 之间到底怎么传数据？
答案就是：State（公共状态机）。
在多智能体系统内部，Agent 之间并不是直接互相吐文本，而是通过共享一个全局的“状态字典”（State）。每个 Agent 执行完，把自己的分析结果追加（Append）或覆盖（Update）到这个 State 里，然后再由路由组件决定下一个接收者。
四、 手把手实战：用 Python 打造一个安全告警路由系统
下面我们用纯 Python 实现一个安全事件分流与处置的多智能体 Demo。
这个 Demo 模拟了这样一个场景： 系统收到一条疑似攻击日志 -> Router 识别威胁类型 -> 分发给对应的专家 Agent（SQL 注入 Agent 或 DDoS Agent）进行处置。
import json from typing import Dict, Any, TypedDict # 1. 定义全局状态 (State) class SecurityAgentState(TypedDict): raw_log: str # 原始日志 threat_type: str # 识别出的威胁类型 analysis_result: str # 分析结果 action_taken: str # 采取的防御措施 status: str # 流程状态 # 2. 定义专家 Agent：SQL 注入分析智能体 class SQLInjectionAgent: def process(self, state: SecurityAgentState) -> SecurityAgentState: print("[SQLi Agent] 正在分析 SQL 注入特征...") log = state['raw_log'] if "UNION SELECT" in log.upper() or "' OR '1'='1" in log.upper(): state['analysis_result'] = "检测到高危 SQL 注入攻击！payload 试图读取数据库敏感信息。" state['action_taken'] = "调用 WAF 规则：已将来源 IP 强制封禁 24 小时。" else: state['analysis_result'] = "未发现明确 SQLi 特征，疑为误报。" state['action_taken'] = "忽略并记录日志。" return state # 3. 定义专家 Agent：DDoS 攻击分析智能体 class DDoSAgent: def process(self, state: SecurityAgentState) -> SecurityAgentState: print("[DDoS Agent] 正在分析流量峰值与请求频率...") state['analysis_result'] = "检测到异常流量并发，QPS 超过阈值 300%。" state['action_taken'] = "已触发 Cloudflare 节点清洗流量，限制单 IP 请求速率。" return state # 4. 定义主控路由 Agent (Supervisor) class SecurityRouter: def __init__(self): self.sqli_agent = SQLInjectionAgent() self.ddos_agent = DDoSAgent() def route_and_execute(self, raw_log: str) -> SecurityAgentState: # 初始化状态 state: SecurityAgentState = { "raw_log": raw_log, "threat_type": "UNKNOWN", "analysis_result": "", "action_taken": "", "status": "PROCESSING" } print(f"\n[Security Router] 收到新日志，开始初步判研: {raw_log}") # 模拟大模型提取意图/分类 (也可以直接对接 LLM API) log_upper = raw_log.upper() if "SELECT" in log_upper or "DROP TABLE" in log_upper: state['threat_type'] = "SQL_INJECTION" print("[Security Router] 路由判定：分发至 [SQLi Agent]") state = self.sqli_agent.process(state) elif "FLOOD" in log_upper or "SYN" in log_upper: state['threat_type'] = "DDOS" print("[Security Router] 路由判定：分发至 [DDoS Agent]") state = self.ddos_agent.process(state) else: state['threat_type'] = "OTHER" state['analysis_result'] = "常规流量，无明显异常。" state['action_taken'] = "放行" state['status'] = "COMPLETED" return state # ---- 运行测试 ---- if __name__ == "__main__": router = SecurityRouter() # 测试用例 1：SQL 注入日志 log1 = "POST /login.php HTTP/1.1 | Input: admin' UNION SELECT null, password FROM users--" res1 = router.route_and_execute(log1) print("处置结果:", json.dumps(res1, ensure_ascii=False, indent=2)) print("-" * 50) # 测试用例 2：DDoS 日志 log2 = "GET /api/v1/resource HTTP/1.1 | Flag: SYN_FLOOD_DETECTED | QPS: 50000" res2 = router.route_and_execute(log2) print("处置结果:", json.dumps(res2, ensure_ascii=False, indent=2))
2. 运行结果示例
我们在终端直接运行该脚本：python security_multi_agent.py
输出效果如下：
[Security Router] 收到新日志，开始初步判研: POST /login.php HTTP/1.1 | Input: admin' UNION SELECT null, password FROM users-- [Security Router] 路由判定：分发至 [SQLi Agent] [SQLi Agent] 正在分析 SQL 注入特征... 处置结果: { "raw_log": "POST /login.php HTTP/1.1 | Input: admin' UNION SELECT null, password FROM users--", "threat_type": "SQL_INJECTION", "analysis_result": "检测到高危 SQL 注入攻击！payload 试图读取数据库敏感信息。", "action_taken": "调用 WAF 规则：已将来源 IP 强制封禁 24 小时。", "status": "COMPLETED" } -------------------------------------------------- [Security Router] 收到新日志，开始初步判研: GET /api/v1/resource HTTP/1.1 | Flag: SYN_FLOOD_DETECTED | QPS: 50000 [Security Router] 路由判定：分发至 [DDoS Agent] [DDoS Agent] 正在分析流量峰值与请求频率... 处置结果: { "raw_log": "GET /api/v1/resource HTTP/1.1 | Flag: SYN_FLOOD_DETECTED | QPS: 50000", "threat_type": "DDOS", "analysis_result": "检测到异常流量并发，QPS 超过阈值 300%。", "action_taken": "已触发 Cloudflare 节点清洗流量，限制单 IP 请求速率。", "status": "COMPLETED" }
五、 总结与进阶建议
从上面的代码可以看出，多智能体的本质其实并不复杂：
解耦： 把原本堆叠在大模型 Prompt 里的复杂任务，拆分成独立的微服务模块（Agent）。
状态传递： 依靠一个标准的 JSON/Dict 结构在 Agent 节点之间进行数据上下文传递。
在实际生产环境中，建议大家使用 LangGraph 或者 AutoGen 这类落地框架，将代码中的简单的条件判断换成真正的 LLM 智能路由，结合企业的 SIEM 系统（如 Splunk、ELK），就能搭建出一套媲美资深安全团队的 AI-SOC 自动响应系统。
资料展示
下面是我整理的AI大模型学习资料和工具包预览，适合收藏后按主题逐步学习。
如果你想看完整资料目录，可以在评论区留言「资料」；也欢迎告诉我你更关注AI大模型里的哪类内容。
