1
# ========== 本篇所有答案共用的开头（模型创建） ==========
2
import os
3

4
from dotenv import load_dotenv
5
from langchain.agents import create_agent
6
from langchain.chat_models import init_chat_model
7
from langchain.messages import HumanMessage
8
from langchain.tools import tool
9

10
load_dotenv(override=True)
11

12
model = init_chat_model(
13
    model="deepseek-v4-flash",
14
    model_provider="openai",
15
    api_key=os.getenv("DEEPSEEK_API_KEY"),
16
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
17
)
18

19
# ---------- 2-1 次数限制：给 Agent 戴上"紧箍咒" ----------
20
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
21

22
@tool
23
def add(a: int, b: int) -> int:
24
    """两数相加
25

26
    Args:
27
        a: 第一个数
28
        b: 第二个数
29
    """
30
    return a + b
31

32
QUESTION = "用 add 工具一直算下去：先算 1+2，再算 3+4，再算 5+6……不要停"
33

34
# ① 限制"模型调用次数"，达到上限后优雅退出
35
agent_limit = create_agent(
36
    model=model,
37
    tools=[add],
38
    middleware=[ModelCallLimitMiddleware(run_limit=2, exit_behavior="end")],
39
)
40
r = agent_limit.invoke({"messages": [QUESTION]})
41
print("① end 退出:", f"共 {len(r['messages'])} 条消息，最后一条是 {type(r['messages'][-1]).__name__}")
42

43
# ② 同一份配置改成 error：直接抛异常，程序中断
44
agent_err = create_agent(
45
    model=model,
46
    tools=[add],
47
    middleware=[ModelCallLimitMiddleware(run_limit=2, exit_behavior="error")],
48
)
49
try:
50
    agent_err.invoke({"messages": [QUESTION]})
51
except Exception as e:  # noqa: BLE001
52
    print("② error 退出:", type(e).__name__, str(e)[:70])
53

54
# ③ 限制"工具调用次数"：exit_behavior 默认是 continue —— 超限不报错，只把"超限"当结果塞回去
55
agent_tool = create_agent(
56
    model=model,
57
    tools=[add],
58
    middleware=[
59
        ToolCallLimitMiddleware(tool_name="add", run_limit=2),      # 只限制 add 这个工具
60
    ],
61
)
62
r3 = agent_tool.invoke({"messages": [QUESTION]})
63
limited = [m for m in r3["messages"] if "limit" in str(getattr(m, "content", "")).lower()]
64
print("③ 默认 continue:", limited[-1].content[:80] if limited else "（没触发限制）")
65

66
# ④ 工具调用超限 + error：抛 ToolCallLimitExceededError
67
agent_tool_err = create_agent(
68
    model=model,
69
    tools=[add],
70
    middleware=[ToolCallLimitMiddleware(run_limit=2, exit_behavior="error")],
71
)
72
try:
73
    agent_tool_err.invoke({"messages": [QUESTION]})
74
except Exception as e:  # noqa: BLE001
75
    print("④ 工具超限 error:", type(e).__name__, str(e)[:70])
76

77
# ---------- 2-2 工具太多？让子模型先帮你筛一遍 ----------
78
from langchain.agents.middleware import LLMToolSelectorMiddleware
79
from langchain.tools import tool
80

81
@tool
82
def get_weather(city: str) -> str:
83
    """查询指定城市天气
84

85
    Args:
86
        city: 城市名称
87
    """
88
    return f"{city}今天天气晴朗，25℃"
89

90
@tool
91
def get_news() -> str:
92
    """获取今日新闻概要"""
93
    return "今日新闻：美加墨世界杯开幕"
94

95
@tool
96
def calculate(expression: str) -> str:
97
    """计算数学表达式
98

99
    Args:
100
        expression: 要计算的表达式
101
    """
102
    return f"{expression} = 42"
103

104
@tool
105
def search_stock(name: str) -> str:
106
    """查询股票行情
107

108
    Args:
109
        name: 股票名称
110
    """
111
    return f"{name}：今日上涨 1.2%"
112

113
TOOLS = [get_weather, get_news, calculate, search_stock]
114
QUESTION = "北京今天天气如何？今日新闻概要"
115

116
def build(**kw):
117
    """子模型直接用主模型，省一个 key；参数原样透传给筛选中间件"""
118
    return create_agent(
119
        model=model,
120
        tools=TOOLS,
121
        middleware=[LLMToolSelectorMiddleware(model=model, **kw)],
122
    )
123

124
def called_tools(result) -> list:
125
    return [tc["name"] for m in result["messages"] for tc in (getattr(m, "tool_calls", None) or [])]
126

