安全引擎:Policy / Rule / Action 护栏

30 秒导读: Upsonic 把"金融级安全"落在一层可插拔护栏上。用户输进来的话(入站)和 Agent 说出去的话(出站)都会先被扫一遍:命中敏感内容(信用卡、SSN、病历……)就按你选的动作处理——直接放行、脱敏改写、替换占位符,或干脆拦截报错。本章讲这层护栏的内部:三段式模型 Policy/Rule/Action,内置的领域策略库,LLM 兜底判定与可逆脱敏,以及它如何挂进第 2 章那条 24 步管线。

1. 这是什么(零基础也能懂)

一句话定义: 安全引擎是一层"内容检查站",在 Agent 真正调用大模型之前检查用户输入、在把回答返回之后检查模型输出。

它解决什么问题。 生产环境里,你不希望:

用户把一整张信用卡号、身份证、病历粘进 prompt,然后被原样发去第三方大模型;
模型的回答里不小心带出了敏感数据、违规内容,直接呈给终端用户。

传统做法是在业务代码里到处写 if "信用卡" in text。Upsonic 把这件事抽象成可复用、可组合、可插拔的护栏对象,一行配置就能挂上。

用起来什么样。 建 Agent 时传一个(或一串)现成策略即可,剩下的管线自动处理:

# 示意,非源码
from upsonic import Agent
from upsonic.safety_engine import PIIAnonymizePolicy, FinancialInfoBlockPolicy

agent = Agent(
    model="openai/gpt-4o",
    user_policy=[PIIAnonymizePolicy, FinancialInfoBlockPolicy],  # 入站护栏
    agent_policy=PIIBlockPolicy,                                  # 出站护栏
)
agent.do("我的卡号是 4111 1111 1111 1111,帮我查账单")
# → FinancialInfoBlockPolicy 命中信用卡,这次运行被拦下,模型根本不会被调用

一句话直觉。 把它想成机场安检:**规则(Rule)**是那台 X 光机——只负责"看出这里有没有违禁品、有多可疑";**动作(Action)**是安检员——决定"放行 / 没收 / 请你重新打包 / 直接报警";**策略(Policy)**就是"这台机器 + 这个安检员"的固定搭配。你要做的只是选一套搭配挂在门口。

2. 顶层全景(三段式怎么转)

安全引擎的骨架只有三个类,职责严格分离。先看它们怎么串起来处理一条文本:

                      PolicyInput(input_texts=[...])
                                 │
                                 ▼
        ┌────────────────────  Policy  ────────────────────┐
        │  (name + 一个 Rule + 一个 Action + 语言/LLM 配置)  │
        │                                                   │
        │   ① rule.process(input)                           │
        │        │                                          │
        │        ▼                                          │
        │   RuleOutput{ confidence, content_type,           │
        │              details, triggered_keywords }        │
        │        │                                          │
        │        ▼                                          │
        │   ② action.execute_action(rule_output, texts)     │
        │        │  按 confidence + 动作类型决定怎么处理     │
        │        ▼                                          │
        │   PolicyOutput{ output_texts,                     │
        │                action_output{action_taken,...},   │
        │                transformation_map }               │
        └───────────────────────────────────────────────────┘
                                 │
                                 ▼
        动作:ALLOW / REPLACE / ANONYMIZE / BLOCK / (raise DisallowedOperation)

三个部件的一句话职责:

部件	干什么	在哪个文件	关键符号
`Rule`(规则)	只做判定:扫文本,给出置信度和命中项,不改内容	`base/rule_base.py`	`RuleBase.process`
`Action`(动作)	只做处置:放行 / 替换 / 脱敏 / 拦截	`base/action_base.py`	`ActionBase.action`
`Policy`(策略)	把一条 Rule + 一个 Action 绑在一起,提供 `check`/`execute`	`base/policy.py`	`Policy.execute`

为什么这样切。 判定和处置解耦后,同一个"信用卡检测规则"可以配不同动作:线上环境配"拦截",测试环境配"脱敏",审计环境配"替换成占位符"。内置策略库正是靠这种排列组合,用几个 Rule × 几个 Action 铺出上百个现成策略(见 §4)。

