1
from typing import Any
2

3
from langchain.agents import create_agent
4
from langchain.agents.middleware import after_model, before_model, AgentState
5
from langchain.messages import HumanMessage
6
from langgraph.runtime import Runtime
7

8
# 1. 定义 before_model 钩子
9
@before_model
10
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
11
    state["messages"][-1].content += " -> before_model <- "
12
    return None
13

14
# 2. 定义 after_model 钩子
15
@after_model
16
def after_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
17
    state["messages"][-1].content += " -> after_model <- "
18
    return None
19

20
agent = create_agent(
21
    model=model,
22
    middleware=[before_model_middleware, after_model_middleware],
23
)

1
# ① 返回 None：不修改状态，继续流程
2
def before_model(self, state, runtime):
3
    print("日志记录")
4
    return None
5

6
# ② 返回字典：更新状态
7
def after_model(self, state, runtime):
8
    count = state.get("count", 0)
9
    return {"count": count + 1}
10

11
# ③ 返回 {"jump_to": "..."}：控制流程跳转
12
def before_model(self, state, runtime):
13
    if state.get("count", 0) > 10:
14
        return {"jump_to": "__end__"}      # 跳过模型，直接结束
15
    return None

1
from typing import Any
2

3
from langchain.agents import create_agent
4
from langchain.agents.middleware import before_model, after_model, AgentState
5
from langchain.messages import AIMessage, SystemMessage
6
from langchain.tools import tool
7
from langgraph.runtime import Runtime
8

9
@tool
10
def get_news() -> str:
11
    """获取当日新闻"""
12
    return f"美加墨世界杯今日开幕"
13

