🎯 课程主题
ModelCallLimitMiddleware（模型调用限制中间件）——限制 Agent 对大模型的调用次数，避免无限循环并控制调用成本。
📝 核心知识点
1. 核心概念与作用
- 概念说明：Agent 在工具调用循环中可能反复调用大模型，ModelCallLimitMiddleware 对模型调用次数设置上限。
- 关键细节：防止模型陷入无限的自循环（如结构化输出校验失败后反复重试），同时控制 API 调用成本。
2. 两种限制维度
- 概念说明：
thread_limit：以会话（线程） 为单位累计限制，整个对话线程的总调用次数不超过上限。run_limit：以单次 agent run 为单位限制，单次 invoke 过程中的模型调用次数不超过上限。
- 关键细节：
thread_limit需要配合checkpointer和config["configurable"]["thread_id"]来追踪同一会话下的累计次数。run_limit不需要 checkpointer，只限制当前这次invoke内的模型调用轮数。
3. 退出行为（exit_behavior）
- 概念说明：达到调用上限后的处理方式。
- 关键细节：
end：优雅结束，不再继续调用模型，静默返回。error：抛出ModelCallLimitExceededError异常。
4. 典型使用场景（run_limit 示例）
- 概念说明：当使用结构化输出（Pydantic Union）时，如果模型返回了多个互斥的结构（如同时返回两个 tool_call 而 union 只允许一个），解析失败会触发重试，导致模型被反复调用。
- 关键细节：通过模拟服务器（80% 概率返回双工具调用）来制造重试场景，演示
run_limit的拦截效果。
🏗️ 架构与工作流
Agent invoke → 模型调用 → 次数累计 → 检查是否超限 → [未超限] 正常执行 → [超限] end/error
Middleware 在每次 Agent 调用模型前检查计数器，若达到上限则按 exit_behavior 决定退出或抛异常。
💻 代码实战
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
# 模型初始化（略）
# === 举例一：thread_limit + end（优雅结束） ===
agent = create_agent(
model,
tools=[],
middleware=[
ModelCallLimitMiddleware(
thread_limit=2, # 每个线程最多 2 次模型调用
exit_behavior="end",
)
],
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "1"}}
# 第 1 次调用（正常）
response1 = agent.invoke(
{"messages": [HumanMessage(content="你好")]},
config=config,
)
# 第 2 次调用（正常，达到上限）
response2 = agent.invoke(
{"messages": [HumanMessage(content="你是谁")]},
config=config,
)
# 第 3 次调用（超限，优雅结束，不抛出异常，静默返回）
response3 = agent.invoke(
{"messages": [HumanMessage(content="你能帮我做什么")]},
config=config,
)
# === 举例二：thread_limit + error（抛异常） ===
agent2 = create_agent(
model,
tools=[],
middleware=[
ModelCallLimitMiddleware(
thread_limit=2,
exit_behavior="error", # 此处改为 error
)
],
checkpointer=InMemorySaver(),
)
try:
for i in range(3):
agent2.invoke(
{"messages": [HumanMessage(content=f"问题{i+1}")]},
config=config2,
)
except Exception as e:
print(f"ModelCallLimitExceededError: {e}")
# === 举例三：run_limit + end ===
agent3 = create_agent(
model,
tools=[],
middleware=[
ModelCallLimitMiddleware(
run_limit=3, # 单次 invoke 最多 3 次模型调用
exit_behavior="end",
)
],
)
# 此处需要配合模拟服务器（返回多工具调用触发重试）来展示效果
# 当模型因结构化输出校验失败反复重试超过 3 次时，自动结束
response = agent3.invoke(
{"messages": [HumanMessage(content="你好")]}
)
# === 举例四：run_limit + error ===
agent4 = create_agent(
model,
tools=[],
middleware=[
ModelCallLimitMiddleware(
run_limit=3,
exit_behavior="error",
)
],
)
# 效果同举例三，但超限后抛出异常
⚠️ 常见问题与避坑指南
thread_limit必须配合checkpointer和相同thread_id的config使用，否则每次都是新线程，限制不生效。run_limit的实际触发取决于模型是否被反复调用（如结构化输出解析失败重试），单次简单问答通常不会触发上限。- 使用模拟服务器测试时，由于 80% 概率返回双工具调用的概率叠加效应，可能需多次运行才能观察到超限效果。
end行为是静默返回，不会报错，调试时可能不易发现；生产环境建议用error方便监控。
💡 个人总结与延伸
ModelCallLimitMiddleware 是成本控制和稳定性保障的基础设施，核心价值在于为 Agent 的模型调用设置了一个「熔断器」。在实际生产环境中，run_limit 适合防止单次请求失控，thread_limit 适合会话级别的资源管控——两者可同时使用，构建多层级的调用保护策略。