三类结果对象都在 models.py 里,是纯 Pydantic 数据类:

入口 PolicyInput(models.py:9):装 input_texts 以及可选的图/音/视频/文件,还有一个关键字段 existing_transformation_map——多策略串行时用来传递已有的脱敏映射。
规则结果 RuleOutput(models.py:21):confidence(0~1)、content_type、details、triggered_keywords。
策略结果 PolicyOutput(models.py:30):output_texts(处理后的文本)、action_output(含 action_taken)、transformation_map(脱敏还原表)。注意 models.py:46 里 ActionOutput = PolicyOutput——两者是同一个类的别名。

3. 核心机制一:三段式的内部

3.1 Policy —— 只是个"绑定器 + 编排器"

Policy 本身很薄。构造时收下一个 rule、一个 action,外加语言和三个可选 LLM(语言识别、基础操作、文本查找),见 base/policy.py:15 Policy.__init__。

它对外只有两个动词:

check(policy_input) —— 只跑规则,拿 RuleOutput(policy.py:67)。
execute(policy_input) —— 先 check 再让动作处置,返回 (rule_result, action_result, policy_output) 三元组(policy.py:79)。

真实编排就这么直白:

# base/policy.py:79  Policy.execute(节选)
rule_result = self.check(policy_input)
action_result = self.action.execute_action(
    rule_result, policy_input.input_texts or [], self.language,
    self.language_identify_llm, self.base_llm, self.text_finder_llm,
    existing_transformation_map=getattr(policy_input, 'existing_transformation_map', None)
)
return rule_result, action_result, action_result

管线实际走的是异步版 execute_async(policy.py:90):它会优先调用 rule/action 各自的 *_async 方法,没有就用 asyncio.to_thread 把同步实现丢进线程池,既不阻塞事件循环,又不强迫每个自定义策略都实现异步——这是全套护栏"同步实现 + 异步外壳"的统一套路。

3.2 Rule —— 只判定,不改内容

RuleBase(base/rule_base.py:14)是抽象基类,唯一必须实现的是 process(policy_input) -> RuleOutput(rule_base.py:25)。它约定了规则只输出判断,绝不修改文本——这条纪律让规则可以随便组合、随便复用。

基类还预置了一个 LLM 兜底工具 _llm_find_keywords_with_input(rule_base.py:33):把输入拼成一段文本,先自动检测语言,再让"文本查找 Agent"抽取指定类型的敏感项。领域规则的 *_LLM_Finder 变体就靠它(见 §4.3)。

3.3 Action —— 五种处置,一个基类全给你

ActionBase(base/action_base.py:16)是护栏里代码量最大的一块,因为所有"怎么处置"的通用能力都沉淀在这里,子类只需在 action() 里挑一个调用。

入口是 execute_action(action_base.py:30):它先把 rule_result、原文、语言、各 LLM 存进实例,再解析目标语言(auto 时用 LLM 检测内容语言),最后调子类的 action()。基类提供的处置原语有五种:

处置原语	方法	`action_taken`	效果
放行	`allow_content` (`action_base.py:208`)	`ALLOW`	原样返回
拦截(带消息)	`raise_block_error` (`action_base.py:232`)	`BLOCK`	用一段(可翻译的)消息替换输出,标记已拦截
替换占位符	`replace_triggered_keywords` (`action_base.py:257`)	`REPLACE`	把命中项统一换成如 `[PII_REDACTED]`
可逆脱敏	`anonymize_triggered_keywords` (`action_base.py:324`)	`ANONYMIZE`	换成同格式随机值,并记录还原表
抛异常	`raise_exception` (`action_base.py:434`)	抛 `DisallowedOperation`	中断整条链路

每种还有 *_LLM 变体,把固定消息交给 LLM 生成更贴合上下文的说辞(如 llm_raise_block_error,action_base.py:401)。

