资讯动态

Guardrails Actions 深度解析:ReAsk、Filter 与 Refrain 的源码级实现原理

发布时间:2026/9/28 3:19:07 来源:尧图企业网站定制
AI 安全治理模型安全AI 应用【免费下载链接】guardrailsAdding guardrails to large language models.项目地址https://gitcode.com/gh_mirrors/gu/guardrails点击查看免费下载本指南以 Guardrails 的 Actions 模块为核心系统讲解大型语言模型LLM输出验证失败后三类核心处置动作ReAsk重新提问、Filter过滤与Refrain克制/拒答的类设计、生命周期与底层实现。读完本文你将掌握ReAsk及其三个子类FieldReAsk、SkeletonReAsk、NonParseableReAsk的语义差异理解apply_filters与apply_refrain的递归处理逻辑并能够从源码层面追踪一次验证失败如何演化为一次重新提问的完整调用链。一、Actions 在 Guardrails 中的定位Guardrails 的核心工作流是生成 → 解析 → 校验 → 处置。当 LLM 的输出无法通过某个验证器时框架必须决定如何处置失败值。这一决策由on_fail描述符驱动而 Actions 模块就是各种处置结果在代码层面的承载对象。在 guardrails/types/on_fail.py 中OnFailAction枚举定义了全部八种处置动作枚举值字面量语义REASKreask校验失败时重新向 LLM 提问FIXfix应用静态修复值FILTERfilter过滤掉非法值REFRAINrefrain克制回答返回空值NOOPnoop不做任何处理EXCEPTIONexception抛出ValidationErrorFIX_REASKfix_reask先静态修复修复值仍失败则 reaskCUSTOMcustom调用自定义函数处理其中与 Actions 模块直接相关的是REASK、FILTER、REFRAIN、FIX_REASK。框架在 validator_service_base.py 的perform_correction方法中把这些枚举值翻译成具体的 Action 对象REASK→ 构造FieldReAsk(incorrectValuevalue, failResults[result])FIX_REASK→ 若修复值复检仍是FailResult则构造FieldReAsk(incorrectValuefixed_value, failResults[result])FILTER→ 返回Filter()实例REFRAIN→ 返回Refrain()实例NOOP→ 原样返回value这些对象随后会混入验证后的输出结构中由 Actions 模块的辅助函数统一识别和处理。所有 Action 对象都通过 guardrails/actions/__init__.py 对外导出Filter、apply_filters、ReAsk、FieldReAsk、SkeletonReAsk、NonParseableReAsk、Refrain、apply_refrain。二、ReAsk 家族重新提问的三类触发场景ReAsk是所有重新提问动作的基类其类型定义来源于guardrails_ai.types外部公共类型包在 guardrails/actions/reask.py 中被具体化。它承载了两个核心属性incorrect_valueAny— 未通过校验的原始值fail_resultsList[FailResult]— 失败校验的结果列表每条FailResult至少包含error_message错误信息与可选的fix_value修复值。在此基础上框架派生出三个语义各异的子类覆盖三种完全不同的失败场景2.1 FieldReAsk字段级重新提问class FieldReAsk(ReAsk): path: Optional[List[Any]] NoneFieldReAsk用于针对某个具体字段发起重新提问。它额外携带path属性——一个键列表标明失败字段在嵌套结构中的位置例如[fees, 1, name]表示fees数组第 2 个元素中的name字段。path的赋值发生在验证后的体检阶段。在 reask.py 的gather_reasks函数中框架递归遍历验证后的输出字典/列表每当发现一个FieldReAsk对象就根据当前遍历深度为其补全路径if isinstance(value, FieldReAsk): value.path path [field] # 字典场景 reasks.append(value) del valid_output[field] # 列表场景 if isinstance(item, FieldReAsk): item.path path [idx] reasks.append(item) del valid_output[idx]这段实现同时揭示了一个重要机制被FieldReAsk标记的字段会从有效输出中删除从而保证最终返回给用户的guarded_output不含非法值。2.2 SkeletonReAsk骨架级重新提问SkeletonReAsk用于结构化数据整体不匹配预期 schema的场景例如 LLM 返回的 JSON 结构残缺、字段缺失或类型不符。它在 runner.py 中被创建校验流程的第一步schema_validation见 guardrails/schema/validator.py先检查解析结果是否符合 JSON Schema若不符合则直接返回SkeletonReAsk此时字段级验证器尚未运行——因为连骨架都不对逐字段校验没有意义。值得注意的是判断一个ReAsk是否应升级为SkeletonReAsk的依据是错误消息内容。在to_reask函数中reask.pyif reask.fail_results and len(reask.fail_results) 1: error_message reask.fail_results[0].error_message if error_message Output is not parseable as JSON: return NonParseableReAsk.model_validate(reask.model_dump()) elif JSON does not match schema in error_message: return SkeletonReAsk.model_validate(reask.model_dump())即错误消息为Output is not parseable as JSON时归为NonParseableReAsk包含JSON does not match schema时归为SkeletonReAsk。2.3 NonParseableReAsk无法解析时的重新提问当 LLM 的原始输出连 JSON 都解析不了时例如输出夹杂了 Markdown 代码块、前言后语或直接是无效 JSON解析器无法产出结构化结果此时产生NonParseableReAsk。它的incorrect_value就是那段无法解析的原始文本。在Runner.steprunner.py中可以看到完整的处置优先级parsed_output, parsing_error self.parse(raw_output, output_schema) if parsing_error or isinstance(parsed_output, ReAsk): iteration.outputs.reasks.append(parsed_output) else: iteration.outputs.parsed_output parsed_output if parsing_error and isinstance(parsed_output, NonParseableReAsk): reasks, _ self.introspect(parsed_output) else: validated_output self.validate(...) reasks, valid_output self.introspect(validated_output)也就是说解析失败直接进入 reask 分支不再执行字段级验证。三、ReAsk 的完整生命周期一次失败如何驱动一次重试理解了三类 ReAsk 之后我们把它们放回框架主循环中看它们如何驱动 LLM 的重新提问。核心调度逻辑在 runner.py 的Runner.__call__for index in range(self.num_reasks 1): iteration self.step(indexindex, ...) if not self.do_loop(index, iteration.reasks): break (output_schema, messages) self.prepare_to_loop( iteration.reasks, output_schema, parsed_outputiteration.outputs.parsed_output, validated_outputcall_log.validation_response, ... )一次循环由四步组成Introspect体检introspect函数reask.py判断验证输出中是否含 ReAsk 对象。单个FieldReAsk/SkeletonReAsk/NonParseableReAsk直接返回嵌套结构则调用gather_reasks收集全部 ReAsk 并剥离有效输出。DoLoop决策do_loop检查存在 ReAsk 且当前尝试次数小于num_reasks预算满足则继续循环。PrepareToLoop重组get_reask_setupreask.py按OutputTypes分派——字符串输出走get_reask_setup_for_stringJSON/结构化输出走get_reask_setup_for_json。Merge合并reask 成功后merge_reask_outputreask.py依据FieldReAsk.path把纠正后的值写回原始输出的对应位置。3.1 三类 reask 的提示词构建差异在get_reask_setup_for_jsonreask.py中三类 ReAsk 的提示词策略完全不同NonParseableReAsk把 LLM 给的无法解析的原始文本原样回填reask_value np_reask.incorrect_value并拼接high_level_json_parsing_reask_prompt要求 LLM只输出合法 JSONSkeletonReAsk回填validation_responseXML 模式或parsing_responseJSON 模式追加high_level_skeleton_reask_prompt和带结构示例的后缀告诉 LLM 按正确骨架重写FieldReAsk普通 reask通过prune_obj_for_reaskingreask.py剪枝——只保留含 ReAsk 的字段及其祖先节点其余已验证正确的字段不再次发送配合get_reask_subschema生成只含失败字段的裁剪 schema实现最小化重试。字段级 reask 的错误信息会被格式化为按路径索引的字典例如{fees.1.name: must be exactly two words}通过json.dumps注入提示词。3.2 修复值的回填sub_reasks_with_fixed_values对于fix_value已提供的校验器sub_reasks_with_fixed_valuesreask.py会递归地把FieldReAsk替换为其fail_results[0].fix_value若没有修复值则保留 ReAsk 对象供上层判断本次调用的最终状态成功、失败或部分成功。一个真实的中间产物示例可参考 validated_output_reask_1.py其中name字段的值为my chase plan校验器报告must be exactly two words并给出fix_valuemy chase同时记录path[fees, 1, name]——这正是gather_reasks补全路径后、merge_reask_output依据路径回填前的典型状态。四、Filter静默剔除非法值Filter是一个标记类filter.py本身不含数据仅用于在输出结构中占位标识此处应被过滤。真正的工作由apply_filters完成def apply_filters(value: Any) - Any: if isinstance(value, Filter): pass # 丢弃 elif isinstance(value, List): # 逐项递归丢弃返回 None 的项 elif isinstance(value, Dict): # 逐值递归丢弃值为 None 的键 else: return value其核心策略是递归下降遇到Filter实例则返回None即剔除列表逐项过滤、字典逐值过滤其余标量原样保留。需要注意的一个实现细节是当字典某键的值为Filter()时apply_filters会同时移除该键返回None导致filtered_dict[k]不被赋值而非只清空值。单测 test_filter.py 完整覆盖了这些行为例如[a, Filter(), b]→[a, b]列表元素剔除{a: Filter()}→{}字典键删除{a: b, c: {d: Filter()}}→{a: b, c: {}}嵌套字典键删除{a: b, c: [d, Filter()]}→{a: b, c: [d]}字典内列表元素剔除五、Refrain克制回答并返回空值Refrainrefrain.py同样是一个标记类。它的语义是当输出中存在任何Refrain标记时整个输出被替换为与输出类型对应的空值——这比 Filter 的逐点剔除更激进属于全有或全无策略。apply_refrain的实现分两步按输出类型确定空值refrain.py。输出类型定义在 output_type.py 的OutputTypes枚举中OutputTypes.STRING→ 空字符串OutputTypes.LIST→ 空列表[]OutputTypes.DICT→ 空字典{}递归检测check_for_refrainrefrain.py深度遍历列表与字典一旦发现任何Refrain实例立即返回True随后整体替换为空值并记录logger.debug(Refrain detected.)。单测 test_refrain.py 的断言清晰地体现了这一一刀切语义输入[a, {b: Refrain(), c: d}, e]时只要输出类型是LIST结果就是[]整个列表清空而不是仅剔除出错分支。相应地check_for_refrain对该输入的返回值为True。六、Filter 与 Refrain 的差异对照两者都源于on_failfilter/on_failrefrain见 validator_service_base.py但处理粒度截然不同下表可帮助快速区分维度FilterRefrain处理粒度逐元素/逐键剔除整体替换发现标记后仅删除该节点保留其余整个输出变为空值空值形态无直接缺失依赖OutputTypes/[]/{}适用场景列表项或字段可安全丢弃任一内容不宜输出时整体拒答核心函数apply_filters(value)apply_refrain(value, output_type)七、从验证失败到 Action 的完整链路把前面各节串联起来一次on_failreask的完整调用链如下可对照 runner.py 与 validator_service_base.py 验证Runner.step调用parse解析 LLM 原始输出解析失败 →NonParseableReAskRunner.validate先做 schema 级校验schema_validation不匹配 →SkeletonReAsk字段级校验由validator_service执行perform_correction依据on_fail_descriptor把FailResult转换为FieldReAsk/Filter()/Refrain()等 Action 对象混入验证输出Runner.introspect经gather_reasks从输出中收集全部 ReAsk 并补全path同时剥离它们得到安全输出Runner.do_loop依据num_reasks预算决定是否继续Runner.prepare_to_loop经get_reask_setup构建裁剪后的 schema 与 reask 提示词进入下一轮stepreask 成功后merge_reask_output按path将纠正值写回原输出。八、小结Actions 模块虽然只是几个轻量类与辅助函数却是 Guardrails失败处置语义的实体化核心ReAsk家族将三类失败场景字段失败、结构不匹配、无法解析映射为三种不同的重试提示词策略Filter与Refrain分别提供局部剔除与整体拒答两种降级方案。理解这些对象的产生时机、路径补全机制与递归处理逻辑是深入阅读 Runner 主循环、自定义验证器乃至调试多轮 reask 行为的前提。相关实现均可直接阅读 guardrails/actions 目录下的三个源文件并通过 test_filter.py、test_refrain.py 及 test_guard.py 中的断言验证其行为。赞分享AI 安全治理模型安全AI 应用【免费下载链接】guardrailsAdding guardrails to large language models.项目地址https://gitcode.com/gh_mirrors/gu/guardrails点击查看免费下载相关推荐gRPC Logging Filter 深度解析调用级日志采集与审计的实现原理gRPC Logging Filter 深度解析调用级日志采集与审计的实现原理 本篇文章围绕 gRPCC 实现中的 Logging Filter 展开后端RPC框架微服务通信如何优雅解决PyMySQL连接瓶颈3种实用连接池实现方案全解析如何优雅解决PyMySQL连接瓶颈3种实用连接池实现方案全解析 PyMySQL作为Python开发者首选的MySQL驱动库在高并发场景下常因频繁创建和销毁连数据库数据库客户端后端Redux-actions源码解析深入理解Flux标准Action工具库的实现原理Redux actions源码解析深入理解Flux标准Action工具库的实现原理 Redux actions是一个专为Redux设计的Flux标准Actio开发工具前端上一篇如何自建免费翻译APILibreTranslate 部署与调用完整指南下一篇终极指南5分钟解决魔兽争霸3在Win10/Win11的所有兼容性问题创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