Coverage for haystack/hooks/human_in_the_loop/strategies.py: 97%
193 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import json
6from dataclasses import replace
7from typing import Any, Literal, cast
9from haystack import tracing
10from haystack.components.agents.state.state import State
11from haystack.core.serialization import (
12 component_to_dict,
13 default_from_dict,
14 default_to_dict,
15 generate_qualified_class_name,
16)
17from haystack.dataclasses import ChatMessage, ToolCall
18from haystack.hooks.human_in_the_loop import ToolExecutionDecision
19from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy, ConfirmationStrategy, ConfirmationUI
20from haystack.tools import Tool
21from haystack.utils.async_utils import _execute_component_async
22from haystack.utils.deserialization import deserialize_component_inplace
24REJECTION_FEEDBACK_TEMPLATE = "Tool execution for '{tool_name}' was rejected by the user."
25MODIFICATION_FEEDBACK_TEMPLATE = (
26 "The parameters for tool '{tool_name}' were updated by the user to:\n{final_tool_params}"
27)
28USER_FEEDBACK_TEMPLATE = "With user feedback: {feedback}"
31class BlockingConfirmationStrategy:
32 """
33 Confirmation strategy that blocks execution to gather user feedback.
34 """
36 def __init__(
37 self,
38 *,
39 confirmation_policy: ConfirmationPolicy,
40 confirmation_ui: ConfirmationUI,
41 reject_template: str = REJECTION_FEEDBACK_TEMPLATE,
42 modify_template: str = MODIFICATION_FEEDBACK_TEMPLATE,
43 user_feedback_template: str = USER_FEEDBACK_TEMPLATE,
44 ) -> None:
45 """
46 Initialize the BlockingConfirmationStrategy with a confirmation policy and UI.
48 :param confirmation_policy:
49 The confirmation policy to determine when to ask for user confirmation.
50 :param confirmation_ui:
51 The user interface to interact with the user for confirmation.
52 :param reject_template:
53 Template for rejection feedback messages. It should include a `{tool_name}` placeholder.
54 :param modify_template:
55 Template for modification feedback messages. It should include `{tool_name}` and `{final_tool_params}`
56 placeholders.
57 :param user_feedback_template:
58 Template for user feedback messages. It should include a `{feedback}` placeholder.
59 """
60 self.confirmation_policy = confirmation_policy
61 self.confirmation_ui = confirmation_ui
62 self.reject_template = reject_template
63 self.modify_template = modify_template
64 self.user_feedback_template = user_feedback_template
66 def run(
67 self,
68 *,
69 tool_name: str,
70 tool_description: str,
71 tool_params: dict[str, Any],
72 tool_call_id: str | None = None,
73 confirmation_strategy_context: dict[str, Any] | None = None, # noqa: ARG002
74 ) -> ToolExecutionDecision:
75 """
76 Run the human-in-the-loop strategy for a given tool and its parameters.
78 :param tool_name:
79 The name of the tool to be executed.
80 :param tool_description:
81 The description of the tool.
82 :param tool_params:
83 The parameters to be passed to the tool.
84 :param tool_call_id:
85 Optional unique identifier for the tool call. This can be used to track and correlate the decision with a
86 specific tool invocation.
87 :param confirmation_strategy_context:
88 Optional dictionary for passing request-scoped resources. Useful in web/server environments
89 to provide per-request objects (e.g., WebSocket connections, async queues, Redis pub/sub clients)
90 that strategies can use for non-blocking user interaction.
92 :returns:
93 A ToolExecutionDecision indicating whether to execute the tool with the given parameters, or a
94 feedback message if rejected.
95 """
96 # Check if we should ask based on policy
97 if not self.confirmation_policy.should_ask(
98 tool_name=tool_name, tool_description=tool_description, tool_params=tool_params
99 ):
100 return ToolExecutionDecision(
101 tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params
102 )
104 # Get user confirmation through UI
105 confirmation_ui_result = self.confirmation_ui.get_user_confirmation(tool_name, tool_description, tool_params)
107 # Pass back the result to the policy for any learning/updating
108 self.confirmation_policy.update_after_confirmation(
109 tool_name, tool_description, tool_params, confirmation_ui_result
110 )
112 # Process the confirmation result
113 final_args = {}
114 if confirmation_ui_result.action == "reject":
115 explanation_text = self.reject_template.format(tool_name=tool_name)
116 if confirmation_ui_result.feedback:
117 explanation_text += " "
118 explanation_text += self.user_feedback_template.format(feedback=confirmation_ui_result.feedback)
119 return ToolExecutionDecision(
120 tool_name=tool_name, execute=False, tool_call_id=tool_call_id, feedback=explanation_text
121 )
122 if confirmation_ui_result.action == "modify" and confirmation_ui_result.new_tool_params:
123 # Update the tool call params with the new params
124 final_args.update(confirmation_ui_result.new_tool_params)
125 explanation_text = self.modify_template.format(tool_name=tool_name, final_tool_params=final_args)
126 if confirmation_ui_result.feedback:
127 explanation_text += " "
128 explanation_text += self.user_feedback_template.format(feedback=confirmation_ui_result.feedback)
129 return ToolExecutionDecision(
130 tool_name=tool_name,
131 tool_call_id=tool_call_id,
132 execute=True,
133 feedback=explanation_text,
134 final_tool_params=final_args,
135 )
136 # action == "confirm"
137 return ToolExecutionDecision(
138 tool_name=tool_name, execute=True, tool_call_id=tool_call_id, final_tool_params=tool_params
139 )
141 async def run_async(
142 self,
143 *,
144 tool_name: str,
145 tool_description: str,
146 tool_params: dict[str, Any],
147 tool_call_id: str | None = None,
148 confirmation_strategy_context: dict[str, Any] | None = None,
149 ) -> ToolExecutionDecision:
150 """
151 Async version of run. Calls the sync run() method by default.
153 :param tool_name:
154 The name of the tool to be executed.
155 :param tool_description:
156 The description of the tool.
157 :param tool_params:
158 The parameters to be passed to the tool.
159 :param tool_call_id:
160 Optional unique identifier for the tool call.
161 :param confirmation_strategy_context:
162 Optional dictionary for passing request-scoped resources.
164 :returns:
165 A ToolExecutionDecision indicating whether to execute the tool with the given parameters.
166 """
167 return self.run(
168 tool_name=tool_name,
169 tool_description=tool_description,
170 tool_params=tool_params,
171 tool_call_id=tool_call_id,
172 confirmation_strategy_context=confirmation_strategy_context,
173 )
175 def to_dict(self) -> dict[str, Any]:
176 """
177 Serializes the BlockingConfirmationStrategy to a dictionary.
179 :returns:
180 Dictionary with serialized data.
181 """
182 return default_to_dict(
183 self,
184 confirmation_policy=self.confirmation_policy.to_dict(),
185 confirmation_ui=self.confirmation_ui.to_dict(),
186 reject_template=self.reject_template,
187 modify_template=self.modify_template,
188 user_feedback_template=self.user_feedback_template,
189 )
191 @classmethod
192 def from_dict(cls, data: dict[str, Any]) -> "BlockingConfirmationStrategy":
193 """
194 Deserializes the BlockingConfirmationStrategy from a dictionary.
196 :param data:
197 Dictionary to deserialize from.
199 :returns:
200 Deserialized BlockingConfirmationStrategy.
201 """
202 deserialize_component_inplace(data["init_parameters"], key="confirmation_policy")
203 deserialize_component_inplace(data["init_parameters"], key="confirmation_ui")
204 return default_from_dict(cls, data)
207def _get_confirmation_strategy(
208 *, tool_name: str, confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy]
209) -> ConfirmationStrategy | None:
210 """
211 Get the confirmation strategy for a given tool name.
213 :param tool_name:
214 The name of the tool to look up.
215 :param confirmation_strategies:
216 Dictionary of confirmation strategies with string or tuple keys. The `"*"` key, if present, is a wildcard
217 applied to any tool without a more specific entry.
218 :returns:
219 The confirmation strategy if found, None otherwise.
220 """
221 if tool_name in confirmation_strategies:
222 return confirmation_strategies[tool_name]
224 for key, strategy in confirmation_strategies.items():
225 if isinstance(key, tuple) and tool_name in key:
226 return strategy
228 # Fall back to the wildcard entry that applies to any tool without a more specific match.
229 return confirmation_strategies.get("*")
232def _passthrough_tool_call(tool_call: ToolCall) -> ToolExecutionDecision:
233 """
234 Build a decision that executes a tool call as-is, bypassing confirmation.
236 Used for tool calls that don't resolve to a known tool (e.g. the model hallucinated the name). Instead of
237 raising here, the call is passed through unchanged so the tool-calling code resolves it and reports the
238 unknown tool uniformly (`ToolNotFoundException`, respecting `raise_on_failure`).
240 :param tool_call: The unresolved tool call to pass through.
241 :returns: A decision that executes the tool call with its original arguments.
242 """
243 return ToolExecutionDecision(
244 tool_call_id=tool_call.id, tool_name=tool_call.tool_name, execute=True, final_tool_params=tool_call.arguments
245 )
248def _classify_decision(tool_call: ToolCall, decision: ToolExecutionDecision) -> Literal["confirm", "modify", "reject"]:
249 """Classify a strategy decision against the tool call that was presented for confirmation."""
250 if not decision.execute:
251 return "reject"
252 if tool_call.arguments != (decision.final_tool_params or {}):
253 return "modify"
254 return "confirm"
257def _create_confirmation_strategy_span(strategy: ConfirmationStrategy, tool_call: ToolCall, parent_span: Any) -> Any:
258 """
259 Create a tracing span for one `ConfirmationStrategy` run.
261 The span covers a single strategy run, so a hook confirming several tool calls produces several sibling spans under
262 the `haystack.agent.hook` span of its invocation. The tool call is identified with the shared `haystack.tool.*`
263 keys so the siblings stay distinguishable when content tracing is disabled.
264 """
265 tags = {
266 "haystack.tool.name": tool_call.tool_name,
267 "haystack.tool.call.id": tool_call.id,
268 "haystack.agent.hook.human_in_the_loop.strategy.type": generate_qualified_class_name(type(strategy)),
269 }
270 return tracing.tracer.trace("haystack.agent.hook.human_in_the_loop.strategy", tags=tags, parent_span=parent_span)
273def _set_strategy_input_tracing_tags(
274 span: Any,
275 tool_name: str,
276 tool_description: str,
277 tool_params: dict[str, Any],
278 tool_call_id: str | None = None,
279 confirmation_strategy_context: dict[str, Any] | None = None, # noqa: ARG001
280) -> None:
281 """
282 Record the arguments a `ConfirmationStrategy` is run with.
284 `confirmation_strategy_context` is left out: it carries request-scoped resources such as WebSocket connections and
285 queues rather than information about the decision.
286 """
287 span.set_content_tag(
288 key="haystack.agent.hook.human_in_the_loop.strategy.input",
289 value={
290 "tool_name": tool_name,
291 "tool_description": tool_description,
292 "tool_params": tool_params,
293 "tool_call_id": tool_call_id,
294 },
295 )
298def _set_strategy_output_tracing_tags(span: Any, tool_call: ToolCall, decision: ToolExecutionDecision) -> None:
299 """Record the decision a `ConfirmationStrategy` returned, plus how it compares to the original tool call."""
300 span.set_tag(
301 "haystack.agent.hook.human_in_the_loop.strategy.decision",
302 _classify_decision(tool_call=tool_call, decision=decision),
303 )
304 span.set_content_tag(key="haystack.agent.hook.human_in_the_loop.strategy.output", value=decision.to_dict())
307def _bind_decision_to_tool_call(decision: ToolExecutionDecision, tool_call: ToolCall) -> ToolExecutionDecision:
308 """Bind a confirmation decision to the tool call being processed by its tool_call_id."""
309 if decision.tool_call_id == tool_call.id:
310 return decision
311 return replace(decision, tool_call_id=tool_call.id)
314def _process_confirmation_strategies(
315 *,
316 confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy],
317 messages_with_tool_calls: list[ChatMessage],
318 tools: list[Tool],
319 state: State,
320 confirmation_strategy_context: dict[str, Any] | None = None,
321) -> list[ChatMessage]:
322 """
323 Run the confirmation strategies and return the updated chat history.
325 The returned history ends with the confirmed/modified tool calls (preceded by any rejection messages), so the
326 pending tool calls to execute are always those on its last message.
328 :param confirmation_strategies: Mapping of tool names to their corresponding confirmation strategies
329 :param messages_with_tool_calls: Chat messages containing tool calls
330 :param tools: The available tools, used to resolve each tool call by name
331 :param state: The current runtime state, used to read the chat history
332 :param confirmation_strategy_context: Optional request-scoped context passed to the strategies
333 :returns:
334 The updated chat history.
335 """
336 # If confirmations strategies is empty, return the chat history unchanged
337 if not confirmation_strategies:
338 return state.data["messages"]
340 # Run confirmation strategies and get tool execution decisions
341 teds = _run_confirmation_strategies(
342 confirmation_strategies=confirmation_strategies,
343 messages_with_tool_calls=messages_with_tool_calls,
344 tools=tools,
345 confirmation_strategy_context=confirmation_strategy_context,
346 )
347 # Apply tool execution decisions to messages_with_tool_calls
348 rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions(
349 tool_call_messages=messages_with_tool_calls, tool_execution_decisions=teds
350 )
352 # Update the chat history with rejection messages and new tool call messages
353 return _update_chat_history(
354 chat_history=state.data["messages"],
355 rejection_messages=rejection_messages,
356 tool_call_and_explanation_messages=modified_tool_call_messages,
357 )
360async def _process_confirmation_strategies_async(
361 *,
362 confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy],
363 messages_with_tool_calls: list[ChatMessage],
364 tools: list[Tool],
365 state: State,
366 confirmation_strategy_context: dict[str, Any] | None = None,
367) -> list[ChatMessage]:
368 """
369 Async version of _process_confirmation_strategies.
371 Run the confirmation strategies and return the updated chat history.
373 The returned history ends with the confirmed/modified tool calls (preceded by any rejection messages), so the
374 pending tool calls to execute are always those on its last message.
376 :param confirmation_strategies: Mapping of tool names to their corresponding confirmation strategies
377 :param messages_with_tool_calls: Chat messages containing tool calls
378 :param tools: The available tools, used to resolve each tool call by name
379 :param state: The current runtime state, used to read the chat history
380 :param confirmation_strategy_context: Optional request-scoped context passed to the strategies
381 :returns:
382 The updated chat history.
383 """
384 # If confirmations strategies is empty, return the chat history unchanged
385 if not confirmation_strategies:
386 return state.data["messages"]
388 # Run confirmation strategies and get tool execution decisions (async version)
389 teds = await _run_confirmation_strategies_async(
390 confirmation_strategies=confirmation_strategies,
391 messages_with_tool_calls=messages_with_tool_calls,
392 tools=tools,
393 confirmation_strategy_context=confirmation_strategy_context,
394 )
395 # Apply tool execution decisions to messages_with_tool_calls
396 rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions(
397 tool_call_messages=messages_with_tool_calls, tool_execution_decisions=teds
398 )
400 # Update the chat history with rejection messages and new tool call messages
401 return _update_chat_history(
402 chat_history=state.data["messages"],
403 rejection_messages=rejection_messages,
404 tool_call_and_explanation_messages=modified_tool_call_messages,
405 )
408def _run_confirmation_strategies(
409 confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy],
410 messages_with_tool_calls: list[ChatMessage],
411 tools: list[Tool],
412 confirmation_strategy_context: dict[str, Any] | None = None,
413) -> list[ToolExecutionDecision]:
414 """
415 Run confirmation strategies for tool calls in the provided chat messages.
417 :param confirmation_strategies: Mapping of tool names to their corresponding confirmation strategies
418 :param messages_with_tool_calls: Messages containing tool calls to process
419 :param tools: The available tools, used to resolve each tool call by name
420 :param confirmation_strategy_context: Optional request-scoped context passed to the strategies
421 :returns:
422 A list of ToolExecutionDecision objects representing the decisions made for each tool call.
423 """
424 tools_with_names = {tool.name: tool for tool in tools}
425 parent_span = tracing.tracer.current_span()
427 teds = []
428 for message in messages_with_tool_calls:
429 if not message.tool_calls:
430 continue
432 for tool_call in message.tool_calls:
433 tool_name = tool_call.tool_name
434 tool_to_invoke = tools_with_names.get(tool_name)
435 if tool_to_invoke is None:
436 # Unknown tool (e.g. the model hallucinated the name): skip confirmation and pass it through.
437 teds.append(_passthrough_tool_call(tool_call=tool_call))
438 continue
440 # Confirm the model-requested arguments
441 final_args = dict(tool_call.arguments)
443 # Get tool execution decisions from confirmation strategies
444 # If no confirmation strategy is defined for this tool, proceed with execution
445 strategy = _get_confirmation_strategy(tool_name=tool_name, confirmation_strategies=confirmation_strategies)
446 if strategy is None:
447 teds.append(
448 ToolExecutionDecision(
449 tool_call_id=tool_call.id, tool_name=tool_name, execute=True, final_tool_params=final_args
450 )
451 )
452 continue
454 # Run the confirmation strategy
455 strategy_inputs: dict[str, Any] = {
456 "tool_name": tool_name,
457 "tool_description": tool_to_invoke.description,
458 "tool_params": final_args,
459 "tool_call_id": tool_call.id,
460 "confirmation_strategy_context": confirmation_strategy_context,
461 }
462 with _create_confirmation_strategy_span(
463 strategy=strategy, tool_call=tool_call, parent_span=parent_span
464 ) as span:
465 _set_strategy_input_tracing_tags(span=span, **strategy_inputs)
466 ted = strategy.run(**strategy_inputs)
467 bound_ted = _bind_decision_to_tool_call(decision=ted, tool_call=tool_call)
468 _set_strategy_output_tracing_tags(span=span, tool_call=tool_call, decision=bound_ted)
469 teds.append(bound_ted)
471 return teds
474async def _run_confirmation_strategies_async(
475 confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy],
476 messages_with_tool_calls: list[ChatMessage],
477 tools: list[Tool],
478 confirmation_strategy_context: dict[str, Any] | None = None,
479) -> list[ToolExecutionDecision]:
480 """
481 Async version of _run_confirmation_strategies.
483 Run confirmation strategies for tool calls in the provided chat messages.
485 :param confirmation_strategies: Mapping of tool names to their corresponding confirmation strategies
486 String keys map individual tools, tuple keys map multiple tools to the same strategy.
487 :param messages_with_tool_calls: Messages containing tool calls to process
488 :param tools: The available tools, used to resolve each tool call by name
489 :param confirmation_strategy_context: Optional request-scoped context passed to the strategies
490 :returns:
491 A list of ToolExecutionDecision objects representing the decisions made for each tool call.
492 """
493 tools_with_names = {tool.name: tool for tool in tools}
494 parent_span = tracing.tracer.current_span()
496 teds = []
497 for message in messages_with_tool_calls:
498 if not message.tool_calls:
499 continue
501 for tool_call in message.tool_calls:
502 tool_name = tool_call.tool_name
503 tool_to_invoke = tools_with_names.get(tool_name)
504 if tool_to_invoke is None:
505 # Unknown tool (e.g. the model hallucinated the name): skip confirmation and pass it through.
506 teds.append(_passthrough_tool_call(tool_call=tool_call))
507 continue
509 # Confirm the model-requested arguments
510 final_args = dict(tool_call.arguments)
512 # Get tool execution decisions from confirmation strategies
513 # If no confirmation strategy is defined for this tool, proceed with execution
514 strategy = _get_confirmation_strategy(tool_name=tool_name, confirmation_strategies=confirmation_strategies)
515 if strategy is None:
516 teds.append(
517 ToolExecutionDecision(
518 tool_call_id=tool_call.id, tool_name=tool_name, execute=True, final_tool_params=final_args
519 )
520 )
521 continue
523 strategy_inputs: dict[str, Any] = {
524 "tool_name": tool_name,
525 "tool_description": tool_to_invoke.description,
526 "tool_params": final_args,
527 "tool_call_id": tool_call.id,
528 "confirmation_strategy_context": confirmation_strategy_context,
529 }
530 with _create_confirmation_strategy_span(
531 strategy=strategy, tool_call=tool_call, parent_span=parent_span
532 ) as span:
533 _set_strategy_input_tracing_tags(span=span, **strategy_inputs)
534 # The _execute_component_async helper supports arbitrary run results, but its return type is currently
535 # component-specific, so we cast it to the expected ToolExecutionDecision type here.
536 ted = cast(ToolExecutionDecision, await _execute_component_async(strategy, **strategy_inputs))
537 bound_ted = _bind_decision_to_tool_call(decision=ted, tool_call=tool_call)
538 _set_strategy_output_tracing_tags(span=span, tool_call=tool_call, decision=bound_ted)
539 teds.append(bound_ted)
541 return teds
544def _apply_tool_execution_decisions(
545 tool_call_messages: list[ChatMessage], tool_execution_decisions: list[ToolExecutionDecision]
546) -> tuple[list[ChatMessage], list[ChatMessage]]:
547 """
548 Apply the tool execution decisions to the tool call messages.
550 :param tool_call_messages: The tool call messages to apply the decisions to.
551 :param tool_execution_decisions: The tool execution decisions to apply.
552 :returns:
553 A tuple containing:
554 - A list of rejection messages for rejected tool calls. These are pairs of tool call and tool call result
555 messages.
556 - A list of tool call messages for confirmed or modified tool calls. If tool parameters were modified,
557 a user message explaining the modification is included before the tool call message.
558 """
559 tool_calls = [tc for message in tool_call_messages for tc in (message.tool_calls or [])]
560 if len(tool_calls) != len(tool_execution_decisions):
561 raise ValueError(
562 f"Expected one ToolExecutionDecision for each tool call, but received {len(tool_execution_decisions)} "
563 f"decisions for {len(tool_calls)} tool calls."
564 )
566 # Create lookup for decisions that have a tool_call_id
567 decision_by_id = {d.tool_call_id: d for d in tool_execution_decisions if d.tool_call_id}
569 # Create a lookup for decisions that don't have a tool_call_id. We use tool name instead.
570 decisions_without_id = [d for d in tool_execution_decisions if not d.tool_call_id]
571 decision_by_name = {d.tool_name: d for d in decisions_without_id if d.tool_name}
573 def make_assistant_message(chat_message: ChatMessage, tool_calls: list[ToolCall]) -> ChatMessage:
574 return ChatMessage.from_assistant(
575 text=chat_message.text,
576 meta=chat_message.meta,
577 name=chat_message.name,
578 tool_calls=tool_calls,
579 reasoning=chat_message.reasoning,
580 )
582 new_tool_call_messages = []
583 rejection_messages = []
584 for chat_msg in tool_call_messages:
585 new_tool_calls = []
586 for tc in chat_msg.tool_calls or []:
587 ted = decision_by_id.pop(tc.id or "", None)
588 if ted is None:
589 ted = decision_by_name.pop(tc.tool_name, None)
590 if ted is None:
591 raise ValueError(
592 f"No unused ToolExecutionDecision matches tool call {tc.tool_name!r} with ID {tc.id!r}. "
593 "Set tool_call_id to match decisions to tool calls reliably."
594 )
596 classified_decision = _classify_decision(tool_call=tc, decision=ted)
597 final_args = ted.final_tool_params or {}
599 if classified_decision == "reject":
600 # rejected tool call
601 tool_result_text = ted.feedback or REJECTION_FEEDBACK_TEMPLATE.format(tool_name=tc.tool_name)
602 rejection_messages.extend(
603 [
604 make_assistant_message(chat_msg, [tc]),
605 ChatMessage.from_tool(tool_result=tool_result_text, origin=tc, error=True),
606 ]
607 )
608 continue
610 # Covers confirm and modify cases
611 if classified_decision == "modify":
612 # In the modify case we add a user message explaining the modification otherwise the LLM won't know
613 # why the tool parameters changed and will likely just try and call the tool again with the
614 # original parameters.
615 user_text = ted.feedback or MODIFICATION_FEEDBACK_TEMPLATE.format(
616 tool_name=tc.tool_name, final_tool_params=final_args
617 )
618 new_tool_call_messages.append(ChatMessage.from_user(text=user_text))
619 new_tool_calls.append(replace(tc, arguments=final_args))
621 # Only add the tool call message if there are any tool calls left (i.e. not all were rejected)
622 if new_tool_calls:
623 new_tool_call_messages.append(make_assistant_message(chat_msg, new_tool_calls))
625 # new_tool_call_messages is a list of assistant messages with an optional preceding user message explaining
626 # modifications
627 # rejection_messages is a list of pairs of assistant and tool messages for rejected tool calls
628 return rejection_messages, new_tool_call_messages
631def _update_chat_history(
632 chat_history: list[ChatMessage],
633 rejection_messages: list[ChatMessage],
634 tool_call_and_explanation_messages: list[ChatMessage],
635) -> list[ChatMessage]:
636 """
637 Update the chat history to include rejection messages and tool call messages at the appropriate positions.
639 Steps:
640 1. Identify the last user message and the last tool message in the current chat history.
641 2. Determine the insertion point as the maximum index of these two messages.
642 3. Create a new chat history that includes:
643 - All messages up to the insertion point.
644 - Any rejection messages (pairs of tool call and tool call result messages).
645 - Any tool call messages for confirmed or modified tool calls, including user messages explaining modifications.
647 :param chat_history: The current chat history.
648 :param rejection_messages: Chat messages to add for rejected tool calls (pairs of tool call and tool call result
649 messages).
650 :param tool_call_and_explanation_messages: Tool call messages for confirmed or modified tool calls, which may
651 include user messages explaining modifications.
652 :returns:
653 The updated chat history.
654 """
655 user_indices = [i for i, message in enumerate(chat_history) if message.is_from("user")]
656 tool_indices = [i for i, message in enumerate(chat_history) if message.is_from("tool")]
658 last_user_idx = max(user_indices) if user_indices else -1
659 last_tool_idx = max(tool_indices) if tool_indices else -1
661 insertion_point = max(last_user_idx, last_tool_idx)
663 return chat_history[: insertion_point + 1] + rejection_messages + tool_call_and_explanation_messages
666def _serialize_confirmation_strategies(
667 confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy],
668) -> dict[str, Any]:
669 """
670 Serialize a confirmation strategies dictionary to a plain, mapping-key-safe dictionary.
672 Mapping keys must be strings, so a tuple of tool names (one strategy shared across several tools) is encoded
673 as a JSON-array string (e.g. `("a", "b")` -> `'["a", "b"]'`); a single tool name is kept as-is.
675 :param confirmation_strategies: Mapping of tool name (or a tuple of tool names) to its strategy.
676 :returns: The same mapping with string keys and each strategy serialized to a dictionary.
677 """
678 return {
679 (json.dumps(list(key)) if isinstance(key, tuple) else key): component_to_dict(
680 obj=strategy, name="confirmation_strategy"
681 )
682 for key, strategy in confirmation_strategies.items()
683 }
686def _deserialize_confirmation_strategies(data: dict[str, Any]) -> dict[str | tuple[str, ...], ConfirmationStrategy]:
687 """
688 Deserialize a confirmation strategies dictionary from its serialized form.
690 Deserializes each strategy component in-place and converts keys that were encoded as JSON-array strings (tuples
691 of tool names) back to tuples; single tool-name string keys are kept as-is.
693 :param data: Raw dictionary of serialized confirmation strategies, keyed by tool name(s).
694 :returns: Deserialized confirmation strategies with proper key types.
695 """
696 for raw_key in list(data):
697 deserialize_component_inplace(data, key=raw_key)
699 return {_decode_strategy_key(raw_key): strategy for raw_key, strategy in data.items()}
702def _decode_strategy_key(raw_key: str | list) -> str | tuple[str, ...]:
703 """Reverse of the key encoding in `_serialize_confirmation_strategies`."""
704 # Backwards-compatibility: an actual list (older in-memory forms) becomes a tuple.
705 if isinstance(raw_key, list):
706 return tuple(raw_key)
707 # A JSON-array string encodes a tuple of tool names; any other string is a single tool name.
708 if raw_key.startswith("["):
709 return tuple(json.loads(raw_key))
710 return raw_key