一个值得记住的细节:带类型前缀的命中项。 规则产出的命中项形如 CREDIT_CARD:4111...。处置时基类会用 keyword.split(":", 1)[1] 剥掉类型前缀,只对真实值做替换(action_base.py:267);而纯检测标记 PII_KEYWORD:xxx 会被显式跳过(action_base.py:346),因为它只是"这里提到了信用卡"这种关键词命中,不是真值,没什么可脱敏的。

4. 核心机制二:内置策略库(挑 PII 与 Financial 看真章)

policies/ 下按领域切了十几个文件:pii、financial、crypto、phishing、fraud_detection、medical、legal、cybersecurity、insider_threat、tool_safety……每个文件的套路一模一样:一个正则规则 + 一个 LLM 规则 + 五个动作 → 组合出 7 个现成 Policy,最后在 safety_engine/__init__.py 里统一惰性导出(__init__.py:69 _get_policy_classes)。

看两个最能体现"金融级"的领域。

4.1 PII 规则:正则矩阵 + 加权置信度

PIIRule(policies/pii_policies.py:12)在构造函数里堆了一整套正则:邮箱、电话(美/国际/无区号多套)、SSN、信用卡、地址、生日、驾照、护照、IP、MAC,外加一串 PII 关键词。

process(pii_policies.py:129)把所有输入拼成一段,逐类 re.findall,命中就打上类型前缀塞进 triggered_items,例如 EMAIL:a@b.com、SSN:123-45-6789。

真正体现"分级"的是加权置信度——不是"命中就 1.0",而是按敏感度打分:

# policies/pii_policies.py:208 (节选)
high_risk_count   = len([...  if any(x in item for x in ["SSN:", "CREDIT_CARD:", "PASSPORT:"])])
medium_risk_count = len([...  if any(x in item for x in ["EMAIL:", "PHONE:", "ADDRESS:", "DOB:"])])
low_risk_count    = len([...  if "PII_KEYWORD:" in item])
confidence = min(1.0, (high_risk_count * 0.9 + medium_risk_count * 0.6 + low_risk_count * 0.3))

一个 SSN 就能把置信度顶到 0.9;而只是文本里出现"phone number"这个词(低危关键词)只加 0.3。

减少误报的巧思。 规则里专门维护了一批 false_positive_patterns(pii_policies.py:97),像 "email system""email server" 这种技术语境里的 "email",会被判为假阳性、不计入命中(pii_policies.py:191)。这就是"专业性":检测器知道"提到 email 这个词"和"贴出一个真邮箱地址"是两回事。

4.2 动作里的阈值门:0.3 起步

领域动作并不无脑处置,而是先看置信度。以 PIIBlockAction(pii_policies.py:268)为例:

# policies/pii_policies.py:275  action(节选)
if rule_result.confidence < 0.3:
    return self.allow_content()          # 太弱,放行
return self.raise_block_error(block_message)  # 够强,拦截

0.3 这个阈值在每个动作里反复出现——它是"低危关键词单独命中(0.3)也刚好够门槛、但空命中(0.0)一定放行"的分界。换成 PIIAnonymizeAction(pii_policies.py:308)就是同样的门后面接 anonymize_triggered_keywords(),PIIReplaceAction(pii_policies.py:325)接 replace_triggered_keywords("[PII_REDACTED]")。

4.3 七种现成组合

文件末尾把规则和动作排列成 7 个开箱即用的 Policy 实例(pii_policies.py:382 起):

策略实例	规则	动作	语义
`PIIBlockPolicy`	`PIIRule`(正则)	Block	命中就拦
`PIIBlockPolicy_LLM`	`PIIRule`	Block(LLM 消息)	拦截,消息由 LLM 生成
`PIIBlockPolicy_LLM_Finder`	`PIIRule_LLM_Finder`	Block	用 LLM 检测再拦
`PIIAnonymizePolicy`	`PIIRule`	Anonymize	可逆脱敏
`PIIReplacePolicy`	`PIIRule`	Replace	换占位符
`PIIRaiseExceptionPolicy`	`PIIRule`	RaiseException	抛异常中断
`PIIRaiseExceptionPolicy_LLM`	`PIIRule`	RaiseException(LLM)	抛异常,消息 LLM 生成

