中间件机制：PII 过滤、摘要压缩与人工介入

from typing import Callable, Protocol

# 一次请求-响应的上下文，中间件可读写它
class MiddlewareContext:
    def __init__(self, messages, metadata=None):
        self.messages = messages        # 送给模型的消息列表
        self.metadata = metadata or {}  # 中间件之间的私人数据
        self.response = None            # 模型返回后填充

# 中间件契约：请求钩子 + 响应钩子
class Middleware(Protocol):
    def process_request(self, ctx: MiddlewareContext) -> MiddlewareContext: ...
    def process_response(self, ctx: MiddlewareContext) -> MiddlewareContext: ...

# 管道：请求正序穿过，响应逆序返回
class MiddlewarePipeline:
    def __init__(self, middlewares: list, model_fn: Callable):
        self.middlewares = middlewares
        self.model_fn = model_fn

    def run(self, ctx: MiddlewareContext) -> MiddlewareContext:
        for mw in self.middlewares:            # 请求方向：正序
            ctx = mw.process_request(ctx)
        ctx.response = self.model_fn(ctx.messages, ctx.metadata)  # 调模型
        for mw in reversed(self.middlewares):  # 响应方向：逆序
            ctx = mw.process_response(ctx)
        return ctx

import re

class PIIMaskMiddleware:
    def __init__(self):
        self.store = {}   # token -> 原文，真实系统里按 session 隔离

    def process_request(self, ctx):
        masked, mapping = self._mask(ctx.messages)
        ctx.messages = masked
        ctx.metadata["pii_map"] = mapping   # 随请求一起带走
        return ctx

    def process_response(self, ctx):
        mapping = ctx.metadata.pop("pii_map", {})
        ctx.response = self._unmask(ctx.response, mapping)
        return ctx

    def _mask(self, messages):
        mapping, counter = {}, 0
        def replace(m):
            nonlocal counter
            token = f"<PII_{counter}>"
            mapping[token] = m.group(0)
            counter += 1
            return token
        for msg in messages:
            msg["content"] = re.sub(r"1[3-9]\d{9}", replace, msg["content"])  # 手机号
        return messages, mapping

    def _unmask(self, text, mapping):
        for token, original in mapping.items():
            text = text.replace(token, original)
        return text

class SummaryCompressionMiddleware:
    def __init__(self, max_messages: int, summarizer_fn):
        self.max_messages = max_messages   # 保留的最近消息条数
        self.summarizer = summarizer_fn    # 调用压缩模型的函数

    def process_request(self, ctx):
        msgs = ctx.messages
        if len(msgs) <= self.max_messages:
            return ctx
        # 最旧的 (总数 - 上限) 条送去压缩
        cut = len(msgs) - self.max_messages
        to_compress, keep = msgs[:cut], msgs[cut:]
        summary = self.summarizer(to_compress)
        ctx.messages = [{"role": "system", "content": f"[历史摘要] {summary}"}] + keep
        return ctx

    def process_response(self, ctx):
        return ctx   # 压缩不可逆，响应方向不做事

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command

def approval_node(state):
    # interrupt 暂停图，把待确认信息交还调用方
    decision = interrupt({
        "action": state["pending_action"],
        "payload": state["payload"],
    })
    return {"status": "approved" if decision["approve"] else "rejected"}

builder = StateGraph(dict)
builder.add_node("approval", approval_node)
builder.add_edge(START, "approval")
builder.add_edge("approval", END)
graph = builder.compile()

# 第一次调用：图在这里暂停，返回需要人工确认的内容
result = graph.invoke({"pending_action": "transfer", "payload": {"amount": 5000}})

# 人工在后台确认后，用 Command 恢复
result = graph.invoke(Command(resume={"approve": True}))