g3fiI8dZddlmZddlmZmZmZmZmZddl m Z m Z m Z ddl mZddlmZddlmZmZddlmZmZmZmZmZerdd lmZed Z Gd d eeeeZdd Z ddZGdde Z!GddeeeefeeefZ"y)z&Tool call limit middleware for agents.) annotations) TYPE_CHECKING AnnotatedAnyGenericLiteral) AIMessageToolCall ToolMessage)UntrackedValue)ContextT) NotRequiredoverride)AgentMiddleware AgentStatePrivateStateAttr ResponseT hook_config)Runtimecontinueerrorendc&eZdZUdZded<ded<y)ToolCallLimitStatea]State schema for `ToolCallLimitMiddleware`. Extends `AgentState` with tool call tracking fields. The count fields are dictionaries mapping tool names to execution counts. This allows multiple middleware instances to track different tools independently. The special key `'__all__'` is used for tracking all tool calls globally. z8NotRequired[Annotated[dict[str, int], PrivateStateAttr]]thread_tool_call_countzHNotRequired[Annotated[dict[str, int], UntrackedValue, PrivateStateAttr]]run_tool_call_countN)__name__ __module__ __qualname____doc____annotations__i/var/www/auto_recruiter/arenv/lib/python3.12/site-packages/langchain/agents/middleware/tool_call_limit.pyrr#sUTaar$rc|rd|dSy)aBuild the error message content for `ToolMessage` when limit is exceeded. This message is sent to the model, so it should not reference thread/run concepts that the model has no notion of. Args: tool_name: Tool name being limited (if specific tool), or `None` for all tools. Returns: A concise message instructing the model not to call the tool again. z'Tool call limit exceeded. Do not call 'z' again.zeZdZdZ d dfd ZxZS)ToolCallLimitExceededErrorzException raised when tool call limits are exceeded. This exception is raised when the configured exit behavior is `'error'` and either the thread or run tool call limit has been exceeded. c||_||_||_||_||_t |||||}t ||y)aInitialize the exception with call count information. Args: thread_count: Current thread tool call count. run_count: Current run tool call count. thread_limit: Thread tool call limit (if set). run_limit: Run tool call limit (if set). tool_name: Tool name being limited (if specific tool), or None for all tools. N)r1r2r3r4r(r8super__init__)selfr1r2r3r4r(msg __class__s r%r=z#ToolCallLimitExceededError.__init__lsN")"(""- )\9i  r$)N) r1intr2rAr3 int | Noner4rBr( str | NonereturnNone)rrr r!r= __classcell__r@s@r%r:r:esQ!% !      r$r:ceZdZdZeZddddd dfdZeddZddZ ddZ dd Z e d g e dd Ze d g  dd ZxZS)ToolCallLimitMiddlewareaTrack tool call counts and enforces limits during agent execution. This middleware monitors the number of tool calls made and can terminate or restrict execution when limits are exceeded. It supports both thread-level (persistent across runs) and run-level (per invocation) call counting. Configuration: - `exit_behavior`: How to handle when limits are exceeded - `'continue'`: Block exceeded tools, let execution continue (default) - `'error'`: Raise an exception - `'end'`: Stop immediately with a `ToolMessage` + AI message for the single tool call that exceeded the limit (raises `NotImplementedError` if there are other pending tool calls (due to parallel tool calling). Examples: !!! example "Continue execution with blocked tools (default)" ```python from langchain.agents.middleware.tool_call_limit import ToolCallLimitMiddleware from langchain.agents import create_agent # Block exceeded tools but let other tools and model continue limiter = ToolCallLimitMiddleware( thread_limit=20, run_limit=10, exit_behavior="continue", # default ) agent = create_agent("openai:gpt-4o", middleware=[limiter]) ``` !!! example "Stop immediately when limit exceeded" ```python # End execution immediately with an AI message limiter = ToolCallLimitMiddleware(run_limit=5, exit_behavior="end") agent = create_agent("openai:gpt-4o", middleware=[limiter]) ``` !!! example "Raise exception on limit" ```python # Strict limit with exception handling limiter = ToolCallLimitMiddleware( tool_name="search", thread_limit=5, exit_behavior="error" ) agent = create_agent("openai:gpt-4o", middleware=[limiter]) try: result = await agent.invoke({"messages": [HumanMessage("Task")]}) except ToolCallLimitExceededError as e: print(f"Search limit exceeded: {e}") ``` Nr)r(r3r4 exit_behaviorct||| d}t|d}||vrd|d|}t|||||kDrd|d|d}t|||_||_||_||_y) aInitialize the tool call limit middleware. Args: tool_name: Name of the specific tool to limit. If `None`, limits apply to all tools. thread_limit: Maximum number of tool calls allowed per thread. `None` means no limit. run_limit: Maximum number of tool calls allowed per run. `None` means no limit. exit_behavior: How to handle when limits are exceeded. - `'continue'`: Block exceeded tools with error messages, let other tools continue. Model decides when to end. - `'error'`: Raise a `ToolCallLimitExceededError` exception - `'end'`: Stop execution immediately with a `ToolMessage` + AI message for the single tool call that exceeded the limit. Raises `NotImplementedError` if there are multiple parallel tool calls to other tools or multiple pending tool calls. Raises: ValueError: If both limits are `None`, if `exit_behavior` is invalid, or if `run_limit` exceeds `thread_limit`. Nz@At least one limit must be specified (thread_limit or run_limit)rzInvalid exit_behavior: z. Must be one of z run_limit (z) cannot exceed thread_limit (zB). The run limit should be less than or equal to the thread limit.)r<r= ValueErrorr(r3r4rJ)r>r(r3r4rJr?valid_behaviorsr@s r%r=z ToolCallLimitMiddleware.__init__s>   I$5TCS/ !6  /+M+< base_names r%namezToolCallLimitMiddleware.names8NN++ >>[$..!13 3r$c|jduxr|dz|jkDxs"|jduxr|dz|jkDS)aCheck if incrementing the counts would exceed any configured limit. Args: thread_count: Current thread call count. run_count: Current run call count. Returns: True if either limit would be exceeded by one more call. N)r3r4)r>r1r2s r%_would_exceed_limitz+ToolCallLimitMiddleware._would_exceed_limit sO!!-V,2BTEVEV2V NN$ & I9q=4>>+I r$cF|jduxs|d|jk(S)zCheck if a tool call matches this middleware's tool filter. Args: tool_call: The tool call to check. Returns: True if this middleware should track this tool call. NrRr')r> tool_calls r%_matches_tool_filterz,ToolCallLimitMiddleware._matches_tool_filters&~~%L6):dnn)LLr$cg}g}|}|}|D]S}|j|s|j||r|j|9|j||dz }|dz }U||||fS)aoSeparate tool calls into allowed and blocked based on limits. Args: tool_calls: List of tool calls to evaluate. thread_count: Current thread call count. run_count: Current run call count. Returns: Tuple of `(allowed_calls, blocked_calls, final_thread_count, final_run_count)`. rT)rXrUr/) r> tool_callsr1r2 allowed_calls blocked_callstemp_thread_counttemp_run_countrWs r%_separate_tool_callsz,ToolCallLimitMiddleware._separate_tool_calls$s)+ (* ("# $I,,Y7''(9>J$$Y/$$Y/!Q&!!# $m->NNr$r) can_jump_toc  |jdg}|syd}t|D]}t|ts|}n|r |jsy|j xsd}|jdij }|jdij }|j|d} |j|d} |j|j| | \} } } }| ||<|t| z||<| s| r||dSy||}||}|jdk(r<|t| z}t|||j|j|j t|j }| Dcgc]#}t||d |jd d %}}|jd k(r|jDcgc]"}|j |d |j k7r|$}}|r3dj|Dchc]}|d  c}}d|d}t!||t| z}t#|||j|j|j }|j%t|||d |dS|||dScc}wcc}wcc}w)aIncrement tool call counts after a model call and check limits. Args: state: The current agent state. runtime: The langgraph runtime. Returns: State updates with incremented tool call counts. If limits are exceeded and exit_behavior is `'end'`, also includes a jump to end with a `ToolMessage` and AI message for the single exceeded tool call. Raises: ToolCallLimitExceededError: If limits are exceeded and `exit_behavior` is `'error'`. NotImplementedError: If limits are exceeded, `exit_behavior` is `'end'`, and there are multiple tool calls. messagesN__all__rrr)rrr)r1r2r3r4r(idrR)content tool_call_idrRstatusrz, zDCannot end execution with other tool calls pending. Found calls to: z-. Use 'continue' or 'error' behavior instead.)re)rrjump_torb)rrrb)getreversed isinstancer rZr(copyr_lenrJr:r3r4r)r r0NotImplementedErrorr8r/)r>stateruntimerblast_ai_messagemessage count_key thread_counts run_countscurrent_thread_countcurrent_run_countr[r\new_thread_count new_run_countfinal_thread_countfinal_run_counthypothetical_thread_counttool_msg_contentrWartificial_messagestc other_tools tool_namesr?final_msg_contents r% after_modelz#ToolCallLimitMiddleware.after_modelDs299Z,) G'9-")  o&@&@NN/i  ":B?DDF YY4b9>>@ ,00A>&NN9a8IMHaHa  & &(<>OI E }&6 $4 i -M0B B 9.;+5+95$Y/    ((:S=O(O %,6)!......  7t~~F+>  (&t_]]6*  > >     &*44>>-"V*2NK !YY['Ir6 'IJ ''1l2_a*#.. );S=O(O % ?)!! !  & &y9J'K L+8'1 /  '4#-+  _> (Js/(I76'I<0 Jc.K|j||Sw)aAsync increment tool call counts after a model call and check limits. Args: state: The current agent state. runtime: The langgraph runtime. Returns: State updates with incremented tool call counts. If limits are exceeded and exit_behavior is `'end'`, also includes a jump to end with a `ToolMessage` and AI message for the single exceeded tool call. Raises: ToolCallLimitExceededError: If limits are exceeded and `exit_behavior` is `'error'`. NotImplementedError: If limits are exceeded, `exit_behavior` is `'end'`, and there are multiple tool calls. )r)r>rorps r% aafter_modelz$ToolCallLimitMiddleware.aafter_models.w//s) r(rCr3rBr4rBrJ ExitBehaviorrDrE)rDstr)r1rAr2rArDbool)rWr rDr)rZzlist[ToolCall]r1rAr2rArDz/tuple[list[ToolCall], list[ToolCall], int, int])rozToolCallLimitState[ResponseT]rpzRuntime[ContextT]rDzdict[str, Any] | None)rrr r!r state_schemar=propertyrRrUrXr_rrrrrFrGs@r%rIrIs8t&L !%#' $&0 4+4+! 4+  4+ $ 4+ 4+l     MO(O8;OHKO 8O@eW% I ,I #I   I &I VeW%0,0#0  0&0r$rIN)r(rCrDr) r1rAr2rAr3rBr4rBr(rCrDr)#r! __future__rtypingrrrrrlangchain_core.messagesr r r "langgraph.channels.untracked_valuer langgraph.typingr typing_extensionsrr!langchain.agents.middleware.typesrrrrrlanggraph.runtimerrrr)r8 Exceptionr:rIr#r$r%rs,"BBDD=%3)12  bI. 0B bJ$==== =  =  =D!!H_0&y18;< Ix  _0r$