写代码跑AI Agent跑得越爽心里越虚——这是很多小白程序员入坑之后的第一反应。模型本身只是生成文本真正危险的是你给了它调用工具的能力它能执行shell命令、读写文件、发HTTP请求甚至操作数据库。我见过太多人兴致勃勃写完一个Agent demo然后看着它在终端里自作主张地删文件、反复重试报错、把密钥写进日志。问题从来不是模型“坏”而是你让它在没有任何护栏的状态下裸奔。这篇文章想聊的就是这件事怎么用Filter Chains这套拦截过滤的架构思路给AI Agent装上一层可控的安全兜底。适合刚开始写Agent、对代码执行和工具调用不太有把握的同学参考原理不复杂代码量也不大但能少踩很多坑。1. AI Agent为什么会“失控”——小白最容易忽略的安全盲区1.1 你以为它在回答问题其实它在操作你的电脑很多新手对Agent的认知还停留在“会聊天的程序”上。实际上Agent和普通聊天机器人的本质区别在于它拥有行动能力它不光能生成文本还能基于模型输出的工具调用指令去真的执行subprocess.run()、open().write()、requests.get()这类代码。也就是说你给它一个目标它可以自己规划步骤、调用工具、看到结果再决定下一步做什么。这个机制听起来很强大但它引入了一个核心问题模型输出的“建议文本”和程序实际执行的“操作指令”之间存在一条被大多数新手忽略的通道。模型只是根据概率生成了一串字符它并不理解这串字符在特定环境里意味着什么。比如它可能生成rm -rf /tmp/cache/*在沙箱里无害在真实目录下可能就是灾难。一个很直观的类比是普通程序像是一把锁好的刀只有你握着它才能切东西而Agent相当于你把这把刀递给了一个想象力丰富但判断力有限的路人他可能真的会拿去削铅笔也可能拿去砍电线杆。重点在于——你不知道他会遇到什么场景遇到之后会做什么选择。而Filter Chains做的事情就是在这条路上每隔一段设一个检查点把不该发生的事拦下来。1.2 三类典型事故每一个都发生在真实项目里类型一直接执行危险命令。有个人开发了一个“系统体检助手”让Agent定期检查服务器的磁盘占用。Agent某次生成的shell命令是find / -name *.log -delete本来想清理日志但匹配范围太宽把系统运行日志也扫掉了。这个命令完全合法、不在任何黑名单里问题在于Agent对操作的影响范围没有概念。类型二敏感信息泄露。另一个场景是让Agent帮忙调试一个第三方接口。Agent很“贴心”地打印了完整的请求头和参数结果请求头里的Access Token就这么进了终端日志又随着日志文件被同步到了远端仓库。模型本身没有安全意识它只知道自己要debug不知道什么能打印什么不能打印。类型三无限循环和连锁调用。让Agent做一个小任务比如“检查所有用户的状态并导出异常列表”结果Agent反复调用同一个函数、每次都因为参数格式不对报错然后自己“修正”参数再试一遍。这类循环既消耗token又可能对线上数据产生重复写入。这三个场景归纳起来就是输入不可信、输出不可控、模型判断不可依赖。你不可能通过修改prompt来根治这些问题因为模型依然是概率输出。所以正确的思路是在模型和工具执行之间加一层程序化的、确定性强的拦截机制——这就是Filter Chains的核心价值。2. Filter Chains到底是什么——一个拿来就能用的安全架构2.1 从“安检通道”说起为什么单个过滤函数不够用先想象一个机场安检有身份核验、行李扫描、液体检查、登机牌复核。如果只设一个“危险品检查”窗口所有旅客都挤过去这个窗口要么特别松漏检要么特别严堵塞。AI Agent的安全检查也是一样的道理——你需要的是一个有序的、分层的检查链条而不是一个试图覆盖所有风险的单点函数。Filter Chains的技术本质是责任链模式Chain of Responsibility。它的核心思想是把安全检查拆成多个独立的过滤器Filter每个过滤器只负责一件事然后按顺序串起来。请求这里指Agent打算执行的每一个动作从链条头进入逐个经过每个过滤器任何一个过滤器判定拒绝请求就被短路拦截全部通过才最终到达真正的工具执行层。这种架构有两个非常直接的好处。第一每个过滤器可以单独测试和演进。你改敏感词表的时候不用动命令白名单的逻辑新加一个“输出脱敏”过滤器也不影响其他环节。第二顺序本身就是一种设计。有些检查必须排在另一些检查前面——比如输入长度校验应该在敏感词扫描之前因为超长输入本身就可能是一种攻击载荷。很多新人会问我写一个check_all()函数把所有规则堆进去不行吗当然可以但一旦规则多了你会发现check_all()迅速膨胀成一个几百行的if-else集合职责混乱、测试困难、性能也不好排查。Filter Chains在工程上最大的价值是让安全逻辑可组合、可观测、可维护。2.2 Filter Chain的三大设计原则原则一顺序敏感先粗后细。成本低的检查放前面成本高的放后面。比如长度检查是一个整数字段比较执行极快而内容AI分类判罚可能涉及一次模型调用成本较高就应放在后面。这样大部分明显异常的请求可以在早期被低成本拦截避免浪费后续检查的资源。原则二拒绝必须“短路”通过必须“留痕”。任何一个Filter判定拒绝后整个请求不得继续向后传递这一点必须严格执行。同时每个Filter的判定结果、耗时、理由都应该记录到上下文对象里这样出了问题能追溯是哪个环节拦下来的、为什么拦下来。原则三检查对象包含输入和输出两端。这一点最容易被忽略。新手往往只想到“Agent要执行什么我得管”却忘了“Agent返回了什么内容我也得管”。模型可能生成包含密钥、内部IP、用户隐私的输出这些内容如果直接返回给上层调用方或者写入日志同样是安全事故。所以Filter Chains的检查对象应该是“输入指令 输出内容”的双向结构。2.3 Filter Chain的执行上下文设计在实际实现中理解“上下文对象”是理解整条链路的钥匙。我用一个FilterContext类来贯穿所有过滤器dataclass class FilterContext: # Agent 请求输入的原始信息 action_type: str # 动作类型are_call / llm_output / http_request tool_name: str # 目标工具或函数名 tool_args: dict # 传给工具的参数 raw_input: str # 模型的原始输入提示词 llm_raw_output: str # 模型生成的原始响应 # 传递给工具的真实参数Filter 可以修改它 allowed: bool True # 是否放行 reason: str # 拒绝原因留给审计和用户提示 metadata: dict field(default_factorydict) # 记录每个 Filter 的判定结果每个过滤器操作的对象就是FilterContext读它、校验它、必要的时候修改它但最终只能有一个决定——放行或拒绝。这种设计的妙处在于后面的过滤器可以拿到前面过滤器的处理结果形成叠加效果。比如第一个过滤器给metadata里写入input_len3200后面的“AI风控”过滤器就能直接依赖这个数据决定是否需要额外检查。3. 实战手写一套轻量Filter ChainsPython实现3.1 定义Filter接口和Chain核心类直接用一个最小可用的Python实现来演示不依赖任何重型框架。先定义Filter的抽象基类和Chain的调度逻辑from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import List, Optional dataclass class FilterContext: 一次Agent行为的完整上下文过滤器之间通过它传递数据 action_type: str # tool_call / llm_output tool_name: str tool_args: dict field(default_factorydict) raw_input: str llm_output: str allowed: bool True reason: str metadata: dict field(default_factorydict) class BaseFilter(ABC): 所有过滤器的基础接口 name: str base abstractmethod def check(self, ctx: FilterContext) - bool: 返回 True 表示通过False 表示拒绝并短路 ... class FilterChain: def __init__(self): self.filters: List[BaseFilter] [] def add(self, f: BaseFilter) - FilterChain: self.filters.append(f) return self def run(self, ctx: FilterContext) - FilterContext: ctx.allowed True ctx.reason for f in self.filters: start_time time.time() passed f.check(ctx) ctx.metadata[f{f.name}_result] passed ctx.metadata[f{f.name}_cost_ms] round((time.time() - start_time) * 1000, 2) if not passed: ctx.allowed False ctx.reason ctx.reason or f{f.name}: 拦截 break return ctx核心逻辑就在run方法里从头到尾遍历过滤器任何一个返回False就立即停止。这样整条Chain的执行成本是可控的——正常放行的请求才需要走完全部检查被拦截的请求在早期就被挡住了。3.2 五个关键Filter实现与参数说明Filter 1: 输入长度与频率控制Agent面对极端长输入时容易生成超出预期的处理逻辑也意味着更高的token成本。我一般会把输入长度上限和单次调用的参数大小限制放在最前面成本极低、收益稳定class InputLengthFilter(BaseFilter): name input_length def __init__(self, max_chars: int 6000, max_args_keys: int 20): self.max_chars max_chars self.max_args_keys max_args_keys def check(self, ctx: FilterContext) - bool: # 检查提示词长度 if len(ctx.raw_input) self.max_chars: ctx.reason f输入过长: {len(ctx.raw_input)} {self.max_chars} return False # 检查工具参数大小防止单个工具被塞进异常巨大的参数 if len(ctx.tool_args) self.max_args_keys: ctx.reason f参数项过多: {len(ctx.tool_args)} {self.max_args_keys} return False return True参数怎么定没有绝对标准但可以从你的实际场景反推。如果Agent主要用于处理代码仓库文件6000字符一般够用如果涉及长文档阅读可以放宽但建议在上层做分片。重要的是所有阈值显式可配置不要硬编码在业务代码里。Filter 2: 敏感信息与命令模式白名单这是最核心的一道关卡。与其试图枚举所有“危险命令”黑名单思维不如只允许明确的、预先定义好的“安全命令”白名单思维。黑名单永远跟不上模型的花样翻新而白名单在限制能力的同时给了你确定性。import re from pathlib import Path class CommandGuardFilter(BaseFilter): name command_guard def __init__(self, allowed_patterns: List[str], blocked_keywords: List[str]): self.allowed_patterns [re.compile(p, re.IGNORECASE) for p in allowed_patterns] self.blocked_keywords blocked_keywords def check(self, ctx: FilterContext) - bool: if ctx.action_type ! tool_call: return True # 1. 任何参数里的危险关键词直接拒绝 arg_text str(ctx.tool_args) for kw in self.blocked_keywords: if kw in arg_text: ctx.reason f出现危险关键词: {kw} return False # 2. 工具名必须在白名单表达式范围内 tool_name ctx.tool_name.lower() for pat in self.allowed_patterns: if pat.search(tool_name): return True ctx.reason f工具不在白名单: {ctx.tool_name} return False白名单的核心价值不是让模型“不能干坏事”而是让它“只能干你允许的事”。模板建议类似这样ALLOWED_TOOLS [ r^read_file$, r^list_dir$, r^grep_search$, r^python_exec\(\s*[\]safe_mode, # 注意exec 必须搭配更严格策略甚至不建议小白开放 ]很多人会在这里犯一个错误为了让Agent“更智能”把bash、rm、docker等强命令也放进了白名单。这在玩具项目里当然跑得爽但你实际上已经用一条if语句消除了整个Filter Chain的意义。我的建议是第一版白名单宁可少不可松。能用read/grep/HTTP GET解决的事一律不开shell。Filter 3: 危险操作审批拦截有些操作即使符合白名单本身依然需要二次确认。比如“删除某个文件”和“读取某个文件”都是合法的文件操作但前者造成的影响无法撤销。这需要一类专门针对“高影响动作”的审批过滤器class SensitiveActionApproval(BaseFilter): name sensitive_action # 参数里出现这些词元时必须走审批 SENSITIVE_WORDS [delete, remove, rm , drop, truncate, overwrite, shutdown, format] def __init__(self, requires_confirmation: bool True): self.requires_confirmation requires_confirmation def check(self, ctx: FilterContext) - bool: if ctx.action_type llm_output: return True arg_text str(ctx.tool_args).lower() for word in self.SENSITIVE_WORDS: if word in arg_text: if self.requires_confirmation: ctx.reason f敏感操作需确认: {word} # 注意这里不返回 False而是设置一个标记交给上层的用户确认回调 ctx.metadata[need_human_confirm] True ctx.metadata[confirm_key] word return False # 先拦截等待外部确认后重新调用 # 如果关闭确认功能则直接拒绝 ctx.reason f危险操作被硬拦截: {word} return False return True这个Filter的设计思路和白名单不同它关注的不是“能不能执行”而是“执行前有没有人看过一眼”。在生产环境里我习惯把它和一个人工确认回调Webhook或终端交互组合使用高危操作进入Pending状态用户在确认面板里点了允许才放行否则超时自动拒绝。简单粗放一点儿的玩法也可以在终端直接input(确认执行(y/n)? )。Filter 4: LLM输出脱敏这个过滤器常常被遗忘。模型生成的响应里可能带上不该出现在用户终端的敏感文本。脱敏Filter和输入校验逻辑一致但作用对象是输出侧class OutputRedactionFilter(BaseFilter): name output_redaction def __init__(self, secret_patterns: List[str]): self.patterns [re.compile(p) for p in secret_patterns] def check(self, ctx: FilterContext) - bool: if ctx.action_type llm_output: original ctx.llm_output for pat in self.patterns: ctx.llm_output pat.sub([已脱敏], ctx.llm_output) if original ! ctx.llm_output: ctx.metadata[redacted] True return True这里有个小细节脱敏Filter通常永远不会返回False——它不拦截请求纯粹是修改输出内容。这就是一种“只处理、不阻断”的过滤器类型。设计Filter Chain时要注意不是所有Filter都在做拦截有的做改写有的做标注有的做记录。统一接口的好处就在于拦截/改写/记录三类逻辑可以混排在一条链里调用方不需要关心内部差异。Filter 5: 速率与配额控制最后是防止Agent“跑飞”的兜底机制。AI Agent的循环特性决定了它可能在一次任务里连续调用几十次工具。限流过滤器保证同一个会话在单位时间内最多调用N次工具、最多消耗M个token超限强制中断。class RateLimitFilter(BaseFilter): name rate_limit def __init__(self, max_calls_per_minute: int 30): self.max_calls max_calls_per_minute self._call_records [] # 每个元素是时间戳 def check(self, ctx: FilterContext) - bool: if ctx.action_type ! tool_call: return True now time.time() # 移除60秒前的记录 self._call_records [t for t in self._call_records if now - t 60] if len(self._call_records) self.max_calls: ctx.reason f一分钟内调用次数超过 {self.max_calls} return False self._call_records.append(now) return True注意真实的限流Filter要配合持久化存储比如内存加Redis因为Agent可能跑在多个并发请求中如果是单机小项目内存版完全够用。3.3 完整组装示例一个带完整护栏的Agent调用把所有Filter串起来的样子长这样chain FilterChain() chain.add(InputLengthFilter(max_chars6000, max_args_keys20)) chain.add(CommandGuardFilter( allowed_patterns[ r^read_file$, r^list_dir$, r^grep_search$, r^http_get$, ], blocked_keywords[rm , mkfs, :(){, curl.*|sh, base64] )) chain.add(SensitiveActionApproval(requires_confirmationTrue)) chain.add(RateLimitFilter(max_calls_per_minute30)) chain.add(OutputRedactionFilter(secret_patterns[ rsk-[A-Za-z0-9]{20,}, rAKIA[0-9A-Z]{16}, rpassword\s*\s*[\][^\]{6,}[\] ])) # 模拟一次 Agent 执行工具的调用 ctx FilterContext( action_typetool_call, tool_nameread_file, tool_args{path: /home/user/config.yaml}, raw_input请读取配置文件并总结, ) result chain.run(ctx) print(result.allowed, result.reason, result.metadata)运行流程里请求先过InputLength再查工具白名单和危险词再查是否需要人工审批同时被限流器计数最终放行到达工具执行层。如果read_file的目标路径是系统敏感目录SensitiveActionApproval会拦下来如果是正常业务文件则全链放行。整个过程从请求进入Chain到拿到结果通常是毫秒级的开销。这里给小白一个建议先跑通上面这段独立脚本不要急着接大模型。你可以mock一个假Agent随机生成tool_call请求观察Filter Chain怎么拦截之后再接入真实的模型输出。先让安全层独立可信再让Agent变聪明这个顺序千万别搞反。4. 小白最容易踩的5个坑——已替你交过学费了4.1 白名单写得太宽松等于没写我见过有人为了“灵活”把.*直接放进白名单还配上“反正没危险命令”的注释。这是核心安全机制上开了天窗。一旦白名单里有贪婪匹配后面SensitiveActionApproval的拦截逻辑就会被直接绕过——比如bash -c echo hello可以被bash.*命中放行继而变成任何你想执行的命令。凡是正则白名单一律先写单元测试用例默认预期一律拒绝每放行一个模式就配一组正向用例和一组反例别靠肉眼判断“应该没事”。4.2 只过滤输入不过滤输出数据照样裸奔很多初版的Filter Chain只检查了“模型准备调用什么工具”完全不管“模型输出里有什么”。但实际事故率高发区恰恰在输出侧模型搜索了API文档然后把文档里的Bearer Token拼在回答里返回给用户或者模型诚实地说出了某个内部服务名和端口号这些信息出去之后就成了社工的素材。输出侧Filter不用很复杂正则脱敏加一个“危险内容扫描”就能挡住大部分低级泄露。4.3 审批机制形同虚设SensitiveActionApproval返回False之后如果上层代码居然无视allowed状态继续执行工具那这个Filter就是摆设。这种Bug通常出在接入层开发者把“过滤链返回结果”和“是否继续执行工具”解耦了Filter链跑完只是打印了一行日志真正执行工具的代码还是无条件跑了。对策是在工具执行的入口加一个硬断言assert result.allowed, f工具执行被拦截: {result.reason}也就是Filter Chain的结果必须是被执行器强制检查的而不是“建议性”的。这条原则贯穿整个安全架构。4.4 日志只记不审计等于白记Filter Chain的metadata里已经记录了每个Filter的判定结果和耗时但如果你不把这些东西落到独立审计日志里事后出问题根本无从排查。我建议每次被拦截的请求除了写应用日志外还要额外写一份WARN级别的结构化日志包含会话ID、输入摘要、触发Filter名称、拦截原因、模型输出片段。这样既方便调试也能帮你持续迭代过滤器——因为你是真的看得到哪些拦截是误伤、哪些拦得漂亮。拦截比例可以作为白名单紧松程度的调节指标比如一周内拦截率高于30%但用户投诉少说明白名单收得过紧拦截率极低则说明Agent行为早被摸透了需要注意。4.5 全链路超时缺失循环拖垮系统Filter Chain管住了“危险动作”但管不住“耗时不正常的动作”。比如Agent去调一个外部接口由于对方服务无响应requests默认会挂很久或者模型反复自我修正同一工具调用10次。给Chain加一个整体超时是必要的从Chain进入开始计时超过15秒强制拒绝不管中间Filter是否通过。这个超时应该在调用模型生成响应之前就开始计时也就是Agent完整执行一步的总时间。千万不要把超时写在某个Filter内部那样只能保障单个环节保障不了整串链路。5. 进阶思路从玩具级安全到生产级防线5.1 不要迷信“模型本身的安全对齐”很多大模型确实做了安全训练比如拒绝违法请求、避免直接输出危险代码。但Agent场景下模型的安全对齐非常脆弱。原因在于模型的判断是可绕过的——只要换一种方式表达它可能就意识不到“这个操作用了我的工具”模型的行为取决于上下文和工具描述工具一多它的“注意力”被分散发生误判的概率直线上升。像“忽略前述所有指令”这类文本在网页嵌入时模型根本分辨不了哪些是用户需求、哪些是外部注入。所以Filter Chain的存在价值就是用确定性的、非模型的程序逻辑来兜底模型负责“能做”Filter负责“能不能做”。5.2 接入可观测性与告警生产级Filter Chain至少要满足“三种可观测性”日志、指标、追踪。日志输出上面说过了指标层面建议统计每类Filter的拦截次数、耗时、放行率追踪层面则要把一次Agent任务中的多次工具调用串成一个Trace包括每次调用的Filter判定结果。搭建起来不需要重型框架——日志用标准logging指标用一个Counter字典配Prometheus客户端也能凑合追踪如果不上分布式系统用一个会话ID手工串也可以。目的只有一个任何一次误拦截或者漏拦截你都能在十分钟内定位到是哪个Filter、哪个条件、哪次调用导致的。5.3 让过滤规则具备热更新能力规则硬编码的Filter Chain有一个现实痛点每次调整白名单或敏感词表都要改代码、发版。成熟一点的做法是把规则抽到配置中心或者简单的YAML文件里Chain启动时加载、定时刷新。敏感信息这种变化频繁的规则特别适合外置。更激进的方案是让Filter的判定结果回流成训练数据用真实拦截样本持续改进规则。但这一步需要谨慎尤其别让用户提交的内容直接进入规则库——先脱敏再抽取成字符串模式。5.4 分层构建自己的安全体系Filter Chains不是安全唯一的解但它是入手成本最低、见效最快的一块拼图。完整的Agent安全体系还需要权限最小化为Agent准备一个受限系统账户而不是让它跑在你的主用户下、容器隔离关键工具跑在沙箱里文件系统只读挂载、数据分级区分可变数据和不可变数据。Filter Chains管的是“应用层逻辑”容器和系统账户管的是“环境层隔离”两者配合使用而不是二选一。如果你的Agent需要执行bash别犹豫先构建一个不带网络权限、只有临时目录的沙箱环境再让Filter Chain管住沙箱内部的动作。最后再分享一个实操中的小技巧每加一个Filter先写五个必拦的测试用例和五个必放的测试用例再写实现。这个习惯帮我避免了很多“看起来没问题、上线就出事”的Filter。比如新加一个“禁止修改系统配置”的Filter先想清楚用户让Agent改hosts文件必须拦用户让Agent读日志必须放Agent自己写的临时配置必须放——把这些用例固定下来Filter的实现方向就不会跑偏。AI Agent的安全建设没有玄学就是一层层确定性的约束叠出来的。Filter Chains不是银弹但它是每个从走通Agent demo到真正敢放心跑Agent的开发者都值得先建起来的防线。