UtRaj/Threat-Intelligence
GitHub: UtRaj/Threat-Intelligence
一个基于 LangGraph 多智能体流水线的生产级威胁情报处理系统，自动摄取实时 CVE 数据并与内部资产关联以生成 AI 辅助修复建议。
Stars: 1 | Forks: 0
# TIAP — Threat Intelligence Agentic Pipeline
[](https://www.python.org/downloads/)
[](LICENSE)
[](https://github.com/astral-sh/ruff)
TIAP 是一个为安全运营中心 (SOC) 构建的生产级 **Threat Intelligence Agentic Pipeline**。它能自主摄取实时 CVE 数据，运行带有人在回路 (HITL) 审查的多智能体 AI pipeline，将漏洞与内部资产相关联，并生成结构化的修复建议，同时提供自动化的质量评估。
**核心能力**：在 Neo4j 知识图谱上执行多跳 GraphRAG 查询，将 CVE、资产、威胁行为者和缓解措施关联起来——从而发现非明显的暴露面。
## ⚠️ 当前状态 / 已知限制
- HITL 被实现为生成后的人工审查，而不是真正的 pipeline 暂停/恢复流程。
- 评估默认使用启发式评分；RAGAS 的集成代码已存在，但需要 OpenAI judge 访问权限以及进一步的集成测试才能完全启用。
- 支持实时的 NVD CVE 关联，但 CPE/软件解析较为简化；在演示场景中，模拟和沙箱关联更加可靠。
- 通过 Qdrant 进行向量搜索的功能已可用并经过健康检查，但尚未成为主要的咨询生成路径。
- 系统支持为 Groq、NVD 和 Shodan 配置可选的 API key，并在未设置 key 时回退到模拟/启发式行为。
## 🎯 核心功能
- **🤖 多智能体 LangGraph Pipeline** — 六节点的有状态智能体图，支持生成后的人工审查和基于置信度的重试循环
- **📡 实时威胁情报源** — 集成 NVD API v2、CISA KEV 目录以及可选的 Shodan，用于实时漏洞富化
- **🕸️ Neo4j 知识图谱** — 3 到 5 跳的遍历查询，揭示非明显的资产暴露链
- **⚡ 生产级 FastAPI** — 异步 REST API，带有 OpenAPI 文档、完整的状态/公告查询以及图形可视化端点
- **📊 交互式 Streamlit 仪表盘** — 多页面分析师 UI，包含威胁地图、图形可视化、pipeline 监控和沙箱威胁注入
- **🧪 评估运行器** — 针对黄金 QA 数据集进行质量评分；目前默认运行启发式评估，已集成 RAGAS 但受限于 OpenAI API 的可用性
- **🐳 Docker Compose 编排** — 四个服务 (Neo4j, Qdrant, API, Dashboard)，包含健康检查和依赖排序
- **💾 持久化 SQLite 后端** — 在 Neo4j/Qdrant 不可用时，作为运行、公告和执行日志的备用存储
## 📐 架构
```
flowchart TD
subgraph Feeds["🔗 Live Intelligence Feeds"]
NVD["NVD API v2
(NIST Vulnerabilities)"] KEV["CISA KEV Catalog
(Known Exploited)"] SHO["Shodan API
(Asset Discovery)"] end subgraph Cache["💾 Cache & Fallback"] Mock["Mock Data
(Dev Mode)"] SQLite["SQLite Persistence
(Offline Mode)"] end subgraph Pipeline["🤖 LangGraph Agent Pipeline"] ING["1️⃣ Ingestor
(Fetch + Normalize)"] ENR["2️⃣ Enricher
(CVSS · KEV · CWE)"] COR["3️⃣ Correlator
(Asset Matching)"] PRI["4️⃣ Prioritizer
(Risk Scoring)"] HITL{"👤 HITL Review
(Post-Generation Analyst Review)"} ADV["5️⃣ Advisory Writer
(Structured Report)"] EVL["6️⃣ Evaluator
(Quality Scoring)" ] end subgraph Storage["🗄️ Data Layer"] N4J[("Neo4j
Knowledge Graph")] QDR[("Qdrant
Vector Store")] SQ[("SQLite
Fallback")] end subgraph Services["🔧 Backend Services"] API["FastAPI :8000
/run · /status · /advisories
/health · /graph · /evaluate"] end subgraph UI["👁️ Frontend"] DASH["Streamlit :8501
Overview · Advisory Center
Review Queue · Pipeline Monitor
Threat Graph · Evaluation · Sandbox"] end Feeds -->|Real-time enrichment| Pipeline Cache -->|Fallback data| Pipeline Pipeline -->|State + Context| Storage Storage -->|Query results| API API -->|JSON responses| UI style Feeds fill:#ff9999 style Pipeline fill:#99ccff style Storage fill:#99ff99 style Services fill:#ffcc99 style UI fill:#cc99ff ``` ## 🚀 快速开始 ### 前置条件 - **Python 3.11+** - **Docker & Docker Compose** (可选，用于包含 Neo4j/Qdrant 的完整技术栈) - **API Keys** (Groq, NVD, 可选的 Shodan) ### 1. 克隆并安装 ``` # Clone 仓库 git clone https://github.com/yourusername/tiap.git cd tiap # 创建虚拟环境 python -m venv venv # 激活 (Windows) venv\Scripts\activate # 或者 (macOS/Linux) source venv/bin/activate # 安装依赖 pip install -r requirements.txt ``` ### 2. 配置环境 在项目根目录创建一个 `.env` 文件： ``` # LLM Provider GROQ_API_KEY=your-groq-api-key-here # Threat Intelligence APIs NVD_API_KEY=your-nist-nvd-api-key SHODAN_API_KEY=your-shodan-key-optional # Neo4j (Docker 服务或外部) NEO4J_URI=bolt://localhost:7687 NEO4J_USER=neo4j NEO4J_PASSWORD=tiap_password # Qdrant (Docker 服务或外部) QDRANT_HOST=localhost QDRANT_PORT=6333 # LangSmith (可选，用于追踪) LANGSMITH_API_KEY=optional LANGSMITH_PROJECT=tiap # API 配置 TIAP_API_URL=http://localhost:8000 TIAP_DB_PATH=./data/tiap.db LOG_LEVEL=INFO ``` ### 3. 启动完整技术栈 (Docker Compose) ``` # 启动所有服务：Neo4j、Qdrant、API、Dashboard docker-compose up -d # 检查服务健康状态 docker-compose ps # 查看日志 docker-compose logs -f api ``` 服务将在以下地址可用： - **API**: http://localhost:8000 (OpenAPI 文档位于 `/docs`) - **Dashboard**: http://localhost:8501 - **Neo4j Browser**: http://localhost:7474 (用户名: neo4j, 密码: tiap_password) - **Qdrant UI**: http://localhost:6333 (如果启用) ### 4. 本地运行 Pipeline (开发环境) ``` # 激活 venv source venv/bin/activate # or venv\Scripts\activate on Windows # 运行主 pipeline 脚本 python main.py # 或直接通过 Python 运行 python -c "from src.agents.pipeline import run_pipeline; run_pipeline()" ``` ## 📊 仪表盘页面 访问 http://localhost:8501 进入分析师仪表盘： | 页面 | 用途 | |------|---------| | **📊 概览** | 实时威胁指标、CVE 时间线、关键资产状态 | | **🛡️ 咨询中心** | 浏览生成的修复公告及其详细信息和跟踪记录 | | **✅ 审查队列** | 等待分析师批准的公告 HITL 队列 | | **🔍 Pipeline 监控** | 实时 pipeline 执行日志、智能体状态转换、错误跟踪 | | **🕸️ 威胁图谱** | 交互式 Neo4j 知识图谱：CVE ↔ 资产 ↔ 威胁行为者 | | **📈 评估** | RAGAS 评分指标：忠实度、相关性、精确度 | | **🧪 沙箱** | 注入自定义 CVE/威胁以测试 pipeline 行为 | ## 🔌 API 端点 ### Pipeline 管理 - `POST /pipeline/run` — 触发 pipeline 执行，支持可选的 `source` 和 `limit` - `POST /sandbox/inject` — 注入沙箱 CVE 或场景以进行临时测试 - `GET /runs` — 列出 pipeline 运行记录 - `GET /runs/{run_id}` — 获取运行详情和执行轨迹 - `GET /advisories` — 列出所有已生成的公告，支持过滤 - `GET /advisories/pending` — 列出待审查的公告 - `POST /pipeline/approve/{advisory_id}` — 在分析师审查后批准公告 - `POST /pipeline/reject/{advisory_id}` — 在分析师审查后拒绝公告 ### 数据查询 - `GET /assets` — 列出内部资产清单 - `GET /graph/topology` — Neo4j 图谱 schema 和节点计数 ### 评估与监控 - `GET /eval/latest` — 来自 `data/eval_results.json` 的最新评估结果 - `GET /health` — 服务健康状态 + 下游依赖状态 **完整 OpenAPI 文档**: http://localhost:8000/docs ## 🧪 测试 运行测试套件以验证 pipeline： ``` # 所有测试 pytest # 仅 Unit tests pytest -m unit # Integration tests（需要 Docker stack 运行） pytest -m integration # 特定的测试文件 pytest tests/test_integration.py -v ``` **测试标记：** - `unit` — 快速单元测试 - `integration` — 完整 pipeline 测试 - `slow` — 耗时 >5s 的测试 ## 📁 项目结构 ``` tiap/ ├── api/ # FastAPI entry point │ ├── main.py # REST service initialization │ └── __init__.py ├── dashboard/ # Streamlit dashboard │ ├── app.py # Multi-page dashboard config │ ├── pages/ # Dashboard pages │ │ ├── 1_overview.py # Threat metrics & timeline │ │ ├── 2_advisory_center.py # Advisory browsing │ │ ├── 3_review_queue.py # HITL review interface │ │ ├── 4_pipeline_monitor.py # Execution monitoring │ │ ├── 5_threat_graph.py # Graph visualization │ │ ├── 6_evaluation.py # RAGAS metrics │ │ └── 7_sandbox.py # Threat injection testing │ └── __init__.py ├── src/ # Core application logic │ ├── agents/ # LangGraph agent nodes │ │ ├── pipeline.py # StatefulGraph orchestration │ │ ├── state.py # Pipeline state schema │ │ ├── ingestion_agent.py # CVE fetching & normalization │ │ ├── extraction_agent.py # Data enrichment │ │ ├── context_agent.py # Asset correlation │ │ ├── remediation_agent.py # Advisory generation │ │ ├── validation_agent.py # Quality validation │ │ ├── hitl_node.py # Human-in-the-loop review │ │ └── __init__.py │ ├── database/ # Persistence & graph layers │ │ ├── graph_client.py # Neo4j wrapper │ │ ├── vector_client.py # Qdrant embeddings │ │ ├── persistence.py # SQLite fallback │ │ └── __init__.py │ ├── feeds/ # Threat intelligence sources │ │ ├── nvd_feed.py # NIST NVD API v2 │ │ ├── cisa_kev_feed.py # CISA KEV catalog │ │ ├── shodan_feed.py # Shodan API │ │ ├── mock_feeds.py # Development mock data │ │ └── __init__.py │ ├── schemas/ # Pydantic data models │ │ ├── models.py # CVE, Asset, Advisory schemas │ │ └── __init__.py │ ├── eval/ # Evaluation & QA │ │ ├── evaluate_pipeline.py # RAGAS evaluation runner │ │ ├── golden_dataset.py # QA ground truth │ │ └── __init__.py │ ├── tools/ # Utility tools │ │ └── mock_feeds.py # Mock data generation │ ├── utils/ # Shared utilities │ │ ├── config.py # Pydantic Settings │ │ ├── logger.py # Structured logging │ │ ├── schemas.py # Shared schema utils │ │ ├── graph_vis.py # Graph visualization helpers │ │ └── __init__.py │ ├── api.py # Legacy API (reference) │ ├── dashboard.py # Legacy dashboard (reference) │ └── __init__.py ├── tests/ # Test suite │ ├── conftest.py # Pytest fixtures │ ├── test_integration.py # End-to-end tests │ ├── test_feeds.py # Feed ingestion tests │ ├── test_persistence.py # Database tests │ ├── test_schemas.py # Schema validation tests │ ├── test_api.py # API endpoint tests │ ├── test_mock_feeds.py # Mock data tests │ └── __init__.py ├── data/ # Runtime data │ ├── tiap.db # SQLite fallback (generated) │ ├── internal_assets.json # Asset inventory (sample) │ └── cache/ # NVD/feed cache ├── reports/ # Generated advisory reports (JSON) ├── docs/ # Documentation │ ├── adr_001_architecture_choices.md │ └── project_action_plan.md ├── docker/ # Docker config │ └── streamlit_config.toml ├── Dockerfile.api # FastAPI container ├── Dockerfile.dashboard # Streamlit container ├── docker-compose.yml # Service orchestration ├── pyproject.toml # Project metadata ├── requirements.txt # Python dependencies ├── pytest.ini # Pytest config ├── Makefile # Development commands ├── main.py # CLI entry point (legacy) ├── run.ps1 # PowerShell launcher (Windows) ├── run.bat # Batch launcher (Windows) └── README.md # This file ``` ## 🔄 智能体 Pipeline 演练 ### 执行流程 ``` sequenceDiagram participant User participant API participant Pipeline participant LLM as Groq LLM participant Neo4j participant DB as SQLite User->>API: POST /run {"cves": [...]} API->>Pipeline: trigger_pipeline() rect rgb(200, 150, 255) Note over Pipeline: 1️⃣ INGESTOR Pipeline->>LLM: Normalize & validate CVEs LLM-->>Pipeline: Structured CVE objects end rect rgb(150, 200, 255) Note over Pipeline: 2️⃣ ENRICHER Pipeline->>LLM: Enrich with CVSS, CWE, KEV status LLM-->>Pipeline: Enriched CVE context end rect rgb(150, 255, 200) Note over Pipeline: 3️⃣ CORRELATOR Pipeline->>Neo4j: Match CVEs to internal assets Neo4j-->>Pipeline: Exposed asset graph end rect rgb(255, 200, 150) Note over Pipeline: 4️⃣ PRIORITIZER Pipeline->>LLM: Risk scoring & ranking LLM-->>Pipeline: Ranked threat list end rect rgb(255, 150, 150) Note over Pipeline: HITL REVIEW Pipeline->>API: Save advisories for analyst review User->>API: Approve/reject advisories API-->>Pipeline: Continue with finalized status end rect rgb(200, 200, 255) Note over Pipeline: 5️⃣ ADVISORY WRITER Pipeline->>LLM: Generate remediation steps LLM-->>Pipeline: Structured advisory reports end rect rgb(255, 255, 150) Note over Pipeline: 6️⃣ EVALUATOR Pipeline->>LLM: RAGAS quality scoring LLM-->>Pipeline: Faithfulness/Relevancy/Precision end Pipeline->>Neo4j: Persist graph relationships Pipeline->>DB: Save advisory & run history Pipeline-->>API: Complete with metrics API-->>User: GET /status/{run_id} ``` ### 状态机 每个智能体节点维护一个 `PipelineState` 对象： ``` class PipelineState(TypedDict): run_id: str # Unique execution ID cves: List[CVE] # Input vulnerabilities enriched_cves: List[CVE] # After enrichment exposed_assets: List[Asset] # Matched internal inventory scored_threats: List[Dict] # Prioritized rankings advisories: List[Advisory] # Final remediation reports evaluation_metrics: Dict # RAGAS scores hitl_approved: bool # Analyst sign-off error_log: List[str] # Pipeline errors/warnings ``` ## 🛠️ 开发 ### 本地运行 (不使用 Docker) ``` # 激活环境 source venv/bin/activate # 仅在 Docker 中启动 Neo4j docker run -d -p 7687:7687 -e NEO4J_AUTH=neo4j/tiap_password neo4j:5.18.0-community # 仅在 Docker 中启动 Qdrant docker run -d -p 6333:6333 qdrant/qdrant:v1.8.0 # 运行 FastAPI uvicorn api.main:app --reload --port 8000 # 在另一个终端中，运行 Streamlit streamlit run dashboard/app.py --server.port 8501 ``` ### 代码风格与检查 ``` # 使用 ruff 格式化代码 ruff format src/ tests/ # 使用 ruff 进行 Lint ruff check src/ tests/ # 使用 mypy 进行类型检查 mypy src/ ``` ### 添加新的情报源 1. 在 `src/feeds/` 中创建情报源类： class CustomFeed: async def fetch(self) -> List[CVE]: """Implement custom feed logic""" pass 2. 在 `src/agents/ingestion_agent.py` 中注册 3. 在 `tests/test_feeds.py` 中添加测试 4. 在 `src/api.py` 中记录 API endpoint ## 🔐 安全与最佳实践 ### API Keys 与密钥 **绝不要提交 `.env` 或敏感文件。** 使用 `.gitignore` 并始终： ``` # Good ✓ export GROQ_API_KEY="sk-..." python main.py # Bad ✗ # GROQ_API_KEY=sk-... 已提交到 Git ``` ### Neo4j 默认凭据 在生产环境中更改 `NEO4J_PASSWORD`： ``` # docker-compose.yml environment: NEO4J_AUTH: neo4j/YOUR_STRONG_PASSWORD ``` ### HITL 审查 HITL 节点将公告标记为生成后需进行分析师审查，而不是暂停 pipeline 执行。关键公告 (CVSS > 9.0) 会被提升审查，其批准状态会被单独记录。 ## 📚 文档 - [架构决策记录](docs/adr_001_architecture_choices.md) — 设计选择与基本原理 - [项目行动计划](docs/project_action_plan.md) — 详细的代码演练 - [API 参考](http://localhost:8000/docs) — 自动生成的 OpenAPI 文档 ## 🐛 故障排除 ### Docker 问题 ``` # Neo4j 健康检查失败？ docker-compose logs neo4j # Qdrant 没有响应？ curl -X GET http://localhost:6333/health # 从头开始 Rebuild docker-compose down -v docker-compose up --build ``` ### HITL 审查状态 HITL 节点记录等待审查的公告，批准操作是通过更新公告状态来处理的，而不是通过 pipeline 恢复 endpoint。 通过以下方式批准： ``` curl -X POST http://localhost:8000/pipeline/approve/{advisory_id} -H "Content-Type: application/json" -d '{"reviewer":"analyst@example.com","notes":"Approved"}' ``` 通过以下方式拒绝： ``` curl -X POST http://localhost:8000/pipeline/reject/{advisory_id} -H "Content-Type: application/json" -d '{"reviewer":"analyst@example.com","reason":"Not applicable"}' ``` ### 缺少 NVD 数据？ 确保已设置 `NVD_API_KEY`。如果情报源不可用，pipeline 将回退到模拟数据： ``` echo $NVD_API_KEY # Should not be empty ``` ## 📈 性能调优 ### Neo4j 查询优化 添加索引以加快遍历速度： ``` CREATE INDEX FOR (c:CVE) ON (c.id); CREATE INDEX FOR (a:Asset) ON (a.hostname); CREATE INDEX FOR (t:ThreatActor) ON (t.name); ``` ### Qdrant 向量搜索 针对更大数据集增加 Hnsw 设置： ``` client.recreate_collection( collection_name="threats", vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE), hnsw_config=models.HnswConfigDiff(m=32, ef_construct=200, ef_search=100) ) ``` ## 📜 许可证 本项目基于 **MIT License** 授权 — 详情请参阅 [LICENSE](LICENSE)。 ## 🙏 致谢 构建工具： - [LangGraph](https://github.com/langchain-ai/langgraph) — 有状态智能体编排 - [FastAPI](https://fastapi.tiangolo.com/) — 高性能 REST API - [Streamlit](https://streamlit.io/) — 交互式数据应用 - [Neo4j](https://neo4j.com/) — 知识图谱数据库 - [Groq](https://groq.com/) — 快速 LLM 推理 - [RAGAS](https://ragas.io/) — RAG 评估框架 ## 📧 联系与支持 如有问题、议题或功能请求： - **GitHub Issues**: [报告 Bug](https://github.com/yourusername/tiap/issues) - **Discussions**: [发起讨论](https://github.com/yourusername/tiap/discussions) **用 ❤️ 为安全社区打造** ## 智能体工作原理 该 pipeline 是一个六节点的 LangGraph `StateGraph`。每个节点都是一个包装在 `try/except` 中的 Python 异步函数，因此任何单点故障都不会导致整个图崩溃。 | # | 智能体 | 职责 | |---|-------|--------------| | 1 | **摄取器 (Ingestor)** | 从 NVD API v2 获取最近的 CVE（默认为过去 7 天）。如果无法访问 API，则回退到本地缓存。将记录标准化为 `CVERecord` Pydantic 模型。 | | 2 | **富化器 (Enricher)** | 为每个 CVE 拉取 CVSS v3.1 基础分数、CWE 分类和 CISA KEV 状态。将富化后的节点写入 Neo4j。 | | 3 | **关联器 (Correlator)** | 查询 Neo4j 以寻找匹配易受攻击 CPE 字符串的资产（主机、服务、软件版本）。可选择调用 Shodan 查找暴露在互联网上的资产。 | | 4 | **优先级排序器 (Prioritiser)** | 使用 CVSS + KEV 权重 + 暴露因子对每个 CVE × 资产对进行评分。返回排序列表。如果最高分超过配置的阈值，图将路由到**重试循环**进行更深度的关联。 | | 5 | **公告编写器 (Advisory Writer)** | 使用 Groq LLM (llama-3 70B) 为前 N 个优先发现生成结构化的 Markdown 公告。将 embedding 存储在 Qdrant 中以用于相似性搜索。 | | 6 | **评估器 (Evaluator)** | 针对生成的公告运行 RAGAS 指标（忠实度、答案相关性、上下文精确度），并将分数写入 SQLite 备用数据库。 | **HITL 审查：** 在优先级排序器和公告编写器之间，图会将公告标记为待审查；批准状态会被记录，但 pipeline 不会执行真正的中断/恢复周期。分析师可以通过仪表盘或 `POST /pipeline/approve/{advisory_id}` 和 `POST /pipeline/reject/{advisory_id}` 批准或拒绝公告。 **重试循环：** 如果关联器返回的暴露资产为零，图将使用更宽的时间窗口路由回摄取器（在放弃并编写“未发现暴露”公告之前最多重试三次）。 ## 实时数据源 | 来源 | Endpoint | 用途 | 回退方案 | |--------|----------|----------|----------| | **NVD API v2** | `https://services.nvd.nist.gov/rest/json/cves/2.0` | CVE 记录、CVSS、CPE | `data/cache/` 中的本地 JSON 缓存 | | **CISA KEV** | `https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json` | 已知被利用标记 | `data/kev_snapshot.json` 中的内置快照 | | **Shodan** | `https://api.shodan.io/shodan/host/search` | 互联网暴露资产查找 | 跳过；暴露分数设为 0 | 所有出站 API 调用均使用带有指数退避的 `tenacity.retry`（初始等待 1 秒，最大 60 秒，最多 5 次尝试），并且被隔离在 `src/ingestors` 中，以便在测试中轻松进行模拟。 ## 评估 TIAP 可以使用 **RAGAS** 来衡量公告质量，但除非配置了可选的 OpenAI judge API，否则实时 pipeline 目前默认采用启发式评估。 | 指标 | 衡量内容 | |--------|-----------------| | **忠实度 (Faithfulness)** | 公告是否仅陈述了出现在检索到的 CVE 上下文中的内容？用于捕捉幻觉。 | | **答案相关性 (Answer Relevancy)** | 公告在多大程度上回答了“SOC 应该如何处理这种暴露？”这一隐含问题？ | | **上下文精确度 (Context Precision)** | 在检索到的图/向量上下文中，有多少比例对公告实际上是有用的？用于衡量检索质量。 | 分数将写入 SQLite 并在 `GET /evaluate/latest` 暴露。仪表盘显示了一个时间序列图表，以便您可以在 LLM 或检索发生变化时跟踪质量漂移。 手动运行评估： ``` make eval ``` ## 服务 URL | 服务 | URL | 备注 | |---------|-----|-------| | **Streamlit Dashboard** | http://localhost:8501 | 主要分析师界面 | | **FastAPI REST API** | http://localhost:8000 | 后端服务 | | **API 交互式文档** | http://localhost:8000/docs | Swagger UI | | **Neo4j Browser** | http://localhost:7474 | 图谱浏览器 (登录: `neo4j` / `tiap_password`) | | **Qdrant Dashboard** | http://localhost:6333/dashboard | 向量集合浏览器 | ## 项目结构 ``` tiap-v2/ │ ├── api/ # FastAPI application │ ├── main.py # App factory, lifespan, CORS, routers │ ├── routers/ # Endpoint modules (run, advisories, graph, eval) │ └── dependencies.py # Shared DI (DB clients, pipeline ref) │ ├── src/ # Core business logic │ ├── config.py # Pydantic-settings config (reads .env) │ ├── state.py # LangGraph PipelineState TypedDict │ ├── pipeline.py # StateGraph definition and compilation │ │ │ ├── agents/ # Six agent node functions │ │ ├── ingestor.py # CVE fetch + normalise │ │ ├── enricher.py # CVSS · KEV · CWE enrichment │ │ ├── correlator.py # Asset exposure lookup │ │ ├── prioritiser.py # Risk scoring + ranking │ │ ├── advisory_writer.py # LLM report generation │ │ └── evaluator.py # RAGAS scoring │ │ │ ├── ingestors/ # External API clients │ │ ├── nvd_client.py # NVD API v2 with tenacity retry │ │ ├── kev_client.py # CISA KEV feed │ │ └── shodan_client.py # Shodan search (optional) │ │ │ ├── database/ # Storage layer │ │ ├── graph_client.py # Neo4j async driver + in-memory fallback │ │ ├── vector_client.py # Qdrant client + in-memory fallback │ │ └── sqlite_client.py # SQLite fallback for run state + scores │ │ │ ├── models/ # Pydantic v2 data models │ │ ├── cve.py # CVERecord, CVSSScore, KEVEntry │ │ ├── asset.py # Asset, Exposure │ │ └── advisory.py # Advisory, EvaluationResult │ │ │ └── eval/ # Evaluation harness │ ├── evaluate_pipeline.py # Entry point (make eval) │ └── golden_dataset.py # Golden QA pairs for RAGAS │ ├── dashboard/ # Streamlit app │ ├── app.py # Main page + navigation │ └── pages/ # Multi-page app modules │ ├── 1_threat_map.py # Plotly exposure map │ ├── 2_graph_view.py # PyVis knowledge graph │ ├── 3_advisories.py # Advisory browser + HITL approval │ └── 4_evaluation.py # RAGAS score history │ ├── tests/ # pytest suite │ ├── unit/ # Fast isolated tests │ └── integration/ # Full pipeline tests (needs Docker) │ ├── data/ # Local data files │ ├── cache/ # NVD response cache (gitignored) │ └── kev_snapshot.json # Bundled KEV fallback │ ├── .streamlit/ │ └── config.toml # Theme + server settings │ ├── docker-compose.yml # Four-service orchestration ├── Dockerfile.api # Multi-stage FastAPI image ├── Dockerfile.dashboard # Multi-stage Streamlit image ├── Makefile # Dev + ops automation ├── pyproject.toml # Build, ruff, mypy, pytest config ├── requirements.txt # Python dependencies ├── .env.example # Environment variable template ├── .gitignore ├── .dockerignore └── README.md # This file ``` ## 贡献 1. `make venv && make install` — 设置您的环境 2. `make lint && make typecheck` — 提交前必须通过 3. `make test` — 所有测试必须通过 4. 向 `main` 提交 PR ## 许可证 MIT — 详情请参阅 `LICENSE`。
(NIST Vulnerabilities)"] KEV["CISA KEV Catalog
(Known Exploited)"] SHO["Shodan API
(Asset Discovery)"] end subgraph Cache["💾 Cache & Fallback"] Mock["Mock Data
(Dev Mode)"] SQLite["SQLite Persistence
(Offline Mode)"] end subgraph Pipeline["🤖 LangGraph Agent Pipeline"] ING["1️⃣ Ingestor
(Fetch + Normalize)"] ENR["2️⃣ Enricher
(CVSS · KEV · CWE)"] COR["3️⃣ Correlator
(Asset Matching)"] PRI["4️⃣ Prioritizer
(Risk Scoring)"] HITL{"👤 HITL Review
(Post-Generation Analyst Review)"} ADV["5️⃣ Advisory Writer
(Structured Report)"] EVL["6️⃣ Evaluator
(Quality Scoring)" ] end subgraph Storage["🗄️ Data Layer"] N4J[("Neo4j
Knowledge Graph")] QDR[("Qdrant
Vector Store")] SQ[("SQLite
Fallback")] end subgraph Services["🔧 Backend Services"] API["FastAPI :8000
/run · /status · /advisories
/health · /graph · /evaluate"] end subgraph UI["👁️ Frontend"] DASH["Streamlit :8501
Overview · Advisory Center
Review Queue · Pipeline Monitor
Threat Graph · Evaluation · Sandbox"] end Feeds -->|Real-time enrichment| Pipeline Cache -->|Fallback data| Pipeline Pipeline -->|State + Context| Storage Storage -->|Query results| API API -->|JSON responses| UI style Feeds fill:#ff9999 style Pipeline fill:#99ccff style Storage fill:#99ff99 style Services fill:#ffcc99 style UI fill:#cc99ff ``` ## 🚀 快速开始 ### 前置条件 - **Python 3.11+** - **Docker & Docker Compose** (可选，用于包含 Neo4j/Qdrant 的完整技术栈) - **API Keys** (Groq, NVD, 可选的 Shodan) ### 1. 克隆并安装 ``` # Clone 仓库 git clone https://github.com/yourusername/tiap.git cd tiap # 创建虚拟环境 python -m venv venv # 激活 (Windows) venv\Scripts\activate # 或者 (macOS/Linux) source venv/bin/activate # 安装依赖 pip install -r requirements.txt ``` ### 2. 配置环境 在项目根目录创建一个 `.env` 文件： ``` # LLM Provider GROQ_API_KEY=your-groq-api-key-here # Threat Intelligence APIs NVD_API_KEY=your-nist-nvd-api-key SHODAN_API_KEY=your-shodan-key-optional # Neo4j (Docker 服务或外部) NEO4J_URI=bolt://localhost:7687 NEO4J_USER=neo4j NEO4J_PASSWORD=tiap_password # Qdrant (Docker 服务或外部) QDRANT_HOST=localhost QDRANT_PORT=6333 # LangSmith (可选，用于追踪) LANGSMITH_API_KEY=optional LANGSMITH_PROJECT=tiap # API 配置 TIAP_API_URL=http://localhost:8000 TIAP_DB_PATH=./data/tiap.db LOG_LEVEL=INFO ``` ### 3. 启动完整技术栈 (Docker Compose) ``` # 启动所有服务：Neo4j、Qdrant、API、Dashboard docker-compose up -d # 检查服务健康状态 docker-compose ps # 查看日志 docker-compose logs -f api ``` 服务将在以下地址可用： - **API**: http://localhost:8000 (OpenAPI 文档位于 `/docs`) - **Dashboard**: http://localhost:8501 - **Neo4j Browser**: http://localhost:7474 (用户名: neo4j, 密码: tiap_password) - **Qdrant UI**: http://localhost:6333 (如果启用) ### 4. 本地运行 Pipeline (开发环境) ``` # 激活 venv source venv/bin/activate # or venv\Scripts\activate on Windows # 运行主 pipeline 脚本 python main.py # 或直接通过 Python 运行 python -c "from src.agents.pipeline import run_pipeline; run_pipeline()" ``` ## 📊 仪表盘页面 访问 http://localhost:8501 进入分析师仪表盘： | 页面 | 用途 | |------|---------| | **📊 概览** | 实时威胁指标、CVE 时间线、关键资产状态 | | **🛡️ 咨询中心** | 浏览生成的修复公告及其详细信息和跟踪记录 | | **✅ 审查队列** | 等待分析师批准的公告 HITL 队列 | | **🔍 Pipeline 监控** | 实时 pipeline 执行日志、智能体状态转换、错误跟踪 | | **🕸️ 威胁图谱** | 交互式 Neo4j 知识图谱：CVE ↔ 资产 ↔ 威胁行为者 | | **📈 评估** | RAGAS 评分指标：忠实度、相关性、精确度 | | **🧪 沙箱** | 注入自定义 CVE/威胁以测试 pipeline 行为 | ## 🔌 API 端点 ### Pipeline 管理 - `POST /pipeline/run` — 触发 pipeline 执行，支持可选的 `source` 和 `limit` - `POST /sandbox/inject` — 注入沙箱 CVE 或场景以进行临时测试 - `GET /runs` — 列出 pipeline 运行记录 - `GET /runs/{run_id}` — 获取运行详情和执行轨迹 - `GET /advisories` — 列出所有已生成的公告，支持过滤 - `GET /advisories/pending` — 列出待审查的公告 - `POST /pipeline/approve/{advisory_id}` — 在分析师审查后批准公告 - `POST /pipeline/reject/{advisory_id}` — 在分析师审查后拒绝公告 ### 数据查询 - `GET /assets` — 列出内部资产清单 - `GET /graph/topology` — Neo4j 图谱 schema 和节点计数 ### 评估与监控 - `GET /eval/latest` — 来自 `data/eval_results.json` 的最新评估结果 - `GET /health` — 服务健康状态 + 下游依赖状态 **完整 OpenAPI 文档**: http://localhost:8000/docs ## 🧪 测试 运行测试套件以验证 pipeline： ``` # 所有测试 pytest # 仅 Unit tests pytest -m unit # Integration tests（需要 Docker stack 运行） pytest -m integration # 特定的测试文件 pytest tests/test_integration.py -v ``` **测试标记：** - `unit` — 快速单元测试 - `integration` — 完整 pipeline 测试 - `slow` — 耗时 >5s 的测试 ## 📁 项目结构 ``` tiap/ ├── api/ # FastAPI entry point │ ├── main.py # REST service initialization │ └── __init__.py ├── dashboard/ # Streamlit dashboard │ ├── app.py # Multi-page dashboard config │ ├── pages/ # Dashboard pages │ │ ├── 1_overview.py # Threat metrics & timeline │ │ ├── 2_advisory_center.py # Advisory browsing │ │ ├── 3_review_queue.py # HITL review interface │ │ ├── 4_pipeline_monitor.py # Execution monitoring │ │ ├── 5_threat_graph.py # Graph visualization │ │ ├── 6_evaluation.py # RAGAS metrics │ │ └── 7_sandbox.py # Threat injection testing │ └── __init__.py ├── src/ # Core application logic │ ├── agents/ # LangGraph agent nodes │ │ ├── pipeline.py # StatefulGraph orchestration │ │ ├── state.py # Pipeline state schema │ │ ├── ingestion_agent.py # CVE fetching & normalization │ │ ├── extraction_agent.py # Data enrichment │ │ ├── context_agent.py # Asset correlation │ │ ├── remediation_agent.py # Advisory generation │ │ ├── validation_agent.py # Quality validation │ │ ├── hitl_node.py # Human-in-the-loop review │ │ └── __init__.py │ ├── database/ # Persistence & graph layers │ │ ├── graph_client.py # Neo4j wrapper │ │ ├── vector_client.py # Qdrant embeddings │ │ ├── persistence.py # SQLite fallback │ │ └── __init__.py │ ├── feeds/ # Threat intelligence sources │ │ ├── nvd_feed.py # NIST NVD API v2 │ │ ├── cisa_kev_feed.py # CISA KEV catalog │ │ ├── shodan_feed.py # Shodan API │ │ ├── mock_feeds.py # Development mock data │ │ └── __init__.py │ ├── schemas/ # Pydantic data models │ │ ├── models.py # CVE, Asset, Advisory schemas │ │ └── __init__.py │ ├── eval/ # Evaluation & QA │ │ ├── evaluate_pipeline.py # RAGAS evaluation runner │ │ ├── golden_dataset.py # QA ground truth │ │ └── __init__.py │ ├── tools/ # Utility tools │ │ └── mock_feeds.py # Mock data generation │ ├── utils/ # Shared utilities │ │ ├── config.py # Pydantic Settings │ │ ├── logger.py # Structured logging │ │ ├── schemas.py # Shared schema utils │ │ ├── graph_vis.py # Graph visualization helpers │ │ └── __init__.py │ ├── api.py # Legacy API (reference) │ ├── dashboard.py # Legacy dashboard (reference) │ └── __init__.py ├── tests/ # Test suite │ ├── conftest.py # Pytest fixtures │ ├── test_integration.py # End-to-end tests │ ├── test_feeds.py # Feed ingestion tests │ ├── test_persistence.py # Database tests │ ├── test_schemas.py # Schema validation tests │ ├── test_api.py # API endpoint tests │ ├── test_mock_feeds.py # Mock data tests │ └── __init__.py ├── data/ # Runtime data │ ├── tiap.db # SQLite fallback (generated) │ ├── internal_assets.json # Asset inventory (sample) │ └── cache/ # NVD/feed cache ├── reports/ # Generated advisory reports (JSON) ├── docs/ # Documentation │ ├── adr_001_architecture_choices.md │ └── project_action_plan.md ├── docker/ # Docker config │ └── streamlit_config.toml ├── Dockerfile.api # FastAPI container ├── Dockerfile.dashboard # Streamlit container ├── docker-compose.yml # Service orchestration ├── pyproject.toml # Project metadata ├── requirements.txt # Python dependencies ├── pytest.ini # Pytest config ├── Makefile # Development commands ├── main.py # CLI entry point (legacy) ├── run.ps1 # PowerShell launcher (Windows) ├── run.bat # Batch launcher (Windows) └── README.md # This file ``` ## 🔄 智能体 Pipeline 演练 ### 执行流程 ``` sequenceDiagram participant User participant API participant Pipeline participant LLM as Groq LLM participant Neo4j participant DB as SQLite User->>API: POST /run {"cves": [...]} API->>Pipeline: trigger_pipeline() rect rgb(200, 150, 255) Note over Pipeline: 1️⃣ INGESTOR Pipeline->>LLM: Normalize & validate CVEs LLM-->>Pipeline: Structured CVE objects end rect rgb(150, 200, 255) Note over Pipeline: 2️⃣ ENRICHER Pipeline->>LLM: Enrich with CVSS, CWE, KEV status LLM-->>Pipeline: Enriched CVE context end rect rgb(150, 255, 200) Note over Pipeline: 3️⃣ CORRELATOR Pipeline->>Neo4j: Match CVEs to internal assets Neo4j-->>Pipeline: Exposed asset graph end rect rgb(255, 200, 150) Note over Pipeline: 4️⃣ PRIORITIZER Pipeline->>LLM: Risk scoring & ranking LLM-->>Pipeline: Ranked threat list end rect rgb(255, 150, 150) Note over Pipeline: HITL REVIEW Pipeline->>API: Save advisories for analyst review User->>API: Approve/reject advisories API-->>Pipeline: Continue with finalized status end rect rgb(200, 200, 255) Note over Pipeline: 5️⃣ ADVISORY WRITER Pipeline->>LLM: Generate remediation steps LLM-->>Pipeline: Structured advisory reports end rect rgb(255, 255, 150) Note over Pipeline: 6️⃣ EVALUATOR Pipeline->>LLM: RAGAS quality scoring LLM-->>Pipeline: Faithfulness/Relevancy/Precision end Pipeline->>Neo4j: Persist graph relationships Pipeline->>DB: Save advisory & run history Pipeline-->>API: Complete with metrics API-->>User: GET /status/{run_id} ``` ### 状态机 每个智能体节点维护一个 `PipelineState` 对象： ``` class PipelineState(TypedDict): run_id: str # Unique execution ID cves: List[CVE] # Input vulnerabilities enriched_cves: List[CVE] # After enrichment exposed_assets: List[Asset] # Matched internal inventory scored_threats: List[Dict] # Prioritized rankings advisories: List[Advisory] # Final remediation reports evaluation_metrics: Dict # RAGAS scores hitl_approved: bool # Analyst sign-off error_log: List[str] # Pipeline errors/warnings ``` ## 🛠️ 开发 ### 本地运行 (不使用 Docker) ``` # 激活环境 source venv/bin/activate # 仅在 Docker 中启动 Neo4j docker run -d -p 7687:7687 -e NEO4J_AUTH=neo4j/tiap_password neo4j:5.18.0-community # 仅在 Docker 中启动 Qdrant docker run -d -p 6333:6333 qdrant/qdrant:v1.8.0 # 运行 FastAPI uvicorn api.main:app --reload --port 8000 # 在另一个终端中，运行 Streamlit streamlit run dashboard/app.py --server.port 8501 ``` ### 代码风格与检查 ``` # 使用 ruff 格式化代码 ruff format src/ tests/ # 使用 ruff 进行 Lint ruff check src/ tests/ # 使用 mypy 进行类型检查 mypy src/ ``` ### 添加新的情报源 1. 在 `src/feeds/` 中创建情报源类： class CustomFeed: async def fetch(self) -> List[CVE]: """Implement custom feed logic""" pass 2. 在 `src/agents/ingestion_agent.py` 中注册 3. 在 `tests/test_feeds.py` 中添加测试 4. 在 `src/api.py` 中记录 API endpoint ## 🔐 安全与最佳实践 ### API Keys 与密钥 **绝不要提交 `.env` 或敏感文件。** 使用 `.gitignore` 并始终： ``` # Good ✓ export GROQ_API_KEY="sk-..." python main.py # Bad ✗ # GROQ_API_KEY=sk-... 已提交到 Git ``` ### Neo4j 默认凭据 在生产环境中更改 `NEO4J_PASSWORD`： ``` # docker-compose.yml environment: NEO4J_AUTH: neo4j/YOUR_STRONG_PASSWORD ``` ### HITL 审查 HITL 节点将公告标记为生成后需进行分析师审查，而不是暂停 pipeline 执行。关键公告 (CVSS > 9.0) 会被提升审查，其批准状态会被单独记录。 ## 📚 文档 - [架构决策记录](docs/adr_001_architecture_choices.md) — 设计选择与基本原理 - [项目行动计划](docs/project_action_plan.md) — 详细的代码演练 - [API 参考](http://localhost:8000/docs) — 自动生成的 OpenAPI 文档 ## 🐛 故障排除 ### Docker 问题 ``` # Neo4j 健康检查失败？ docker-compose logs neo4j # Qdrant 没有响应？ curl -X GET http://localhost:6333/health # 从头开始 Rebuild docker-compose down -v docker-compose up --build ``` ### HITL 审查状态 HITL 节点记录等待审查的公告，批准操作是通过更新公告状态来处理的，而不是通过 pipeline 恢复 endpoint。 通过以下方式批准： ``` curl -X POST http://localhost:8000/pipeline/approve/{advisory_id} -H "Content-Type: application/json" -d '{"reviewer":"analyst@example.com","notes":"Approved"}' ``` 通过以下方式拒绝： ``` curl -X POST http://localhost:8000/pipeline/reject/{advisory_id} -H "Content-Type: application/json" -d '{"reviewer":"analyst@example.com","reason":"Not applicable"}' ``` ### 缺少 NVD 数据？ 确保已设置 `NVD_API_KEY`。如果情报源不可用，pipeline 将回退到模拟数据： ``` echo $NVD_API_KEY # Should not be empty ``` ## 📈 性能调优 ### Neo4j 查询优化 添加索引以加快遍历速度： ``` CREATE INDEX FOR (c:CVE) ON (c.id); CREATE INDEX FOR (a:Asset) ON (a.hostname); CREATE INDEX FOR (t:ThreatActor) ON (t.name); ``` ### Qdrant 向量搜索 针对更大数据集增加 Hnsw 设置： ``` client.recreate_collection( collection_name="threats", vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE), hnsw_config=models.HnswConfigDiff(m=32, ef_construct=200, ef_search=100) ) ``` ## 📜 许可证 本项目基于 **MIT License** 授权 — 详情请参阅 [LICENSE](LICENSE)。 ## 🙏 致谢 构建工具： - [LangGraph](https://github.com/langchain-ai/langgraph) — 有状态智能体编排 - [FastAPI](https://fastapi.tiangolo.com/) — 高性能 REST API - [Streamlit](https://streamlit.io/) — 交互式数据应用 - [Neo4j](https://neo4j.com/) — 知识图谱数据库 - [Groq](https://groq.com/) — 快速 LLM 推理 - [RAGAS](https://ragas.io/) — RAG 评估框架 ## 📧 联系与支持 如有问题、议题或功能请求： - **GitHub Issues**: [报告 Bug](https://github.com/yourusername/tiap/issues) - **Discussions**: [发起讨论](https://github.com/yourusername/tiap/discussions) **用 ❤️ 为安全社区打造** ## 智能体工作原理 该 pipeline 是一个六节点的 LangGraph `StateGraph`。每个节点都是一个包装在 `try/except` 中的 Python 异步函数，因此任何单点故障都不会导致整个图崩溃。 | # | 智能体 | 职责 | |---|-------|--------------| | 1 | **摄取器 (Ingestor)** | 从 NVD API v2 获取最近的 CVE（默认为过去 7 天）。如果无法访问 API，则回退到本地缓存。将记录标准化为 `CVERecord` Pydantic 模型。 | | 2 | **富化器 (Enricher)** | 为每个 CVE 拉取 CVSS v3.1 基础分数、CWE 分类和 CISA KEV 状态。将富化后的节点写入 Neo4j。 | | 3 | **关联器 (Correlator)** | 查询 Neo4j 以寻找匹配易受攻击 CPE 字符串的资产（主机、服务、软件版本）。可选择调用 Shodan 查找暴露在互联网上的资产。 | | 4 | **优先级排序器 (Prioritiser)** | 使用 CVSS + KEV 权重 + 暴露因子对每个 CVE × 资产对进行评分。返回排序列表。如果最高分超过配置的阈值，图将路由到**重试循环**进行更深度的关联。 | | 5 | **公告编写器 (Advisory Writer)** | 使用 Groq LLM (llama-3 70B) 为前 N 个优先发现生成结构化的 Markdown 公告。将 embedding 存储在 Qdrant 中以用于相似性搜索。 | | 6 | **评估器 (Evaluator)** | 针对生成的公告运行 RAGAS 指标（忠实度、答案相关性、上下文精确度），并将分数写入 SQLite 备用数据库。 | **HITL 审查：** 在优先级排序器和公告编写器之间，图会将公告标记为待审查；批准状态会被记录，但 pipeline 不会执行真正的中断/恢复周期。分析师可以通过仪表盘或 `POST /pipeline/approve/{advisory_id}` 和 `POST /pipeline/reject/{advisory_id}` 批准或拒绝公告。 **重试循环：** 如果关联器返回的暴露资产为零，图将使用更宽的时间窗口路由回摄取器（在放弃并编写“未发现暴露”公告之前最多重试三次）。 ## 实时数据源 | 来源 | Endpoint | 用途 | 回退方案 | |--------|----------|----------|----------| | **NVD API v2** | `https://services.nvd.nist.gov/rest/json/cves/2.0` | CVE 记录、CVSS、CPE | `data/cache/` 中的本地 JSON 缓存 | | **CISA KEV** | `https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json` | 已知被利用标记 | `data/kev_snapshot.json` 中的内置快照 | | **Shodan** | `https://api.shodan.io/shodan/host/search` | 互联网暴露资产查找 | 跳过；暴露分数设为 0 | 所有出站 API 调用均使用带有指数退避的 `tenacity.retry`（初始等待 1 秒，最大 60 秒，最多 5 次尝试），并且被隔离在 `src/ingestors` 中，以便在测试中轻松进行模拟。 ## 评估 TIAP 可以使用 **RAGAS** 来衡量公告质量，但除非配置了可选的 OpenAI judge API，否则实时 pipeline 目前默认采用启发式评估。 | 指标 | 衡量内容 | |--------|-----------------| | **忠实度 (Faithfulness)** | 公告是否仅陈述了出现在检索到的 CVE 上下文中的内容？用于捕捉幻觉。 | | **答案相关性 (Answer Relevancy)** | 公告在多大程度上回答了“SOC 应该如何处理这种暴露？”这一隐含问题？ | | **上下文精确度 (Context Precision)** | 在检索到的图/向量上下文中，有多少比例对公告实际上是有用的？用于衡量检索质量。 | 分数将写入 SQLite 并在 `GET /evaluate/latest` 暴露。仪表盘显示了一个时间序列图表，以便您可以在 LLM 或检索发生变化时跟踪质量漂移。 手动运行评估： ``` make eval ``` ## 服务 URL | 服务 | URL | 备注 | |---------|-----|-------| | **Streamlit Dashboard** | http://localhost:8501 | 主要分析师界面 | | **FastAPI REST API** | http://localhost:8000 | 后端服务 | | **API 交互式文档** | http://localhost:8000/docs | Swagger UI | | **Neo4j Browser** | http://localhost:7474 | 图谱浏览器 (登录: `neo4j` / `tiap_password`) | | **Qdrant Dashboard** | http://localhost:6333/dashboard | 向量集合浏览器 | ## 项目结构 ``` tiap-v2/ │ ├── api/ # FastAPI application │ ├── main.py # App factory, lifespan, CORS, routers │ ├── routers/ # Endpoint modules (run, advisories, graph, eval) │ └── dependencies.py # Shared DI (DB clients, pipeline ref) │ ├── src/ # Core business logic │ ├── config.py # Pydantic-settings config (reads .env) │ ├── state.py # LangGraph PipelineState TypedDict │ ├── pipeline.py # StateGraph definition and compilation │ │ │ ├── agents/ # Six agent node functions │ │ ├── ingestor.py # CVE fetch + normalise │ │ ├── enricher.py # CVSS · KEV · CWE enrichment │ │ ├── correlator.py # Asset exposure lookup │ │ ├── prioritiser.py # Risk scoring + ranking │ │ ├── advisory_writer.py # LLM report generation │ │ └── evaluator.py # RAGAS scoring │ │ │ ├── ingestors/ # External API clients │ │ ├── nvd_client.py # NVD API v2 with tenacity retry │ │ ├── kev_client.py # CISA KEV feed │ │ └── shodan_client.py # Shodan search (optional) │ │ │ ├── database/ # Storage layer │ │ ├── graph_client.py # Neo4j async driver + in-memory fallback │ │ ├── vector_client.py # Qdrant client + in-memory fallback │ │ └── sqlite_client.py # SQLite fallback for run state + scores │ │ │ ├── models/ # Pydantic v2 data models │ │ ├── cve.py # CVERecord, CVSSScore, KEVEntry │ │ ├── asset.py # Asset, Exposure │ │ └── advisory.py # Advisory, EvaluationResult │ │ │ └── eval/ # Evaluation harness │ ├── evaluate_pipeline.py # Entry point (make eval) │ └── golden_dataset.py # Golden QA pairs for RAGAS │ ├── dashboard/ # Streamlit app │ ├── app.py # Main page + navigation │ └── pages/ # Multi-page app modules │ ├── 1_threat_map.py # Plotly exposure map │ ├── 2_graph_view.py # PyVis knowledge graph │ ├── 3_advisories.py # Advisory browser + HITL approval │ └── 4_evaluation.py # RAGAS score history │ ├── tests/ # pytest suite │ ├── unit/ # Fast isolated tests │ └── integration/ # Full pipeline tests (needs Docker) │ ├── data/ # Local data files │ ├── cache/ # NVD response cache (gitignored) │ └── kev_snapshot.json # Bundled KEV fallback │ ├── .streamlit/ │ └── config.toml # Theme + server settings │ ├── docker-compose.yml # Four-service orchestration ├── Dockerfile.api # Multi-stage FastAPI image ├── Dockerfile.dashboard # Multi-stage Streamlit image ├── Makefile # Dev + ops automation ├── pyproject.toml # Build, ruff, mypy, pytest config ├── requirements.txt # Python dependencies ├── .env.example # Environment variable template ├── .gitignore ├── .dockerignore └── README.md # This file ``` ## 贡献 1. `make venv && make install` — 设置您的环境 2. `make lint && make typecheck` — 提交前必须通过 3. `make test` — 所有测试必须通过 4. 向 `main` 提交 PR ## 许可证 MIT — 详情请参阅 `LICENSE`。
标签：AI智能体, GitHub, GPT, Kubernetes, Web报告查看器, 威胁情报, 安全运营, 实时处理, 密码管理, 开发者工具, 扫描框架, 漏洞管理, 请求拦截, 逆向工具