数据截至 (上游 commit 101f0313b0dd)
安全引擎: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:[email protected]、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 # 其它原样保留
所以 [email protected] 会变成 [email protected] 这种——格式一致、语义抹掉。同时把 {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:25vsbase/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),对其它语言不是对称支持。
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 |