CTI-Agent-Project

# CTI 智能体 一款自动化网络威胁情报分析师日常工作的多智能体平台。它对新报告进行分类，分析威胁行为者的特征，丰富外部数据源的IOC，并关联多个数据存储中的证据。用户可以像与初级分析师一样与其聊天，每个智能体采取的步骤都在答案下方的可折叠时间线中可见。 本仓库是一个公开的、合成数据的生产环境Agentic AI栈的版本，我在工作中构建和运行过。它与生产环境具有相同的智能体架构、提示模式、MCP服务器设计和验证循环，但实际数据源、付费数据源和客户特定连接器已被RSS数据源、MITRE ATT&CK和确定性模拟器取代，以便该项目可以作为作品集交付，而不会泄露任何敏感信息。 ## CTI 智能体演示 https://github.com/user-attachments/assets/9e71959c-be0c-4350-b612-42294c320317 ## 分析师在系统中的工作流程 询问“APT41使用哪些工具？”请求会经过四个智能体。 编排智能体会读取问题，确定意图是威胁行为者配置文件，并选择下一个智能体要调用的工具（`graph_query`、`vector_search`、`keyword_search`）。检索智能体会运行一个ReACT循环，依次调用这些工具，读取每个结果，并决定是否调用另一个。编写智能体会根据编排计划选择五个精选提示模板之一，并草拟一个Markdown答案。验证智能体会检查答案是否基于证据；如果不是，图形会将反馈循环回编写智能体，进行一两次重试。 每个工具调用、每个计划、每个重试都会在聊天气泡下的UI中显示。在答案渲染后，用户可以点击竖起大拇指或竖起小拇指，这会向带有运行ID的Postgres反馈表中写入一行。该表是离线RLHF或DPO风格训练的种子数据集。 ## 架构 ```mermaid flowchart LR User([用户]) -->|HTTP/SSE| FE[Next.js 聊天界面] FE -->|POST /chat| API[FastAPI + LangGraph] subgraph 智能体 [LangGraph StateGraph] ORCH[编排<br/>意图 + 计划] --> RET[检索<br/>ReACT循环] RET --> WRT[编写<br/>一次性模板] WRT --> VAL[验证] VAL -.无效 + 重试<最大.-> WRT end API --> 智能体 RET -->|MCP/SSE| MCP1[检索-mcp] RET -->|MCP/SSE| MCP2[搜索-mcp] RET -->|MCP/SSE| MCP3[丰富-mcp] MCP1 --> Milvus[(Milvus<br/>向量数据库)] MCP1 --> ES[(Elasticsearch)] MCP1 --> Neo[(Neo4j<br/>STIX 知识图谱)] MCP1 --> PG[(Postgres<br/>IOC 数据库)] MCP2 --> Tavily[Tavily 或模拟器] MCP3 --> VT[VirusTotal] MCP3 --> Abuse[AbuseIPDB] MCP3 --> RF[Recorded Future] API --> FB[(Postgres<br/>反馈 + 跟踪)] FE -.竖起大拇指/小拇指.-> FB ``` 四个智能体通过共享的LangGraph状态进行通信。每个节点读取前一个节点写入的内容，贡献自己的增量，编排框架将所有内容合并在一起。验证器到编写器的条件边是自纠正循环的核心；只有在验证失败并且有重试次数限制时才会触发。 检索MCP服务器是与所有四个知识存储通信的服务器。搜索MCP服务器包装Tavily进行开放网络搜索。丰富MCP服务器为每个第三方声誉服务暴露一个工具。以这种方式拆分它们意味着可以通过编写一个新的MCP服务器来添加新的工具类别，而无需触及现有的其他工具。 ## 智能体在代码中的实现 编排器首先运行。它的任务是读取用户的问题并输出一个JSON计划，供其余图形遵循。 ```python # agents/orchestrator.py（缩略版） _VALID_TEMPLATES = { "摘要", "威胁行为者配置文件", "IOC报告", "关联", "常规", } async def orchestrator_node(state): prompt = load_prompt("orchestrator").format( user_query=state["user_query"], available_tools=", ".join(state["available_tools"]), selected_tools=", ".join(state["selected_tools"]) or "ALL", ) llm = get_llm(temperature=0.0) resp = await llm.ainvoke([ SystemMessage(content=prompt), HumanMessage(content=state["user_query"]), ]) plan = _parse_json(resp.content) # 安全网：如果模型出现幻觉，则回落到常规。 if plan.get("writer_template") not in _VALID_TEMPLATES: plan["writer_template"] = "常规" # 用户在UI侧边栏的选择始终优先。 if state["selected_tools"]: plan["tools_to_use"] = [ t for t in plan.get("tools_to_use", []) if t in state["selected_tools"] ] return {"plan": plan, "trace": trace} ``` 检索智能体会获取计划并针对MCP工具集运行ReACT循环。它使用`langgraph.prebuilt.create_react_agent`，通过OpenAI函数调用接口处理Thought / Action / Observation循环。该系统对始终调用至少一个工具有严格要求，因为本地LLM有时会尝试从一般知识中回答。 ```python # agents/retrieval.py（缩略版） async def retrieval_node(state): all_tools = await load_all_tools() # MCP发现 tools = filter_tools(all_tools, state["plan"]["tools_to_use"]) system_prompt = load_prompt("retrieval_react").format( user_query=state["user_query"], plan=json.dumps(state["plan"]), ) react_agent = create_react_agent(get_llm(0.0), tools, prompt=system_prompt) result = await react_agent.ainvoke( {"messages": [HumanMessage(content=state["user_query"])]}, config={"recursion_limit": 12}, ) evidence = _evidence_from_messages(result["messages"]) return {"evidence": evidence, "trace": trace} ``` 编写智能体会获取编排器的计划，查找要使用的模板，并渲染答案。 ```python # agents/writer.py（缩略版） _TEMPLATE_MAP = { "摘要": "writer_summary", "威胁行为者配置文件": "writer_threat_actor", "IOC报告": "writer_ioc_report", "关联": "writer_correlation", "常规": "writer_general", } async def writer_node(state): template_name = _TEMPLATE_MAP.get( state["plan"].get("writer_template", "常规"), "writer_general", ) prompt = load_prompt(template_name).format( user_query=state["user_query"], evidence=_format_evidence(state.get("evidence")), ) if feedback := state.get("validation", {}).get("feedback"): prompt += f"\n\n上一次尝试验证失败。修复此问题：{feedback}" resp = await get_llm(0.3).ainvoke([ SystemMessage(content=prompt), HumanMessage(content=state["user_query"]), ]) return {"answer": resp.content, "trace": trace} ``` 验证器返回一个小的JSON裁决，图形使用它来决定是否重试。 ```python # agents/validator.py（缩略版） async def validator_node(state): prompt = load_prompt("validator").format( user_query=state["user_query"], evidence=_format_evidence(state.get("evidence")), answer=state.get("answer", ""), ) resp = await get_llm(0.0).ainvoke([SystemMessage(content=prompt), HumanMessage(content=state["user_query"])]) verdict = _parse_json(resp.content) # {"valid": bool, "issues": [...], "feedback": "..."} return { "validation": verdict, "retry_count": state.get("retry_count", 0) + (0 if verdict["valid"] else 1), "trace": trace, } ``` 以及连接它们的图形。验证器的条件边是唯一的分支点。 ```python # agents/graph.py（缩略版） def build_graph(): g = StateGraph(CTIState) g.add_node("orchestrator", orchestrator_node) g.add_node("retrieval", retrieval_node) g.add_node("writer", writer_node) g.add_node("validator", validator_node) g.add_edge(START, "orchestrator") g.add_edge("orchestrator", "retrieval") g.add_edge("retrieval", "writer") g.add_edge("writer", "validator") g.add_conditional_edges("validator", _route_after_validator, { "writer": "writer", END: END, }) return g.compile() ``` ## 编排器如何选择正确的编写器模板 这在第一次阅读代码时很容易引起混淆，因此有必要详细说明。`agents/prompts/writer_*.txt`中的五个提示模板每个都有不同的输出形状。摘要模板只提供一个报告的简要摘要。威胁行为者配置文件使用MITRE样式的章节（别名、来源、TTP、工具、最近活动）。IOC报告有一条裁决线和一个比较表。模板的选择控制了最终答案的整个形状。 编排器负责决定使用哪一个。它的提示列出了五个有效的模板名称，并显示了一个示例，因此模型将用户的意图分类为五个类别之一，并将其作为JSON计划的`writer_template`字段返回。然后，编排器代码将该值与白名单（`_VALID_TEMPLATES`）进行验证，如果模型返回一个无法识别的值，则回落到`general`。编写器读取`state["plan"]["writer_template"]`，在`_TEMPLATE_MAP`中查找字符串，并从`agents/prompts/`加载相应的`.txt`文件。 要添加第六个模板，例如`vulnerability_brief`，您需要创建`agents/prompts/writer_vulnerability_brief.txt`，添加`"vulnerability_brief": "writer_vulnerability_brief"`到`writer.py`中的`_TEMPLATE_MAP`，将`vulnerability_brief`添加到`orchestrator.py`中的`_VALID_TEMPLATES`，并将新值追加到`agents/prompts/orchestrator.txt`中的管道分隔列表中。重新构建API容器，新模板即可上线。 ## MCP服务器设计允许您轻松添加新工具 最干净的演示是在丰富服务器中。要添加新的声誉提供者（例如GreyNoise），您只需在`mcp_servers/enrichment_mcp/adapters.py`中添加一个类： ```python class GreyNoiseAdapter(EnrichmentAdapter): name = "greynoise" def __init__(self): self.api_key = os.getenv("GREYNOISE_API_KEY", "") def available(self): return bool(self.api_key) def enrich(self, value, ioc_type): if not self.available(): return _mock_response("greynoise", value, ioc_type, classification="benign") with httpx.Client() as c: r = c.get(f"https://api.greynoise.io/v3/community/{value}", headers={"key": self.api_key}) return r.json() ADAPTERS.append(GreyNoiseAdapter()) ``` 丰富MCP服务器已经迭代`ADAPTERS`以注册每个提供者的一个工具，因此名为`greynoise_lookup`的工具会自动公开。无需更改代理代码，无需编写MCP服务器重启逻辑。检索代理在每次运行时都会重新发现工具，因此它会立即拾取新工具。 检索MCP服务器使用相同的模式。要添加新的检索源，只需将一个函数装饰为`@registry.register()`并放入`mcp_servers/retrieval_mcp/server.py`： ```python @registry.register() def shodan_lookup(ip: str, top_k: int = 10) -> list: """在Shodan中查找IP并返回端口、服务、横幅。""" return shodan_client.host(ip) ``` 就是这样。该工具会在下一次请求时出现在编排器的`available_tools`列表中，UI侧边栏会增加一个新的复选框。 ## 知识存储 四个存储驱动检索端。 Milvus存储分段CTI报告文本的句子转换器嵌入以及元数据（来源、URL、威胁行为者、提到的恶意软件、发布时间）。`vector_search`工具在那里运行HNSW搜索。分段使用带有重叠的标记感知句子分割器，以保留段落边界。 Elasticsearch以其完整的形式存储相同的文档，并以BM25索引。`keyword_search`工具在代理需要精确匹配检索时使用，例如CVE ID、哈希值或特定工具名称。语义相似性不适用于精确标记；BM25适用于它们。 Neo4j存储从MITRE ATT&CK企业包中填充的STIX 2.1知识图谱。威胁行为者、恶意软件、工具、攻击模式、漏洞以及它们之间的关系。`graph_query`工具通过Cypher查询回答诸如“APT41使用哪些工具？”之类的结构问题。 Postgres存储IOC数据库（IP、域名、哈希、URL、CVE行，具有置信度、标签、首次看到、最后看到）以及另外两个表：`agent_runs`记录每个完整运行及其跟踪，`feedback`记录每个竖起大拇指或竖起小拇指，键入相应的运行ID。`ioc_lookup`工具从`iocs`读取。API写入`agent_runs`和`feedback`。 ## 使用的提示技术 两种技术在不同地方使用。 **ReACT**驱动检索代理。模型在推理、调用工具和观察结果之间交替。我们使用OpenAI函数调用风格的ReACT，而不是文本解析风格，因为Qwen3更可靠地遵循函数调用规范，并且我们可以从LangChain消息日志中免费获得工具调用的可观察性。`agents/prompts/retrieval_react.txt`中的系统提示明确使用函数调用机制，并强制至少调用一次工具。 **一次性提示**在需要结构的所有地方使用。编排器提示有一个示例，显示我们期望的确切JSON形状。五个编写器模板中的每一个都有一个示例，显示该模板的章节结构。验证器有一个示例，显示JSON裁决形状。这里使用一次性提示的目的不是教模型领域（它知道领域），而是固定输出结构，以便下游代码可以确定性地解析它。对于本地LLM且没有微调，这比GPT-4更重要。 ## 技术栈 后端是Python 3.11，FastAPI带有`sse-starlette`用于流式处理，Pydantic v2用于验证，LangChain 0.3和LangGraph 0.2用于代理层。MCP服务器在`mcp`Python SDK上运行，带有FastMCP和SSE传输。代理到MCP的桥梁是一个70行的直接适配器（`agents/tools/mcp_loader.py`），它从MCP工具描述符构建LangChain `StructuredTool`实例，并使用MCP JSON模式动态构建Pydantic参数模式。我们故意不依赖`langchain-mcp-adapters`，因为其langchain-core下限漂移，每隔几周就会破坏安装。 LLM是本地的。默认使用通过其OpenAI兼容API由vLLM提供的Qwen3。任何vLLM、Ollama或TGI端点，只要支持相同的协议，都可以使用。`agents/llm.py`中的包装器剥去了几个请求体字段，这些字段是严格的vLLM构建所拒绝的（更多信息见下文）。 嵌入来自`sente-nce-transformers/multi-qa-mpnet-base-dot-v1`。数据库是Milvus 2.4、Elasticsearch 8、Neo4j 5，带有APOC插件和Postgres 16。前端是Next.js 14，带有App Router、React 18、Tailwind CSS和lucide-react。所有内容都打包在一个docker-compose文件中，无需外部服务依赖。 ## 仓库布局 ``` cti-agent/ docker-compose.yml 一键部署：4个数据库 + 3个MCP + API + UI .env.example 所有配置 agents/ LangGraph多智能体系统 orchestrator.py retrieval.py ReACT循环 writer.py 一次性模板 validator.py graph.py 状态图连接 + 条件重试边 prompts/ 8个提示模板 tools/mcp_loader.py 直接MCP到LangChain桥梁（70行） llm.py 容忍vLLM的ChatOpenAI工厂 state.py config.py ingestion/ RAG管道 rss_ingestor.py The Hacker News、Bleeping、Krebs chunker.py 标记感知句子分段 + 重叠 extractors.py 正则表达式IOC + 行为者/恶意软件字典 embedder.py writers.py Milvus + Elasticsearch pipeline.py 端到端运行器 databases/ milvus/client.py elasticsearch/client.py neo4j/ stix_loader.py MITRE ATT&CK到Neo4j queries.py Cypher助手 schema.cypher postgres/ init.sql IOC + agent_runs + feedback模式 client.py mcp_servers/ 每个工具类别一个服务器 retrieval_mcp/ vector_search、keyword_search、graph_query、ioc_lookup search_mcp/ tavily_search（混合实时/模拟） enrichment_mcp/ VT、AbuseIPDB、Recorded Future（混合） api/main.py FastAPI + SSE流式处理 frontend/ Next.js聊天机器人 app/page.tsx components/ lib/api.ts SSE客户端 scripts/ generate_synthetic_reports.py ``` ## 快速开始 先决条件是Docker Desktop和一个可从Docker网络访问的OpenAI兼容LLM端点。主机上的vLLM是默认假设。 复制示例环境文件，并将其指向您的LLM。 ```bash cp .env.example .env # 编辑LLM_BASE_URL和LLM_MODEL ``` # 可选：为实时模式设置 TAVILY_API_KEY、VIRUSTOTAL_API_KEY、ABUSEIPDB_API_KEY 启动整个系统。 ```bash docker compose up -d ``` 这将启动 Milvus、Elasticsearch、Neo4j、Postgres、三个 MCP 服务器、FastAPI 后端和 Next.js UI。UI 地址为 `http://localhost:3000`。 初始化知识库。第一个命令将 RSS 文章嵌入到 Milvus 和 Elasticsearch 中（如果 RSS 不可用，则会回退到合成的 CTI 报告）。第二个命令拉取 MITRE ATT&CK 并将其加载到 Neo4j 中。 ```bash docker compose exec api python -m ingestion.pipeline docker compose exec api python -m databases.neo4j.stix_loader ``` 在一切运行后，可以尝试以下几个查询： ``` APT41 使用哪些工具？ 总结最近的 LockBit 活动 185.12.45.78 是否恶意？ FIN7 和 LockBit 是否有关联？ 给我一个关于 CVE-2024-3400 的 IOC 报告 ``` ## 生产经验 这个项目的大部分时间都花在了让本地 LLM 系统可靠运行上。 vLLM 拒绝每一次聊天调用，返回 HTTP 400，直到我发现 `langchain-openai` 在非流式请求中附加 `stream_options={"include_usage": True}`，而更严格的 vLLM 版本拒绝接受它。在 `agents/llm.py` 中的修复是一个 httpx 请求钩子，它在每个输出的请求体中去掉这个字段（以及 `parallel_tool_calls` 和多余的 `n=1`），然后发送请求。在 langchain-openai 的每个补丁版本中都有效，因为在 HTTP 层进行剥离。 vLLM 也拒绝没有用户回合的请求，返回 "No user query found in messages." 标准 LangChain 模式将所有内容放入单个 `SystemMessage` 中不起作用。现在每个代理节点发送 `[SystemMessage(prompt), HumanMessage(user_query)]` 以满足 Qwen3 聊天模板。 Milvus 2.4 的表达式解析器拒绝带有前导通配符的 `LIKE` 过滤器（`failed to create query plan`）。我将威胁参与者过滤器从 Milvus 中拉出，改为在 Python 中进行软后过滤。嵌入排名已经将相关块拉到顶部，因此后过滤器更多的是一个 tiebreaker 而不是硬约束。 检索代理最初跳过所有工具调用，并根据一般知识回答。原因是提示混合了文本格式 ReACT 指令（Thought / Action / Observation）与 OpenAI 函数调用接口。Qwen3 查看冲突的指令并生成文本响应。重写提示以删除文本格式并明确要求函数调用修复了它。 ## 路线图 缓存是最大的空白。外部丰富 API 调用（VirusTotal、AbuseIPDB）很慢，并且速率有限；同一个 IOC 在运行过程中被查找数十次。在丰富 MCP 服务器前添加一个小的 Redis 或 Postgres 缓存将把延迟从大约每调用一秒降低到热指标的单数字毫秒，并防止 API 配额被大量消耗。相同的想法但对 `vector_search` 结果在会话内使用更短的 TTL。