PIIRule_LLM_Finder(pii_policies.py:223)有个稳健设计:没配 text_finder_llm,或 LLM 调用抛错时,自动回落到正则版 PIIRule(pii_policies.py:236、262)——LLM 是增强,不是单点故障。

4.4 Financial 规则:同一套骨架,金融特化

FinancialInfoRule(policies/financial_policies.py:12)结构与 PII 如出一辙,但正则更专业:按卡种精确匹配 Visa/MasterCard/Amex/Discover(financial_policies.py:23),还有 CVV、IBAN、SWIFT、路由号、EIN/TIN 税号、余额/利率等财务语句,以及比特币/以太坊钱包地址(financial_policies.py:110)。

它的加权分级比 PII 多一档"critical",权重直接给到 1.0:

# policies/financial_policies.py:198 (节选)
critical_count = len([... if any(x in item for x in ["CREDIT_CARD:", "SSN:", "BANK_ACCOUNT:", "ROUTING_NUMBER:"])])
confidence = min(1.0, (critical_count*1.0 + high_risk_count*0.8 + medium_risk_count*0.6 + low_risk_count*0.3))

也就是说,单独一个信用卡号就足以让置信度到 1.0、越过所有动作阈值。同样导出 7 个现成策略(financial_policies.py:372 起,如 FinancialInfoBlockPolicy、FinancialInfoAnonymizePolicy)。

注意别和第 4 章的工具策略搞混:policies/tool_safety_policies.py 里的 HarmfulToolBlockPolicy / MaliciousToolCallBlockPolicy 是针对工具调用的护栏,由独立的 ToolPolicyManager 执行;本章讲的是针对文本消息的护栏。

5. 核心机制三:LLM 判定与可逆脱敏

正则能抓"长得像敏感数据"的东西,但抓不住"语义上敏感"的表达。安全引擎为此配了一个 LLM 侧车,和一套可逆脱敏工具。

5.1 UpsonicLLMProvider —— 护栏专用的小 Agent 封装

UpsonicLLMProvider(llm/upsonic_llm.py:75)内部就是包了一个 Upsonic Agent,对每种任务用结构化输出(Pydantic response_format)约束返回:

能力	方法	结构化返回
抽取敏感项	`find_keywords` (`upsonic_llm.py:85`)	`KeywordDetectionResponse`
生成拦截消息	`generate_block_message` (`upsonic_llm.py:149`)	`BlockMessageResponse`
语义脱敏	`anonymize_content` (`upsonic_llm.py:197`)	`AnonymizationResponse`
语言识别	`detect_language` (`upsonic_llm.py:258`)	`LanguageDetectionResponse`
翻译	`translate_text` (`upsonic_llm.py:299`)	`TranslationResponse`
工具安全分析	`analyze_tool_safety` (`upsonic_llm.py:567`)	`ToolSafetyAnalysisResponse`
违规反馈	`generate_policy_feedback` (`upsonic_llm.py:754`)	`PolicyFeedbackResponse`

置信度闸门。 LLM 抽取不是照单全收——find_keywords 只在模型自报 confidence >= 0.7 时才采纳结果,否则返回空(upsonic_llm.py:111);detect_language 的阈值是 0.6(upsonic_llm.py:273)。而且每个方法都有 try/except 兜底:LLM 挂了就回退到安全默认(空列表、原文、英文),绝不因为侧车故障把整条护栏搞崩。

多语言支持。 translate_text(upsonic_llm.py:299)带一张 ISO 639-1 → 语言全名的大映射表,拦截消息因此能按检测到的用户语言本地化返回——这也是 ActionBase._translate(action_base.py:126)在处置时会被调用的原因。

5.2 可逆脱敏:换成随机同格式值,还能还原

ANONYMIZE 和 REPLACE 的本质区别是可逆。replace_triggered_keywords 把命中项统一换成一个固定占位符(信息全丢);anonymize_triggered_keywords 则换成同格式随机值并记录映射,事后能还原。

核心生成逻辑在 ActionBase._generate_unique_replacement(action_base.py:157):

