diff --git a/haystack/hooks/human_in_the_loop/strategies.py b/haystack/hooks/human_in_the_loop/strategies.py index a324387b69..21af34fc6f 100644 --- a/haystack/hooks/human_in_the_loop/strategies.py +++ b/haystack/hooks/human_in_the_loop/strategies.py @@ -4,14 +4,21 @@ import json from dataclasses import replace -from typing import Any +from typing import Any, Literal, cast +from haystack import tracing from haystack.components.agents.state.state import State -from haystack.core.serialization import component_to_dict, default_from_dict, default_to_dict +from haystack.core.serialization import ( + component_to_dict, + default_from_dict, + default_to_dict, + generate_qualified_class_name, +) from haystack.dataclasses import ChatMessage, ToolCall from haystack.hooks.human_in_the_loop import ToolExecutionDecision from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy, ConfirmationStrategy, ConfirmationUI from haystack.tools import Tool +from haystack.utils.async_utils import _execute_component_async from haystack.utils.deserialization import deserialize_component_inplace REJECTION_FEEDBACK_TEMPLATE = "Tool execution for '{tool_name}' was rejected by the user." @@ -238,6 +245,65 @@ def _passthrough_tool_call(tool_call: ToolCall) -> ToolExecutionDecision: ) +def _classify_decision(tool_call: ToolCall, decision: ToolExecutionDecision) -> Literal["confirm", "modify", "reject"]: + """Classify a strategy decision against the tool call that was presented for confirmation.""" + if not decision.execute: + return "reject" + if tool_call.arguments != (decision.final_tool_params or {}): + return "modify" + return "confirm" + + +def _create_confirmation_strategy_span(strategy: ConfirmationStrategy, tool_call: ToolCall, parent_span: Any) -> Any: + """ + Create a tracing span for one `ConfirmationStrategy` run. + + The span covers a single strategy run, so a hook confirming several tool calls produces several sibling spans under + the `haystack.agent.hook` span of its invocation. The tool call is identified with the shared `haystack.tool.*` + keys so the siblings stay distinguishable when content tracing is disabled. + """ + tags = { + "haystack.tool.name": tool_call.tool_name, + "haystack.tool.call.id": tool_call.id, + "haystack.agent.hook.human_in_the_loop.strategy.type": generate_qualified_class_name(type(strategy)), + } + return tracing.tracer.trace("haystack.agent.hook.human_in_the_loop.strategy", tags=tags, parent_span=parent_span) + + +def _set_strategy_input_tracing_tags( + span: Any, + tool_name: str, + tool_description: str, + tool_params: dict[str, Any], + tool_call_id: str | None = None, + confirmation_strategy_context: dict[str, Any] | None = None, # noqa: ARG001 +) -> None: + """ + Record the arguments a `ConfirmationStrategy` is run with. + + `confirmation_strategy_context` is left out: it carries request-scoped resources such as WebSocket connections and + queues rather than information about the decision. + """ + span.set_content_tag( + key="haystack.agent.hook.human_in_the_loop.strategy.input", + value={ + "tool_name": tool_name, + "tool_description": tool_description, + "tool_params": tool_params, + "tool_call_id": tool_call_id, + }, + ) + + +def _set_strategy_output_tracing_tags(span: Any, tool_call: ToolCall, decision: ToolExecutionDecision) -> None: + """Record the decision a `ConfirmationStrategy` returned, plus how it compares to the original tool call.""" + span.set_tag( + "haystack.agent.hook.human_in_the_loop.strategy.decision", + _classify_decision(tool_call=tool_call, decision=decision), + ) + span.set_content_tag(key="haystack.agent.hook.human_in_the_loop.strategy.output", value=decision.to_dict()) + + def _bind_decision_to_tool_call(decision: ToolExecutionDecision, tool_call: ToolCall) -> ToolExecutionDecision: """Bind a confirmation decision to the tool call being processed by its tool_call_id.""" if decision.tool_call_id == tool_call.id: @@ -278,7 +344,6 @@ def _process_confirmation_strategies( tools=tools, confirmation_strategy_context=confirmation_strategy_context, ) - # Apply tool execution decisions to messages_with_tool_calls rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=messages_with_tool_calls, tool_execution_decisions=teds @@ -327,7 +392,6 @@ async def _process_confirmation_strategies_async( tools=tools, confirmation_strategy_context=confirmation_strategy_context, ) - # Apply tool execution decisions to messages_with_tool_calls rejection_messages, modified_tool_call_messages = _apply_tool_execution_decisions( tool_call_messages=messages_with_tool_calls, tool_execution_decisions=teds @@ -358,6 +422,7 @@ def _run_confirmation_strategies( A list of ToolExecutionDecision objects representing the decisions made for each tool call. """ tools_with_names = {tool.name: tool for tool in tools} + parent_span = tracing.tracer.current_span() teds = [] for message in messages_with_tool_calls: @@ -369,7 +434,7 @@ def _run_confirmation_strategies( tool_to_invoke = tools_with_names.get(tool_name) if tool_to_invoke is None: # Unknown tool (e.g. the model hallucinated the name): skip confirmation and pass it through. - teds.append(_passthrough_tool_call(tool_call)) + teds.append(_passthrough_tool_call(tool_call=tool_call)) continue # Confirm the model-requested arguments @@ -387,14 +452,21 @@ def _run_confirmation_strategies( continue # Run the confirmation strategy - ted = strategy.run( - tool_name=tool_name, - tool_description=tool_to_invoke.description, - tool_params=final_args, - tool_call_id=tool_call.id, - confirmation_strategy_context=confirmation_strategy_context, - ) - teds.append(_bind_decision_to_tool_call(decision=ted, tool_call=tool_call)) + strategy_inputs: dict[str, Any] = { + "tool_name": tool_name, + "tool_description": tool_to_invoke.description, + "tool_params": final_args, + "tool_call_id": tool_call.id, + "confirmation_strategy_context": confirmation_strategy_context, + } + with _create_confirmation_strategy_span( + strategy=strategy, tool_call=tool_call, parent_span=parent_span + ) as span: + _set_strategy_input_tracing_tags(span=span, **strategy_inputs) + ted = strategy.run(**strategy_inputs) + bound_ted = _bind_decision_to_tool_call(decision=ted, tool_call=tool_call) + _set_strategy_output_tracing_tags(span=span, tool_call=tool_call, decision=bound_ted) + teds.append(bound_ted) return teds @@ -419,6 +491,7 @@ async def _run_confirmation_strategies_async( A list of ToolExecutionDecision objects representing the decisions made for each tool call. """ tools_with_names = {tool.name: tool for tool in tools} + parent_span = tracing.tracer.current_span() teds = [] for message in messages_with_tool_calls: @@ -430,7 +503,7 @@ async def _run_confirmation_strategies_async( tool_to_invoke = tools_with_names.get(tool_name) if tool_to_invoke is None: # Unknown tool (e.g. the model hallucinated the name): skip confirmation and pass it through. - teds.append(_passthrough_tool_call(tool_call)) + teds.append(_passthrough_tool_call(tool_call=tool_call)) continue # Confirm the model-requested arguments @@ -447,24 +520,23 @@ async def _run_confirmation_strategies_async( ) continue - # Use run_async if available, otherwise fall back to sync run - if hasattr(strategy, "run_async"): - ted = await strategy.run_async( - tool_name=tool_name, - tool_description=tool_to_invoke.description, - tool_params=final_args, - tool_call_id=tool_call.id, - confirmation_strategy_context=confirmation_strategy_context, - ) - else: - ted = strategy.run( - tool_name=tool_name, - tool_description=tool_to_invoke.description, - tool_params=final_args, - tool_call_id=tool_call.id, - confirmation_strategy_context=confirmation_strategy_context, - ) - teds.append(_bind_decision_to_tool_call(decision=ted, tool_call=tool_call)) + strategy_inputs: dict[str, Any] = { + "tool_name": tool_name, + "tool_description": tool_to_invoke.description, + "tool_params": final_args, + "tool_call_id": tool_call.id, + "confirmation_strategy_context": confirmation_strategy_context, + } + with _create_confirmation_strategy_span( + strategy=strategy, tool_call=tool_call, parent_span=parent_span + ) as span: + _set_strategy_input_tracing_tags(span=span, **strategy_inputs) + # The _execute_component_async helper supports arbitrary run results, but its return type is currently + # component-specific, so we cast it to the expected ToolExecutionDecision type here. + ted = cast(ToolExecutionDecision, await _execute_component_async(strategy, **strategy_inputs)) + bound_ted = _bind_decision_to_tool_call(decision=ted, tool_call=tool_call) + _set_strategy_output_tracing_tags(span=span, tool_call=tool_call, decision=bound_ted) + teds.append(bound_ted) return teds @@ -509,7 +581,6 @@ def _apply_tool_execution_decisions( new_tool_call_messages = [] rejection_messages = [] - for chat_msg in tool_call_messages: new_tool_calls = [] for tc in chat_msg.tool_calls or []: @@ -522,7 +593,10 @@ def _apply_tool_execution_decisions( "Set tool_call_id to match decisions to tool calls reliably." ) - if not ted.execute: + classified_decision = _classify_decision(tool_call=tc, decision=ted) + final_args = ted.final_tool_params or {} + + if classified_decision == "reject": # rejected tool call tool_result_text = ted.feedback or REJECTION_FEEDBACK_TEMPLATE.format(tool_name=tc.tool_name) rejection_messages.extend( @@ -534,8 +608,7 @@ def _apply_tool_execution_decisions( continue # Covers confirm and modify cases - final_args = ted.final_tool_params or {} - if tc.arguments != final_args: + if classified_decision == "modify": # In the modify case we add a user message explaining the modification otherwise the LLM won't know # why the tool parameters changed and will likely just try and call the tool again with the # original parameters. @@ -550,7 +623,7 @@ def _apply_tool_execution_decisions( new_tool_call_messages.append(make_assistant_message(chat_msg, new_tool_calls)) # new_tool_call_messages is a list of assistant messages with an optional preceding user message explaining - # modifications + # modifications # rejection_messages is a list of pairs of assistant and tool messages for rejected tool calls return rejection_messages, new_tool_call_messages diff --git a/releasenotes/notes/trace-human-in-the-loop-hook-decisions-ebd13eaf68f9cbb7.yaml b/releasenotes/notes/trace-human-in-the-loop-hook-decisions-ebd13eaf68f9cbb7.yaml new file mode 100644 index 0000000000..094c07ba6c --- /dev/null +++ b/releasenotes/notes/trace-human-in-the-loop-hook-decisions-ebd13eaf68f9cbb7.yaml @@ -0,0 +1,10 @@ +--- +enhancements: + - | + Add a tracing span per ``ConfirmationStrategy`` run to ``ConfirmationHook``. Each + ``haystack.agent.hook.human_in_the_loop.strategy`` span identifies the tool call it confirms and records the + strategy type and the applied confirm, modify, or reject decision. When content tracing is enabled, spans also + carry the arguments the strategy was run with and the ``ToolExecutionDecision`` it returned under + ``haystack.agent.hook.human_in_the_loop.strategy.input`` and + ``haystack.agent.hook.human_in_the_loop.strategy.output``; chat messages, the confirmation strategy context, and + the Agent ``State`` are not recorded by the hook. diff --git a/test/hooks/human_in_the_loop/test_hooks.py b/test/hooks/human_in_the_loop/test_hooks.py index 97f6aff304..0db8291fae 100644 --- a/test/hooks/human_in_the_loop/test_hooks.py +++ b/test/hooks/human_in_the_loop/test_hooks.py @@ -7,6 +7,7 @@ from typing import Any import pytest +from haystack import tracing from haystack.components.agents.state.state import State from haystack.components.agents.state.state_utils import merge_lists, replace_values from haystack.dataclasses import ChatMessage, ToolCall @@ -20,6 +21,7 @@ from haystack.hooks.human_in_the_loop import ( ToolExecutionDecision, ) from haystack.hooks.human_in_the_loop.types import ConfirmationUI +from haystack.hooks.invocation import _run_hooks, _run_hooks_async from haystack.tools import Tool, create_tool_from_function @@ -278,6 +280,199 @@ class TestConfirmationHookAsync: ] +class TestConfirmationHookTracing: + def test_adds_one_tool_span_per_confirmation_strategy(self, spying_tracer): + tools = [_echo_tool(name) for name in ("confirmed_tool", "modified_tool", "rejected_tool")] + messages = [ + ChatMessage.from_user("sensitive user message"), + ChatMessage.from_assistant( + tool_calls=[ + ToolCall(id="tc-confirm", tool_name="confirmed_tool", arguments={"x": 1}), + ToolCall(id="tc-modify", tool_name="modified_tool", arguments={"x": 2}), + ToolCall(id="tc-reject", tool_name="rejected_tool", arguments={"x": 3}), + ] + ), + ] + state = _state_with(messages, tools) + hook = _multi_tool_hook( + { + "confirmed_tool": ConfirmationUIResult(action="confirm"), + "modified_tool": ConfirmationUIResult( + action="modify", new_tool_params={"x": 99}, feedback="sensitive modification feedback" + ), + "rejected_tool": ConfirmationUIResult(action="reject", feedback="sensitive rejection feedback"), + } + ) + + _run_hooks(hooks={"before_tool": [hook]}, hook_point="before_tool", state=state) + + hook_span, *tool_spans = spying_tracer.spans + assert hook_span.tags == { + "haystack.agent.hook.point": "before_tool", + "haystack.agent.hook.name": "ConfirmationHook", + "haystack.agent.hook.type": "haystack.hooks.human_in_the_loop.hooks.ConfirmationHook", + } + assert len(tool_spans) == 3 + assert all(span.operation_name == "haystack.agent.hook.human_in_the_loop.strategy" for span in tool_spans) + assert all(span.parent_span is hook_span for span in tool_spans) + assert [span.tags for span in tool_spans] == [ + { + "haystack.tool.name": "confirmed_tool", + "haystack.tool.call.id": "tc-confirm", + "haystack.agent.hook.human_in_the_loop.strategy.type": ( + "haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy" + ), + "haystack.agent.hook.human_in_the_loop.strategy.input": { + "tool_name": "confirmed_tool", + "tool_description": "Echo tool confirmed_tool.", + "tool_params": {"x": 1}, + "tool_call_id": "tc-confirm", + }, + "haystack.agent.hook.human_in_the_loop.strategy.decision": "confirm", + "haystack.agent.hook.human_in_the_loop.strategy.output": { + "tool_name": "confirmed_tool", + "execute": True, + "tool_call_id": "tc-confirm", + "feedback": None, + "final_tool_params": {"x": 1}, + }, + }, + { + "haystack.tool.name": "modified_tool", + "haystack.tool.call.id": "tc-modify", + "haystack.agent.hook.human_in_the_loop.strategy.type": ( + "haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy" + ), + "haystack.agent.hook.human_in_the_loop.strategy.input": { + "tool_name": "modified_tool", + "tool_description": "Echo tool modified_tool.", + "tool_params": {"x": 2}, + "tool_call_id": "tc-modify", + }, + "haystack.agent.hook.human_in_the_loop.strategy.decision": "modify", + "haystack.agent.hook.human_in_the_loop.strategy.output": { + "tool_name": "modified_tool", + "execute": True, + "tool_call_id": "tc-modify", + "feedback": "The parameters for tool 'modified_tool' were updated by the user to:\n{'x': 99} " + "With user feedback: sensitive modification feedback", + "final_tool_params": {"x": 99}, + }, + }, + { + "haystack.tool.name": "rejected_tool", + "haystack.tool.call.id": "tc-reject", + "haystack.agent.hook.human_in_the_loop.strategy.type": ( + "haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy" + ), + "haystack.agent.hook.human_in_the_loop.strategy.input": { + "tool_name": "rejected_tool", + "tool_description": "Echo tool rejected_tool.", + "tool_params": {"x": 3}, + "tool_call_id": "tc-reject", + }, + "haystack.agent.hook.human_in_the_loop.strategy.decision": "reject", + "haystack.agent.hook.human_in_the_loop.strategy.output": { + "tool_name": "rejected_tool", + "execute": False, + "tool_call_id": "tc-reject", + "feedback": "Tool execution for 'rejected_tool' was rejected by the user. " + "With user feedback: sensitive rejection feedback", + "final_tool_params": None, + }, + }, + ] + + metadata = { + key: value + for span in spying_tracer.spans + for key, value in span.tags.items() + if not key.endswith((".strategy.input", ".strategy.output")) + } + assert "sensitive" not in str(metadata) + assert "99" not in str(metadata) + + def test_omits_decision_content_when_content_tracing_is_disabled(self, eager_spying_tracer, monkeypatch, tools): + monkeypatch.setattr(tracing.tracer, "is_content_tracing_enabled", False) + messages = [ + ChatMessage.from_user("sensitive user message"), + ChatMessage.from_assistant( + tool_calls=[ToolCall(id="tc-1", tool_name="addition_tool", arguments={"a": 1, "b": 2})] + ), + ] + state = _state_with(messages, tools) + hook = _confirm_hook(ConfirmationUIResult(action="reject", feedback="sensitive rejection feedback")) + + _run_hooks(hooks={"before_tool": [hook]}, hook_point="before_tool", state=state) + + hook_span, tool_span = eager_spying_tracer.spans + assert tool_span.parent_span is hook_span + assert tool_span.tags == { + "haystack.tool.name": "addition_tool", + "haystack.tool.call.id": "tc-1", + "haystack.agent.hook.human_in_the_loop.strategy.type": ( + "haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy" + ), + "haystack.agent.hook.human_in_the_loop.strategy.decision": "reject", + } + assert "sensitive" not in str(eager_spying_tracer.spans) + + @pytest.mark.asyncio + async def test_adds_decision_tags_to_tool_span_async(self, spying_tracer, tools): + messages = [ + ChatMessage.from_user("sensitive user message"), + ChatMessage.from_assistant( + tool_calls=[ToolCall(id="tc-1", tool_name="addition_tool", arguments={"a": 1, "b": 2})] + ), + ] + state = _state_with(messages, tools) + hook = _confirm_hook(ConfirmationUIResult(action="reject", feedback="sensitive rejection feedback")) + + await _run_hooks_async(hooks={"before_tool": [hook]}, hook_point="before_tool", state=state) + + hook_span, tool_span = spying_tracer.spans + assert tool_span.parent_span is hook_span + assert tool_span.tags["haystack.agent.hook.human_in_the_loop.strategy.decision"] == "reject" + strategy_input = tool_span.tags["haystack.agent.hook.human_in_the_loop.strategy.input"] + assert strategy_input["tool_params"] == {"a": 1, "b": 2} + strategy_output = tool_span.tags["haystack.agent.hook.human_in_the_loop.strategy.output"] + assert "sensitive rejection feedback" in strategy_output["feedback"] + + def test_tool_span_can_close_without_a_decision(self, spying_tracer, monkeypatch, tools): + messages = [ + ChatMessage.from_user("add"), + ChatMessage.from_assistant(tool_calls=[ToolCall(tool_name="addition_tool", arguments={"a": 1, "b": 2})]), + ] + state = _state_with(messages, tools) + strategy = _confirm_strat(ConfirmationUIResult(action="confirm")) + + def suspend(**kwargs: Any) -> ToolExecutionDecision: + del kwargs + raise RuntimeError("execution suspended") + + monkeypatch.setattr(strategy, "run", suspend) + hook = ConfirmationHook(confirmation_strategies={"addition_tool": strategy}) + + with pytest.raises(RuntimeError, match="execution suspended"): + _run_hooks(hooks={"before_tool": [hook]}, hook_point="before_tool", state=state) + + hook_span, tool_span = spying_tracer.spans + assert tool_span.parent_span is hook_span + assert tool_span.tags == { + "haystack.tool.name": "addition_tool", + "haystack.tool.call.id": None, + "haystack.agent.hook.human_in_the_loop.strategy.type": ( + "haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy" + ), + "haystack.agent.hook.human_in_the_loop.strategy.input": { + "tool_name": "addition_tool", + "tool_description": "A tool that adds two integers together.", + "tool_params": {"a": 1, "b": 2}, + "tool_call_id": None, + }, + } + + def _echo(x: int) -> dict[str, int]: return {"echoed": x}