PatelKrunali/Conversational-Threat-Intelligence-Analyst

# TIA — 对话式威胁情报分析师 一个专为 SOC 分析师设计的基于聊天界面的 Agent。直接用自然语言提问；该 Agent 会自动将请求路由到合适的威胁情报工具（VirusTotal、AbuseIPDB、AlienVault OTX、NVD、Shodan 等），关联查询结果，并给出带有证据支持的回答——绝不胡编乱造。 专为 EC-Council “对话式威胁情报分析师”技术评估（ECC-ASMT-2026）而构建。 ## 功能说明 | 能力 | 示例 | |---|---| | IOC 查询 | "Is 45.83.122.10 malicious?" | | 威胁组织与 TTP 分析 | "What TTPs is APT29 known for?" | | 漏洞暴露面分析 (CVEs) | "We run Confluence 7.13 — are we exposed?" | | 主机暴露面分析 (Shodan) | "What does this server expose to the internet?" | | 扩展追踪 | "Pivot from that IP to related domains." | | 多轮追问 | "Is 45.83.122.10 malicious?" → "who owns it?" → "its ASN?" → "should I block it?" | 回答将以二/三级（Tier-2/3）SOC 分析师简报的形式结构化呈现——包括摘要、调查发现、证据（标明来源）、风险评估、建议操作、置信度、参考来源——而不是直接抛出原始的 JSON 数据。 ## 架构 ``` app/ api/ FastAPI routers (chat, sessions, health, metrics) + DI providers agents/ LangGraph state graph: classify -> agent (LLM+tools) -> synthesize nodes/ classify.py, agent.py (model + tool binding), tools_executor.py tools/ Typed async adapters: virustotal, abuseipdb, otx, nvd, shodan + registry.py (the 5 LangChain tools the LLM actually sees) memory/ Session/conversation repository pattern (SQLite-backed, survives restarts), TTL expiration, cleanup security/ Injection scanning (direct + indirect), input validation, sanitization services/ ConversationService (turn orchestration + streaming), typed friendly errors, conversation titles, Prometheus metrics models/ Pydantic domain + API contracts -- the only shapes that cross LLM <-> tool <-> API boundaries prompts/ system_prompt.py, security_prompt.py, tool_prompt.py, response_prompt.py, classification_prompt.py config/ pydantic-settings environment configuration utils/ structured logging, validators, Langfuse tracing frontend/ React (Vite) chat console: sidebar, streaming progress, markdown reports tests/ pytest suite: memory, validation, security, tools, routing, sessions, streaming, error mapping, SQLite persistence ``` ### 系统架构 ``` flowchart LR User(["SOC Analyst"]) --> FE["React Frontend
(Vite)"] FE -->|"POST /chat
POST /chat/stream"| API["FastAPI
app/api"] API --> CS["ConversationService
app/services"] CS --> Mem[("SQLite
app/memory")] CS --> Sec["Security Layer
app/security"] CS --> Graph["LangGraph Agent
app/agents"] Graph --> Tools["Tool Adapters
app/tools"] Tools --> Ext[("VirusTotal · AbuseIPDB
OTX · NVD · Shodan")] Graph -.->|traces| Langfuse["Langfuse
(POST /chat only)"] CS -.->|metrics / logs| Obs["Prometheus · structured logs"] ``` ### Agent 工作流（位于 `app/agents` 内） ``` flowchart LR Start(["User message"]) --> Classify["classify
keyword heuristic,
observability only"] Classify --> AgentNode["agent
LLM + 5 bound tools"] AgentNode -->|tool_calls present| ToolsNode["tools
parallel execution"] ToolsNode -->|"results wrapped as
untrusted_data"| AgentNode AgentNode -->|no tool_calls| Answer(["Final answer
Summary / Findings / Evidence /
Risk / Actions / Confidence / Sources"]) ``` 该 Agent 是一个基于 **LangGraph** 的状态图，而非手工编写的循环：`classify`（轻量级关键词启发式算法，仅用于可观测性） → `agent`（一个 LLM 节点，默认使用 OpenRouter，或通过 `LLM_PROVIDER=groq` 使用 Groq，并通过 `.bind_tools()` 绑定了 5 个工具） → 有条件地触发 `tools`（一轮对话中的每一次工具调用都是并发执行的） → 返回至 `agent`，如此循环直到模型不再请求调用工具。关于为何将意图路由交由模型的原生工具调用（而不是分类器）处理，以及 prompt 注入防御的工作原理，请参阅 `DESIGN_NOTE.md`。 **LLM 绝不直接调用外部 API。** 它仅能识别五个强类型的 LangChain 工具（`app/tools/registry.py`）；每个工具都会验证其输入，调用 `app/tools/*.py` 中的一个或多个适配器，并返回一个 Pydantic 模型。外部 API 是事实来源——模型只是位于其上的推理与编排层。 每个 LangGraph 节点都会显式地将其 `RunnableConfig` 转发给内部的 `.ainvoke()` 调用（不仅仅依赖隐式的 contextvar 传播）——这正是使得 Langfuse 的 span 归因以及 SSE 工具进度流传输保持可靠的原因；在此功能接入前，已直接针对 `astream_events` 的输出进行了验证。 ## 安装说明 要求使用 **Python 3.12** 和 **Node 18+**（用于前端）。 ``` cd threat-intel-agent python3.12 -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install -r requirements.txt cp .env.example .env ``` 编辑 `.env`： ``` LLM_PROVIDER=openrouter # "openrouter" (default) or "groq" -- see below OPENROUTER_API_KEY=sk-or-... # required if LLM_PROVIDER=openrouter — https://openrouter.ai/keys VIRUSTOTAL_API_KEY=... # optional — free tier at virustotal.com ABUSEIPDB_API_KEY=... # optional — free tier at abuseipdb.com OTX_API_KEY=... # optional — OTX pulse search works keyless too NVD_API_KEY=... # optional — NVD works keyless (rate-limited) SHODAN_API_KEY=... # optional — free tier at shodan.io LANGFUSE_PUBLIC_KEY=... # optional — tracing, see Observability below LANGFUSE_SECRET_KEY=... LANGFUSE_HOST=... # e.g. https://us.cloud.langfuse.com for the US region ``` **切换至 Groq。** 设置 `LLM_PROVIDER=groq` 和 `GROQ_API_KEY=gsk-...`（可在 https://console.groq.com/keys 获取）—— `GROQ_MODEL` 默认为 `llama-3.3-70b-versatile`。OpenRouter 依然是默认选项，除此之外无需更改任何内容；这只是在同一个指向不同 OpenAI 兼容端点（`app/agents/nodes/agent.py`）的 `ChatOpenAI` 客户端背后进行的直接替换。如果 `openrouter/free` 的单次调用模型自动路由（参见已知局限性）在演示时显得太不稳定，此功能会很有用。 **你不需要配置所有的 API Key。** 缺失的 Key 会触发优雅降级：该来源会被跳过，Agent 会告知你其不可用，并且依然会基于*已配置的*可用数据源给出回答。`GET /health` 会显示当前处于激活状态的配置，前端的实时状态指示也会与其保持同步。 **务必将真实的 API Key 保存在 `.env` 中，绝不要放在 `.env.example` 里。** `.gitignore` 仅排除了 `.env` 文件——`.env.example` 是受版本控制的模板。如果将真实密钥粘贴到示例文件中，将会导致它们被提交到 git 历史记录中。 ## 运行说明 后端： ``` uvicorn app.main:app --reload --port 8000 ``` 前端（在单独的终端中运行）： ``` cd frontend npm install npm run dev ``` 打开 http://localhost:5173 —— Vite 开发服务器会将 `/chat`、`/sessions`、`/health` 和 `/metrics` 代理至运行在 8000 端口的后端，因此在开发环境下无需进行 CORS 或 base-URL 配置。若是构建静态生产包：在 `frontend/` 目录下运行 `npm run build`（输出目录为 `frontend/dist/`，可部署在任何静态主机或位于 FastAPI 后端前方的反向代理服务器之后）。关于组件结构以及客户端如何消费 SSE 流，请参阅 [`frontend/README.md`](frontend/README.md)。 对话会持久化存储到 `data/sessions.db` 中（SQLite 文件，已被 git 忽略，会自动创建）并且能在服务器重启后保留——侧边栏会列出过往的对话，点击即可重新加载其完整历史记录。 ## API 接口 | Endpoint | 用途 | |---|---| | `POST /chat` | 发送消息；返回回答、证据来源、置信度、追踪记录 | | `POST /chat/stream` | 同上，以 Server-Sent Events (SSE) 格式返回：`turn_start`、`intent`、`tool_start`/`tool_end`、`token`、`security`、`done`（完整响应）、`error` | | `POST /sessions` | 提前创建一个空会话（可选——chat 接口也会按需延迟创建） | | `GET /sessions` | 列出过往的对话（包含标题、轮次计数、时间戳），用于侧边栏展示 | | `GET /sessions/{id}` | 会话元数据：标题、轮次计数、实体、消息数量 | | `GET /sessions/{id}/messages` | 完整的用户/助手历史记录 + 追踪记录 + 实体，用于重新加载对话 | | `DELETE /sessions/{id}` | 结束一个会话 | | `GET /health` | 查看已配置的数据源/LLM 以及当前的活动会话数 | | `GET /metrics` | Prometheus 展示格式 | ## 运行测试 ``` pytest tests/ -v ``` 共包含 106 个测试，不需要网络连接或 API Key。测试套件在进程内驱动真实的 FastAPI 应用（会话内存、安全包装、路由、流传输等均会真实执行）；只有 LLM 调用本身被替换为了预先编排好的 `GraphRunner`/`GraphEventStreamer` 桩件（详见 `tests/conftest.py`），从而保证测试套件保持快速且结果确定。工具适配器针对模拟的 HTTP 响应（使用 `respx`）进行了独立测试，覆盖了成功、404、429 以及缺少 API Key 的场景。`test_llm_provider.py` 验证了 OpenRouter/Groq 的切换机制确实能为各个提供商构造出指向正确端点和模型的客户端，且全程无需触碰任何一方的网络。`conftest.py` 还专门针对测试进程禁用了 dotenv 加载，因此开发者的真实 `.env` 密钥绝不会泄漏到测试程序认为的“未配置 API Key”环境中——这是本项目曾遇到并修复过的真实 Bug。`test_sse.py` 锁定了 SSE 心跳行为机制（对于响应缓慢、静默的数据源，依然能确保其真实事件被准确送达，且不会丢失或重复任何内容），同时 `test_conversations.py` 覆盖了针对最终缺失的 graph 事件的 token 累积降级策略——这两个 Bug 都是通过针对真实模型进行实测时发现的，而非凭空臆测（参见下文的“流传输可靠性”）。 ## 设计说明（摘要——完整文档请见 DESIGN_NOTE.md） - **意图路由**：通过 `.bind_tools()` 利用模型原生的工具调用（函数调用）能力，而非手工编写分类器——在处理各种自然语言表述时，其泛化能力优于正则匹配。一个轻量级的关键词分类器（`app/agents/nodes/classify.py`）会并行运行，但纯粹用于可观测性目的——它只会在追踪记录/指标中为每一轮对话标注可能的意图，但绝不干涉路由走向。 - **多轮记忆**：完整的对话历史（包括之前的工具调用结果）会在每一轮对话中通过 `app/memory` 回放给模型，*并且*每轮对话都会向系统提示词注入明确的实体记忆提示（最近触及的 5 个 IOC/组织/产品），作为一种确定性的兜底手段。系统提示词明确指示模型要在整个连续追问的链条中解析代词，而不仅仅是回溯一环，并将判断性问题（“我应该拦截它吗？”）视为对已确立上下文的推理分析，而不是一次全新的查询——该逻辑已通过真实模型在包含四个代词链接场景下的实际测试中验证。 - **Prompt 注入防御**：工具的输出会以结构独立的方式传输于 LangChain 的 `ToolMessage` 中（绝不会直接拼接在指令文本内），被包裹在显式的 `` 标记中，并接受启发式扫描——匹配到的内容会被内联标记并显示在追踪记录中，而不是被悄悄过滤。详见 `DESIGN_NOTE.md` 中的完整说明。 - **工具编排**：在单次模型对话回合内的所有工具调用都是并发的（`asyncio.gather`）；通过单次调用的预算限制来控制成本；每个适配器仅针对瞬时的网络故障进行重试（不重试 4xx 错误），并始终返回类型化结果而非抛出异常，从而确保单个数据源的故障绝不会阻塞其他数据源。 - **置信度评分**：每个工具结果会根据数据源的覆盖范围和数据完整性携带一个 High（高）/Medium（中）/Low（低）的置信度评级；整轮对话的综合置信度会采取保守策略，即取所有引用数据源中的最低置信度。 - **错误处理**：原生的 LLM 提供商异常（速率限制、认证失败、超时）会被映射为类型化的、对分析师友好的错误（`app/services/errors.py`），并附带正确的 HTTP 状态码——绝不向 UI 抛出堆栈追踪。全局异常处理器作为应对所有未知突发情况的最后一道安全网。 - **可观测性**：每一次工具调用（名称、参数、延迟、置信度、错误、注入标志）都会流入追踪面板并增加 Prometheus 计数器（`GET /metrics`）。结构化的 JSON 日志会为每一轮对话记录会话 ID、意图、工具耗时以及最终置信度。Langfuse 追踪（使用一个全局共享的 `CallbackHandler`，而不是为每个请求单独分配）会为每一轮对话创建追踪记录，并为分类、LLM 调用以及各个工具执行过程生成嵌套的 span——这是在真实的 Langfuse 项目中实测验证过的，而非仅靠代码审查。**该功能在 `POST /chat` 上已启用，但在 `POST /chat/stream` 上被刻意禁用**——具体原因请参阅下文的“流传输可靠性”。 - **会话存储**：`SqliteSessionRepository`（基于文件的 `data/sessions.db`）是默认的生产环境配置——对话记录可在重启后保留，并支撑侧边栏的对话列表显示。`InMemorySessionRepository` 则用于支撑测试套件。两者均实现了相同的 `SessionRepository` 接口，因此要将其替换为多进程生产环境的 Redis 只需要在一个局部范围内进行代码修改。 - **流传输可靠性**：`POST /chat/stream` 驱动 LangGraph 的 `astream_events` 接口，并将原始的链/工具/token 事件转换为面向 UI 的精简事件词汇表，该映射逻辑在接入服务层之前，已根据真实事件输出进行了验证，而非盲目遵从文档说明。在实测中发现了两个可靠性问题，目前已予以修复，而非仅仅停留在评估层面： - 一个普通的（非工具调用的）LLM 轮次可能会在 30-60 秒甚至更长时间内保持零字节数据发送，某些浏览器或代理会将其视为死连接，并在真正的结果返回前切断连接。`app/api/sse.py` 在后台任务中运行数据源，并基于定时器发送 SSE 心跳注释，该心跳机制完全独立于数据源本身——如果在心跳超时时直接取消数据源自身的迭代，将会中断正在执行中的 LLM 调用，而不仅仅是跳过一次心跳，这正是早期尝试时踏入的错误方向。 - Langfuse 的旧版（v2）LangChain 回调偶尔会导致 `astream_events` 无法发出携带最终结果的事件——现象已被直接复现（在挂载该回调的情况下，3 次真实调用中有 2 次失败；去除该回调后，3 次调用全部成功），这看起来像是 SDK 后台刷新线程中的线程安全竞态条件。修复方法是在流传输路径上专门禁用追踪（非流式传输的 `/chat` 依然保留完整的追踪），并且让 `stream_turn` 在最终事件因为任何原因缺失时，通过已经流式传输的 token 重新拼装出最终答案，而不是丢弃掉分析人员已经亲眼看着显示出来的真实回答。 ## 已知局限性 - NVD 的免 Key 层级受速率限制（5 次请求/30秒）；Agent 会对其自身的请求进行节流以维持在此限制之下，这可能会给漏洞暴露面分析的查询增加几秒钟的延迟。 - OTX 的免费 pulse 搜索来源于社区，对于一些冷门的威胁组织可能缺乏数据；Agent 针对知名组织（APT29、APT28、Lazarus、FIN7）利用一个小型静态的 MITRE ATT&CK 参考表来进行补充。 - `SqliteSessionRepository` 不支持 `db_path=":memory:"` —— 为了保证线程安全，每次调用都会打开自己的独立连接，而内存中的 SQLite 数据库仅隶属于创建它的那个连接。请使用真实的文件路径。 - 仅支持单进程：SQLite 和基于进程的指标无法在多个 uvicorn worker 之间共享状态。对于本项目所针对的演示规模来说没有问题；但需注意，这是实现横向扩展时必须首先改造的地方（引入共享的 Postgres/Redis 存储库、Prometheus pushgateway 或针对每个 worker 的抓取目标）。 - Langfuse 追踪仅覆盖了非流式的 `POST /chat` 端点。`POST /chat/stream`（前端实际使用的接口）在运行时关闭了追踪，因为 Langfuse v2 SDK 的回调在实际负载下会间歇性地中断事件流（参见上文的“流传输可靠性”）——这是真实可复现的问题，而非理论假设。这个权衡是经过深思熟虑的：丢失一轮对话的追踪记录是可以接受的，但悄悄丢弃分析人员的回答则是致命的。如果重新考虑此方案，意味着要么使用 Langfuse 基于更新版 OTel 构建的 v3 SDK，要么编写自定义的轻量级 span 发射器，而在本项目中这两者均未尝试。 - `openrouter/free`（OpenRouter 的免费自动路由层）非常便于零成本测试，但它每次调用都会路由到不同的底层免费模型，其中某些模型在工具调用方面的表现明显不如其他模型稳定——在使用该特定模型时，请预料到偶尔会出现 `ModelUnavailableError` 的响应。付费的或单一指定的 OpenRouter 模型会更为稳定一致；如果你更倾向于使用单一模型的快速免费端点，而非自动路由，`LLM_PROVIDER=groq`（参见安装说明）则是另一个备选方案。

标签：AI智能体, AV绕过, FastAPI, LangGraph, Python, 威胁情报, 安全运营(SOC), 开发者工具, 无后门, 自定义请求头, 逆向工具