# base/action_base.py:188 (节选)—— 逐字符保形替换
for char in original:
    if char.isdigit():   replacement += str(random.randint(0, 9))   # 数字→随机数字
    elif char.isalpha(): replacement += random.choice(...ascii...)  # 字母→随机字母
    else:                replacement += char                        # 其它原样保留

所以 john@example.com 会变成 xkqp@fhsmwtr.lzn 这种——格式一致、语义抹掉。同时把 {original, anonymous} 存进 transformation_map。

anonymization.py 是这套逻辑的独立、可复用版本(供管线在把脱敏文本发去大模型、再从响应里还原时使用):

Anonymizer(anonymization.py:53)带缓存,保证同一个原值在多次出现时映射到同一个随机值。
deanonymize_content(anonymization.py:225)反向替换,还原时按匿名串长度从长到短排序(anonymization.py:257),避免短串是长串子串导致的错误还原。
StreamDeanonymizer(anonymization.py:312)是流式还原器:边收 token 边缓冲,只吐出"确定不会跨到某个匿名串中间"的安全前缀——这样流式输出也能安全还原敏感值。

一个防绕过的小心思。 _generate_unique_replacement(action_base.py:164)会处理"带前导空格的变体":如果 ' 555-123-4567' 已经映射过,那么裸值 '555-123-4567' 会直接派生自它(去掉空格),防止 LLM 因分词差异产出一个还原不回来的匿名值。

6. 核心机制四:如何接入一次运行

前面都是"护栏本身",这一节讲它怎么挂进第 2 章那条管线。链路上有两个挂点,分别在调用大模型的前后:

  用户输入
    │
    ▼
 [UserPolicyStep]  ── 入站护栏(在 ModelExecutionStep 之前)
    │   agent._apply_user_policy(task, context, sys_prompt_mgr)
    │   扫: description / context / system_prompt / chat_history
    ▼
 ...(模型选择、工具装配、消息组装)...
    │
    ▼
 [ModelExecutionStep]  ── 真正调用 LLM
    │
    ▼
 ...(响应处理、反思、可靠性层)...
    │
    ▼
 [AgentPolicyStep]  ── 出站护栏
    │   agent._apply_agent_policy(task, context)
    │   扫: 模型的最终回答
    ▼
  返回用户

6.1 构造时:两个 PolicyManager

Agent 构造函数把 user_policy / agent_policy 各自包进一个 PolicyManager(agent/agent.py:508):

# agent/agent.py:509 (节选)
self.user_policy_manager  = PolicyManager(policies=user_policy,  policy_type="user_policy", ...)
self.agent_policy_manager = PolicyManager(policies=agent_policy, policy_type="agent_policy", ...)

紧接着 _setup_policy_models(agent/agent.py:690)把 Agent 自己的 model 灌给每条策略的 base_llm——这样策略不必单独配模型,默认复用 Agent 的模型(policy_manager.py:451 setup_policy_models)。

6.2 入站:_apply_user_policy 的"带源标签扫描"

_apply_user_policy(agent/agent.py:3026)由 UserPolicyStep(pipeline/steps.py:327)调用。它的巧妙在于不只扫用户那句话,而是把这次要发给模型的所有文本都收集起来,每条带一个来源标签:

# agent/agent.py:3069 (节选)—— 收集输入 + 来源
input_texts.append(task.description);   source_keys.append(("description", None))
if task.context_formatted:  input_texts.append(...);  source_keys.append(("context", None))
if system_prompt_text:      input_texts.append(...);  source_keys.append(("system_prompt", None))
# ...再逐条追加 chat_history 的每个 part

然后交给 execute_policies_async,并带上 source_keys、task、agent(agent.py:3102)。这开启了按作用域过滤:每条策略可声明只作用于某些来源(如"只脱敏聊天历史,不动系统提示"),由 resolve_policy_scope(policy_manager.py:34)按 Policy > Task > Agent 的优先级解析。

处置结果回来后:命中 BLOCK/异常就 task.task_end() 并把结果写成响应、返回 should_continue=False,让管线提前收尾、根本不调用模型(agent.py:3148);命中 REPLACE/ANONYMIZE 则把脱敏后的文本按来源写回 task 和 chat_history(agent.py:3174 起)。