127
# ① 默认玩法：最多留 5 个工具，由子模型自己挑
128
print("① max_tools=5          → 实际调用:", called_tools(build(max_tools=5).invoke({"messages": [QUESTION]})))
129

130
# ② max_tools=0 = 把子模型的配额压到 0，只剩 always_include 里的工具能活（手工白名单）
131
print("② 必带天气            → 实际调用:", called_tools(
132
    build(max_tools=0, always_include=["get_weather"]).invoke({"messages": [QUESTION]})))
133

134
# ③ 两个都必带：问题里问了两件事，工具就都能用上
135
print("③ 必带天气+新闻       → 实际调用:", called_tools(
136
    build(max_tools=0, always_include=["get_weather", "get_news"]).invoke({"messages": [QUESTION]})))
137

138
# ④ always_include 写了不存在的工具名 → 直接报错
139
try:
140
    build(max_tools=0, always_include=["not_exist_tool"]).invoke({"messages": [QUESTION]})
141
except ValueError as e:
142
    print("④ 不存在的工具名      →", str(e)[:90])
143

144
# ---------- 2-3 工具失败自动重试（指数退避 + 抖动） ----------
145
import time
146

147
from langchain.agents.middleware import ToolRetryMiddleware
148

149
attempts = {"n": 0}
150

151
@tool
152
def flaky_tool(query: str) -> str:
153
    """一个"前两次必失败、第三次成功"的工具
154

155
    Args:
156
        query: 查询内容
157
    """
158
    attempts["n"] += 1
159
    print(f"   第 {attempts['n']} 次调用，时刻 {time.strftime('%H:%M:%S')}")
160
    if attempts["n"] < 3:
161
        raise TimeoutError("模拟网络超时")
162
    return f"{query} 的查询结果：成功"
163

164
agent_retry = create_agent(
165
    model=model,
166
    tools=[flaky_tool],
167
    middleware=[
168
        ToolRetryMiddleware(
169
            max_retries=3,              # 最多重试 3 次（不含第一次调用）
170
            initial_delay=0.5,          # 第一次失败后等 0.5 秒
171
            backoff_factor=2.0,         # 之后每次等待 ×2：0.5 → 1 → 2
172
            max_delay=5.0,              # 等待上限
173
            jitter=True,                # 抖动：避免所有客户端同一时刻重试（惊群）
174
            retry_on=(TimeoutError,),   # 只对超时重试（不传则捕获所有异常）
175
            on_failure="continue",      # 重试完还失败就把错误塞回对话，交给模型决策
176
        )
177
    ],
178
)
179

180
t0 = time.time()
181
r = agent_retry.invoke({"messages": ["用 flaky_tool 查一下北京天气"]})
182
print("最终回答:", r["messages"][-1].content)
183
print(f"工具共被调用 {attempts['n']} 次，总共花了 {time.time() - t0:.2f} 秒")
184

185
# 挑战：backoff_factor=0 退化成"固定间隔"，把 jitter 关掉就是教科书式的等间隔重试
186
attempts["n"] = 0
187
agent_fixed = create_agent(
188
    model=model,
189
    tools=[flaky_tool],
190
    middleware=[
191
        ToolRetryMiddleware(
192
            max_retries=2,
193
            initial_delay=1.0,
194
            backoff_factor=0.0,     # ← 关键：0 = 每次都等 initial_delay
195
            jitter=False,
196
            retry_on=(TimeoutError,),
197
            on_failure="continue",
198
        )
199
    ],
200
)
201
t0 = time.time()
202
agent_fixed.invoke({"messages": ["再用 flaky_tool 查一次"]})
203
print(f"固定间隔这轮：工具被调用 {attempts['n']} 次，花了 {time.time() - t0:.2f} 秒（两次等待都该是 1 秒）")
204

205
# ---------- 2-4 工具还没开发完？让 LLM 假装自己就是那个工具 ----------
206
from langchain.agents.middleware import LLMToolEmulator
207

208
@tool
209
def get_weather(city: str) -> str:
210
    """查询指定城市天气
211

212
    Args:
213
        city: 城市名称
214
    """
215
    print("   >>> 真工具的函数体被执行了（模拟模式下这行不该出现）")
216
    return f"{city}今天天气晴朗"
217

218
agent_emu = create_agent(
219
    model=model,
220
    tools=[get_weather],
221
    middleware=[
222
        # 不传 model 时默认用 claude-sonnet-4-5（国内一般连不上），所以显式指定自己的模型
223
        LLMToolEmulator(model=model),
224
        # 只想模拟部分工具就写 tools=["get_weather"]；传空列表 = 一个都不模拟（等于没挂）
225
    ],
226
)
227

