很多团队把 PII 脱敏、历史压缩、人工确认这些逻辑全塞进一个超长 system prompt或者散落在业务函数里。结果代码改不动、测不了顺序一乱就出 bug。这篇聊中间件middleware——一类在请求进入模型前、响应离开模型后做横切处理的机制。它不是 prompt 里的杂烩而是有顺序、可组合的管道。我们讲透三个典型中间件PII 过滤、对话摘要压缩、人工介入HITL。第一节中间件到底是什么一句话中间件是在请求进模型前、响应出模型后做横切处理的管道节点。它和 Web 框架里的 middleware、Spring 的拦截器是同构的东西——横切关注点cross-cutting concern不该写进业务主干应该抽出来挂在请求-响应的路径上。一个普通函数调用是输入 → 函数 → 输出中间件把它变成输入 → 中间件A → 中间件B → 模型 → 中间件B → 中间件A → 输出。注意请求方向和响应方向是两趟独立的钩子process_request在模型前process_response在模型后。请求钩子request hook在消息送去模型之前改写、校验、脱敏输入响应钩子response hook在模型吐字之后恢复、审计、拦截输出可组合多个中间件按顺序叠成一条栈前一个的输出是后一个的输入这跟把规则写进 prompt本质不同。prompt 里写请勿泄露隐私模型听不听是两码事中间件在管道里强制把明文换成占位符模型这辈子都看不到原始 PII。管道是确定性的工程控制prompt 只是软约束。下面这段是中间件的最小契约和一个按正序进、逆序出的管道实现后面三个中间件都挂在它上面。fromtypingimportCallable,Protocol# 一次请求-响应的上下文中间件可读写它classMiddlewareContext:def__init__(self,messages,metadataNone):self.messagesmessages# 送给模型的消息列表self.metadatametadataor{}# 中间件之间的私人数据self.responseNone# 模型返回后填充# 中间件契约请求钩子 响应钩子classMiddleware(Protocol):defprocess_request(self,ctx:MiddlewareContext)-MiddlewareContext:...defprocess_response(self,ctx:MiddlewareContext)-MiddlewareContext:...# 管道请求正序穿过响应逆序返回classMiddlewarePipeline:def__init__(self,middlewares:list,model_fn:Callable):self.middlewaresmiddlewares self.model_fnmodel_fndefrun(self,ctx:MiddlewareContext)-MiddlewareContext:formwinself.middlewares:# 请求方向正序ctxmw.process_request(ctx)ctx.responseself.model_fn(ctx.messages,ctx.metadata)# 调模型formwinreversed(self.middlewares):# 响应方向逆序ctxmw.process_response(ctx)returnctx逆序返回这一点很重要PII 脱敏在请求方向第一个把明文换掉响应方向它就最后一个把占位符换回来——这样中间所有环节包括模型看到的都是脱敏后的内容。顺序即语义。第二节PII 过滤——进脱敏出恢复PII个人身份信息过滤要解决一个具体问题用户输入里常有手机号、身份证、邮箱你不能把它们送进第三方模型但又得把答案里的真实值返回给用户。解法是两端脱敏-恢复请求方向把敏感字段替换成占位符PII_0把映射存在上下文里随请求带走响应方向再把占位符还原。这样模型自始至终看不到明文合规面和泄露面都大幅缩小。占位符到原文的映射绝不能写进日志也绝不能进模型的上下文——它只活在中件间的私有 metadata 里。importreclassPIIMaskMiddleware:def__init__(self):self.store{}# token - 原文真实系统里按 session 隔离defprocess_request(self,ctx):masked,mappingself._mask(ctx.messages)ctx.messagesmasked ctx.metadata[pii_map]mapping# 随请求一起带走returnctxdefprocess_response(self,ctx):mappingctx.metadata.pop(pii_map,{})ctx.responseself._unmask(ctx.response,mapping)returnctxdef_mask(self,messages):mapping,counter{},0defreplace(m):nonlocalcounter tokenfPII_{counter}mapping[token]m.group(0)counter1returntokenformsginmessages:msg[content]re.sub(r1[3-9]\d{9},replace,msg[content])# 手机号returnmessages,mappingdef_unmask(self,text,mapping):fortoken,originalinmapping.items():texttext.replace(token,original)returntext几个工程要点占位符要可逆且唯一用计数器生成PII_n比固定字符串安全避免不同用户的 PII 串台恢复只在响应出口做模型、下游工具、日志拿到的都是占位符别只靠正则真实生产用 Presidio、Microsoft 识别器或自训 NER地址、姓名、银行卡正则覆盖不全第三节对话摘要压缩——历史太长时的取舍多轮对话一久历史消息会无限涨token 直接爆表还挤占模型的有效注意力。摘要压缩中间件干的事是当历史超过阈值把最旧的若干条交给一个压缩模型生成摘要用一条摘要消息替换它们。这是有损降维不是无损存储。设计时必须回答三个问题压哪些通常最旧的、留哪些最近的 Few-shot 和当前轮、压完往哪放压缩结果要写回持久化历史否则下一轮又从头涨。classSummaryCompressionMiddleware:def__init__(self,max_messages:int,summarizer_fn):self.max_messagesmax_messages# 保留的最近消息条数self.summarizersummarizer_fn# 调用压缩模型的函数defprocess_request(self,ctx):msgsctx.messagesiflen(msgs)self.max_messages:returnctx# 最旧的 (总数 - 上限) 条送去压缩cutlen(msgs)-self.max_messages to_compress,keepmsgs[:cut],msgs[cut:]summaryself.summarizer(to_compress)ctx.messages[{role:system,content:f[历史摘要]{summary}}]keepreturnctxdefprocess_response(self,ctx):returnctx# 压缩不可逆响应方向不做事压缩是单向的所以process_response通常是空实现这点要和 PII 区分开别搞混阈值按 token 算按消息条数拍脑袋不稳一条消息长度差异巨大压缩子调用也要过脱敏否则历史里的隐私会在 summarizer 那一步裸奔这是顺序问题下一节细说它和记忆memory不是一回事memory 是把历史存下来压缩是把历史变小。两者常配合但职责不同。第四节人工介入——在关键步骤踩刹车有些动作不能让模型自己拍板发邮件、转账、删库、对外发消息。这时候需要 HITLHuman-In-The-Loop人工介入在管道或状态机里暂停把决策权交给人人确认后再继续。LangGraph 的interrupt()是这个模式的标杆实现它不是在代码里写个input()卡住线程而是在状态机层面暂停图执行把需要确认的内容交还给调用方外部用Command(resume...)恢复。这意味着暂停可以跨进程、跨网络、带超时、可审计。fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,Commanddefapproval_node(state):# interrupt 暂停图把待确认信息交还调用方decisioninterrupt({action:state[pending_action],payload:state[payload],})return{status:approvedifdecision[approve]elserejected}builderStateGraph(dict)builder.add_node(approval,approval_node)builder.add_edge(START,approval)builder.add_edge(approval,END)graphbuilder.compile()# 第一次调用图在这里暂停返回需要人工确认的内容resultgraph.invoke({pending_action:transfer,payload:{amount:5000}})# 人工在后台确认后用 Command 恢复resultgraph.invoke(Command(resume{approve:True}))断点要落在副作用之前转账节点之前先 approval而不是转完再问要不要撤回恢复必须带上下文人确认了什么、谁确认的、什么时候这些要进审计日志图的 interrupt 优于同步阻塞如果只是简单阻塞中间件在process_request置blocked_pending_human也行但可恢复性和持久化远不如 interrupt第五节组合与顺序以及反模式中间件最容易被低估的一点顺序即语义。它们叠成一条栈前一个的输出是后一个的输入换个顺序行为完全不同。最经典的坑是日志中间件和 PII 中间件的位置关系。如果日志在脱敏之前运行明文 PII 就写进了日志——合规直接爆雷。正确顺序是脱敏先行日志后行日志里只有占位符。脱敏要覆盖压缩的子调用压缩中间件会调一个子模型去总结历史如果 PII 脱敏只包住主调用没包住子调用历史里的隐私会在压缩那一步裸奔。要么让脱敏也包住 summarizer要么保证压缩发生在脱敏之后、且子调用复用同一套脱敏HITL 要挡在副作用最前面审批节点必须早于任何写操作顺序错了就是先斩后奏响应方向逆序恢复PII 恢复必须是最后一个响应钩子否则恢复出的明文会被后面的中间件比如再次日志的环节看到反例与反模式这些是我在 code review 里打回最多的写法反模式一巨型 prompt把脱敏规则、压缩指令、校验逻辑、人工确认提示全写进一个超长 system prompt。模型听不听随缘最关键的是没法单测——你永远无法确认脱敏这个关注点真的生效了只能靠肉眼看输出反模式二逻辑散落业务代码在 handler 里随手写if 1 in text: hide()脱敏、压缩、校验各写各的顺序藏在调用时序里新人根本看不到全貌一改就串反模式三顺序随意先打日志再脱敏导致明文落盘先压缩再脱敏导致子调用泄露恢复明文的钩子排在审计钩子之前导致审计看到明文反模式四HITL 写死同步阻塞在推理路径里直接input()或同步等待没法在异步服务、分布式部署里恢复重启即丢状态也无法超时和审计回到本质中间件的价值不是多一层包装而是把横切关注点变成可测试、可组合、顺序明确的单元。当脱敏是一个中间件、压缩是一个中间件、人工确认是一个中间件你就能像搭积木一样组合它们每个都能单独写单测组合起来行为可预测。这才是 AI 工程化该有的样子。