Streaming bypasses guardrails/middleware

快速结论：当你在 LangChain 的 create_agent 中启用输出侧中间件（如 PIIMiddleware）并用 agent.stream(..., stream_mode="messages") 流式输出时，模型返回的 token 会在中间件生效之前就直接打印给用户，导致 Streaming bypasses guardrails/middleware。优先排查你的流式消费逻辑是否绕过了中间件的后处理。

适用环境：Issue 中确认涉及 langchain、langchain-core、LangGraph；复现环境为 macOS（/Users/… 路径、Homebrew Python），Python 3.10.16，模型 gpt-4o-mini，使用 PIIMiddleware 配合 create_agent 的 stream_mode="messages"。CUDA、显卡未在 Issue 中提及，无需确认。

最快修复方案：暂无确认的一步修复方案。Issue 中维护者明确指出中间件节点只在整段流式输出结束后才执行，当前 Middleware 架构下没有逐 token 执行中间件的直接方式。可优先尝试 Issue 评论中提供的“伪流式”变通方案：先用 agent.invoke() 非流式获取完整回复，确保其通过 PII 中间件，再从最后一条 message 用 tokenizer 解码后逐块打印。

注意事项：伪流式方案牺牲了真正的流式 UX（首 token 延迟等于整段生成时间），且 Issue 中未标记为官方验证修复，仅作为 workaround 提出。根本修复需要在 LangGraph / Middleware 架构层面支持流式感知，Issue 已被标记为需要重大架构变更而保持开放，后续有 PR #35470 尝试修复但尚未确认落地。

问题场景

用户在 LangChain 中使用 create_agent 创建 agent，并挂载了输出侧保护中间件 PIIMiddleware("email", strategy="block", apply_to_input=False, apply_to_output=True)。随后通过 agent.stream({...}, stream_mode="messages") 逐 token 消费模型输出并直接 print。此时模型生成的邮件地址等 PII 内容会在中间件拦截或脱敏之前就暴露给终端用户，保护形同虚设。

报错原文

Streaming bypasses guardrails/middleware

Sure! Here's a funny greeting incorporating your email:

---

🎉 Hey there, awesome human! 🎉

Just popping in to say, if you ever need a digital superhero, you can reach me at my secret lair: **john.doe@example.com**! But beware, sending a message may result in spontaneous laughter and a 100% increase in good vibes!

Chirp, chirp, and let the email adventures begin! 🦸‍♂️💻

---

Feel free to tweak it however you like!Traceback (most recent call last):
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/tests/test_hiddenlayer.py", line 79, in <module>
    test_streaming_malicious()
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/tests/test_hiddenlayer.py", line 64, in test_streaming_malicious
    for msg, metadata in agent.stream(
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/.venv/lib/python3.10/site-packages/langgraph/pregel/main.py", line 2646, in stream
    for _ in runner.tick(
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/.venv/lib/python3.10/site-packages/langgraph/pregel/_runner.py", line 258, in tick
    _panic_or_proceed(
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/.venv/lib/python3.10/site-packages/langgraph/pregel/_runner.py", line 520, in _panic_or_proceed
    raise exc
  File "/Users/sidhu/hiddenlayer-langchain-guardrails/.venv/lib/python3.10/site-packages/langgraph/pregel/_executor.py", line 80, in done
    task.result()
  File "/opt/homebrew/Cellar/python@3.10/3.10.16/Frameworks/Python.framework/Versions/3.10/lib/python3.10/concurrent/futures/_base.py", line 451, in result
    return self.__get_result()
  File "/opt/homebrew/Cellar/python@3.10/3.10.16/Frameworks/Python.framework/Versions/3.10/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/opt/homebrew/Cellar/python@3.10/3.10.16/Frameworks/Python.framework/Versions/3.10/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
...

原因分析

最可能的原因是 Middleware 的执行时序与流式输出不匹配：在现有 AgentMiddleware 架构下，中间件节点（包括 after_model 一类的输出后处理）只有在整段流式内容全部 yield 回用户之后才会被触发。因此 stream_mode="messages" 下逐 token 抵达调用方的原始输出，会在 PII 等敏感内容被 PIIMiddleware 拦截或脱敏之前就直接展示给用户。维护者在评论中确认这是架构层面的问题——任何作为“事后处理步骤”运行的中间件系统，在流式场景下都会遭遇同样的绕过问题。

环境排查

确认 Python 版本（Issue 中为 3.10.16，Homebrew 安装）。
确认 langchain 与 langchain-core 版本；Issue 提到“更新到最新稳定版也未解决”，因此仅升级可能无效。
确认 LangGraph 版本及 agent.stream 的调用栈是否为 langgraph/pregel/main.py → _runner.py → _executor.py 路径。
确认所用模型（Issue 中为 gpt-4o-mini）以及是否使用了 fake streaming model 进行本地复现。
确认中间件配置：strategy="block"、apply_to_output=True 是否按预期传入。
确认流式消费方是否在 for msg, metadata in agent.stream(...) 循环内直接 print(msg.content)，这会绕过任何后置中间件。

解决步骤

先确认问题可稳定复现：使用 Issue 中的最小复现代码，观察 PII（如邮箱）是否在中间件被执行前就出现在终端输出。
若需立即降低泄露风险，可优先尝试 Issue 评论中的“伪流式”变通方案，思路如下：
- 改用 agent.invoke() 非流式获取完整结果，确保整段文本经过 PIIMiddleware 处理。
- 从 output["messages"][-1] 取出最终内容。
- 用与模型匹配的 tokenizer（如 tiktoken.encoding_for_model("gpt-4o-mini")）将内容编码为 token，再逐 token 解码并用小幅 time.sleep 模拟流式打印。
若必须在真实流式下做防护，可考虑在客户端/消费侧增加带缓冲窗口的增量扫描逻辑：累积 N 个 token，扫描缓冲区，干净则 flush，命中敏感模式则阻断。注意此方案 Issue 中仅作为思路提出，未在 LangChain 内实现或验证。
关注 Issue 后续的架构级修复（评论中提到 PR #35470 尝试解决，需在 sync/async 流式及不同输出处理策略下验证是否仍有暴露面）。
在问题未修复前，避免在生产环境仅依赖流式 + 输出中间件来保证 PII 保护，必要时改用非流式路径或增加独立的输出审查层。

验证方法

用原复现代码重新运行：若终端在整个响应完成前不再出现 john.doe@example.com 等被保护的 PII 内容，或该内容被阻断/脱敏后才出现，说明防护生效。若采用伪流式方案，可通过日志确认打印内容来自中间件处理后的 output["messages"][-1]，并验证同步与异步流式、不同 strategy 下均无原始敏感 token 提前泄漏。