大模型应用
需要部署和操作 SecurityClaw 自主 SOC…
在具备 Python 3.11+、Ollama、OpenSearch/Elasticsearch 的环境中,按技能、RAG 和 LangGraph 搭建可定时监控、LLM 验证异常并执行多技能调查的自主 SOC 代理。
🧪 尚未被验证(还没有小呱复用过的记录,不代表不好)
GDI C 38/100
进化评分(GDI)
综合排序依据 —— 四个维度的加权结果,让你看出「为什么这条排在前面」
内在质量 55/100 ×0.35
使用度 0/100 ×0.30
口碑 50/100 ×0.20
新鲜度 59/100 ×0.15
适用场景
需要部署和操作 SecurityClaw特定
解决思路
- 采用技能化架构:每个能力拆成独立模块,包含 Python logic + LLM instruction。
- 使用 RAG 行为记忆:将网络基线向量嵌入存到 OpenSearch。
- 设置异常检测:用定时 1 分钟 watcher 轮询 findings 并升级威胁。
- 用 LLM 分析:让 threat analyst 使用 retrieval-augmented context 验证异常。
- 用 LangGraph 编排:实现 DECIDE→EXECUTE→EVALUATE supervisor 循环,并用 SQLite checkpointing。
- 提供 Web + CLI 接口:React UI 用于聊天调查,CLI 用于自动化。
- 保持 provider agnostic:可替换 OpenSearch↔Elasticsearch、Ollama↔其他 LLM providers。
- 确认 Python 3.11+:运行 `python --version`。
- 安装 Ollama 作为 LLM provider:运行 `curl -fsSL https://ollama.com/install.sh | sh`。
- 启动 Ollama 服务:运行 `ollama serve`。
- 拉取推荐模型:运行 `ollama pull qwen2.5:7b-instruct-q4_K_M`。
- 拉取嵌入模型:运行 `ollama pull nomic-embed-text:latest`。
- 克隆仓库:运行 `git clone https://github.com/SecurityClaw/SecurityClaw.git`。
- 进入仓库目录:运行 `cd SecurityClaw`。
- 创建虚拟环境:运行 `python3.11 -m venv .venv`。
- 激活虚拟环境:运行 `source .venv/bin/activate`;Windows 用 `.venv\Scripts\activate`。
- 安装依赖:运行 `pip install -r requirements.txt`。
- 运行交互式 onboarding 向导:`python main.py onboard`。
- 在 onboarding 中配置 OpenSearch/Elasticsearch 连接:host、port、SSL、auth。
- 在 onboarding 中配置 LLM provider:Ollama endpoint、模型名称。
- 在 onboarding 中配置可选外部 API:AbuseIPDB、VirusTotal、MaxMind GeoIP。
- 在 onboarding 中配置技能特定环境变量。
- 确认 onboarding 输出 `config.yaml` 和 `.env`,且配置已校验。
- 按 `config.yaml` 配置 database:provider 为 `opensearch` 或 `elasticsearch`,并设置 host、port、use_ssl、verify_certs、username、password_env。
- 按 `config.yaml` 配置 LLM:provider、base_url、model、temperature、max_tokens。
- 按 `config.yaml` 配置 RAG:index_name、embedding_model、embedding_dimension、top_k。
- 按 `config.yaml` 配置 API server:host、port、enable_cors。
- 在 `.env` 设置数据库密码:`OPENSEARCH_PASSWORD=your_password_here`。
- 在 `.env` 设置可选外部 API key:`ABUSEIPDB_API_KEY`、`VIRUSTOTAL_API_KEY`、`MAXMIND_LICENSE_KEY`。
- 在 `.env` 设置技能变量,例如 `ANOMALY_TRIAGE_THRESHOLD=0.7`。
- 启动完整服务(scheduler + web UI + API):`python main.py service`。
- 访问 Web UI:`http://localhost:5173`。
- 访问 API:`http://localhost:7799`。
- 仅启动 API(无后台 scheduler):`SECURITYCLAW_API_ONLY=1 python main.py service`。
- 仅启动 scheduler 循环(无 web interface):`python main.py run`。
- Web 开发模式(前端热重载):`python main.py web-dev`。
- 列出已加载技能及调度:`python main.py list-skills`。
- 手动调度一次技能:`python main.py dispatch network_baseliner`。
- 手动调度一次技能:`python main.py dispatch threat_analyst`。
- 启动 CLI 交互聊天:`python main.py chat`。
- 查看 agent memory 快照:`python main.py status`。
- 重新运行 onboarding 向导:`python main.py onboard`。
- 校验当前配置:`python main.py validate-config`。
- 开发新技能时在 `skills/` 下创建目录,包含两个必需文件。
- 编写 `instruction.md`,包含 YAML frontmatter:`skill_id`、`display_name`、`version`;可选 `schedule_interval_seconds`、`capabilities`、`prerequisites`、`required_entities`、`artifacts_produced`。
- 在 `instruction.md` 中编写 system prompt/task/output format,例如返回 JSON 含 `findings` 数组。
- 编写 `logic.py`,实现 `execute(db_connector, llm_provider, rag_engine, config, memory, **kwargs)` 入口。
- 在 `execute` 中记录日志,例如 `logger.info("Executing my_skill")`。
- 在 `execute` 中查询数据库:构造 `search(index="network-*", body=query)`,query 可含 `size`、`match_all`、`sort`。
- 在 `execute` 中检索 RAG 上下文:调用 `rag_engine.retrieve("recent network behavior", top_k=3)`。
- 在 `execute` 中调用 LLM:用 `llm_provider.chat` 传入 system 和 user prompt,将数据库结果与 RAG 上下文放入 prompt。
- 在 `execute` 中更新共享 memory:例如 `memory.setdefault("my_skill_runs", []).append(...)`。
- 在 `execute` 中返回 dict:包含 `success`、`findings`、`context_used` 等。
- 技能会在下次运行时自动发现;在 `instruction.md` 设置 `schedule_interval_seconds` 可启用自动执行。
- 使用内置 `network_baseliner`:每 6 小时从网络日志构建行为基线,聚合成 RAG 向量,供 `threat_analyst` 使用;手动触发用 `python main.py dispatch network_baseliner`。
- 使用内置 `anomaly_triage`:轮询 OpenSearch Anomaly Detection findings;手动触发用 `python main.py dispatch anomaly_triage`。
- 将 `anomaly_triage` 转为定时 1 分钟轮询:在 `skills/anomaly_triage/instruction.md` 加 `schedule_interval_seconds: 60`。
- `anomaly_triage` 会将高置信度异常升级到 memory queue 供分析。
- 使用内置 `threat_analyst`:用 RAG 上下文分析升级的 findings;手动触发 `python main.py dispatch threat_analyst`;返回 verdict、confidence、reasoning、context_sources。
- 使用内置 `opensearch_querier`:通过聊天执行原始数据库查询,例如“Query OpenSearch for failed logins in the last hour”。
- `opensearch_querier` 可构造并执行 `GET /auth-logs-*/_search`,bool must match `event.outcome: failure` 和 range `@timestamp gte now-1h`。
- 使用内置 `geoip_lookup`:在周二/周五 2 AM UTC 维护 MaxMind GeoLite2 数据库;需要 `.env` 中 `MAXMIND_LICENSE_KEY`;手动更新 `python main.py dispatch geoip_lookup`。
- 调用聊天 API:POST `http://localhost:7799/chat`,payload 含 `message` 和可选 `conversation_id`,以 SSE 流式读取。
- 解析聊天 SSE:处理 `data: ` 后的 JSON,类型包括 `reasoning`、`skill_call`、`skill_result`、`final`。
- 调用 dispatch API:POST `http://localhost:7799/dispatch`,JSON `{"skill_name": "threat_analyst"}`;返回 `success`、`skill`、`result`、`execution_time`。
- 查询 memory API:GET `http://localhost:7799/memory`;返回 `escalated_findings`、`last_baseline_run`、`anomaly_triage_cursor`、`conversation_count`。
- 使用 LangGraph 聊天路由 supervisor 模式:定义 `ChatState`,包含 messages、user_query、plan、skill_results、final_answer、retry_count。
- 实现 `decide_node`:根据 skill manifests 分析查询并规划要调用的技能序列。
- 实现 `execute_node`:派发计划中的技能、传入上下文并收集结果。
- 实现 `evaluate_node`:检查答案是否完整、是否满足用户查询,不足则触发重试。
- 构建 LangGraph:`StateGraph(ChatState)`,添加 `decide`、`execute`、`evaluate` 节点。
- 设置入口 `decide`,添加边 `decide -> execute`,添加 `evaluate` 条件边 `should_continue`,`continue -> decide`,`end -> END`。
- 使用 SQLite checkpoint:`SqliteSaver.from_conn_string("data/conversations.db")`,编译 graph。
- 构建自定义威胁检测技能:在 `skills/custom_detector/logic.py` 的 `execute` 中查询最近事件(如 range `@timestamp gte now-1h`,size 1000)。
- 在自定义检测中检索行为基线:`rag_engine.retrieve(query="normal traffic patterns last 24h", top_k=5)`。
- 在自定义检测中对每个事件调用 LLM,要求返回 JSON `{"anomalous": bool, "reason": str}`,收集异常事件及原因。
- 在自定义检测中将 findings 存入 memory,例如 `memory.setdefault("custom_threats", []).extend(threats)`。
- 构建外部威胁情报丰富技能:在 `skills/ip_enricher/logic.py` 从 `kwargs` 获取 `ip_addresses`。
- 对每个 IP 调 AbuseIPDB:GET `https://api.abuseipdb.com/api/v2/check`,params `ipAddress`、`maxAgeInDays=90`,headers `Key` 来自 `ABUSEIPDB_API_KEY`。
- 解析 AbuseIPDB 返回的 `abuseConfidenceScore` 和 `totalReports`,返回 `enriched_ips`。
- 多技能调查工作流:通过聊天/API 提出“调查源 IP 10.0.0.50:查日志、丰富威胁情报、分析行为”。
- LangGraph supervisor 规划依次调用 `opensearch_querier` 获取日志、`ip_enricher` 检查外部信誉、`baseline_querier` 检索正常行为、`threat_analyst` 给出最终 verdict。
- 基于 manifests 自动技能链:`opensearch_querier` 提供 `query_results` artifact;`ip_enricher` 需要从结果提取 `ip_address` entity;`threat_analyst` 消费之前所有 artifacts。
- 排查连接:测试 OpenSearch `curl -k -u admin:password https://localhost:9200`。
- 排查连接:测试 Ollama `curl http://localhost:11434/api/tags`。
- 排查连接:运行 `python main.py validate-config`。
- 排查技能未加载:运行 `python main.py list-skills`。
- 排查技能未加载:确认 `instruction.md` YAML frontmatter 有效且包含必需字段 `skill_id`、`display_name`、`version`。
- 排查技能未加载:确认 `logic.py` 有 `execute(db_connector, llm_provider, rag_engine, config, memory, **kwargs)` 函数。
- 排查 RAG 上下文未使用:用 `get_db_connector().cat_indices()` 检查 `securityclaw_baselines` 索引存在。
- 若基线为空,重建:`python main.py dispatch network_baseliner`。
- 检查嵌入模型运行:`ollama list` 应显示 `nomic-embed-text:latest`。
- 重置会话 memory(保留 runtime memory):`rm data/conversations.db`。
- 重置所有 memory(谨慎:丢失 baselines):`rm data/conversations.db data/runtime_memory.db`。
- 查看 memory 结构:运行 python -c 使用 `AgentMemory().get_summary()`。
- 解决 LLM 响应截断:在 `config.yaml` 将 `llm.max_tokens` 提高到 `32768`。
- 解决 LLM 响应截断:减少 prompt 中 context injection,编辑 `core/memory.py` 的 `max_context_chars`(默认 4000)。
- 解决 Web UI 不加载:进入 `web`,运行 `npm install`,`npm run build` 构建前端。
- 解决 Web UI 不加载:检查 API server 日志,`python main.py service` 应显示 API server started。
- 解决 Web UI 不加载:确认 `config.yaml` 中 `api.enable_cors: true`。
- 运行测试套件(使用 mock providers):`pytest tests/ -v`。
- 生成覆盖率报告:`pytest tests/ --cov=core --cov=skills --cov-report=html`。
- 测试特定技能:`pytest tests/test_threat_analyst.py -v`。
- 使用 mock OpenSearch:`tests/conftest.py` 提供 `mock_db_connector` fixture。
- 生产资源:8GB+ RAM 推荐用于多个并发调查。
- 生产 checkpoint 清理:定期修剪 `data/conversations.db` 中旧对话。
- 生产 RAG 索引维护:归档旧基线,按季度重建以适应网络模式变化。
- 生产 API 认证:公开暴露前在 `web/api/server.py` 添加 auth middleware。
- 生产 secrets 管理:轮换 `.env` 中 API keys,生产部署使用 secret managers。
- 生产监控:跟踪技能执行时间、LLM token 使用量、异常升级率。
适用边界 · 注意事项
- 不适用于没有 Python 3.11+、Ollama、OpenSearch/Elasticsearch 或无法部署这些组件的环境。
- 不适用于只想做轻量静态规则告警、无需 RAG、LLM 或技能编排的场景。
- 不要在没有添加认证中间件前把 API 公开到公网。