跳到主要内容

输入/输出护栏的执行与内置护栏库

30 秒导读： 本章讲拦截型护栏怎么跑——用户消息进模型前的「输入护栏」、机器人回复出去前的 「输出护栏」，它们本质上是一串检查动作（action）。Colang flow 决定这些动作是顺序还是并行 执行，动作分发器（ActionDispatcher）负责真正调用，每个动作靠一个 output_mapping 把返回值翻译成 「放行 / 拦截」。命中拦截时要么回一句预设拒答、要么抛出 rails 异常并 abort。最后还有一套内置护栏库 （self-check、content-safety、jailbreak、敏感数据、正则、注入检测）和流式输出的缓冲策略。

本章是 nemo-guardrails 系列的一章，聚焦「拦截」这条支线。相关章节：

本章不重复对话生成主线，只讲拦截型 rail 的执行机制与 library/ 生态。

1. 这是什么（先建立直觉）​

拦截型护栏 = 在数据进出模型的两个关口上，挂一串「检查函数」。

想象一个安检门：

  • 输入护栏：用户说的话在被送去生成回复之前先过一遍安检。发现越狱提示、辱骂、SQL 注入 就当场拦下，模型根本不会看到这句话。
  • 输出护栏：模型生成的回复在发给用户之前再过一遍安检。发现事实幻觉、泄露敏感信息、 违反内容安全策略就拦下，用户看到的是一句「抱歉，我无法回应」。

每一「道」安检就是一个 rail flow（Colang 里的一段流程），它内部调用一个或多个 action （真正干活的 Python 函数）。一个典型配置长这样：

# config.yml —— 示意
rails:
  input:
    flows:
      - self check input        # 输入关口挂两道安检
      - detect sensitive data on input
  output:
    flows:
      - self check output       # 输出关口挂一道安检

一句话直觉： rail 就是「关口上的一串 if 检查」，任何一个检查说「不行」，这一轮就被短路—— 输入被拦就不生成，输出被拦就替换成拒答。本章拆解的就是「这串检查具体怎么被调度、怎么被调用、 怎么判定放行还是拦截」。

2. 顶层全景：一次输入护栏怎么转​

先看大盘。用户消息进来后，Colang 运行时会触发 process user input 这个 flow，它内部 do run input rails（nemoguardrails/rails/llm/llm_flows.co:16），把配置里 rails.input.flows 列的每一道护栏跑一遍。

用户消息
   │
   ▼
┌─────────────────────────────────────────────┐
│  subflow: run input rails                    │   llm_flows.co:46
│                                              │
│   config.rails.input.parallel ?              │
│      │否(默认)              │是              │
│      ▼                      ▼                │
│  顺序 for 循环          run_input_rails_      │
│  逐个 do <flow>         in_parallel(flows)   │
│      │                      │                │
│      │                      ▼                │
│      │              asyncio 并发跑所有 flow  │
│      │              谁先喊 stop 就取消其余   │
└──────┼──────────────────────┼────────────────┘
       │                      │
       ▼                      ▼
   每个 flow 内部 do <action>（如 self check input）
       │
       ▼
   ActionDispatcher.execute_action  ← 真正调用 Python 函数
       │
       ▼
   is_output_blocked(result, action) ← 用 output_mapping 判定
       │
   放行 ┤ 拦截
       │     └→ bot refuse to respond / 抛 InputRailException + abort
       ▼
   继续生成回复

怎么读这张图： 从上到下是数据流。关键的三个决策点是——(1) 顺序还是并行跑护栏； (2) 每道护栏调用哪个 action；(3) action 返回值怎么被翻译成「放行/拦截」。输出护栏结构完全对称， 只是关口在「回复发出去之前」。

各部件职责一览：

