SkillSpector 架构深潜并发模型、线程安全与 contrib 批量扫描层的设计实践【免费下载链接】SkillSpectorSecurity scanner for AI agent skills. Detect vulnerabilities, malicious patterns, security risks, prompt injection, data exfiltration, and supply-chain risks in Claude Code, Codex, and MCP skills before you install them.项目地址: https://gitcode.com/GitHub_Trending/sk/SkillSpector本文面向 SkillSpector 的上游维护者与新贡献者基于仓库内 contrib/batch_scan/docs/archive/ARCHITECTURE_DEEP_DIVE.md 展开。文章完整覆盖上游架构、三层并行、线程安全、API 限流、Provider 系统与 contrib 集成方式并结合仓库源码给出可验证的实现证据。SkillSpector 是面向 AI Agent 技能skill包的安全扫描器用于在安装 Claude Code、Codex、MCP 等技能之前检测漏洞、恶意模式、Prompt 注入、数据外泄与供应链风险。本篇文章要回答的核心问题是当扫描一个 skill被建模为一次无状态的graph.invoke()纯函数调用之后如何在不改动上游一行代码的前提下用线程池把单 skill 扫描安全地扩展到任意规模的批量扫描读完本文你将掌握 SkillSpector 的三层并行金字塔结构、7 个线程安全兼容补丁的来历与原理、基于多 API Key 的水平限流方案以及 contrib 层长出来而非塞进去的集成哲学。1. 核心洞察graph.invoke()是一个纯函数SkillSpector 的整个架构建立在一个关键抽象之上把扫描一个 skill建模为无状态的纯函数state → graph.invoke(state) → result纯函数意味着相同的输入永远产生相同的输出且调用过程不依赖任何外部可变全局状态。一旦接受这个前提扫描 N 个 skill就只是对这个函数做mapresults map(graph.invoke, states)并行化也顺理成章地成为并行 mapwith ThreadPoolExecutor(max_workers4) as pool: results pool.map(graph.invoke, states)由此引出整个 contrib 层contrib/batch_scan/的设计纲领围绕这个 map 函数做加法——加语言检测、加 API 池、加比对标记——但永远不触碰函数本身。这一设计之所以成立是因为上游的 LangGraph 编译产物graph是只读的执行计划。从 graph.py 可以看到create_graph()把 7 个固定节点resolve_input、build_context、20 个 analyzer 节点、meta_analyzer、finalize_inspection_ledger、report与若干边组装成StateGraph后调用workflow.compile()并在 graph.py 处模块加载时执行一次graph create_graph()。此后该对象只被读取、从不被写入天然支持多线程并发调用。2. 无状态性证明逐层剖析纯函数不是口号而是要在每一层都能被证实的性质。文档逐层拆解了四个层次我们结合源码逐一验证。State 层每次invoke()创建新字典class SkillspectorState(TypedDict, totalFalse): input_path: str | None file_cache: dict[str, str] findings: Annotated[list[Finding], operator.add] ...真实定义位于 state.py要点有三totalFalse意味着所有字段均可选初始化没有任何约束contrib 层只需注入input_path、output_format、use_llm三个字段即可启动一次扫描见 runner.py 的scan_state。findings使用 reducer 合并实际为merge_findings_by_id按 finding_id 原地替换富化实例见 state.py但这种合并只发生在一次invoke()调用内部——LangGraph 在同一份 state 字典上执行 reducer 归并。每次invoke()都会创建一个全新的 state 字典input_path、file_cache等字段不会跨调用共享引用。Provider 层每次调用全新实例def create_openai_compatible_chat_model(*, model, credentials, max_tokens, timeout): return ChatOpenAI(modelmodel, api_keySecretStr(...), timeouttimeout)从 providers/init.py 的create_chat_model实现可以看到每次构造模型时都会调用provider.create_chat_model(...)生成全新的ChatOpenAI实例——不做连接池缓存。凭证通过参数传递来自resolve_credentials()而非读取全局状态。这意味着每个线程持有的 LLM 客户端彼此完全隔离。Analyzer 层实例局部状态class LLMAnalyzerBase: def __init__(self, base_prompt, model): self._llm get_chat_model(modelmodel) # fresh instance self._structured_llm ... # fresh instance见 llm_analyzer_base.py构造函数只接收base_prompt和model两个参数_llm与_structured_llm都是实例局部属性不跨 analyzer 共享。response_schema虽然在类上以类属性形式存在第 629 行但第 7 节会说明 contrib 层如何通过补丁把它实例化以彻底消除跨线程共享这正是历史上竞态 bug 的根源。Graph 层蓝图与素材分离graph create_graph() # compiled once at module load # Each invoke creates a new state; graph is a read-only execution plangraph 拓扑蓝图只读、无状态state 注入流水线的素材每次调用独立。线程安全检查Thread-1: graph.invoke(state_1) → reads/writes state_1 only Thread-2: graph.invoke(state_2) → reads/writes state_2 only Thread-3: graph.invoke(state_3) → reads/writes state_3 only结论安全。线程之间不存在共享的可变状态唯一的共享对象graph是只读的编译执行计划。测试 test_pool_wiring.py 与 test_runner_patches.py 从不同侧面验证了这条无状态边界。3. 三层并行金字塔SkillSpector 的并发能力由三个互不知情的并行层级叠加而成Layer 3 — batch_scan.py: ThreadPoolExecutor(max_workersN) across skills [CONTRIB] Layer 2 — llm_analyzer_base: asyncio.Semaphore(10) per-analyzer [UPSTREAM] Layer 1 — graph.py: 20 analyzers fan-out per-skill [UPSTREAM]每一层都对其他层无知Graph 不知道自己在被多个 worker 线程并发调用Worker 线程不知道 graph 内部会扇出 20 个 analyzerLLMAnalyzerBase不知道是哪个 worker 在调用自己。这种解耦正是纯函数模型的直接收益——各层只对自己的输入输出负责从而可以在不协调的情况下独立演进。Layer 1图内扇出上游LangGraph 语义当一个节点有多条出边时目标节点并行执行。从 graph.py 可以看到所有 analyzer 节点都从build_context扇出、再汇聚到meta_analyzer形成经典的 fan-out / fan-in 结构。这批 analyzer 按类型分为两组15 个静态分析器CPU 密集、毫秒级模式匹配、AST、YARA、供应链检查等5 个 LLM 分析器网络密集、秒级SSD语义安全发现、SDI语义开发者意图、SQP语义质量策略、TP4工具投毒与 meta 分析器。静态分析器与 LLM 分析器的数量划分来自 src/skillspector/nodes/analyzers/ 目录20 个static_patterns_*.py、behavioral_*.py、semantic_*.py等模块的自动发现注册。需要说明的是analyzer 集合是动态装配的graph.py 会检查每个模块的is_available()与requires_api_key缺失 API Key 的 LLM 分析器会被跳过而不报错。Layer 2单分析器批处理上游# llm_analyzer_base.py:387 sem asyncio.Semaphore(max_concurrency10) async def _process(batch): async with sem: response await self._structured_llm.ainvoke(prompt) return self.parse_response(response, batch) return list(await asyncio.gather(*[_process(b) for b in batches]))源码中的对应实现是 llm_analyzer_base.py 的arun_batches_detailed当未显式传入max_concurrency时使用进程级共享限流器_GlobalLLMLimiter默认并发 10常量见 llm_analyzer_base.py每个 batch 通过async with sem获取许可后并发执行_ainvoke_batch_with_retries。失败按 batch 隔离单个失败只丢弃自己的 batch不影响其他结果。Token 预算感知的分块超过模型上下文窗口的文件按行拆分相邻 chunk 之间保留50 行重叠常量CHUNK_OVERLAP_LINES 50见 llm_analyzer_base.py防止分块边界处的发现被遗漏。分块实现见chunk_file_by_linesllm_analyzer_base.py行号从 1 开始计数number_lines会为 chunk 内容加上真实文件行号前缀L1:、L2:等让 LLM 在输出中能引用准确的源码行号llm_analyzer_base.py。Layer 3跨 skill 并行contrib# batch_scan.py with ThreadPoolExecutor(max_workersargs.workers) as executor: futures {executor.submit(_scan_skill, dir, root, ...): idx for idx, dir in enumerate(skill_dirs)} for future in as_completed(futures): entry, error, name future.result(timeout90)对应实现见 batch_scan.py--workers控制并发数默认 4每个 skill 在一个专用线程中执行完整graph.invoke()流水线future.result(timeout90)提供每 skill 90 秒超时。超时或崩溃的 skill 被跳过、记录错误不重试也不阻塞其他 worker——这正是崩溃恢复的落地方式。语言检测在提交任务前预先完成_resolve_languagebatch_scan.py避免 worker 线程在文件 I/O 上互相竞争。4. 并发与限流上游只有Semaphore(10)上游唯一的并发控制是每个 analyzer 内部的Semaphore(10)更准确地说是上面提到的进程级共享限流器。没有重试、没有退避、没有 429 处理——LangChain 的ChatOpenAI为网络错误提供默认 2 次重试。对于单次扫描这足够但对批量扫描来说限制在错误的位置。批量放大的问题当 4 个 skill 通过ThreadPoolExecutor并行时每个 skill 内部都会创建独立的Semaphore(10)实例且每个 skill 内部还有 5 个 LLM 分析器并发。理论峰值并发请求数可达4 × 40 160路并发打向同一个 API 端点——对于免费层或 RPM 受限的账号这几乎必然触发 429 限流而被 429 的 batch 会直接从结果中丢弃。contrib 解法通过--workers水平节流与其给上游加一个全局信号量那需要改动上游代码contrib 层选择控制同时运行的 skill 数量ThreadPoolExecutor(max_workersN) ├─ skill_1 → graph.invoke() (upstream untouched) ├─ skill_2 → graph.invoke() (upstream untouched) └─ ...--workers与 API 档位对应TierWorkersPeak concurrent requestsFree tier110-15Paid basic4 (default)25-40Enterprise850-80命令行参数定义见 batch_scan.py帮助文本明确写着 Reduce to 1 for free-tier API keys, increase for enterprise tiers。另外上游还提供一个进程级兜底开关SKILLSPECTOR_MAX_LLM_CONCURRENCY见 llm_analyzer_base.py 的resolve_max_concurrency免费层用户可以把它设为1让所有 analyzer 的 LLM 请求全局串行化避免瞬时突发被限流。这也是上游自身提供限流旋钮的一个证据。所有 LLM 调用统一走 ApiKeyPool文档强调所有 LLM 调用——既包括图内分析器SSD/SDI/SQP/meta每个 skill 约 20 次调用也包括 gap-fill 补充扫描——都通过set_api_pool()路由到一个共享的、K8s 调度器风格的 Key 池。池通过替换全局的get_chat_model工厂生效因此每个ChatOpenAI实例都从同一个 key 环中取号。真实实现位于 api_pool.py获取acquire选择负载最低的空闲 key。ApiKeyPool.acquire的调度优先级为优先恢复退避期已过的 key → 在可用 key 中选active_requests最少的 → 全部满载时阻塞等待api_pool.py。每个 key 默认 5 个并发槽位_DEFAULT_MAX_CONCURRENT_PER_KEY 5。限流恢复指数退避30s × 2^n上限 300s常量_BACKOFF_BASE_S 30.0、_BACKOFF_CAP_S 300.0见 api_pool.py。自动故障转移429 → 标记该 key 限流 → 下次 acquire 选择不同 key。重试PooledChatModel包装 LangChainBaseChatModel遇到限流错误时透明切换 key 重试最多 5 次_MAX_RATE_LIMIT_RETRIES 5见 api_pool.py 与 api_pool.py 的_invoke_with_retry。池的注入点是 runner.py 的set_api_pool它把skillspector.llm_utils.get_chat_model与skillspector.llm_analyzer_base.get_chat_model同时替换为返回PooledChatModel的工厂从而保证包括图内分析器在内的每一次 LLM 调用都经过池set_api_pool(None)则恢复原始工厂。多 key 配置方式SKILLSPECTOR_API_KEYS环境变量新行或分号分隔的key|base_url|model条目export SKILLSPECTOR_API_KEYS sk-or-xxx1|https://api.openai.com/v1|gpt-5.4 sk-or-xxx2|https://api.openai.com/v1|gpt-5.4 单 key 模式保持向后兼容OPENAI_API_KEY此时create_api_key_pool_from_env返回None回退到普通 Provider 路径api_pool.py。批量扫描结束时池的snapshot()会输出请求总数、峰值并发槽位、限流次数与重试成功数作为报告的元数据batch_scan.py。相关行为有专门测试覆盖test_api_pool.py。5. 线程安全7 个兼容补丁调用setup_deepseek_compat()会应用 7 个针对性的 monkey-patch。补丁通过一个跟踪嵌套深度的上下文管理器显式应用——只有最外层退出时才恢复原始实现嵌套的with deepseek_compat()块在内层退出时不会提前还原。为什么需要补丁DeepSeek 的 API 不支持response_format结构化输出。上游LLMAnalyzerBase在response_schema is not None时会无条件调用with_structured_output(response_schema)见 llm_analyzer_base.py。把response_format发给 DeepSeek 会返回 HTTP 400并且会破坏 httpx 连接池——这是第 6 节 Bug 历史的起点。补丁设计原则所有补丁遵循同一模式在原始构造函数运行之前通过__init__包装器注入。由于每个实例在self.__dict__中拥有自己的值这种注入天然是线程隔离的。#TargetWhatWhy1LLMAnalyzerBase.__init__self.response_schema None实例属性禁用结构化输出实例隔离、无竞态2LLMAnalyzerBase.parse_response手动 JSON 解析 Pydantic 校验处理裸字符串响应无response_format3LLMMetaAnalyzer.parse_response同上 清洗 null→、none→low处理 LLM 输出怪癖4LLMAnalyzerBase.build_prompt追加 JSON 输出指令无response_format时模型需要显式 JSON 格式5LLMMetaAnalyzer.build_prompt为 meta 分析器追加同样指令同上6ChatOpenAI.__init__httpx.Timeout(connect8s, read30s)防止挂起的连接无限期阻塞 worker7asyncio.run静默Event loop is closed异常抑制无害的 httpx 清理噪音补丁 1实例属性而非类属性这是解决竞态问题的关键洞察。最初的方案是直接改写LLMAnalyzerBase.response_schema所有线程共享的类属性。修正后的方案是在每个实例的__dict__上设置self.response_schema None——Python 的 MRO 保证实例属性先于类属性被找到因此每个 analyzer 实例独立配置。这是语言层面的保证不依赖任何库的内部实现细节天然免疫上游类层次结构的演进。实现见 runner.py。补丁 6Pydantic 别名透传ChatOpenAI.timeout是request_timeout的别名。OpenAI 客户端在__init__中被急切地缓存root_client/root_async_client。Pydantic v2 在别名值与规范名同时存在时优先采用别名值。补丁在__init__运行前覆盖kwargs[timeout]别名从而确保超时从创建起就流入每一个root_client/async_client。实现见 runner.py——出于稳妥它同时设置了timeout与request_timeout两个键不依赖别名优先级这一 Pydantic v2 内部行为。补丁的健壮性设计补丁并非盲目替换。runner.py 的_verify_patch_targets()会在应用前用inspect.signature校验所有补丁目标的方法签名与深层依赖如LLMAnalysisResult.model_validate、Batchdataclass 字段、asyncio.new_event_loop一旦上游 API 演进导致假设失效会在补丁应用时立刻抛出明确错误而不是在运行时静默降级。深度嵌套计数_patches_depth保证可重入安全runner.py。这些设计被 test_runner_patches.py 系统性验证。6. Bug 历史一次关键竞态的调试全过程时间线症状--no-llm完全正常LLM 路径却间歇性出现 400 错误或在cleanup_result中挂起。根因四个线程并发读写LLMAnalyzerBase.response_schema类属性。线程 A 恢复了原始值而线程 B 的 meta 分析器还在创建实例。为什么偏偏是 meta 分析器它在图中运行得最晚在所有分析器扇出之后。当它的实例创建时另一个线程可能已经恢复了 schema。为什么 400 会导致清理挂起DeepSeek 对response_format返回 400。httpx 连接池在部分 400 响应后没有被正确清理。shutil.rmtree在 macOS 上会因临时目录里存在持有 dangling fd 的文件而阻塞。修复补丁 1实例属性 补丁 6httpx 超时cleanup_result的子进程回退。从源码印证修复实例属性_patched_base_initrunner.py在调用原始__init__前写入self.response_schema None。httpx 超时_patched_chatopenai_initrunner.py强制httpx.Timeout(30.0, connect8.0)从根上杜绝连接挂起。清理容错cleanup_result使用shutil.rmtree(temp_dir, ignore_errorsTrue)runner.py临时目录清理失败不会导致批量任务整体失败run_one的finally块保证无论成功失败都会尝试清理runner.py。这条 Bug 历史的价值在于它解释了为什么补丁 1 一定要用实例属性而非类属性、为什么超时配置必须从连接建立那一刻生效、以及为什么清理逻辑必须可容错——每一个设计决策背后都是一次线上故障。7. Provider 系统三层抽象Protocol (base.py) Implementation (per-provider) ───────────────── ──────────────────────────── ModelMetadataProvider openai / anthropic / nv_build ├─ get_context_length() ├─ provider.py ├─ get_max_output_tokens() └─ model_registry.yaml └─ resolve_model(slot) CredentialsProvider └─ resolve_credentials() ChatModelProvider └─ create_chat_model()三个 Protocol 定义在 providers/base.py并通过组合 ProtocolLLMProviderproviders/base.py统一暴露。它们是结构子类型structural subtyping而非 ABC 继承——任何满足方法签名的对象都可以充当 Provider无需显式继承。此外还有一个可选的AgentCLICapable协议providers/base.py供claude_cli、codex_cli、gemini_cli这类本地 CLI 提供方实现绕过 HTTP 传输。每个 provider 都是独立子包包含provider.py与model_registry.yaml模型上下文长度、最大输出 token 等元数据例如 openai/、anthropic/、bedrock/、ollama/ 等。选择链SKILLSPECTOR_PROVIDER env var ├─ openai → OpenAIProvider → OPENAI_API_KEY ├─ anthropic → AnthropicProvider → ANTHROPIC_API_KEY ├─ nv_build → NvBuildProvider → NVIDIA key └─ unset → NvInferenceProvider (→ NvBuildProvider fallback)真实的选择逻辑在 providers/init.py 的_select_active_provider按SKILLSPECTOR_PROVIDER的值分发到对应 Provider 类环境变量未设置或为nv_inference时优先尝试可选的nv_inference子包ImportError时回退到NvBuildProvider。当前仓库支持的 Provider 比文档列出的更多包括anthropic_proxy、bedrock、azure_openai、openai_compatibleGroq、Together AI、Mistral 等以及三种本地 CLI Provider。凭证解析带有回退链resolve_chat_model_credentialsproviders/init.py在活动 Provider 未配置时回退到OPENAI_API_KEY保持历史兼容。结合第 4 节可知ApiKeyPool 正是通过替换这一模型构造链路的入口get_chat_model来接管全部 LLM 调用的。8. Contrib 集成长出来的而不是塞进去的src/skillspector/ 零修改contrib 层完全位于上游之外。它以上游类为父类进行继承、对上游函数做包装而不是修改上游源码contrib/batch_scan/ ├── batch_scan.py ← CLI ThreadPoolExecutor ├── runner.py ← graph.invoke() wrapper 7 safety patches ├── gap_fill.py ← GapFillAnalyzer(LLMAnalyzerBase) ├── api_pool.py ← ApiKeyPool PooledChatModel ├── detection.py ← Unicode script-ratio language detection ├── annotation.py ← finding language-compatibility labeling ├── discovery.py ← recursive SKILL.md finder └── reports.py ← Terminal / JSON / Markdown formatters各模块职责与文档一致。其中detection.py的语言检测实现值得展开它零外部依赖只用标准库unicodedatadetection.py按 CJK 统一表意文字、假名平/片假名、谚文字母的码点区间占比判定zh/ja/ko/en阈值分别为 0.10 / 0.05 / 0.10detect_skill_language再对全 skill 文件做多数投票detection.py。设计原则继承而非重写。GapFillAnalyzer继承LLMAnalyzerBase——自动获得 token 预算、批处理、并发能力gap_fill.py。包装而非钻孔。API 池包装ChatOpenAI而非修改其构造方式。打标签而非重构。增加language_compatible、scan_mode、enhancements等字段但不改变Finding的结构见 runner.py 的entry_from_resultscan_mode固定标记为multilingual-enhancedenhancements记录gap_fill_applied、gap_fill_findings、跳过的英文关键词规则数。可比对而非隐藏。单 skill 扫描skillspector scan与批量扫描batch_scan产出可 diff 的输出scan_mode标签跟踪结果来源。此外batch_scan.py 展示了非英文 skill 的gap-fill 后处理当检测到非英文且启用 LLM 时图扫描完成后对 8 类没有语义分析器对应物的漏洞类别做一次定向 LLM 补充扫描run_gap_fill并把新发现追加进 issues 列表同时把enhancements.gap_fill_applied置为真。该行为由 test_gap_fill.py 覆盖。何时上游化如果批量扫描、多语言支持与 API 池被证明有广泛价值文档给出的演进路径是ApiKeyPool→src/skillspector/providers/pool.py语言检测 →build_context节点GapFill→ 注册为第 21 个 analyzer 节点批量扫描 → 合并进 CLI 的scan命令在此之前的原则是先用价值证明自己再谈合并。附录 A关键文件索引FileRolesrc/skillspector/graph.pyGraph 拓扑7 节点、20 分析器扇出src/skillspector/state.pyState schemaTypedDict与资源预算src/skillspector/llm_analyzer_base.pyLLM 分析器基类token 预算 批处理 并发src/skillspector/providers/init.pyProvider 工厂与凭证回退链src/skillspector/providers/base.pyProvider Protocol 定义src/skillspector/llm_utils.pyLLM 工具get_chat_model、chat_completionsrc/skillspector/cli.pyCLI 入口scan命令src/skillspector/nodes/analyzers/20 个分析器实现src/skillspector/nodes/meta_analyzer.pyMeta 分析器LLM 验证contrib/batch_scan/batch_scan.py批量扫描 CLI ThreadPoolExecutorcontrib/batch_scan/runner.pygraph.invoke() 包装 7 个安全补丁contrib/batch_scan/api_pool.pyApiKeyPool PooledChatModelcontrib/batch_scan/detection.pyUnicode 文字比例语言检测contrib/batch_scan/gap_fill.py非英文 skill 的 gap-fill 补充扫描contrib/batch_scan/tests/tests-pro/test_api_pool.pyAPI 池测试contrib/batch_scan/tests/tests-pro/test_runner_patches.py兼容补丁测试附录 B术语表TermMeaningSkillAI Agent 技能包目录或 zipFinding一条安全发现rule_id severity line ...Batch一次 LLM 调用单元一个文件或一个 chunkState一次graph.invoke()的完整输入/输出ProviderLLM 后端抽象OpenAI / Anthropic / NVIDIA / CLI 等Meta-analyzerLLM 验证/过滤节点Fan-out一个节点 → 多个并行节点Fan-in多个节点 → 一个聚合节点Chunk超限文件按行拆分带重叠Semaphoreasyncio 并发门闩API Pool多 Key 资源调度器Gap-fill对非英文 skill 的定向 LLM 补充扫描8 类漏洞类别PooledChatModel透明 key 切换的 LangChain 兼容模型包装结语SkillSpector 的架构证明了一个朴素而强大的命题——只要把核心扫描建模为无状态纯函数并发、限流、兼容性这些复杂问题都可以在不触碰核心的前提下通过外围包装包装函数、替换工厂、注入实例属性逐一解决。三层并行金字塔让每个 skill 内部的图扇出、单分析器的异步批处理与跨 skill 的线程池各司其职7 个兼容补丁以语言级保证的实例隔离化解了竞态ApiKeyPool 以调度器思路把限流从单请求重试升级为多 key 水平扩展。这套长出来而非塞进去的 contrib 哲学为 SkillSpector 后续的批量扫描、多语言支持与多 key 调度能力铺平了道路。【免费下载链接】SkillSpectorSecurity scanner for AI agent skills. Detect vulnerabilities, malicious patterns, security risks, prompt injection, data exfiltration, and supply-chain risks in Claude Code, Codex, and MCP skills before you install them.项目地址: https://gitcode.com/GitHub_Trending/sk/SkillSpector创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考