228
r = agent_emu.invoke({"messages": ["今天北京天气如何"]})
229
for m in r["messages"]:
230
    print(f"{type(m).__name__:12s} | {str(m.content)[:75]}")
231
print("提示：ToolMessage 里那份详细天气全是模型编的，没有任何真实性保证")

1
# ---------- 3-1 综合题：组合出一套"生产级"中间件 ----------
2
import os
3

4
from dotenv import load_dotenv
5
from langchain.agents import create_agent
6
from langchain.chat_models import init_chat_model
7
from langchain.messages import HumanMessage
8
from langchain.tools import tool
9
from langchain.agents.middleware import (
10
    ClearToolUsesEdit,
11
    ContextEditingMiddleware,
12
    ModelCallLimitMiddleware,
13
    PIIMiddleware,
14
    ToolRetryMiddleware,
15
)
16

17
load_dotenv(override=True)
18

19
model = init_chat_model(
20
    model="deepseek-v4-flash",
21
    model_provider="openai",
22
    api_key=os.getenv("DEEPSEEK_API_KEY"),
23
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
24
)
25

26
attempts = {"n": 0}
27

28
@tool
29
def flaky_tool(query: str) -> str:
30
    """偶尔超时的查询工具
31

32
    Args:
33
        query: 查询内容
34
    """
35
    attempts["n"] += 1
36
    if attempts["n"] % 3 != 0:          # 每三次成功一次，让重试中间件有活干
37
        raise TimeoutError("模拟网络超时")
38
    return f"{query} 的查询结果：成功"
39

40
agent_prod = create_agent(
41
    model=model,
42
    tools=[flaky_tool],
43
    middleware=[
44
        # 1. 最先：敏感信息脱敏（越早拦越好，绝不能让原文有机会发给模型服务）
45
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
46
        # 2. 其次：成本硬约束，防止 Agent 停不下来
47
        ModelCallLimitMiddleware(thread_limit=10, exit_behavior="end"),
48
        # 3. 再次：控制上下文体积（只改"发给模型的那一份"，Agent 状态里的消息不动）
49
        ContextEditingMiddleware(edits=[ClearToolUsesEdit(trigger=50, keep=0)]),
50
        # 4. 最后：失败后的兜底重试 —— 它离真正的调用最近，所以垫在洋葱的最里层
51
        ToolRetryMiddleware(
52
            max_retries=3, initial_delay=0.5, backoff_factor=2.0, on_failure="continue"
53
        ),
54
    ],
55
)
56

57
r = agent_prod.invoke({"messages": ["我的邮箱是 a@b.com，用它查一下北京天气"]})
58
for m in r["messages"]:
59
    print(f"{type(m).__name__:12s} | {str(m.content)[:70]}")
60
print("工具共被调用:", attempts["n"], "次（说明重试中间件真的在兜底）")
61

62
# 排顺序的道理（一句话）：先做"检查/拦截"，再做"增强/兜底"。
63
# - PII 放最前：它管的是"别把敏感信息发出去"，必须在大模型看到请求之前生效
64
# - 限流第二：在任何昂贵动作之前先把闸门关小
65
# - 上下文编辑第三：内容已经安全了，再动体积
66
# - 重试最后：只有前面的检查都放行了，才谈得上"失败了怎么办"
67

68
# ---------- 第五步：亲手验证"洋葱顺序"和"同名惩罚" ----------
69
from langchain.agents.middleware import AgentMiddleware
70

71
class MarkerA(AgentMiddleware):
72
    def before_model(self, state, runtime):
73
        print("  [A] before_model")
74
        return None
75

76
    def after_model(self, state, runtime):
77
        print("  [A] after_model")
78
        return None
79

80
class MarkerB(AgentMiddleware):
81
    def before_model(self, state, runtime):
82
        print("  [B] before_model")
83
        return None
84

85
    def after_model(self, state, runtime):
86
        print("  [B] after_model")
87
        return None
88

89
for order, mws in [("[A(), B()]", [MarkerA(), MarkerB()]), ("[B(), A()]", [MarkerB(), MarkerA()])]:
90
    print(f"--- 书写顺序 {order} ---")
91
    create_agent(model=model, middleware=mws).invoke({"messages": [HumanMessage("测试顺序")]})
92
# before_* 跟着书写顺序走（从前到后），after_* 反过来（从后往前）——整体是个洋葱
93

94
# 同一个类实例化两次会被直接拒绝，因为 name 默认取类名
95
try:
96
    create_agent(model=model, middleware=[MarkerA(), MarkerA()])
97
except AssertionError as e:
98
    print("同类实例化两次 →", e)