部件干什么在哪
run input rails / run output rails 子流程决定顺序 or 并行，逐个触发护栏 flowrails/llm/llm_flows.co:46 / :198
_run_flows_in_parallel用 asyncio 并发跑多个 flow，首个 stop 即取消其余colang/v1_0/runtime/runtime.py:263
ActionDispatcher.execute_action按名字找到并调用 action 函数（支持 sync/async/类/LangChain）actions/action_dispatcher.py:180
@action 装饰器 / ActionResult给函数打上元数据、承载返回值+事件+上下文更新actions/actions.py:41 / :85
is_output_blocked用 action 的 output_mapping 把返回值判成「拦/放」actions/output_mapping.py:34
library/ 各目录内置护栏的 action + flow 定义nemoguardrails/library/*

3. 核心机制一：顺序 vs 并行执行​

3.1 决策点在 Colang flow 里​

顺序还是并行，不是硬编码在 Python 里，而是写在 Colang 的子流程里，由配置 rails.input.parallel （默认 False）决定。看输入护栏的子流程：

define subflow run input rails
  """Runs all the input rails in a sequential order. """
  $input_flows = $config.rails.input.flows

  if $config.rails.input.parallel
    execute run_input_rails_in_parallel(flows=$input_flows)
  else
    $i = 0
    while $i < len($input_flows)
      $triggered_input_rail = $input_flows[$i]
      create event StartInputRail(flow_id=$triggered_input_rail)
      event StartInputRail
      do $input_flows[$i]           # ← 逐个跑，前一个没拦才轮到后一个
      $i = $i + 1
      create event InputRailFinished(flow_id=$triggered_input_rail)
      event InputRailFinished

真源码 nemoguardrails/rails/llm/llm_flows.co:46-68。输出护栏的子流程 run output rails （:198-220）结构完全对称，只是用 StartOutputRail / OutputRailFinished 事件和 rails.output.parallel 开关。

要点：

  • 顺序模式是默认，一个 while 循环逐个 do <flow>。前一道护栏若 abort 了，后面的根本不会跑 ——所以顺序模式里护栏顺序有意义，把「便宜的检查」排前面能省钱。
  • 每道护栏前后包一对 StartInputRail / InputRailFinished 标记事件，供日志和可观测性追踪 「现在跑到哪道护栏了」（见 06 章）。
  • parallel=True 时改为把整批 flow 名交给一个内置 action run_input_rails_in_parallel。

3.2 并行执行：起独立事件循环 + 首个 stop 即熔断​

run_input_rails_in_parallel / run_output_rails_in_parallel 这两个 action 只是薄封装， 分别造好一批 StartInputRail / StartOutputRail（前置）和 *Finished（后置）标记事件， 然后都委托给同一个 _run_flows_in_parallel：

真源码 nemoguardrails/colang/v1_0/runtime/runtime.py:440-460。核心逻辑在 _run_flows_in_parallel（:263）：

  • 每道护栏起一个 asyncio task。 给每个 flow 复制一份 events、追加一个 start_flow 事件， 然后 asyncio.create_task(...) 在同一个事件循环内并发触发 self.generate_events （runtime.py:307-341）。
  • 用 as_completed 收结果，谁先喊停谁赢。 用 asyncio.as_completed(tasks) 按完成先后 取结果（:348）。一旦某个 flow 的结果里出现 BotIntent(intent="stop") 或任何 *Exception 事件（has_stop，:353），立刻把其余还没完成的 task 全部 cancel()（:360-377）并 break。
  • 收尾清理。 finally 块再兜一次底，把所有未完成 task 取消，避免「Task was destroyed but it is pending」告警（:388-392）。
        flow A ─┐
        flow B ─┼─► asyncio.create_task 并发跑
        flow C ─┘
                     │
       as_completed 按完成顺序取：
          B 先完成 → 检查有没有 stop？
                       │有                │无
                       ▼                  ▼
              cancel(A) cancel(C)     存 B 的结果，继续等 A、C
              带着 B 的 stop 返回

为什么这么设计： 并行模式下多道护栏（尤其是各自要调 LLM 的那些）同时跑，总延迟 ≈ 最慢的一道， 而不是所有护栏耗时之和。代价是——只要有一道拦截，其余护栏的 LLM 调用就白花了（被 cancel）。 所以并行适合「护栏都较贵、且大多数请求会放行」的场景。

3.3 task_call_helper 里的 stop 判定细节​

并行时每个 task 包了一层 task_call_helper（runtime.py:292-303）：它跑完 flow 后先看结果里有没有 stop / *Exception，只有没 stop 时才追加 InputRailFinished 这类后置标记事件——因为一道被拦下的 护栏不该发出「正常完成」的信号。这个 has_stop 判定和外层 as_completed 里的判定用的是同一套条件 （event["type"].endswith("Exception") 或 BotIntent + intent=="stop"），保证顺序/并行两条路 对「什么叫拦截」理解一致。

4. 核心机制二：动作如何被解析、调用​

护栏 flow 里的 do self check input / execute run_input_rails_in_parallel 最终都落到 动作分发器 ActionDispatcher 上。这一节讲一个 action 从「名字」到「被调用」的完整链路。

4.1 @action 装饰器：给函数贴元数据​

任何要被 Colang 调用的 Python 函数，都要用 @action 装饰。它做的事很简单——把一个 action_meta 字典塞到函数对象上：

真源码 nemoguardrails/actions/actions.py:41-82。action_meta（一个 TypedDict，:30）含四个字段：

字段含义
nameaction 名（默认取函数名），Colang 按这个名字找它
is_system_action是否是内置系统 action（系统 action 永远本地跑，不走 action server）
execute_async是否异步执行
output_mapping一个函数，把返回值翻译成「是否拦截」——这是拦截判定的核心，见 §4.3

装饰器把这份元数据 setattr 到函数上（:79），分发器加载时就靠 hasattr(obj, "action_meta") 把它认出来（action_dispatcher.py:295）。

4.2 ActionDispatcher.execute_action：一个调用器吃四种形态​

分发器初始化时会递归扫盘加载 action：先扫包内 actions/，再遍历 library/ 下每个带 actions.py 或 actions/ 子目录的护栏目录，再扫当前工作目录和 config 路径 （action_dispatcher.py:53-88）。这就是为什么 library/ 里每个护栏放一个 actions.py 就能被自动发现。

真正调用在 execute_action（:180-250）。它先把名字归一化（_normalize_action_name：去掉 Action 后缀、驼峰转下划线，:155），再按 action 的形态分派：

execute_action(name, params)
   │ 归一化名字，查 registered_actions
   ▼
 是 class ？ ──是──► 惰性实例化 fn()，缓存回去   (:204)
   │否
   ▼
 是 function/method ？ ──是──► fn(**params)，是协程就 await   (:211)
   │否
   ▼
 有 ainvoke ？ ──是──► await fn.ainvoke(input=params)  (LangChain Runnable, :219)
   │否
   ▼
 有 run 方法 ？ ──是──► fn.run(**params)   (:225)
   │
   ▼
 返回 (result, status)   status ∈ {"success","failed"}

几个巧妙点：

  • 鸭子类型认 LangChain。 不 import langchain，只检查对象有没有 ainvoke（:219-222）， 避免核心代码依赖 LangChain。
  • 系统 action 永远本地跑。 若配了 actions_server_url，非系统 action 会被发到远程 action server 执行（runtime.py:622、_get_action_resp），但 is_system_action=True 的护栏动作始终本地执行。
  • 异常分级。 LLMCallException 被原样 re-raise（:237）——让上层能区分「护栏说不安全」和 「护栏调 LLM 时崩了」；其它异常被吞掉并记 log，返回 (None, "failed")（:240-250）。

4.3 output_mapping + is_output_blocked：返回值 → 拦/放​

这是整个拦截机制的语义核心。护栏 action 各自返回不同形状（bool、float、dict、TypedDict）， 系统怎么统一判断「这算拦截吗」？答案：每个 action 自带一个 output_mapping 函数，is_output_blocked 去调它。

真源码 nemoguardrails/actions/output_mapping.py:34-51：

  • 取出 action 的 action_meta["output_mapping"]；没有就用 default_output_mapping（:19）。
  • 把结果规整成元组、取第一个元素，交给 mapping，mapping 返回 True 表示「应拦截」。

default_output_mapping（:19-31）的兜底约定值得记住：

返回值类型判定
boolnot result —— 即 True=安全/放行，False=拦截
数字result < 0.5 —— 低于 0.5 视为拦截
其它一律放行（False）

关键约定：布尔护栏里 True 是「安全」而不是「命中」。 这解释了为什么 self_check_output 显式写 output_mapping=lambda value: not value（content_safety/self_check 里都能看到这个反转）， 以及为什么 self_check_input 返回 is_safe（安全为 True）。搞反了就会把安全内容全拦下。

顺序模式下这个判定藏在 Colang flow 的 if not $allowed 里；并行和流式模式下则由 Python 直接调 is_output_blocked（runtime.py:495、llmrails.py:2022）——所以内置护栏必须把 output_mapping 配对，否则并行/流式路径判定会出错。

4.4 ActionResult：不止返回值，还能改事件流和上下文​

护栏 action 不一定只返回一个布尔。它可以返回一个 ActionResult（actions/actions.py:85-102）：

字段用途
return_value真正的返回值（后续被塞进 context 的 action_result_key）
events要追加进事件流的事件（例如 self_check_input 返回一个 mask_prev_user_message 事件去打码用户消息）
context_updates对上下文变量的更新

_process_start_action（runtime.py:659-691）负责把 ActionResult 拆开，把 events 追加到下一步、 把 context_updates 变成 ContextUpdate 事件。这让护栏不仅能「拦」，还能「改」——比如给敏感数据打码、 掩盖上一条用户消息。

5. 深挖代表性内置护栏​

library/ 下有二十多个护栏目录（很多是第三方集成：autoalign、patronusai、cleanlab…）。 这里挑几个自带、无需外部服务或用法有代表性的，看它们各自的检测思路。

5.1 self-check：最经典的「LLM 自检」模式​

self_check_input / self_check_output 是最基础的护栏——用一次额外的 LLM 调用，让模型给自己的 输入/输出打「安全/不安全」标签。这套「渲染 prompt → 调 LLM → 解析输出」是整个库里反复出现的模板。

看 self_check_input（library/self_check/input_check/actions.py:33-97）的骨架：

# 示意，贴近源码结构
@action(is_system_action=True)
async def self_check_input(llm_task_manager, context, llm, config, **kwargs):
    user_input = context.get("user_message")
    task = Task.SELF_CHECK_INPUT
    # 1. 用 check_input 任务模板 + 用户输入渲染出 prompt
    prompt = llm_task_manager.render_task_prompt(task, context={"user_input": user_input})
    # 2. 调 LLM（低温度，把随机性压到最低）
    llm_response = await llm_call(llm, prompt, stop=..., llm_params={"temperature": config.lowest_temperature, ...})
    # 3. 解析输出为 is_safe（默认解析器 is_content_safe）
    result = llm_task_manager.parse_task_output(task, output=llm_response.content, forced_output_parser="is_content_safe")
    is_safe = result[0]
    if not is_safe:
        # 不安全：返回 False，并发一个事件把用户消息打码
        return ActionResult(return_value=False, events=[new_event_dict("mask_prev_user_message", ...)])
    return is_safe

重点看三步： render_task_prompt（把安全策略问题拼成 prompt）→ llm_call（真去问 LLM）→ parse_task_output（把 "yes"/"no" 之类回答解析成布尔）。llm_call / render_task_prompt 的机制详见 05 章。

self_check_output（library/self_check/output_check/actions.py:32）几乎一样，差别有二：

  • 装饰器带 output_mapping=lambda value: not value——因为它返回 is_safe，要反转成「拦截」。
  • prompt 上下文多了 bot_response 和 bot_thinking（把模型的回复和推理内容一起送去自检）。

对应的 Colang flow（library/self_check/input_check/flows.co）把「不安全时怎么办」写死了：

flow self check input
  $allowed = await SelfCheckInputAction
  if not $allowed
    if $system.config.enable_rails_exceptions
      send InputRailException(message="Input not allowed...")
    else
      bot refuse to respond
    abort

5.2 content_safety：多模型 + 缓存 + 多语言拒答​

content_safety_check_input / _output（library/content_safety/actions.py:42 / :143）是 self-check 的加强版，专为 Llama Guard、NemoGuard 这类专用内容安全分类模型设计：

  • 按名字选模型。 接收 llms: Dict[str, LLMModel] 和 model_name，从字典里取指定模型 （:65）——一套护栏能挂多个不同的安全模型。
  • 结果带策略违规明细。 解析成 is_safe, *violated_policies，返回 {"allowed": is_safe, "policy_violations": [...]}（:109-111）。对应的 content_safety_check_output_mapping（:126）读 allowed 字段判拦截。
  • 可缓存。 若配了 model_caches，用归一化 prompt 当 key 命中缓存直接返回，省一次 LLM 调用 （:91-98、:113-121）。
  • 多语言拒答。 配套的 detect_language（:272）用 fast-langdetect 检测用户语言，从 DEFAULT_REFUSAL_MESSAGES（覆盖 en/es/zh/de/fr/hi/ja/ar/th，:232）挑一句对应语言的拒答。

flow 里的分支逻辑（library/content_safety/flows.co）体现了这些开关：不安全时先看 enable_rails_exceptions 决定抛异常还是回消息，回消息时再看 content_safety.multilingual.enabled 决定用多语言拒答还是通用 bot refuse to respond，最后 abort。

5.3 其它内置护栏的检测思路对比​

剩下几个自带护栏各用不同技术，一表看清：

护栏检测思路关键 action / 配置
jailbreak_detection两条路：①启发式（困惑度 perplexity + 前缀/后缀长度阈值）；②模型分类器（本地 embedding 分类，或走 NIM 微服务）。可本地算，也可打远程 APIlibrary/jailbreak_detection/actions.py:56（heuristics）/ :92（model）；配置 JailbreakDetectionConfig（config.py:888）
sensitive_data_detection基于微软 Presidio 分析器识别 PII 实体（人名、卡号…），detect_* 判有无、mask_* 用匿名化引擎打码library/sensitive_data_detection/actions.py:94 / :137；配置 SensitiveDataDetection（config.py:229），分 input/output/retrieval 三源、可设 score_threshold
regex纯正则匹配。配置加载时就预编译 patterns（RegexDetectionOptions.compile_patterns），运行时逐个 compiled.search(text)，命中即拦library/regex/actions.py:37（detect_regex_pattern）；配置 RegexDetection（config.py:284），支持 case_insensitive
injection_detection用 YARA 规则扫代码/模板/SQL/XSS 注入（injections: [sqli, template, code, xss]）。action 决定命中后是 reject（拒答）还是 omit（抠掉命中片段）；sanitize 尚未实现library/injection_detection/actions.py:299；配置 InjectionDetection（config.py:183），可自带 yara_rules 或指向 yara_path

这些护栏的共同套路：一个 @action 函数做检测、返回结构化结果（TypedDict 或 dict），配一个 output_mapping 把结果的某个字段（is_match / allowed / is_injection）翻译成拦截布尔。例如：

  • detect_regex_pattern 带 output_mapping=_regex_blocked_mapping，读 is_match（regex/actions.py:31-33）。
  • detect_sensitive_data 带 output_mapping=detect_sensitive_data_mapping（sensitive_data_detection/actions.py:83）。

边界提醒： 很多护栏需要可选依赖（presidio-analyzer、yara-python、fast-langdetect）， 库用 try/except ImportError 惰性导入，缺了就在首次调用时报错或告警——它们不在默认安装路径里。

6. 核心机制三：流式输出护栏的缓冲策略​

普通输出护栏在完整回复生成完后才检查。但如果开了流式（token 边生成边推给用户），护栏没法等 整段——它得在 token 流上「边流边查」。难点是：单个 token 太碎，检查没意义（"暴" "力" 分开看都不违规）。解法是缓冲：攒够一批 token 再送去检查。

6.1 RollingBuffer：滚动窗口攒 token​

BufferStrategy（rails/llm/buffer.py:56）是抽象基类，唯一实现是 RollingBuffer（:168）。 思路很直接：

  • 攒 token 进 buffer，攒够 buffer_chunk_size（默认 10）个就吐出一批去检查。
  • 吐出去检查的那批多带 buffer_context_size（默认 5）个上文 token（processing_context）， 让护栏看到跨批的连续语义，避免违规词正好卡在批边界被漏掉。
  • 但真正推给用户的只是新 token（user_output_chunks），不重复推上文。

真源码 process_stream（buffer.py:256-325）。它 yield 的是 ChunkBatch(processing_context, user_output_chunks) 这个具名元组（:24）——一个给护栏看（带上文），一个给用户看（只新内容）。

token 流: A B C D E F G H I J | K L ...
               chunk_size=10 攒满
   ┌──────────────────────────────┐
   │ processing_context (给护栏)   │  = 前 context_size 个 + 这批10个
   │  ...(上文) A B C D E F G H I J │
   └──────────────────────────────┘
   ┌──────────────────────────────┐
   │ user_output_chunks (给用户)   │  = 只有这批新 token
   │  A B C D E F G H I J           │
   └──────────────────────────────┘
   之后 buffer 只保留末尾 context_size 个 token，滚动继续

get_buffer_strategy（:349）是个工厂，目前固定返回 RollingBuffer（源码注释里留了将来做成注册表的 TODO）。参数来自 OutputRailsStreamingConfig 的 chunk_size / context_size。

6.2 逐批检查：顺序 or 并行两条路​

流式护栏主循环在 _run_output_rails_in_streaming（rails/llm/llmrails.py:1795）。它对 buffer_strategy(streaming_handler) 吐出的每个 ChunkBatch 做检查，同样分并行/顺序两条路 （llmrails.py:1905 的 parallel_mode 开关）：

  • 顺序（:2003-2043）：对每个输出护栏 flow，取出 action、execute_action，再 is_output_blocked(result, action_func) 判定（:2022）。命中就不再往下 yield，而是吐出一个 JSON 错误对象（{"error": {"type": "guardrails_violation", "code": "content_blocked", ...}}） 然后 return 终止流。
  • 并行（:1928-1998）：调 run_output_rails_in_parallel_streaming action（见下）。若返回的 stop 事件带 error_type == "internal_error"，错误码用 rail_execution_failure；否则是 content_blocked。

stream_first 开关决定「先推给用户再检查」还是「先检查再推」（:1923 vs :2045）——前者延迟低但可能 让用户瞥见被拦内容的开头，后者更安全但每批要等护栏跑完。

6.3 流式并行的精简版：run_single_rail​

流式并行走的是 _run_output_rails_in_parallel_streaming（runtime.py:462），它不复用上面那套完整的 _run_flows_in_parallel（后者有完整 flow 状态管理，源码注释说会和流式的 hide_prev_turn 逻辑打架）。 取而代之是一个精简闭包 run_single_rail（:476-502）：

  • 直接 action_dispatcher.execute_action(action_name, params) 调 action，跳过 Colang flow。
  • 用 is_output_blocked(result, action_func) 判拦截（:495）。
  • action 崩了就返回 (flow_id, "internal_error", error_msg)（:499-502）。

外层同样 asyncio.as_completed + 首个命中即 cancel 其余（:512-550），返回带 BotIntent(intent="stop") 的事件（内部错误还带 error_type="internal_error" 和 error_message，:517-527）。

7. 核心机制四：拦截后怎么收场——拒答、异常、abort​

护栏判定「拦」之后，有两种收场方式，由全局开关 enable_rails_exceptions（默认 False， config.py:1818）切换。

默认：回预设消息。 flow 走 bot refuse to respond（返回「I'm sorry, I can't respond to that.」 之类）再 abort（Colang 2.x）或 stop（Colang 1.0）。用户拿到一句正常的拒答。

开启异常：发 *Exception 事件。 flow 改走 send InputRailException(...) / ContentSafetyCheckInputException(...) 再 abort。这些 *Exception 事件被专门识别：

  • 并行执行里，event["type"].endswith("Exception") 和 BotIntent stop 一样触发熔断 （runtime.py:296、:355）。
  • 生成结果处理时，扫到 *Exception 事件就把它存进 exception 变量（llmrails.py:1113-1114）， 最终产出的消息 role 变成 "exception" 而非 "assistant"（:1146-1147）——调用方由此知道 「这不是正常回复，是护栏拦下了」。
护栏判定拦截
   │
   ├─ enable_rails_exceptions = False（默认）
   │     └→ bot refuse to respond  →  abort
   │           用户收到：一句拒答字符串（role=assistant）
   │
   └─ enable_rails_exceptions = True
         └→ send XxxRailException(message=...)  →  abort
               调用方收到：role="exception" 的消息，content 是异常事件

为什么给两种： 面向终端用户的聊天场景要「优雅拒答」；面向程序集成的场景（如把 guardrails 当中间件）要「明确的异常信号」好让上层代码分支处理。abort / stop 保证——不管哪种收场，被拦的这一轮 都立即短路，不会继续生成或把内容发出去。

8. 巧妙之处（可借鉴）​

  • 拦截判定「外置」成 output_mapping。 action 只管返回自己的自然结果（bool/dict/分数）， 「这算不算拦截」由一个附在元数据上的小函数决定（output_mapping.py:34）。新护栏只要写好检测逻辑 + 一个 mapping，就能接入统一的拦截框架，无需碰调度代码。
  • 顺序/并行共用一套 stop 语义。 无论哪条执行路径，「拦截」都统一表示成 BotIntent(intent="stop") 或 *Exception 事件（runtime.py:295-297），judge 逻辑一处定义、多处复用。
  • 并行首个命中即熔断。 as_completed + cancel 让并行护栏「一票否决、立即止损」，不浪费其余 护栏的 LLM 调用（runtime.py:348-377）。
  • 鸭子类型接 LangChain 而不引入依赖。 只探 ainvoke 属性（action_dispatcher.py:219）， 核心保持零 LangChain 依赖。
  • 流式的「双份 chunk」。 ChunkBatch 把「给护栏看的（带上文）」和「给用户推的（只新内容）」 分成两份（buffer.py:24），既保证跨批检测的连续性，又不重复推送。

9. 边界与局限​

  • 顺序模式下护栏顺序影响成本与行为。 前一道 abort 后面就不跑；把贵的（调 LLM 的）排后面、 便宜的（正则）排前面能省钱。
  • 并行模式下被拦时其余护栏的 LLM 调用白花。 熔断只是 cancel task，已发出的请求未必能撤回。
  • 很多护栏依赖可选包。 Presidio、YARA、fast-langdetect、jailbreak 的分类模型/NIM 端点都是可选， 缺了会 ImportError 告警或需要外部服务；它们不在默认安装里。
  • sanitize 类动作未实现。 注入检测的 sanitize 会抛 NotImplementedError （injection_detection/actions.py:364）；SensitiveDataDetectionOptions.mask_token 也标注了 "not currently in use"（config.py:217）。
  • 流式 stream_first=True 可能让用户瞥见被拦内容的开头——先推后查是延迟与安全的取舍。
  • 本章讲的是 Colang 1.0 运行时的拦截路径。现代 IORails 引擎（Colang 2.x / 引擎注册表）的等价机制 见 06-iorails-engine-server-observability.md。

10. 代码地图（导航索引）​

主题文件路径符号名
输入护栏子流程（顺序/并行决策）nemoguardrails/rails/llm/llm_flows.cosubflow run input rails（:46）
输出护栏子流程nemoguardrails/rails/llm/llm_flows.cosubflow run output rails（:198）
输入护栏并行封装nemoguardrails/colang/v1_0/runtime/runtime.py_run_input_rails_in_parallel（:440）
输出护栏并行封装nemoguardrails/colang/v1_0/runtime/runtime.py_run_output_rails_in_parallel（:451）
并行执行核心（asyncio + 首个 stop 熔断）nemoguardrails/colang/v1_0/runtime/runtime.py_run_flows_in_parallel（:263）、task_call_helper（:292）
流式并行执行nemoguardrails/colang/v1_0/runtime/runtime.py_run_output_rails_in_parallel_streaming（:462）、run_single_rail（:476）
动作分发调用nemoguardrails/actions/action_dispatcher.pyActionDispatcher.execute_action（:180）
动作加载扫盘nemoguardrails/actions/action_dispatcher.pyload_actions_from_path（:102）、_find_actions（:316）
动作装饰器与结果类型nemoguardrails/actions/actions.pyaction（:41）、ActionResult（:85）、ActionMeta（:30）
拦截判定nemoguardrails/actions/output_mapping.pyis_output_blocked（:34）、default_output_mapping（:19）
self-check 输入/输出nemoguardrails/library/self_check/input_check/actions.py、.../output_check/actions.pyself_check_input（:33）、self_check_output（:32）
content-safetynemoguardrails/library/content_safety/actions.pycontent_safety_check_input（:42）、content_safety_check_output（:143）、detect_language（:272）
jailbreak 检测nemoguardrails/library/jailbreak_detection/actions.pyjailbreak_detection_heuristics（:56）、jailbreak_detection_model（:92）
敏感数据检测nemoguardrails/library/sensitive_data_detection/actions.pydetect_sensitive_data（:94）、mask_sensitive_data（:137）
正则检测nemoguardrails/library/regex/actions.pydetect_regex_pattern（:37）、_regex_blocked_mapping（:31）
注入检测（YARA）nemoguardrails/library/injection_detection/actions.pyinjection_detection（:299）、_reject_injection（:263）、_omit_injection（:185）
检测型护栏配置项nemoguardrails/rails/llm/config.pyInjectionDetection（:183）、SensitiveDataDetection（:229）、RegexDetection（:284）
流式缓冲策略nemoguardrails/rails/llm/buffer.pyBufferStrategy（:56）、RollingBuffer（:168）、get_buffer_strategy（:349）
流式输出护栏主循环nemoguardrails/rails/llm/llmrails.py_run_output_rails_in_streaming（:1795）
rails 异常开关nemoguardrails/rails/llm/config.pyenable_rails_exceptions（:1818）