6.3 出站:_apply_agent_policy 与反馈重试环

_apply_agent_policy(agent/agent.py:3529)由 AgentPolicyStep(pipeline/steps.py:2623)调用,扫的是 task.response。

它多了一个反馈循环:如果开了 agent_policy_feedback,违规时不直接拦,而是让 LLM 生成一段"你哪里违规了、该怎么改"的反馈,把它当成新的 user 消息重新执行模型,最多试 feedback_loop_count + 1 轮(steps.py:2699 的 max_iterations、agent.py:3616 的 should_retry_with_feedback)。轮次耗尽还不过,才落回最终的拦截/改写。

6.4 PolicyManager:顺序执行 + 取最严

一个挂点可以挂多条策略,PolicyManager.execute_policies_async(policy_manager.py:198)负责编排。核心规则:

逐条执行,把命中的策略名和 RuleOutput 累积进 PolicyResult。
BLOCK 最霸道:任一策略 BLOCK,立即 break,不再跑后面的(policy_manager.py:292)。
REPLACE/ANONYMIZE 会链式叠加:前一条脱敏后的文本喂给后一条,transformation_map 也累积合并(policy_manager.py:306),并通过 accumulated_map 传给下一条的 existing_transformation_map——保证多条脱敏策略共用一致的映射。
DisallowedOperation 等价于 BLOCK:捕获后立即停(policy_manager.py:341)。
未预期异常则跳过该策略、继续(policy_manager.py:372)——一条坏策略不拖垮整组。

结果对象 PolicyResult(policy_manager.py:58)用 should_block() / should_retry_with_feedback() 两个判断方法,把"要不要拦、要不要带反馈重试"暴露给上层的两个 apply 方法。

7. 巧妙之处(可借鉴的技术)

判定与处置彻底解耦。 Rule 只出 RuleOutput、绝不改文本;Action 只处置。于是一个规则能配五种动作,内置库靠"规则 × 动作"排列组合铺出上百个策略(base/rule_base.py:25 vs base/action_base.py:149)。
加权分级置信度,而非布尔命中。 敏感度不同权重不同,配 0.3 阈值门,天然区分"贴出真值"和"只是提到关键词"(policies/pii_policies.py:208、financial_policies.py:198)。
LLM 是增强而非依赖。 *_LLM_Finder 规则在未配 LLM 或调用失败时自动回落正则版;所有 LLM 调用都 try/except 兜底到安全默认(pii_policies.py:236、upsonic_llm.py:116)。
可逆脱敏 + 保形随机值 + 一致映射缓存。 脱敏后仍能发去大模型、再从响应里精确还原,靠的是逐字符保形替换、按长度排序的反向替换、以及同值一致映射(base/action_base.py:157、anonymization.py:106/:257)。
带源标签的作用域扫描。 入站不只扫用户输入,而是把 description/context/system_prompt/chat_history 一起收集、逐来源可控处置(agent/agent.py:3069、policy_manager.py:34)。
出站反馈重试环。 违规不必立刻拦,可让模型带着"错在哪"的反馈重答几轮,兼顾安全与可用(pipeline/steps.py:2699)。

8. 边界与局限(诚实)

正则规则有固有误报/漏报。 例如 SSN/驾照/护照的正则会互相重叠(policies/pii_policies.py:57~66 的驾照与护照 pattern 高度相似),纯正则模式下靠加权和假阳性表勉强兜底;要更准得开 *_LLM_Finder,但那要付出一次额外 LLM 调用的延迟与成本。
脱敏基于精确子串替换。 anonymize/replace 用 re.escape(target) 做大小写不敏感替换(base/action_base.py:271),如果敏感值在文本里被换行、空格或格式打断,可能替换不到。代码只对"前导空格变体"做了特判(action_base.py:164),其它变形不保证覆盖。
LLM 判定的确定性有限。 命中与否取决于模型自报的 confidence 是否过 0.7/0.6 阈值(llm/upsonic_llm.py:111、:273),不同模型、不同措辞结果可能不同;失败时静默回落,不会报错提示。
翻译逻辑里有硬编码痕迹。 translate_text 的 prompt 写死了 "cryptocurrency → kripto para" 的土耳其语规则、并带土耳其语兜底翻译表(llm/upsonic_llm.py:401、:423),对其它语言不是对称支持。
本章不覆盖工具护栏与管线骨架。 工具调用的 pre/post 策略见第 4 章;24 步管线整体见第 2 章。