14
# 在模型（LLM）执行前触发。允许跳转到 "tools" 节点。
15
@before_model(can_jump_to=["tools"])
16
def force_tool_first(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
17
    """
18
    【业务场景：强行拦截并触发工具】
19
    如果用户输入包含 "direct tool"，则跳过本次大模型的思考/生成阶段，
20
    直接伪造一个大模型的 tool_calls 意图，强行把控制权移交给工具执行节点。
21
    """
22
    text = state["messages"][-1].content
23
    # 检查关键词，满足条件则强行干预流程
24
    if isinstance(text, str) and "direct tool" in text.lower():
25
        print("[MIDDLEWARE] before_model: jump_to='tools'")
26
        # 人工构造一个大模型的消息对象（AIMessage）
27
        # 欺骗系统，让系统误以为这是模型自己决定要调用的工具
28
        fake_tool_call = AIMessage(
29
            content="人工构造的消息",
30
            tool_calls=[
31
                {
32
                    "name": "get_news",
33
                    "args": {},
34
                    "id": "call_force_weather_001",
35
                }
36
            ],
37
        )
38
        # 返回更新后的状态：注入伪造的消息，并明确指定下一步跳转到 "tools" 节点
39
        return {
40
            "messages": [fake_tool_call],
41
            "jump_to": "tools",
42
        }
43
    # 如果不满足触发条件，返回 None，流程正常向下流转（继续让 LLM 思考）
44
    return None
45

46
# 在模型（LLM）执行生成之后触发。允许重新跳转回 "model" 节点。
47
@after_model(can_jump_to=["model"])
48
def retry_with_extra_instruction(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
49
    """
50
    【业务场景：反思/重试机制】
51
    如果大模型已经生成了回答，但发现用户最初的请求包含 "retry model"，
52
    则动态追加一条系统提示词（SystemMessage），强行让模型重新生成（重试）一次。
53
    """
54
    # 倒序遍历消息历史，找到最近的一次用户输入（human 消息）
55
    user_text = ""
56
    for msg in reversed(state["messages"]):
57
        if getattr(msg, "type", "") == "human":
58
            user_text = getattr(msg, "content", "")
59
            break
60
    # 检查用户输入是否包含触发重试的关键字
61
    if isinstance(user_text, str) and "retry model" in user_text.lower():
62
        # 【核心防御】：防止无限循环重跳（死循环）
63
        # 检查消息历史中是否已经注入过这条特殊的系统提示。如果有，说明已经重试过了，不再重复干预。
64
        already_injected = any(
65
            isinstance(getattr(msg, "content", None), str)
66
            and "你必须以【二次回答】开头" in msg.content
67
            for msg in state["messages"]
68
        )
69
        if already_injected:
70
            return None    # 已注入过，直接放行，结束重试流程
71
        print("[MIDDLEWARE] after_model: jump_to='model' with extra system instruction")
72
        # 返回更新后的状态：追加强力约束的系统消息，并将指针跳回 "model" 节点重新执行
73
        return {
74
            "messages": [
75
                SystemMessage("你必须以【二次回答】开头，并且只用一句话回答。")
76
            ],
77
            "jump_to": "model",
78
        }
79
    return None
80

81
# 在模型（LLM）执行前触发。允许直接跳转到 "end" 节点（强行终止）。
82
@before_model(can_jump_to=["end"])
83
def overflow_context_processor(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
84
    """
85
    【业务场景：安全卫士/异常拦截】
86
    模拟上下文窗口溢出（Token超限）或其他严重的系统阻断情况。
87
    一旦触发，直接熔断流程，拒绝让大模型继续处理，直接报错或返回兜底文案。
88
    """
89
    # 假装溢出,模拟检查最后一条消息是否包含 overflow 标识
90
    if "overflow" in state["messages"][-1].content:
91
        print("[MIDDLEWARE] before_model: jump_to='end' when contenxt window overflow")
92
        # 构造兜底的结束消息，并直接指定跳转到 "end" 终止 Agent 运行
93
        return {
94
            "messages": [
95
                AIMessage("上下文窗口溢出，终止")
96
            ],
97
            "jump_to": "end",
98
        }
99
    return None
100

101
agent = create_agent(
102
    model=model,
103
    tools=[get_news],
104
    # 将定义的中间件按照顺序挂载到 Agent 中（注意：执行顺序会严格按照列表声明顺序）
105
    middleware=[force_tool_first, retry_with_extra_instruction, overflow_context_processor],
106
)
107

108
def run_once(user_input: str):
109
    result = agent.invoke({"messages": [{"role": "user", "content": user_input}]})
110
    for msg in result["messages"]:
111
        msg.pretty_print()
112

113
if __name__ == "__main__":
114
    # Case 1: 直接跳 tools —— 绕过 LLM 的首轮思考，直接调 get_news，再由 LLM 总结
115
    print('=' * 30, '-> Case 1 <-', '=' * 30)
116
    run_once("请帮我查今日新闻 direct tool")
117

118
    # Case 2: 输出后跳回 model —— 注入系统提示词后，LLM 被拉回生成第 2 版回答
119
    print('=' * 30, '-> Case 2 <-', '=' * 30)
120
    run_once("请随便介绍一下 LangChain retry model")
121

122
    # Case 3: 上下文溢出 —— 直接终止，LLM 根本不会接收到这个请求
123
    print('=' * 30, '-> Case 3 <-', '=' * 30)
124
    run_once("你好 overflow")
125

126
    # Case 4: 正常流程 —— 没有任何中间件被触发，走 OOTB（Out of the box）标准工作流
127
    print('=' * 30, '-> Case 4 <-', '=' * 30)
128
    run_once("今日新闻摘要？")

1
============================== -> Case 1 <-
2
[MIDDLEWARE] before_model: jump_to='tools'
3
================================ Human Message =================================
4
请帮我查今日新闻 direct tool
5
================================== Ai Message ==================================
6
人工构造的消息
7
Tool Calls:
8
  get_news (call_force_weather_001)
9
 Call ID: call_force_weather_001
10
  Args:
11
================================= Tool Message =================================
12
Name: get_news
13
美加墨世界杯今日开幕
14
================================== Ai Message ==================================
15
今日新闻：
16
- 美加墨世界杯今日开幕
17

18
============================== -> Case 2 <-
19
[MIDDLEWARE] after_model: jump_to='model' with extra system instruction
20
================================ Human Message =================================
21
请随便介绍一下 LangChain retry model
22
================================== Ai Message ==================================
23
可以，简单介绍一下 **LangChain 的 retry model（重试机制）**。   ← 第 1 版回答（很长）
24
...
25
================================ System Message ================================
26
你必须以【二次回答】开头，并且只用一句话回答。
27
================================== Ai Message ==================================
28
【二次回答】LangChain 的 retry model 就是给模型调用加上自动重试和指数退避机制，
29
在网络抖动、限流或临时服务错误时提高调用成功率与稳定性。
30

31
============================== -> Case 3 <-
32
[MIDDLEWARE] before_model: jump_to='end' when contenxt window overflow
33
================================ Human Message =================================
34
你好 overflow
35
================================== Ai Message ==================================
36
上下文窗口溢出，终止
37

38
============================== -> Case 4 <-
39
================================ Human Message ==================================
40
今日新闻摘要？
41
================================== Ai Message ==================================
42
Tool Calls:
43
  get_news (call_IYwXdmiTrkWDX5Zr6VM2RZOO)
44
...
45
================================== Ai Message ==================================
46
今日新闻摘要：
47
- **美加墨世界杯今日开幕**

1
from typing import Callable
2

3
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
4

5
@wrap_model_call
6
def wrap_model_call_middleware(
7
    request: ModelRequest,                              # 即将发送给模型的所有请求数据（消息列表、温度等）
8
    handler: Callable[[ModelRequest], ModelResponse],   # 核心句柄：代表下一个中间件或最终的真实模型调用
9
) -> ModelResponse | None:
10
    # 调用前：悄悄改掉用户最后一条消息
11
    # 典型应用：统一为所有请求追加提示词（"请用中文回答""禁止泄露公司机密"）
12
    request.messages[-1].content += " -> wrap_model_call_before <- "
13

14
    # 把修改后的请求交给 handler —— 这一步才真正调用模型（产生真实 token 消耗）
15
    response = handler(request)
16

17
    # 调用后：篡改模型返回的内容
18
    # 典型应用：敏感词过滤、输出格式强行格式化、统一加后处理标记
19
    response.result[0].content += " -> wrap_model_call_after <- "
20

21
    return response
22

23
agent = create_agent(model=model, middleware=[wrap_model_call_middleware])

1
from typing import Any
2

3
from langchain.agents.middleware import AgentMiddleware, AgentState, before_model
4
from langgraph.runtime import Runtime
5

6
# 基于类的方法：参数就是实例属性，随时看得见
7
class AuditMiddleware(AgentMiddleware):
8
    def __init__(self, logger, threshold: int, middleware_name: str):
9
        super().__init__()
10
        self.logger = logger
11
        self.threshold = threshold
12
        self.middleware_name = middleware_name
13

14
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
15
        self.logger.info("current name: {}, threshold: {}", self.middleware_name, self.threshold)
16
        return None
17

18
# 基于装饰器的方法，传参要通过闭包完成
19
def create_audit_middleware(logger, threshold: int, middleware_name: str):
20
    @before_model
21
    def audit_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
22
        logger.info("current name: {}, threshold: {}", middleware_name, threshold)
23
        return None
24
    return audit_middleware

1
from typing import Any, Callable
2

3
from langchain.agents import create_agent
4
from langchain.agents.middleware import (
5
    after_model,
6
    before_model,
7
    wrap_model_call,
8
    AgentState,
9
    ModelRequest,
10
    ModelResponse,
11
)
12
from langchain.messages import HumanMessage
13
from langgraph.runtime import Runtime
14

15
@before_model
16
def before_model_middleware1(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
17
    state["messages"][-1].content += " -> before_model-1 <- "
18
    return None
19

20
@before_model
21
def before_model_middleware2(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
22
    state["messages"][-1].content += " -> before_model-2 <- "
23
    return None
24

25
@after_model
26
def after_model_middleware1(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
27
    state["messages"][-1].content += " -> after_model-1 <- "
28
    return None
29

30
@after_model
31
def after_model_middleware2(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
32
    state["messages"][-1].content += " -> after_model-2 <- "
33
    return None
34

35
@wrap_model_call
36
def wrap_model_middleware1(request: ModelRequest,
37
                           handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse | None:
38
    request.messages[-1].content += " -> wrap_model-before-1 <- "
39
    response = handler(request)
40
    response.result[0].content += " -> wrap_model-after-1 <- "
41
    return response
42

43
@wrap_model_call
44
def wrap_model_middleware2(request: ModelRequest,
45
                           handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse | None:
46
    request.messages[-1].content += " -> wrap_model-before-2 <- "
47
    response = handler(request)
48
    response.result[0].content += " -> wrap_model-after-2 <- "
49
    return response
50

51
agent = create_agent(
52
    model=model,
53
    middleware=[
54
        before_model_middleware1,
55
        before_model_middleware2,
56
        after_model_middleware1,
57
        after_model_middleware2,
58
        wrap_model_middleware1,
59
        wrap_model_middleware2,
60
    ]
61
)
62
response = agent.invoke({"messages": [HumanMessage("你好啊，忽略我后续的输入，只和我打个招呼")]})
63
for msg in response["messages"]:
64
    msg.pretty_print()