3、其它内置中间件这里为大部分中间件提供测试代码和输出感兴趣的同学自行研究。、3.1 ModelCallLimitMiddleware中间件限制模型调用次数避免无限循环控制调用成本举例1整个会话限制-优雅退出from dotenv import load_dotenv from langchain.agents.middleware import PIIMiddleware from langchain.chat_models import init_chat_model load_dotenv(overrideTrue) #定义模型,调用的是ChatDeepseek() modelinit_chat_model( modeldeepseek:deepseek-v4-pro, extra_body{thinking: {type: disabled}}, )from langchain.agents import create_agent from langchain.agents.middleware import ModelCallLimitMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) from typing import List agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ModelCallLimitMiddleware( thread_limit2, # 每个线程最多 2 次模型调用 exit_behaviorend, # 达到限制后退出 ), ], ) def pretty_iterate_msg( messages: List[SystemMessage | HumanMessage | AIMessage | ToolMessage], ): for msg in messages: msg.pretty_print() config { configurable: { thread_id: 1, } } # 第一次调用 response_first agent.invoke( { messages: [ HumanMessage(你好), ] }, configconfig, ) print( * 30, first , * 30) pretty_iterate_msg(response_first[messages]) # 第二次调用 response_second agent.invoke( { messages: [ HumanMessage(你是谁), ] }, configconfig, ) print( * 30, second , * 30) pretty_iterate_msg(response_second[messages]) # 第三次调用 # 由于 thread_limit2这次调用会达到线程模型调用限制 response_third agent.invoke( { messages: [ HumanMessage(你能帮我做什么), ] }, configconfig, ) print( * 30, third , * 30) pretty_iterate_msg(response_third[messages]) first Human Message 你好 Ai Message 你好 很高兴见到你有什么我可以帮忙的吗 second Human Message 你好 Ai Message 你好 很高兴见到你有什么我可以帮忙的吗 Human Message 你是谁 Ai Message 我是DeepSeek由深度求索公司创造的AI助手我是一个纯文本模型可以帮你解答问题、处理信息、进行对话等等。我的知识截止日期是2025年5月目前是免费使用的支持阅读链接、上传文件图像、txt、pdf、ppt、word、excel等还有1M的超长上下文可以一次性处理像《三体》三部曲那么大体量的内容不过需要说明的是我不支持多模态识别功能但可以读取上传文件中的文字信息。另外我还可以联网搜索不过需要你在Web或App端手动点开联网搜索按键哦。有什么我可以帮你的吗 third Human Message 你好 Ai Message 你好 很高兴见到你有什么我可以帮忙的吗 Human Message 你是谁 Ai Message 我是DeepSeek由深度求索公司创造的AI助手我是一个纯文本模型可以帮你解答问题、处理信息、进行对话等等。我的知识截止日期是2025年5月目前是免费使用的支持阅读链接、上传文件图像、txt、pdf、ppt、word、excel等还有1M的超长上下文可以一次性处理像《三体》三部曲那么大体量的内容不过需要说明的是我不支持多模态识别功能但可以读取上传文件中的文字信息。另外我还可以联网搜索不过需要你在Web或App端手动点开联网搜索按键哦。有什么我可以帮你的吗 Human Message 你能帮我做什么 Ai Message Model call limits exceeded: thread limit (2/2)举例2整个会话限制-抛异常agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ModelCallLimitMiddleware( thread_limit2, # 每个线程最多 2 次模型调用 exit_behaviorerror, # 整个会话限制-抛异常 ), ], )ModelCallLimitExceededError: Model call limits exceeded: thread limit (2/2)During task with name ModelCallLimitMiddleware.before_model and id 91eb247d-38bc-3c46-a257-f70218b6590b举例3单次调用限制-优雅退出需要fake-server重复触发工具调用代码如下 Author:shkstart Desc: 模拟 DeepSeek/OpenAI 兼容接口用于测试 LangChain 工具调用Tool Calling import json import random import time from http.server import BaseHTTPRequestHandler, HTTPServer class FakeDeepSeekHandler(BaseHTTPRequestHandler): def do_POST(self): # 获取请求体的字节长度 content_length int(self.headers.get(Content-Length, 0)) # 读取请求体并将字节数据解码为字符串 raw_body self.rfile.read(content_length).decode(utf-8) print(\n * 100) # 用于保存解析后的 JSON 请求数据 json_body None try: # 将请求体字符串解析为 Python 字典 json_body json.loads(raw_body) print([JSON BODY]) # 格式化打印请求参数方便调试 print(json.dumps(json_body, ensure_asciiFalse, indent2)) except Exception as e: # JSON 解析失败时打印异常信息 print([JSON PARSE ERROR]) print(repr(e)) # 构造一个 OpenAI Chat Completions 格式的响应 response { # 模拟响应 ID id: chatcmpl-test, # 表示这是一个聊天补全响应 object: chat.completion, # 当前时间戳 created: int(time.time()), # 模拟模型名称 model: any, # 模型生成的结果 choices: [ { # 当前结果的索引 index: 0, # Assistant 消息 message: { role: assistant, # 因为这里主要测试 Tool Calling所以普通文本内容为空 content: , # 模拟模型返回的工具调用 tool_calls: [ { # 第一个工具调用的 ID id: call_1, # 表示这是一个函数调用 type: function, # 函数调用的具体信息 function: { # 从请求中的第一个工具获取函数名称 name: json_body[tools][0][function][name], # 函数参数必须是 JSON 字符串 arguments: json.dumps( { name: 康师傅, email: songhongkangatguigu.cn, phone: 12345678912, }, ensure_asciiFalse, ), }, } ], }, # 模拟正常结束 finish_reason: stop, } ], # 模拟 Token 使用量 usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2, }, } # 构造第二个工具调用 append_val { # 第二个工具调用的 ID id: call_2, # 表示这是一个函数调用 type: function, # 函数调用信息 function: { # 从请求中的第二个工具获取函数名称 name: json_body[tools][1][function][name], # 第二个工具的参数 arguments: json.dumps( { event_name: 问数项目启动会, date: 2026-03-27, }, ensure_asciiFalse, ), }, } # 模拟随机返回一个或两个工具调用 # 大约 80% 的概率添加第二个工具调用 if random.randint(1, 10) 2: response[choices][0][message][tool_calls].append(append_val) # 如果需要测试错误参数可以取消下面的注释 # response[choices][0][message][tool_calls][0][function][arguments] json.dumps( # { # name1: 康师傅, # email2: songhongkangatguigu.cn, # phone: 12345678912, # }, # ensure_asciiFalse # ) print(\n * 100) print([RESPONSE]) # 打印最终返回给客户端的响应 print(json.dumps(response, ensure_asciiFalse, indent2)) # 将 Python 字典序列化成 JSON 字符串再编码成字节 body json.dumps(response, ensure_asciiFalse).encode(utf-8) # 返回 HTTP 200 状态码 self.send_response(200) # 设置响应类型为 JSON并声明 UTF-8 编码 self.send_header( Content-Type, application/json; charsetutf-8, ) # 设置响应体长度 self.send_header( Content-Length, str(len(body)), ) # 结束响应头 self.end_headers() # 将 JSON 响应数据写回客户端 self.wfile.write(body) def log_message(self, format, *args): # 禁止 HTTPServer 默认输出访问日志 pass def main(): # 创建 HTTP 服务监听本机 9876 端口 server HTTPServer( (127.0.0.1, 9876), FakeDeepSeekHandler, ) # 打印服务启动地址 print(Fake DeepSeek server running at http://127.0.0.1:9876) # 持续运行服务器等待客户端请求 server.serve_forever() if __name__ __main__: # 只有直接运行当前文件时才启动服务器 main()注意服务端代码逻辑是80%概率输出非法响应所以不一定会导致单次请求的工具调用超过限制尝 试几次即可看到效果。客户端代码from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver from langchain.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel, Field, SecretStr from typing import List, Union from dotenv import load_dotenv load_dotenv(overrideTrue) model ChatDeepSeek( modelany, api_basehttp://127.0.0.1:9876, api_keySecretStr(KEY), ) class ContactInfo(BaseModel): 用户的联系方式 name: str Field(description用户姓名) email: str Field(description用户邮箱地址) phone: str Field(description用户的手机号) class EventInfo(BaseModel): event_name: str Field(description事件名称) date: str Field(description事件发生日期) agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ModelCallLimitMiddleware( run_limit3, exit_behaviorend, ), ], response_formatUnion[ContactInfo, EventInfo], ) def pretty_iterate_msg( messages: List[SystemMessage | HumanMessage | AIMessage | ToolMessage], ): for msg in messages: msg.pretty_print() config { configurable: { thread_id: 1, } } # seen set() response agent.invoke( { messages: [ HumanMessage(你好), ], }, configconfig, ) pretty_iterate_msg(response[messages]) Human Message 你好 Ai Message Tool Calls:ContactInfo (call_1)Call ID: call_1Args:name: 康师傅email: songhongkangatguigu.cnphone: 12345678912EventInfo (call_2)Call ID: call_2Args:event_name: 问数项目启动会date: 2026-03-27 Tool Message Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: EventInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Ai Message Tool Calls:ContactInfo (call_1)Call ID: call_1Args:name: 康师傅email: songhongkangatguigu.cnphone: 12345678912EventInfo (call_2)Call ID: call_2Args:event_name: 问数项目启动会date: 2026-03-27 Tool Message Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: EventInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Ai Message Tool Calls:ContactInfo (call_1)Call ID: call_1Args:name: 康师傅email: songhongkangatguigu.cnphone: 12345678912EventInfo (call_2)Call ID: call_2Args:event_name: 问数项目启动会date: 2026-03-27 Tool Message Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: EventInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Ai Message Model call limits exceeded: run limit (3/3)举例4单次调用限制-抛异常agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ModelCallLimitMiddleware( run_limit3, # exit_behaviorend, exit_behaviorerror, ), ], response_formatUnion[ContactInfo, EventInfo], )ModelCallLimitExceededError: Model call limits exceeded: run limit (3/3)During task with name ModelCallLimitMiddleware.before_model and id 319d96aa-f4d2-3c75-459b-39e9aa70cf483.2 ToolCallLimitMiddleware中间件限制工具调用次数可以 限制所有工具 调用的总次数也可以 限制特定工具 的调用次数。作用如下避免过多调用某些昂贵的外部API限制网络爬虫或数据库查询请求的数量避免Agent陷入无限循环退出行为有三种模式error直接抛异常end结束整个会话continue继续运行Agent这是默认行为此时Agent会将工具调用超出限制的信息传递给模 型后者自主决定后续行为如果模型能力不足可能导致死循环为了避免这种情况我们实现 的fake server会以20%的概率输出正确响应从而能终止循环。举例1整个会话限制-优雅结束from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver from langchain.agents.middleware import ToolCallLimitMiddleware from langchain.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ) from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel, Field, SecretStr from typing import List, Union from dotenv import load_dotenv load_dotenv(overrideTrue) model ChatDeepSeek( modelany, api_basehttp://127.0.0.1:9876, api_keySecretStr(KEY), ) class ContactInfo(BaseModel): 用户的联系方式 name: str Field(description用户姓名) email: str Field(description用户邮箱地址) phone: str Field(description用户的手机号) class EventInfo(BaseModel): event_name: str Field(description事件名称) date: str Field(description事件发生日期) agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ToolCallLimitMiddleware( run_limit2, # 每次运行最多2次 exit_behaviorend, ), ], response_formatUnion[ContactInfo, EventInfo], ) def pretty_iterate_msg( messages: List[SystemMessage | HumanMessage | AIMessage | ToolMessage], ): for msg in messages: msg.pretty_print() config { configurable: { thread_id: 1, } } # seen set() response agent.invoke( { messages: [ HumanMessage(你好), ], }, configconfig, ) pretty_iterate_msg(response[messages]) Human Message 你好 Ai Message Tool Calls:ContactInfo (call_1)Call ID: call_1Args:name: 康师傅email: songhongkangatguigu.cnphone: 12345678912EventInfo (call_2)Call ID: call_2Args:event_name: 问数项目启动会date: 2026-03-27 Tool Message Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: EventInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Ai Message Tool Calls:ContactInfo (call_1)Call ID: call_1Args:name: 康师傅email: songhongkangatguigu.cnphone: 12345678912EventInfo (call_2)Call ID: call_2Args:event_name: 问数项目启动会date: 2026-03-27 Tool Message Name: ContactInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: EventInfoError: Model incorrectly returned multiple structured responses (ContactInfo, EventInfo) when only one is expected.Please fix your mistakes. Tool Message Name: ContactInfoTool call limit exceeded. Do not make additional tool calls. Tool Message Name: EventInfoTool call limit exceeded. Do not make additional tool calls. Ai Message Tool call limit reached: run limit exceeded (4/2 calls).举例2整个会话限制-抛异常agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ToolCallLimitMiddleware( run_limit2, # 每次运行最多2次 exit_behaviorerror, ), ], response_formatUnion[ContactInfo, EventInfo], )ToolCallLimitExceededError: Tool call limit reached: run limit exceeded (4/2 calls).During task with name ToolCallLimitMiddleware.after_model and id ac15f386-5415-6c60-3f37-e53df36bf02c案例3单次调用限制-继续运行agent create_agent( modelmodel, checkpointerInMemorySaver(), # Required for thread limiting tools[], middleware[ ToolCallLimitMiddleware( run_limit2, # 每次运行最多2次 exit_behaviorcontinue, ), ], response_formatUnion[ContactInfo, EventInfo], )