9. 代码地图(导航索引)

主题	文件路径	符号名
三段式:策略绑定与编排	`src/upsonic/safety_engine/base/policy.py`	`Policy.check` / `Policy.execute` / `Policy.execute_async`
三段式:规则基类(只判定)	`src/upsonic/safety_engine/base/rule_base.py`	`RuleBase.process` / `RuleBase._llm_find_keywords_with_input`
三段式:动作基类(五种处置)	`src/upsonic/safety_engine/base/action_base.py`	`ActionBase.execute_action` / `allow_content` / `raise_block_error` / `replace_triggered_keywords` / `anonymize_triggered_keywords` / `_generate_unique_replacement`
数据模型	`src/upsonic/safety_engine/models.py`	`PolicyInput` / `RuleOutput` / `PolicyOutput`
异常	`src/upsonic/safety_engine/exceptions.py`	`DisallowedOperation`
PII 规则与策略	`src/upsonic/safety_engine/policies/pii_policies.py`	`PIIRule.process` / `PIIRule_LLM_Finder` / `PIIBlockAction` / `PIIBlockPolicy`
金融规则与策略	`src/upsonic/safety_engine/policies/financial_policies.py`	`FinancialInfoRule.process` / `FinancialInfoBlockPolicy` / `FinancialInfoAnonymizePolicy`
策略统一导出	`src/upsonic/safety_engine/__init__.py`	`_get_policy_classes`
LLM 侧车	`src/upsonic/safety_engine/llm/upsonic_llm.py`	`UpsonicLLMProvider.find_keywords` / `generate_block_message` / `detect_language` / `analyze_tool_safety` / `generate_policy_feedback`
可逆脱敏工具	`src/upsonic/safety_engine/anonymization.py`	`Anonymizer` / `deanonymize_content` / `StreamDeanonymizer`
多策略编排	`src/upsonic/agent/policy_manager.py`	`PolicyManager.execute_policies_async` / `resolve_policy_scope` / `PolicyResult` / `setup_policy_models`
接入点:构造与模型灌入	`src/upsonic/agent/agent.py`	`Agent._setup_policy_models`(构造处 `agent.py:508`)
接入点:入站护栏	`src/upsonic/agent/agent.py`	`Agent._apply_user_policy`
接入点:出站护栏 + 反馈环	`src/upsonic/agent/agent.py`	`Agent._apply_agent_policy`
管线挂点	`src/upsonic/agent/pipeline/steps.py`	`UserPolicyStep` / `AgentPolicyStep`

1. 这是什么(零基础也能懂)​

2. 顶层全景(三段式怎么转)​

3. 核心机制一:三段式的内部​

3.1 Policy —— 只是个"绑定器 + 编排器"​

3.2 Rule —— 只判定,不改内容​

3.3 Action —— 五种处置,一个基类全给你​

4. 核心机制二:内置策略库(挑 PII 与 Financial 看真章)​

4.1 PII 规则:正则矩阵 + 加权置信度​

4.2 动作里的阈值门:0.3 起步​

4.3 七种现成组合​

4.4 Financial 规则:同一套骨架,金融特化​

5. 核心机制三:LLM 判定与可逆脱敏​

5.1 UpsonicLLMProvider —— 护栏专用的小 Agent 封装​

5.2 可逆脱敏:换成随机同格式值,还能还原​

6. 核心机制四:如何接入一次运行​

6.1 构造时:两个 PolicyManager​

6.2 入站:_apply_user_policy 的"带源标签扫描"​

6.3 出站:_apply_agent_policy 与反馈重试环​

6.4 PolicyManager:顺序执行 + 取最严​

7. 巧妙之处(可借鉴的技术)​

8. 边界与局限(诚实)​

9. 代码地图(导航索引)​