diff --git a/README.md b/README.md index 8a75917a..baa3d5c2 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,6 @@ that includes it. Once it reaches the end of its lifespan, the experiment will b |----------------------------------------------------------------------------------------|---------------------------------------|-------------------|--------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------| | [`OpenAIChatGenerator`][9] | Chat Generator Component | November 2025 | None | Open In Colab | [Discuss][10] | | [`MarkdownHeaderLevelInferrer`][15] | Preprocessor | January 2025 | None | None | [Discuss][16] | -| [`Agent`][17]; [`BreakpointConfirmationStrategy`][20]; [`HITLBreakpointException`][22] | Human in the Loop via Breakpoints | December 2025 | None | None | [Discuss][23] | | [`LLMSummarizer`][24] | Document Summarizer | January 2025 | None | None | [Discuss][25] | | [`InMemoryChatMessageStore`][1]; [`ChatMessageRetriever`][2]; [`ChatMessageWriter`][3] | Chat Message Store, Retriever, Writer | February 2025 | None | Open In Colab | [Discuss][4] | @@ -57,10 +56,6 @@ that includes it. Once it reaches the end of its lifespan, the experiment will b [10]: https://github.com/deepset-ai/haystack-experimental/discussions/361 [15]: https://github.com/deepset-ai/haystack-experimental/blob/main/haystack_experimental/components/preprocessors/md_header_level_inferrer.py [16]: https://github.com/deepset-ai/haystack-experimental/discussions/376 -[17]: https://github.com/deepset-ai/haystack-experimental/blob/main/haystack_experimental/components/agents/agent.py -[20]: https://github.com/deepset-ai/haystack-experimental/blob/main/haystack_experimental/components/agents/human_in_the_loop/strategies.py -[22]: https://github.com/deepset-ai/haystack-experimental/blob/main/haystack_experimental/components/agents/human_in_the_loop/errors.py -[23]: https://github.com/deepset-ai/haystack-experimental/discussions/381 [24]: https://github.com/deepset-ai/haystack-experimental/blob/main/haystack_experimental/components/sumarizers/llm_summarizer.py [25]: https://github.com/deepset-ai/haystack-experimental/discussions/382 @@ -88,6 +83,7 @@ that includes it. Once it reaches the end of its lifespan, the experiment will b |------------------------|----------------------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------| | `OpenAIFunctionCaller` | Function Calling Component | 0.3.0 | None | | `OpenAPITool` | OpenAPITool component | 0.3.0 | [Notebook](https://github.com/deepset-ai/haystack-experimental/blob/fe20b69b31243f8a3976e4661d9aa8c88a2847d2/examples/openapitool.ipynb) | +| `Agent`; `BreakpointConfirmationStrategy`; `HITLBreakpointException` | Human in the Loop via Breakpoints | 0.19.0 | None | | `EvaluationHarness` | Evaluation orchestrator | 0.7.0 | None | ## Usage diff --git a/haystack_experimental/components/agents/__init__.py b/haystack_experimental/components/agents/__init__.py deleted file mode 100644 index 4170413e..00000000 --- a/haystack_experimental/components/agents/__init__.py +++ /dev/null @@ -1,16 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -import sys -from typing import TYPE_CHECKING - -from lazy_imports import LazyImporter - -_import_structure = {"agent": ["Agent"]} - -if TYPE_CHECKING: - from .agent import Agent as Agent - -else: - sys.modules[__name__] = LazyImporter(name=__name__, module_file=__file__, import_structure=_import_structure) diff --git a/haystack_experimental/components/agents/agent.py b/haystack_experimental/components/agents/agent.py deleted file mode 100644 index 536d8223..00000000 --- a/haystack_experimental/components/agents/agent.py +++ /dev/null @@ -1,825 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -# ruff: noqa: I001 - -import inspect -from typing import Any, Literal - -# Monkey patch Haystack's AgentSnapshot with our extended version -import haystack.dataclasses.breakpoints as hdb -from haystack_experimental.dataclasses.breakpoints import AgentSnapshot - -hdb.AgentSnapshot = AgentSnapshot # type: ignore[misc] - -# Monkey patch Haystack's breakpoint functions with our extended versions -import haystack.core.pipeline.breakpoint as hs_breakpoint -import haystack_experimental.core.pipeline.breakpoint as exp_breakpoint - -hs_breakpoint._create_agent_snapshot = exp_breakpoint._create_agent_snapshot -hs_breakpoint._create_pipeline_snapshot_from_tool_invoker = exp_breakpoint._create_pipeline_snapshot_from_tool_invoker # type: ignore[assignment] - -from haystack import component, logging -from haystack.components.agents.agent import Agent as HaystackAgent, _ExecutionContext, _schema_from_dict -from haystack.human_in_the_loop.strategies import ( - ConfirmationStrategy, - _process_confirmation_strategies, - _process_confirmation_strategies_async, -) -from haystack.components.agents.state import replace_values -from haystack.components.generators.chat.types import ChatGenerator -from haystack.core.errors import BreakpointException, PipelineRuntimeError -from haystack.core.pipeline import AsyncPipeline, Pipeline -from haystack.core.pipeline.breakpoint import ( - _create_pipeline_snapshot_from_chat_generator, - _create_pipeline_snapshot_from_tool_invoker, - _save_pipeline_snapshot, - _should_trigger_tool_invoker_breakpoint, -) -from haystack.core.pipeline.utils import _deepcopy_with_exceptions -from haystack.core.serialization import default_from_dict -from haystack.dataclasses import ChatMessage -from haystack.dataclasses.breakpoints import AgentBreakpoint, ToolBreakpoint -from haystack.dataclasses.streaming_chunk import StreamingCallbackT -from haystack.tools import ToolsType, deserialize_tools_or_toolset_inplace -from haystack.utils.callable_serialization import deserialize_callable -from haystack.utils.deserialization import deserialize_component_inplace - -from haystack_experimental.chat_message_stores.types import ChatMessageStore -from haystack_experimental.components.agents.human_in_the_loop import HITLBreakpointException -from haystack_experimental.components.retrievers import ChatMessageRetriever -from haystack_experimental.components.writers import ChatMessageWriter - -logger = logging.getLogger(__name__) - - -@component -class Agent(HaystackAgent): - """ - A Haystack component that implements a tool-using agent with provider-agnostic chat model support. - - NOTE: This class extends Haystack's Agent component to add support for human-in-the-loop confirmation strategies. - - The component processes messages and executes tools until an exit condition is met. - The exit condition can be triggered either by a direct text response or by invoking a specific designated tool. - Multiple exit conditions can be specified. - - When you call an Agent without tools, it acts as a ChatGenerator, produces one response, then exits. - - ### Usage example - ```python - from haystack.components.generators.chat import OpenAIChatGenerator - from haystack.dataclasses import ChatMessage - from haystack.tools.tool import Tool - - from haystack_experimental.components.agents import Agent - from haystack_experimental.components.agents.human_in_the_loop import ( - HumanInTheLoopStrategy, - AlwaysAskPolicy, - NeverAskPolicy, - SimpleConsoleUI, - ) - - calculator_tool = Tool(name="calculator", description="A tool for performing mathematical calculations.", ...) - search_tool = Tool(name="search", description="A tool for searching the web.", ...) - - agent = Agent( - chat_generator=OpenAIChatGenerator(), - tools=[calculator_tool, search_tool], - confirmation_strategies={ - calculator_tool.name: HumanInTheLoopStrategy( - confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() - ), - search_tool.name: HumanInTheLoopStrategy( - confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() - ), - }, - ) - - # Run the agent - result = agent.run( - messages=[ChatMessage.from_user("Find information about Haystack")] - ) - - assert "messages" in result # Contains conversation history - ``` - """ - - def __init__( # noqa: PLR0913 - self, - *, - chat_generator: ChatGenerator, - tools: ToolsType | None = None, - system_prompt: str | None = None, - user_prompt: str | None = None, - required_variables: list[str] | Literal["*"] | None = None, - exit_conditions: list[str] | None = None, - state_schema: dict[str, Any] | None = None, - max_agent_steps: int = 100, - streaming_callback: StreamingCallbackT | None = None, - raise_on_tool_invocation_failure: bool = False, - confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy] | None = None, - tool_invoker_kwargs: dict[str, Any] | None = None, - chat_message_store: ChatMessageStore | None = None, - ) -> None: - """ - Initialize the agent component. - - :param chat_generator: An instance of the chat generator that your agent should use. It must support tools. - :param tools: List of Tool objects or a Toolset that the agent can use. - :param system_prompt: System prompt for the agent. - :param user_prompt: User prompt for the agent. If provided this is appended to the messages provided at runtime. - :param required_variables: - List variables that must be provided as input to user_prompt. - If a variable listed as required is not provided, an exception is raised. - If set to `"*"`, all variables found in the prompt are required. Optional. - :param exit_conditions: List of conditions that will cause the agent to return. - Can include "text" if the agent should return when it generates a message without tool calls, - or tool names that will cause the agent to return once the tool was executed. Defaults to ["text"]. - :param state_schema: The schema for the runtime state used by the tools. - :param max_agent_steps: Maximum number of steps the agent will run before stopping. Defaults to 100. - If the agent exceeds this number of steps, it will stop and return the current state. - :param streaming_callback: A callback that will be invoked when a response is streamed from the LLM. - The same callback can be configured to emit tool results when a tool is called. - :param raise_on_tool_invocation_failure: Should the agent raise an exception when a tool invocation fails? - If set to False, the exception will be turned into a chat message and passed to the LLM. - :param tool_invoker_kwargs: Additional keyword arguments to pass to the ToolInvoker. - :param chat_message_store: The ChatMessageStore that the agent can use to store - and retrieve chat messages history. - :raises TypeError: If the chat_generator does not support tools parameter in its run method. - :raises ValueError: If the exit_conditions are not valid. - """ - super(Agent, self).__init__( - chat_generator=chat_generator, - tools=tools, - system_prompt=system_prompt, - user_prompt=user_prompt, - required_variables=required_variables, - exit_conditions=exit_conditions, - state_schema=state_schema, - max_agent_steps=max_agent_steps, - streaming_callback=streaming_callback, - raise_on_tool_invocation_failure=raise_on_tool_invocation_failure, - tool_invoker_kwargs=tool_invoker_kwargs, - confirmation_strategies=confirmation_strategies, - ) - self._chat_message_store = chat_message_store - self._chat_message_retriever = ( - ChatMessageRetriever(chat_message_store=chat_message_store) if chat_message_store else None - ) - self._chat_message_writer = ( - ChatMessageWriter(chat_message_store=chat_message_store) if chat_message_store else None - ) - - def _initialize_fresh_execution( - self, - messages: list[ChatMessage], - streaming_callback: StreamingCallbackT | None, - requires_async: bool, - *, - system_prompt: str | None = None, - user_prompt: str | None = None, - generation_kwargs: dict[str, Any] | None = None, - tools: ToolsType | list[str] | None = None, - confirmation_strategy_context: dict[str, Any] | None = None, - chat_message_store_kwargs: dict[str, Any] | None = None, - **kwargs: dict[str, Any], - ) -> _ExecutionContext: - """ - Initialize execution context for a fresh run of the agent. - - :param messages: List of ChatMessage objects to start the agent with. - :param streaming_callback: Optional callback for streaming responses. - :param requires_async: Whether the agent run requires asynchronous execution. - :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt. - :param user_prompt: User prompt for the agent. If provided, it overrides the default user prompt and is - appended to the messages provided at runtime. - :param tools: Optional list of Tool objects, a Toolset, or list of tool names to use for this run. - When passing tool names, tools are selected from the Agent's originally configured tools. - - :param confirmation_strategy_context: Optional dictionary for passing request-scoped resources - to confirmation strategies. - :param chat_message_store_kwargs: Optional dictionary of keyword arguments to pass to the ChatMessageStore. - For example, it can include the `chat_history_id` and `last_k` parameters for retrieving chat history. - :param kwargs: Additional data to pass to the State used by the Agent. - """ - exe_context = super(Agent, self)._initialize_fresh_execution( - messages=messages, - streaming_callback=streaming_callback, - requires_async=requires_async, - system_prompt=system_prompt, - user_prompt=user_prompt, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - chat_message_store_kwargs=chat_message_store_kwargs, - **kwargs, - ) - - # NOTE: difference with parent method to add chat message retrieval - if self._chat_message_retriever: - retriever_kwargs = _select_kwargs(self._chat_message_retriever, chat_message_store_kwargs or {}) - if "chat_history_id" in retriever_kwargs: - updated_messages = self._chat_message_retriever.run( - current_messages=exe_context.state.get("messages", []), **retriever_kwargs - )["messages"] - # We replace the messages in state with the updated messages including chat history - exe_context.state.set("messages", updated_messages, handler_override=replace_values) - - return _ExecutionContext( - state=exe_context.state, - component_visits=exe_context.component_visits, - chat_generator_inputs=exe_context.chat_generator_inputs, - tool_invoker_inputs=exe_context.tool_invoker_inputs, - confirmation_strategy_context=exe_context.confirmation_strategy_context, - ) - - def _initialize_from_snapshot( # type: ignore[override] - self, - snapshot: AgentSnapshot, - streaming_callback: StreamingCallbackT | None, - requires_async: bool, - *, - generation_kwargs: dict[str, Any] | None = None, - tools: ToolsType | list[str] | None = None, - confirmation_strategy_context: dict[str, Any] | None = None, - ) -> _ExecutionContext: - """ - Initialize execution context from an AgentSnapshot. - - :param snapshot: An AgentSnapshot containing the state of a previously saved agent execution. - :param streaming_callback: Optional callback for streaming responses. - :param requires_async: Whether the agent run requires asynchronous execution. - :param generation_kwargs: Additional keyword arguments for chat generator. These parameters will - override the parameters passed during component initialization. - :param tools: Optional list of Tool objects, a Toolset, or list of tool names to use for this run. - When passing tool names, tools are selected from the Agent's originally configured tools. - :param confirmation_strategy_context: Optional dictionary for passing request-scoped resources - to confirmation strategies. - """ - exe_context = super(Agent, self)._initialize_from_snapshot( - snapshot=snapshot, - streaming_callback=streaming_callback, - requires_async=requires_async, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - ) - # NOTE: Only difference is to use pass tool_execution_decisions to _ExecutionContext - return _ExecutionContext( - state=exe_context.state, - component_visits=exe_context.component_visits, - chat_generator_inputs=exe_context.chat_generator_inputs, - tool_invoker_inputs=exe_context.tool_invoker_inputs, - counter=exe_context.counter, - skip_chat_generator=exe_context.skip_chat_generator, - confirmation_strategy_context=exe_context.confirmation_strategy_context, - tool_execution_decisions=snapshot.tool_execution_decisions, - ) - - def run( # type: ignore[override] # noqa: PLR0915 PLR0912 C901 - self, - messages: list[ChatMessage], - streaming_callback: StreamingCallbackT | None = None, - *, - generation_kwargs: dict[str, Any] | None = None, - break_point: AgentBreakpoint | None = None, - snapshot: AgentSnapshot | None = None, - system_prompt: str | None = None, - user_prompt: str | None = None, - tools: ToolsType | list[str] | None = None, - confirmation_strategy_context: dict[str, Any] | None = None, - chat_message_store_kwargs: dict[str, Any] | None = None, - **kwargs: Any, - ) -> dict[str, Any]: - """ - Process messages and execute tools until an exit condition is met. - - :param messages: List of Haystack ChatMessage objects to process. - :param streaming_callback: A callback that will be invoked when a response is streamed from the LLM. - The same callback can be configured to emit tool results when a tool is called. - :param generation_kwargs: Additional keyword arguments for LLM. These parameters will - override the parameters passed during component initialization. - :param break_point: An AgentBreakpoint, can be a Breakpoint for the "chat_generator" or a ToolBreakpoint - for "tool_invoker". - :param snapshot: A dictionary containing a snapshot of a previously saved agent execution. The snapshot contains - the relevant information to restart the Agent execution from where it left off. - :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt. - :param user_prompt: User prompt for the agent. If provided, it overrides the default user prompt and is - appended to the messages provided at runtime. - :param tools: Optional list of Tool objects, a Toolset, or list of tool names to use for this run. - When passing tool names, tools are selected from the Agent's originally configured tools. - :param confirmation_strategy_context: Optional dictionary for passing request-scoped resources - to confirmation strategies. Useful in web/server environments to provide per-request - objects (e.g., WebSocket connections, async queues, Redis pub/sub clients) that strategies - can use for non-blocking user interaction. - :param chat_message_store_kwargs: Optional dictionary of keyword arguments to pass to the ChatMessageStore. - For example, it can include the `chat_history_id` and `last_k` parameters for retrieving chat history. - :param kwargs: Additional data to pass to the State schema used by the Agent. - The keys must match the schema defined in the Agent's `state_schema`. - :returns: - A dictionary with the following keys: - - "messages": List of all messages exchanged during the agent's run. - - "last_message": The last message exchanged during the agent's run. - - Any additional keys defined in the `state_schema`. - :raises RuntimeError: If the Agent component wasn't warmed up before calling `run()`. - :raises BreakpointException: If an agent breakpoint is triggered. - """ - - agent_inputs = { - "messages": messages, - "streaming_callback": streaming_callback, - "break_point": break_point, - "snapshot": snapshot, - **kwargs, - } - # TODO Probably good to add a warning in runtime checks that BreakpointConfirmationStrategy will take - # precedence over passing a ToolBreakpoint - # Support both old signature (break_point) and new signature (break_point, tools) - _runtime_checks_kwargs: dict[str, Any] = {"break_point": break_point} - if "tools" in inspect.signature(HaystackAgent._runtime_checks).parameters: - _runtime_checks_kwargs["tools"] = tools - self._runtime_checks(**_runtime_checks_kwargs) - - if snapshot: - exe_context = self._initialize_from_snapshot( - snapshot=snapshot, - streaming_callback=streaming_callback, - requires_async=False, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - ) - else: - exe_context = self._initialize_fresh_execution( - messages=messages, - streaming_callback=streaming_callback, - requires_async=False, - system_prompt=system_prompt, - user_prompt=user_prompt, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - chat_message_store_kwargs=chat_message_store_kwargs, - **kwargs, - ) - - with self._create_agent_span() as span: - span.set_content_tag("haystack.agent.input", _deepcopy_with_exceptions(agent_inputs)) - - while exe_context.counter < self.max_agent_steps: - # We skip the chat generator when restarting from a snapshot from a ToolBreakpoint - if exe_context.skip_chat_generator: - llm_messages = exe_context.state.get("messages", [])[-1:] - # Set to False so the next iteration will call the chat generator - exe_context.skip_chat_generator = False - else: - try: - result = Pipeline._run_component( - component_name="chat_generator", - component={"instance": self.chat_generator}, - inputs={ - "messages": exe_context.state.data["messages"], - **exe_context.chat_generator_inputs, - }, - component_visits=exe_context.component_visits, - parent_span=span, - break_point=break_point.break_point if isinstance(break_point, AgentBreakpoint) else None, - ) - except (BreakpointException, PipelineRuntimeError) as e: - if isinstance(e, BreakpointException): - agent_name = break_point.agent_name if break_point else None - saved_bp = break_point - else: - agent_name = getattr(self, "__component_name__", None) - saved_bp = None - - e.pipeline_snapshot = _create_pipeline_snapshot_from_chat_generator( - agent_name=agent_name, execution_context=exe_context, break_point=saved_bp - ) - if isinstance(e, BreakpointException): - e._break_point = e.pipeline_snapshot.break_point - # If Agent is not in a pipeline, we save the snapshot to a file. - # Checked by __component_name__ not being set. - if getattr(self, "__component_name__", None) is None: - full_file_path = _save_pipeline_snapshot(pipeline_snapshot=e.pipeline_snapshot) - e.pipeline_snapshot_file_path = full_file_path - raise e - - llm_messages = result["replies"] - exe_context.state.set("messages", llm_messages) - - # Check if any of the LLM responses contain a tool call or if the LLM is not using tools - if not any(msg.tool_call for msg in llm_messages) or self._tool_invoker is None: - exe_context.counter += 1 - break - - # We only pass down the breakpoint if the tool name matches the tool call in the LLM messages - resolved_break_point = None - break_point_to_pass = None - if ( - break_point - and isinstance(break_point.break_point, ToolBreakpoint) - and _should_trigger_tool_invoker_breakpoint( - break_point=break_point.break_point, llm_messages=llm_messages - ) - ): - resolved_break_point = break_point - break_point_to_pass = resolved_break_point.break_point - - # NOTE: difference with parent method to add support HITLBreakpointException - try: - # Apply confirmation strategies and update State and messages sent to ToolInvoker - # Run confirmation strategies to get updated tool call messages and modified chat history - modified_tool_call_messages, new_chat_history = _process_confirmation_strategies( - confirmation_strategies=self._confirmation_strategies, - messages_with_tool_calls=llm_messages, - execution_context=exe_context, - ) - # Replace the chat history in state with the modified one - exe_context.state.set(key="messages", value=new_chat_history, handler_override=replace_values) - except HITLBreakpointException as tbp_error: - # We create a break_point to pass to Pipeline._run_component - resolved_break_point = AgentBreakpoint( - agent_name=getattr(self, "__component_name__", ""), - break_point=ToolBreakpoint( - component_name="tool_invoker", - tool_name=tbp_error.tool_name, - visit_count=exe_context.component_visits["tool_invoker"], - snapshot_file_path=tbp_error.snapshot_file_path, - ), - ) - break_point_to_pass = resolved_break_point.break_point - # If we hit a HITL breakpoint, we skip passing modified messages to ToolInvoker - modified_tool_call_messages = llm_messages - - # Run ToolInvoker - try: - # We only send the messages from the LLM to the tool invoker - tool_invoker_result = Pipeline._run_component( - component_name="tool_invoker", - component={"instance": self._tool_invoker}, - inputs={ - "messages": modified_tool_call_messages, - "state": exe_context.state, - **exe_context.tool_invoker_inputs, - }, - component_visits=exe_context.component_visits, - parent_span=span, - break_point=break_point_to_pass, - ) - except (BreakpointException, PipelineRuntimeError) as e: - if isinstance(e, BreakpointException): - agent_name = resolved_break_point.agent_name if resolved_break_point else None - tool_name = e.break_point.tool_name if isinstance(e.break_point, ToolBreakpoint) else None - saved_bp = resolved_break_point - else: - agent_name = getattr(self, "__component_name__", None) - tool_name = getattr(e.__cause__, "tool_name", None) - saved_bp = None - - e.pipeline_snapshot = _create_pipeline_snapshot_from_tool_invoker( - tool_name=tool_name, agent_name=agent_name, execution_context=exe_context, break_point=saved_bp - ) - if isinstance(e, BreakpointException): - e._break_point = e.pipeline_snapshot.break_point - # If Agent is not in a pipeline, we save the snapshot to a file. - # Checked by __component_name__ not being set. - if getattr(self, "__component_name__", None) is None: - full_file_path = _save_pipeline_snapshot(pipeline_snapshot=e.pipeline_snapshot) - e.pipeline_snapshot_file_path = full_file_path - raise e - - # Set execution context tool execution decisions to empty after applying them b/c they should only - # be used once for the current tool calls - exe_context.tool_execution_decisions = None - tool_messages = tool_invoker_result["tool_messages"] - exe_context.state = tool_invoker_result["state"] - exe_context.state.set("messages", tool_messages) - - # Check if any LLM message's tool call name matches an exit condition - if self.exit_conditions != ["text"] and self._check_exit_conditions(llm_messages, tool_messages): - exe_context.counter += 1 - break - - # Increment the step counter - exe_context.counter += 1 - - if exe_context.counter >= self.max_agent_steps: - logger.warning( - "Agent reached maximum agent steps of {max_agent_steps}, stopping.", - max_agent_steps=self.max_agent_steps, - ) - span.set_content_tag("haystack.agent.output", exe_context.state.data) - span.set_tag("haystack.agent.steps_taken", exe_context.counter) - - result = {**exe_context.state.data} - if msgs := result.get("messages"): - result["last_message"] = msgs[-1] - - # Write messages to ChatMessageStore if configured - if self._chat_message_writer: - writer_kwargs = _select_kwargs(self._chat_message_writer, chat_message_store_kwargs or {}) - if "chat_history_id" in writer_kwargs: - self._chat_message_writer.run(messages=result["messages"], **writer_kwargs) - - return result - - async def run_async( # type: ignore[override] # noqa: PLR0915 PLR0912 - self, - messages: list[ChatMessage], - streaming_callback: StreamingCallbackT | None = None, - *, - generation_kwargs: dict[str, Any] | None = None, - break_point: AgentBreakpoint | None = None, - snapshot: AgentSnapshot | None = None, - system_prompt: str | None = None, - user_prompt: str | None = None, - tools: ToolsType | list[str] | None = None, - confirmation_strategy_context: dict[str, Any] | None = None, - chat_message_store_kwargs: dict[str, Any] | None = None, - **kwargs: Any, - ) -> dict[str, Any]: - """ - Asynchronously process messages and execute tools until the exit condition is met. - - This is the asynchronous version of the `run` method. It follows the same logic but uses - asynchronous operations where possible, such as calling the `run_async` method of the ChatGenerator - if available. - - :param messages: List of Haystack ChatMessage objects to process. - :param streaming_callback: An asynchronous callback that will be invoked when a response is streamed from the - LLM. The same callback can be configured to emit tool results when a tool is called. - :param generation_kwargs: Additional keyword arguments for LLM. These parameters will - override the parameters passed during component initialization. - :param break_point: An AgentBreakpoint, can be a Breakpoint for the "chat_generator" or a ToolBreakpoint - for "tool_invoker". - :param snapshot: A dictionary containing a snapshot of a previously saved agent execution. The snapshot contains - the relevant information to restart the Agent execution from where it left off. - :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt. - :param user_prompt: User prompt for the agent. If provided, it overrides the default user prompt and is - appended to the messages provided at runtime. - :param tools: Optional list of Tool objects, a Toolset, or list of tool names to use for this run. - :param confirmation_strategy_context: Optional dictionary for passing request-scoped resources - to confirmation strategies. Useful in web/server environments to provide per-request - objects (e.g., WebSocket connections, async queues, Redis pub/sub clients) that strategies - can use for non-blocking user interaction. - :param chat_message_store_kwargs: Optional dictionary of keyword arguments to pass to the ChatMessageStore. - For example, it can include the `chat_history_id` and `last_k` parameters for retrieving chat history. - :param kwargs: Additional data to pass to the State schema used by the Agent. - The keys must match the schema defined in the Agent's `state_schema`. - :returns: - A dictionary with the following keys: - - "messages": List of all messages exchanged during the agent's run. - - "last_message": The last message exchanged during the agent's run. - - Any additional keys defined in the `state_schema`. - :raises RuntimeError: If the Agent component wasn't warmed up before calling `run_async()`. - :raises BreakpointException: If an agent breakpoint is triggered. - """ - - agent_inputs = { - "messages": messages, - "streaming_callback": streaming_callback, - "break_point": break_point, - "snapshot": snapshot, - **kwargs, - } - _runtime_checks_kwargs: dict[str, Any] = {"break_point": break_point} - if "tools" in inspect.signature(HaystackAgent._runtime_checks).parameters: - _runtime_checks_kwargs["tools"] = tools - self._runtime_checks(**_runtime_checks_kwargs) - - if snapshot: - exe_context = self._initialize_from_snapshot( - snapshot=snapshot, - streaming_callback=streaming_callback, - requires_async=True, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - ) - else: - exe_context = self._initialize_fresh_execution( - messages=messages, - streaming_callback=streaming_callback, - requires_async=True, - system_prompt=system_prompt, - user_prompt=user_prompt, - generation_kwargs=generation_kwargs, - tools=tools, - confirmation_strategy_context=confirmation_strategy_context, - chat_message_store_kwargs=chat_message_store_kwargs, - **kwargs, - ) - - with self._create_agent_span() as span: - span.set_content_tag("haystack.agent.input", _deepcopy_with_exceptions(agent_inputs)) - - while exe_context.counter < self.max_agent_steps: - # We skip the chat generator when restarting from a snapshot from a ToolBreakpoint - if exe_context.skip_chat_generator: - llm_messages = exe_context.state.get("messages", [])[-1:] - # Set to False so the next iteration will call the chat generator - exe_context.skip_chat_generator = False - else: - try: - result = await AsyncPipeline._run_component_async( - component_name="chat_generator", - component={"instance": self.chat_generator}, - component_inputs={ - "messages": exe_context.state.data["messages"], - **exe_context.chat_generator_inputs, - }, - component_visits=exe_context.component_visits, - parent_span=span, - break_point=break_point.break_point if isinstance(break_point, AgentBreakpoint) else None, - ) - except BreakpointException as e: - e.pipeline_snapshot = _create_pipeline_snapshot_from_chat_generator( - agent_name=break_point.agent_name if break_point else None, - execution_context=exe_context, - break_point=break_point, - ) - e._break_point = e.pipeline_snapshot.break_point - # We check if the agent is part of a pipeline by checking for __component_name__ - # If it is not in a pipeline, we save the snapshot to a file. - in_pipeline = getattr(self, "__component_name__", None) is not None - if not in_pipeline: - full_file_path = _save_pipeline_snapshot(pipeline_snapshot=e.pipeline_snapshot) - e.pipeline_snapshot_file_path = full_file_path - raise e - - llm_messages = result["replies"] - exe_context.state.set("messages", llm_messages) - - # Check if any of the LLM responses contain a tool call or if the LLM is not using tools - if not any(msg.tool_call for msg in llm_messages) or self._tool_invoker is None: - exe_context.counter += 1 - break - - # We only pass down the breakpoint if the tool name matches the tool call in the LLM messages - resolved_break_point = None - break_point_to_pass = None - if ( - break_point - and isinstance(break_point.break_point, ToolBreakpoint) - and _should_trigger_tool_invoker_breakpoint( - break_point=break_point.break_point, llm_messages=llm_messages - ) - ): - resolved_break_point = break_point - break_point_to_pass = resolved_break_point.break_point - - # NOTE: difference with parent method to add support HITLBreakpointException - try: - # Apply confirmation strategies and update State and messages sent to ToolInvoker - # Run confirmation strategies to get updated tool call messages and modified chat history (async) - modified_tool_call_messages, new_chat_history = await _process_confirmation_strategies_async( - confirmation_strategies=self._confirmation_strategies, - messages_with_tool_calls=llm_messages, - execution_context=exe_context, - ) - # Replace the chat history in state with the modified one - exe_context.state.set(key="messages", value=new_chat_history, handler_override=replace_values) - except HITLBreakpointException as tbp_error: - # We create a break_point to pass into _check_tool_invoker_breakpoint - resolved_break_point = AgentBreakpoint( - agent_name=getattr(self, "__component_name__", ""), - break_point=ToolBreakpoint( - component_name="tool_invoker", - tool_name=tbp_error.tool_name, - visit_count=exe_context.component_visits["tool_invoker"], - snapshot_file_path=tbp_error.snapshot_file_path, - ), - ) - break_point_to_pass = resolved_break_point.break_point - # If we hit a HITL breakpoint, we skip passing modified messages to ToolInvoker - modified_tool_call_messages = llm_messages - - try: - # We only send the messages from the LLM to the tool invoker - tool_invoker_result = await AsyncPipeline._run_component_async( - component_name="tool_invoker", - component={"instance": self._tool_invoker}, - component_inputs={ - "messages": modified_tool_call_messages, - "state": exe_context.state, - **exe_context.tool_invoker_inputs, - }, - component_visits=exe_context.component_visits, - parent_span=span, - break_point=break_point_to_pass, - ) - except BreakpointException as e: - e.pipeline_snapshot = _create_pipeline_snapshot_from_tool_invoker( - tool_name=e.break_point.tool_name if isinstance(e.break_point, ToolBreakpoint) else None, - agent_name=resolved_break_point.agent_name if resolved_break_point else None, - execution_context=exe_context, - break_point=resolved_break_point, - ) - e._break_point = e.pipeline_snapshot.break_point - # If Agent is not in a pipeline, we save the snapshot to a file. - # Checked by __component_name__ not being set. - if getattr(self, "__component_name__", None) is None: - full_file_path = _save_pipeline_snapshot(pipeline_snapshot=e.pipeline_snapshot) - e.pipeline_snapshot_file_path = full_file_path - raise e - - # Set execution context tool execution decisions to empty after applying them b/c they should only - # be used once for the current tool calls - exe_context.tool_execution_decisions = None - tool_messages = tool_invoker_result["tool_messages"] - exe_context.state = tool_invoker_result["state"] - exe_context.state.set("messages", tool_messages) - - # Check if any LLM message's tool call name matches an exit condition - if self.exit_conditions != ["text"] and self._check_exit_conditions(llm_messages, tool_messages): - exe_context.counter += 1 - break - - # Increment the step counter - exe_context.counter += 1 - - if exe_context.counter >= self.max_agent_steps: - logger.warning( - "Agent reached maximum agent steps of {max_agent_steps}, stopping.", - max_agent_steps=self.max_agent_steps, - ) - span.set_content_tag("haystack.agent.output", exe_context.state.data) - span.set_tag("haystack.agent.steps_taken", exe_context.counter) - - result = {**exe_context.state.data} - if msgs := result.get("messages"): - result["last_message"] = msgs[-1] - - # Write messages to ChatMessageStore if configured - if self._chat_message_writer: - writer_kwargs = _select_kwargs(self._chat_message_writer, chat_message_store_kwargs or {}) - if "chat_history_id" in writer_kwargs: - self._chat_message_writer.run(messages=result["messages"], **writer_kwargs) - - return result - - def to_dict(self) -> dict[str, Any]: - """ - Serialize the component to a dictionary. - - :return: Dictionary with serialized data - """ - data = super(Agent, self).to_dict() - # NOTE: This is different from the base Agent class to handle ChatMessageStore serialization - data["init_parameters"]["chat_message_store"] = ( - self._chat_message_store.to_dict() if self._chat_message_store is not None else None - ) - return data - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "Agent": - """ - Deserialize the agent from a dictionary. - - :param data: Dictionary to deserialize from - :return: Deserialized agent - """ - init_params = data.get("init_parameters", {}) - - deserialize_component_inplace(init_params, key="chat_generator") - - if init_params.get("state_schema") is not None: - init_params["state_schema"] = _schema_from_dict(init_params["state_schema"]) - - if init_params.get("streaming_callback") is not None: - init_params["streaming_callback"] = deserialize_callable(init_params["streaming_callback"]) - - deserialize_tools_or_toolset_inplace(init_params, key="tools") - - if init_params.get("confirmation_strategies") is not None: - restored: dict[str | tuple[str, ...], Any] = {} - - for raw_key in init_params["confirmation_strategies"].keys(): - deserialize_component_inplace(init_params["confirmation_strategies"], key=raw_key) - strategy = init_params["confirmation_strategies"][raw_key] - - if isinstance(raw_key, list): - key = tuple(raw_key) - else: - key = raw_key - restored[key] = strategy - - init_params["confirmation_strategies"] = restored - - # NOTE: This is different from the base Agent class to handle ChatMessageStore deserialization - if "chat_message_store" in init_params and init_params["chat_message_store"] is not None: - deserialize_component_inplace(init_params, key="chat_message_store") - - return default_from_dict(cls, data) - - -def _select_kwargs(obj: Any, source: dict) -> dict[str, Any]: - """ - Select only those key-value pairs from source dict that are valid parameters for obj.run() method. - """ - sig = inspect.signature(obj.run) - allowed = set(sig.parameters.keys()) - return {k: v for k, v in source.items() if k in allowed} diff --git a/haystack_experimental/components/agents/human_in_the_loop/__init__.py b/haystack_experimental/components/agents/human_in_the_loop/__init__.py deleted file mode 100644 index 07c9711b..00000000 --- a/haystack_experimental/components/agents/human_in_the_loop/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -import sys -from typing import TYPE_CHECKING - -from lazy_imports import LazyImporter - -_import_structure = {"errors": ["HITLBreakpointException"], "strategies": ["BreakpointConfirmationStrategy"]} - -if TYPE_CHECKING: - from .errors import HITLBreakpointException as HITLBreakpointException - from .strategies import BreakpointConfirmationStrategy as BreakpointConfirmationStrategy - -else: - sys.modules[__name__] = LazyImporter(name=__name__, module_file=__file__, import_structure=_import_structure) diff --git a/haystack_experimental/components/agents/human_in_the_loop/breakpoint.py b/haystack_experimental/components/agents/human_in_the_loop/breakpoint.py deleted file mode 100644 index 8dd948e1..00000000 --- a/haystack_experimental/components/agents/human_in_the_loop/breakpoint.py +++ /dev/null @@ -1,64 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from copy import deepcopy - -from haystack.dataclasses.breakpoints import AgentSnapshot, ToolBreakpoint -from haystack.human_in_the_loop.strategies import _prepare_tool_args -from haystack.utils import _deserialize_value_with_schema - - -def get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot: AgentSnapshot, breakpoint_tool_only: bool = True -) -> tuple[list[dict], dict[str, str]]: - """ - Extract tool calls and tool descriptions from an AgentSnapshot. - - By default, only the tool call that caused the breakpoint is processed and its arguments are reconstructed. - This is useful for scenarios where you want to present the relevant tool call and its description - to a human for confirmation before execution. - - :param agent_snapshot: The AgentSnapshot from which to extract tool calls and descriptions. - :param breakpoint_tool_only: If True, only the tool call that caused the breakpoint is returned. If False, all tool - calls are returned. - :returns: - A tuple containing a list of tool call dictionaries and a dictionary of tool descriptions - """ - break_point = agent_snapshot.break_point.break_point - if not isinstance(break_point, ToolBreakpoint): - raise TypeError("The provided AgentSnapshot does not contain a ToolBreakpoint.") - - tool_caused_break_point = break_point.tool_name - - # Deserialize the tool invoker inputs from the snapshot - tool_invoker_inputs = _deserialize_value_with_schema(deepcopy(agent_snapshot.component_inputs["tool_invoker"])) - tool_call_messages = tool_invoker_inputs["messages"] - state = tool_invoker_inputs["state"] - tool_name_to_tool = {t.name: t for t in tool_invoker_inputs["tools"]} - - tool_calls = [] - for msg in tool_call_messages: - if msg.tool_calls: - tool_calls.extend(msg.tool_calls) - serialized_tcs = [tc.to_dict() for tc in tool_calls] - - # Reconstruct the final arguments for each tool call - tool_descriptions = {} - updated_tool_calls = [] - for tc in serialized_tcs: - # Only process the tool that caused the breakpoint if breakpoint_tool_only is True - if breakpoint_tool_only and tc["tool_name"] != tool_caused_break_point: - continue - - final_args = _prepare_tool_args( - tool=tool_name_to_tool[tc["tool_name"]], - tool_call_arguments=tc["arguments"], - state=state, - streaming_callback=tool_invoker_inputs.get("streaming_callback", None), - enable_streaming_passthrough=tool_invoker_inputs.get("enable_streaming_passthrough", False), - ) - updated_tool_calls.append({**tc, "arguments": final_args}) - tool_descriptions[tc["tool_name"]] = tool_name_to_tool[tc["tool_name"]].description - - return updated_tool_calls, tool_descriptions diff --git a/haystack_experimental/components/agents/human_in_the_loop/errors.py b/haystack_experimental/components/agents/human_in_the_loop/errors.py deleted file mode 100644 index c45b30a7..00000000 --- a/haystack_experimental/components/agents/human_in_the_loop/errors.py +++ /dev/null @@ -1,24 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - - -class HITLBreakpointException(Exception): - """ - Exception raised when a tool execution is paused by a ConfirmationStrategy (e.g. BreakpointConfirmationStrategy). - """ - - def __init__(self, message: str, tool_name: str, snapshot_file_path: str, tool_call_id: str | None = None) -> None: - """ - Initialize the HITLBreakpointException. - - :param message: The exception message. - :param tool_name: The name of the tool whose execution is paused. - :param snapshot_file_path: The file path to the saved pipeline snapshot. - :param tool_call_id: Optional unique identifier for the tool call. This can be used to track and correlate - the decision with a specific tool invocation. - """ - super().__init__(message) - self.tool_name = tool_name - self.snapshot_file_path = snapshot_file_path - self.tool_call_id = tool_call_id diff --git a/haystack_experimental/components/agents/human_in_the_loop/strategies.py b/haystack_experimental/components/agents/human_in_the_loop/strategies.py deleted file mode 100644 index 14c15c22..00000000 --- a/haystack_experimental/components/agents/human_in_the_loop/strategies.py +++ /dev/null @@ -1,128 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from typing import Any - -from haystack.core.serialization import default_from_dict, default_to_dict -from haystack.human_in_the_loop.dataclasses import ToolExecutionDecision - -from haystack_experimental.components.agents.human_in_the_loop import HITLBreakpointException - -_REJECTION_FEEDBACK_TEMPLATE = "Tool execution for '{tool_name}' was rejected by the user." -_MODIFICATION_FEEDBACK_TEMPLATE = ( - "The parameters for tool '{tool_name}' were updated by the user to:\n{final_tool_params}" -) - - -class BreakpointConfirmationStrategy: - """ - Confirmation strategy that raises a tool breakpoint exception to pause execution and gather user feedback. - - This strategy is designed for scenarios where immediate user interaction is not possible. - When a tool execution requires confirmation, it raises an `HITLBreakpointException`, which is caught by the Agent. - The Agent then serialize its current state, including the tool call details. This information can then be used to - notify a user to review and confirm the tool execution. - """ - - def __init__(self, snapshot_file_path: str) -> None: - """ - Initialize the BreakpointConfirmationStrategy. - - :param snapshot_file_path: The path to the directory that the snapshot should be saved. - """ - self.snapshot_file_path = snapshot_file_path - - def run( - self, - *, - tool_name: str, - tool_description: str, # noqa: ARG002 - tool_params: dict[str, Any], # noqa: ARG002 - tool_call_id: str | None = None, - confirmation_strategy_context: dict[str, Any] | None = None, # noqa: ARG002 - ) -> ToolExecutionDecision: - """ - Run the breakpoint confirmation strategy for a given tool and its parameters. - - :param tool_name: - The name of the tool to be executed. - :param tool_description: - The description of the tool. - :param tool_params: - The parameters to be passed to the tool. - :param tool_call_id: - Optional unique identifier for the tool call. This can be used to track and correlate the decision with a - specific tool invocation. - :param confirmation_strategy_context: - Optional dictionary for passing request-scoped resources. Not used by this strategy but included for - interface compatibility. - - :raises HITLBreakpointException: - Always raises an `HITLBreakpointException` exception to signal that user confirmation is required. - - :returns: - This method does not return; it always raises an exception. - """ - raise HITLBreakpointException( - message=f"Tool execution for '{tool_name}' requires user confirmation.", - tool_name=tool_name, - tool_call_id=tool_call_id, - snapshot_file_path=self.snapshot_file_path, - ) - - async def run_async( - self, - *, - 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, - ) -> ToolExecutionDecision: - """ - Async version of run. Calls the sync run() method. - - :param tool_name: - The name of the tool to be executed. - :param tool_description: - The description of the tool. - :param tool_params: - The parameters to be passed to the tool. - :param tool_call_id: - Optional unique identifier for the tool call. - :param confirmation_strategy_context: - Optional dictionary for passing request-scoped resources. - - :raises HITLBreakpointException: - Always raises an `HITLBreakpointException` exception to signal that user confirmation is required. - - :returns: - This method does not return; it always raises an exception. - """ - return self.run( - tool_name=tool_name, - tool_description=tool_description, - tool_params=tool_params, - tool_call_id=tool_call_id, - confirmation_strategy_context=confirmation_strategy_context, - ) - - def to_dict(self) -> dict[str, Any]: - """ - Serializes the BreakpointConfirmationStrategy to a dictionary. - """ - return default_to_dict(self, snapshot_file_path=self.snapshot_file_path) - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "BreakpointConfirmationStrategy": - """ - Deserializes the BreakpointConfirmationStrategy from a dictionary. - - :param data: - Dictionary to deserialize from. - - :returns: - Deserialized BreakpointConfirmationStrategy. - """ - return default_from_dict(cls, data) diff --git a/haystack_experimental/components/generators/chat/openai.py b/haystack_experimental/components/generators/chat/openai.py index 154ff1eb..e37b5bec 100644 --- a/haystack_experimental/components/generators/chat/openai.py +++ b/haystack_experimental/components/generators/chat/openai.py @@ -7,6 +7,7 @@ from haystack import component from haystack.components.generators.chat.openai import OpenAIChatGenerator as BaseOpenAIChatGenerator +from haystack.components.generators.utils import _normalize_messages from haystack.dataclasses import ChatMessage, StreamingCallbackT from haystack.tools import ToolsType @@ -55,7 +56,7 @@ class OpenAIChatGenerator(BaseOpenAIChatGenerator): @component.output_types(replies=list[ChatMessage]) def run( self, - messages: list[ChatMessage], + messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, *, @@ -67,7 +68,8 @@ def run( Invokes chat completion based on the provided messages and generation parameters. :param messages: - A list of ChatMessage instances representing the input messages. + A list of ChatMessage instances representing the input messages. If a string is provided, it is + converted to a list containing a ChatMessage with user role. :param streaming_callback: A callback function that is called when a new token is received from the stream. :param generation_kwargs: @@ -97,6 +99,8 @@ def run( - `hallucination_risk`: The EDFL hallucination risk bound. - `hallucination_rationale`: The rationale behind the hallucination decision. """ + messages = _normalize_messages(messages) + if len(messages) == 0: return {"replies": []} @@ -122,7 +126,7 @@ def run( @component.output_types(replies=list[ChatMessage]) async def run_async( self, - messages: list[ChatMessage], + messages: list[ChatMessage] | str, streaming_callback: StreamingCallbackT | None = None, generation_kwargs: dict[str, Any] | None = None, *, @@ -137,7 +141,8 @@ async def run_async( but can be used with `await` in async code. :param messages: - A list of ChatMessage instances representing the input messages. + A list of ChatMessage instances representing the input messages. If a string is provided, it is + converted to a list containing a ChatMessage with user role. :param streaming_callback: A callback function that is called when a new token is received from the stream. Must be a coroutine. @@ -168,6 +173,8 @@ async def run_async( - `hallucination_risk`: The EDFL hallucination risk bound. - `hallucination_rationale`: The rationale behind the hallucination decision. """ + messages = _normalize_messages(messages) + if len(messages) == 0: return {"replies": []} diff --git a/haystack_experimental/core/pipeline/breakpoint.py b/haystack_experimental/core/pipeline/breakpoint.py deleted file mode 100644 index 9c300890..00000000 --- a/haystack_experimental/core/pipeline/breakpoint.py +++ /dev/null @@ -1,118 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from dataclasses import replace -from datetime import datetime -from typing import Any - -from haystack import logging -from haystack.components.agents.agent import _ExecutionContext -from haystack.core.pipeline.utils import _deepcopy_with_exceptions -from haystack.dataclasses.breakpoints import AgentBreakpoint, PipelineSnapshot, PipelineState, ToolBreakpoint -from haystack.human_in_the_loop import ToolExecutionDecision -from haystack.utils.base_serialization import _serialize_value_with_schema -from haystack.utils.misc import _get_output_dir - -from haystack_experimental.dataclasses.breakpoints import AgentSnapshot - -logger = logging.getLogger(__name__) - - -def _create_agent_snapshot( - *, - component_visits: dict[str, int], - agent_breakpoint: AgentBreakpoint, - component_inputs: dict[str, Any], - tool_execution_decisions: list[ToolExecutionDecision] | None = None, -) -> AgentSnapshot: - """ - Create a snapshot of the agent's state. - - NOTE: Only difference to Haystack's native implementation is the addition of tool_execution_decisions to the - AgentSnapshot. - - :param component_visits: The visit counts for the agent's components. - :param agent_breakpoint: AgentBreakpoint object containing breakpoints - :param component_inputs: The inputs to the agent's components. - :param tool_execution_decisions: Optional list of ToolExecutionDecision objects representing decisions made - regarding tool executions. - :return: An AgentSnapshot containing the agent's state and component visits. - """ - return AgentSnapshot( - component_inputs={ - "chat_generator": _serialize_value_with_schema( - _deepcopy_with_exceptions(component_inputs["chat_generator"]) - ), - "tool_invoker": _serialize_value_with_schema(_deepcopy_with_exceptions(component_inputs["tool_invoker"])), - }, - component_visits=component_visits, - break_point=agent_breakpoint, - timestamp=datetime.now(), - tool_execution_decisions=tool_execution_decisions, - ) - - -def _create_pipeline_snapshot_from_tool_invoker( - *, - execution_context: "_ExecutionContext", - tool_name: str | None = None, - agent_name: str | None = None, - break_point: AgentBreakpoint | None = None, - parent_snapshot: PipelineSnapshot | None = None, -) -> PipelineSnapshot: - """ - Create a pipeline snapshot when a tool invoker breakpoint is raised or an exception during execution occurs. - - :param execution_context: The current execution context of the agent. - :param tool_name: The name of the tool that triggered the breakpoint, if available. - :param agent_name: The name of the agent component if present in a pipeline. - :param break_point: An optional AgentBreakpoint object. If provided, it will be used instead of creating a new one. - A scenario where a new breakpoint is created is when an exception occurs during tool execution and we want to - capture the state at that point. - :param parent_snapshot: An optional parent PipelineSnapshot to build upon. - :returns: - A PipelineSnapshot containing the state of the pipeline and agent at the point of the breakpoint or exception. - """ - if break_point is None: - agent_breakpoint = AgentBreakpoint( - agent_name=agent_name or "agent", - break_point=ToolBreakpoint( - component_name="tool_invoker", - visit_count=execution_context.component_visits["tool_invoker"], - tool_name=tool_name, - snapshot_file_path=_get_output_dir("pipeline_snapshot"), - ), - ) - else: - agent_breakpoint = break_point - - messages = execution_context.state.data["messages"] - agent_snapshot = _create_agent_snapshot( - component_visits=execution_context.component_visits, - agent_breakpoint=agent_breakpoint, - component_inputs={ - "chat_generator": {"messages": messages[:-1], **execution_context.chat_generator_inputs}, - "tool_invoker": { - "messages": messages[-1:], # tool invoker consumes last msg from the chat_generator, contains tool call - "state": execution_context.state, - **execution_context.tool_invoker_inputs, - }, - }, - tool_execution_decisions=execution_context.tool_execution_decisions, - ) - if parent_snapshot is None: - # Create an empty pipeline snapshot if no parent snapshot is provided - final_snapshot = PipelineSnapshot( - pipeline_state=PipelineState(inputs={}, component_visits={}, pipeline_outputs={}), - timestamp=agent_snapshot.timestamp, - break_point=agent_snapshot.break_point, - agent_snapshot=agent_snapshot, - original_input_data={}, - ordered_component_names=[], - include_outputs_from=set(), - ) - else: - final_snapshot = replace(parent_snapshot, agent_snapshot=agent_snapshot) - - return final_snapshot diff --git a/haystack_experimental/dataclasses/breakpoints.py b/haystack_experimental/dataclasses/breakpoints.py deleted file mode 100644 index eb3fa15d..00000000 --- a/haystack_experimental/dataclasses/breakpoints.py +++ /dev/null @@ -1,52 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from dataclasses import dataclass -from datetime import datetime -from typing import Any - -from haystack.dataclasses import AgentBreakpoint -from haystack.dataclasses import AgentSnapshot as HaystackAgentSnapshot -from haystack.human_in_the_loop.dataclasses import ToolExecutionDecision - - -@dataclass -class AgentSnapshot(HaystackAgentSnapshot): - tool_execution_decisions: list[ToolExecutionDecision] | None = None - - def to_dict(self) -> dict[str, Any]: - """ - Convert the AgentSnapshot to a dictionary representation. - - :return: A dictionary containing the agent state, timestamp, and breakpoint. - """ - return { - "component_inputs": self.component_inputs, - "component_visits": self.component_visits, - "break_point": self.break_point.to_dict(), - "timestamp": self.timestamp.isoformat() if self.timestamp else None, - "tool_execution_decisions": [ted.to_dict() for ted in self.tool_execution_decisions] - if self.tool_execution_decisions - else None, - } - - @classmethod - def from_dict(cls, data: dict) -> "AgentSnapshot": - """ - Populate the AgentSnapshot from a dictionary representation. - - :param data: A dictionary containing the agent state, timestamp, and breakpoint. - :return: An instance of AgentSnapshot. - """ - return cls( - component_inputs=data["component_inputs"], - component_visits=data["component_visits"], - break_point=AgentBreakpoint.from_dict(data["break_point"]), - timestamp=datetime.fromisoformat(data["timestamp"]) if data.get("timestamp") else None, - tool_execution_decisions=[ - ToolExecutionDecision.from_dict(ted) for ted in data.get("tool_execution_decisions", []) - ] - if data.get("tool_execution_decisions") - else None, - ) diff --git a/hitl_breakpoint_example.py b/hitl_breakpoint_example.py deleted file mode 100644 index a4f2a8ce..00000000 --- a/hitl_breakpoint_example.py +++ /dev/null @@ -1,181 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -import os -from pathlib import Path -from typing import Any - -from haystack.components.generators.chat import OpenAIChatGenerator -from haystack.core.errors import BreakpointException -from haystack.core.pipeline.breakpoint import load_pipeline_snapshot -from haystack.dataclasses import ChatMessage -from haystack.dataclasses.breakpoints import PipelineSnapshot -from haystack.human_in_the_loop import ( - AlwaysAskPolicy, - BlockingConfirmationStrategy, - RichConsoleUI, - ToolExecutionDecision, -) -from haystack.tools import create_tool_from_function -from rich.console import Console - -from haystack_experimental.components.agents.agent import Agent -from haystack_experimental.components.agents.human_in_the_loop import BreakpointConfirmationStrategy -from haystack_experimental.components.agents.human_in_the_loop.breakpoint import ( - get_tool_calls_and_descriptions_from_snapshot, -) - - -def get_bank_balance(account_id: str) -> str: - """ - Simulate fetching a bank balance for a given account ID. - - :param account_id: The ID of the bank account. - :returns: - A string representing the bank balance. - """ - return f"Balance for account {account_id} is $1,234.56" - - -def addition(a: float, b: float) -> float: - """ - A simple addition function. - - :param a: First float. - :param b: Second float. - :returns: - Sum of a and b. - """ - return a + b - - -def get_latest_snapshot(snapshot_file_path: str) -> PipelineSnapshot: - """ - Load the latest pipeline snapshot from the 'pipeline_snapshots' directory. - """ - snapshot_dir = Path(snapshot_file_path) - possible_snapshots = [snapshot_dir / f for f in os.listdir(snapshot_dir)] - latest_snapshot_file = str(max(possible_snapshots, key=os.path.getctime)) - return load_pipeline_snapshot(latest_snapshot_file) - - -def frontend_simulate_tool_decision( - tool_calls: list[dict[str, Any]], tool_descriptions: dict[str, str], console: Console -) -> list[dict]: - """ - Simulate front-end receiving tool calls, prompting user, and sending back decisions. - - :param tool_calls: - A list of tool call dictionaries containing tool_name, id, and arguments. - :param tool_descriptions: - A dictionary mapping tool names to their descriptions. - :param console: - A Rich Console instance for displaying prompts and messages. - :returns: - A list of serialized ToolExecutionDecision dictionaries. - """ - - confirmation_strategy = BlockingConfirmationStrategy( - confirmation_policy=AlwaysAskPolicy(), confirmation_ui=RichConsoleUI(console=console) - ) - - tool_execution_decisions = [] - for tc in tool_calls: - tool_execution_decisions.append( - confirmation_strategy.run( - tool_name=tc["tool_name"], - tool_description=tool_descriptions[tc["tool_name"]], - tool_call_id=tc["id"], - tool_params=tc["arguments"], - ) - ) - return [ted.to_dict() for ted in tool_execution_decisions] - - -def run_agent( - agent: Agent, - messages: list[ChatMessage], - console: Console, - snapshot_file_path: str | None = None, - tool_execution_decisions: list[dict[str, Any]] | None = None, -) -> dict[str, Any] | None: - """ - Run the agent with the given messages and optional snapshot. - """ - # Load the latest snapshot if a path is provided - snapshot = None - if snapshot_file_path: - snapshot = get_latest_snapshot(snapshot_file_path=snapshot_file_path) - - # Add any new tool execution decisions to the snapshot - if tool_execution_decisions: - teds = [ToolExecutionDecision.from_dict(ted) for ted in tool_execution_decisions] - existing_decisions = snapshot.agent_snapshot.tool_execution_decisions or [] - snapshot.agent_snapshot.tool_execution_decisions = existing_decisions + teds - - try: - return agent.run(messages=messages, snapshot=snapshot.agent_snapshot if snapshot else None) - except BreakpointException as e: - console.print("[bold red]Execution paused by Breakpoint Confirmation Strategy:[/bold red]", str(e)) - return None - - -def main(user_message: str): - """ - Main function to demonstrate the Breakpoint Confirmation Strategy with an agent. - """ - cons = Console() - cons.print("\n[bold blue]=== Breakpoint Confirmation Strategy Example ===[/bold blue]\n") - cons.print(f"[bold yellow]User Message:[/bold yellow] {user_message}\n") - - # Define agent with both tools and breakpoint confirmation strategies - addition_tool = create_tool_from_function( - function=addition, name="addition", description="Add two floats together." - ) - balance_tool = create_tool_from_function( - function=get_bank_balance, name="get_bank_balance", description="Get the bank balance for a given account ID." - ) - snapshot_fp = "pipeline_snapshots" - bank_agent = Agent( - chat_generator=OpenAIChatGenerator(model="gpt-4.1"), - tools=[balance_tool, addition_tool], - system_prompt="You are a helpful financial assistant. Use the provided tool to get bank balances when needed.", - confirmation_strategies={ - balance_tool.name: BreakpointConfirmationStrategy(snapshot_file_path=snapshot_fp), - addition_tool.name: BreakpointConfirmationStrategy(snapshot_file_path=snapshot_fp), - }, - ) - - # Step 1: Initial run - result = run_agent(bank_agent, [ChatMessage.from_user(user_message)], cons) - - # Step 2: Loop to handle break point confirmation strategy until agent completes - while result is None: - # Load the latest snapshot from disk and prep data for front-end - loaded_snapshot = get_latest_snapshot(snapshot_file_path=snapshot_fp) - serialized_tool_calls, tool_descripts = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=loaded_snapshot.agent_snapshot, breakpoint_tool_only=True - ) - - # Simulate front-end interaction - serialized_teds = frontend_simulate_tool_decision(serialized_tool_calls, tool_descripts, cons) - - # Re-run the agent with the new tool execution decisions - result = run_agent(bank_agent, [], cons, snapshot_fp, serialized_teds) - - # Step 3: Final result - last_message = result["last_message"] - cons.print(f"\n[bold green]Agent Result:[/bold green] {last_message.text}") - - -if __name__ == "__main__": - for usr_msg in [ - # Single tool call question --> Works - "What's the balance of account 56789?", - # Two tool call question --> Works - "What's the balance of account 56789 and what is 5.5 + 3.2?", - # Multiple sequential tool calls question --> Works - "What's the balance of account 56789? If it's lower than $2000, what's the balance of account 12345?", - ]: - main(usr_msg) diff --git a/pydoc/agents_api.yml b/pydoc/agents_api.yml deleted file mode 100644 index aee7856b..00000000 --- a/pydoc/agents_api.yml +++ /dev/null @@ -1,16 +0,0 @@ -loaders: - - search_path: [../] - modules: - - haystack_experimental.components.agents.agent - - haystack_experimental.components.agents.human_in_the_loop.breakpoint - - haystack_experimental.components.agents.human_in_the_loop.errors - - haystack_experimental.components.agents.human_in_the_loop.strategies -processors: - - type: filter - documented_only: true - skip_empty_modules: true -renderer: - title: Agents - id: experimental-agents-api - description: Tool-using agents with provider-agnostic chat model support. - filename: experimental_agents_api.md diff --git a/pyproject.toml b/pyproject.toml index 84bb5acb..fc000bf4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,6 @@ docs = "haystack-pydoc pydoc tmp_api_reference" extra-dependencies = [ "tiktoken", # LLM-based Summarizer "nltk>=3.9.1", # LLM-based Summarizer - "rich", # for hitl_breakpoint_example.py # Type check "mypy", "pip", diff --git a/test/components/agents/__init__.py b/test/components/agents/__init__.py deleted file mode 100644 index c1764a6e..00000000 --- a/test/components/agents/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 diff --git a/test/components/agents/human_in_the_loop/__init__.py b/test/components/agents/human_in_the_loop/__init__.py deleted file mode 100644 index c1764a6e..00000000 --- a/test/components/agents/human_in_the_loop/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 diff --git a/test/components/agents/human_in_the_loop/test_breakpoint.py b/test/components/agents/human_in_the_loop/test_breakpoint.py deleted file mode 100644 index 503d2a1c..00000000 --- a/test/components/agents/human_in_the_loop/test_breakpoint.py +++ /dev/null @@ -1,148 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from haystack.dataclasses.breakpoints import AgentBreakpoint, ToolBreakpoint - -from haystack_experimental.dataclasses.breakpoints import AgentSnapshot -from haystack_experimental.components.agents.human_in_the_loop.breakpoint import ( - get_tool_calls_and_descriptions_from_snapshot, -) - - -def get_bank_balance(account_id: str) -> str: - return f"The balance for account {account_id} is $1,234.56." - - -def addition(a: float, b: float) -> float: - return a + b - - -def test_get_tool_calls_and_descriptions_from_snapshot(): - agent_snapshot = AgentSnapshot( - component_inputs={ - "chat_generator": {}, - "tool_invoker": { - "serialization_schema": { - "type": "object", - "properties": { - "messages": { - "type": "array", - "items": {"type": "haystack.dataclasses.chat_message.ChatMessage"}, - }, - "state": {"type": "haystack.components.agents.state.state.State"}, - "tools": {"type": "array", "items": {"type": "haystack.tools.tool.Tool"}}, - "enable_streaming_callback_passthrough": {"type": "boolean"}, - }, - }, - "serialized_data": { - "messages": [ - { - "role": "assistant", - "content": [ - { - "tool_call": { - "tool_name": "get_bank_balance", - "arguments": {"account_id": "56789"}, - "id": None, - } - } - ], - } - ], - "state": { - "schema": { - "messages": { - "type": "list[haystack.dataclasses.chat_message.ChatMessage]", - "handler": "haystack.components.agents.state.state_utils.merge_lists", - } - }, - "data": { - "serialization_schema": { - "type": "object", - "properties": { - "messages": { - "type": "array", - "items": {"type": "haystack.dataclasses.chat_message.ChatMessage"}, - } - }, - }, - "serialized_data": { - "messages": [ - { - "role": "system", - "content": [ - { - "text": "You are a helpful financial assistant. Use the provided tool to get bank balances when needed." - } - ], - }, - { - "role": "user", - "content": [{"text": "What's the balance of account 56789?"}], - }, - { - "role": "assistant", - "content": [ - { - "tool_call": { - "tool_name": "get_bank_balance", - "arguments": {"account_id": "56789"}, - "id": None, - } - } - ], - }, - ] - }, - }, - }, - "tools": [ - { - "type": "haystack.tools.tool.Tool", - "data": { - "name": "get_bank_balance", - "description": "Get the bank balance for a given account ID.", - "parameters": { - "properties": {"account_id": {"type": "string"}}, - "required": ["account_id"], - "type": "object", - }, - "function": "test.components.agents.human_in_the_loop.test_breakpoint.get_bank_balance", - }, - }, - { - "type": "haystack.tools.tool.Tool", - "data": { - "name": "addition", - "description": "Add two floats together.", - "parameters": { - "properties": {"a": {"type": "number"}, "b": {"type": "number"}}, - "required": ["a", "b"], - "type": "object", - }, - "function": "test.components.agents.human_in_the_loop.test_breakpoint.addition", - }, - }, - ], - "enable_streaming_callback_passthrough": False, - }, - }, - }, - component_visits={"chat_generator": 1, "tool_invoker": 0}, - break_point=AgentBreakpoint( - agent_name="agent", - break_point=ToolBreakpoint( - tool_name="get_bank_balance", component_name="tool_invoker", visit_count=0, snapshot_file_path=None - ), - ), - ) - - tool_calls, tool_descriptions = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=agent_snapshot, breakpoint_tool_only=True - ) - - assert len(tool_calls) == 1 - assert tool_calls[0]["tool_name"] == "get_bank_balance" - assert tool_calls[0]["arguments"] == {"account_id": "56789"} - assert tool_descriptions == {"get_bank_balance": "Get the bank balance for a given account ID."} diff --git a/test/components/agents/human_in_the_loop/test_strategies.py b/test/components/agents/human_in_the_loop/test_strategies.py deleted file mode 100644 index 54cb4b35..00000000 --- a/test/components/agents/human_in_the_loop/test_strategies.py +++ /dev/null @@ -1,108 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -from typing import Any, Optional - -import pytest -from haystack.human_in_the_loop.strategies import ( - _run_confirmation_strategies, - _run_confirmation_strategies_async, -) -from haystack.components.agents.agent import _ExecutionContext -from haystack.components.agents.state.state import State -from haystack.dataclasses import ChatMessage, ToolCall -from haystack.tools import Tool, create_tool_from_function - -from haystack_experimental.components.agents.human_in_the_loop import ( - BreakpointConfirmationStrategy, - HITLBreakpointException, -) - - -def addition_tool(a: int, b: int) -> int: - return a + b - - -@pytest.fixture -def tools() -> list[Tool]: - tool = create_tool_from_function( - function=addition_tool, name="addition_tool", description="A tool that adds two integers together." - ) - return [tool] - - -@pytest.fixture -def execution_context(tools: list[Tool]) -> _ExecutionContext: - return _ExecutionContext( - state=State(schema={"messages": {"type": list[ChatMessage]}}), - component_visits={"chat_generator": 0, "tool_invoker": 0}, - chat_generator_inputs={}, - tool_invoker_inputs={"tools": tools}, - counter=0, - skip_chat_generator=False, - tool_execution_decisions=None, - ) - - -class TestBreakpointConfirmationStrategy: - def test_initialization(self): - strategy = BreakpointConfirmationStrategy(snapshot_file_path="test") - assert strategy.snapshot_file_path == "test" - - def test_to_dict(self): - strategy = BreakpointConfirmationStrategy(snapshot_file_path="test") - strategy_dict = strategy.to_dict() - assert strategy_dict == { - "type": "haystack_experimental.components.agents.human_in_the_loop.strategies.BreakpointConfirmationStrategy", - "init_parameters": {"snapshot_file_path": "test"}, - } - - def test_from_dict(self): - strategy_dict = { - "type": "haystack_experimental.components.agents.human_in_the_loop.strategies.BreakpointConfirmationStrategy", - "init_parameters": {"snapshot_file_path": "test"}, - } - strategy = BreakpointConfirmationStrategy.from_dict(strategy_dict) - assert isinstance(strategy, BreakpointConfirmationStrategy) - assert strategy.snapshot_file_path == "test" - - def test_run(self): - strategy = BreakpointConfirmationStrategy(snapshot_file_path="test") - with pytest.raises(HITLBreakpointException): - strategy.run(tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"}) - - def test_run_confirmation_strategies_hitl_breakpoint(self, tmp_path, tools, execution_context): - with pytest.raises(HITLBreakpointException): - _run_confirmation_strategies( - confirmation_strategies={tools[0].name: BreakpointConfirmationStrategy(str(tmp_path))}, - messages_with_tool_calls=[ - ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"param1": "value1"})]), - ], - execution_context=execution_context, - ) - - -class TestAsyncConfirmationStrategies: - @pytest.mark.asyncio - async def test_breakpoint_strategy_run_async(self): - strategy = BreakpointConfirmationStrategy(snapshot_file_path="test_path") - - with pytest.raises(HITLBreakpointException) as exc_info: - await strategy.run_async( - tool_name="test_tool", tool_description="A test tool", tool_params={"param1": "value1"} - ) - - assert exc_info.value.tool_name == "test_tool" - assert exc_info.value.snapshot_file_path == "test_path" - - @pytest.mark.asyncio - async def test_run_confirmation_strategies_async_hitl_breakpoint(self, tmp_path, tools, execution_context): - with pytest.raises(HITLBreakpointException): - await _run_confirmation_strategies_async( - confirmation_strategies={tools[0].name: BreakpointConfirmationStrategy(str(tmp_path))}, - messages_with_tool_calls=[ - ChatMessage.from_assistant(tool_calls=[ToolCall(tools[0].name, {"a": 1, "b": 2})]), - ], - execution_context=execution_context, - ) diff --git a/test/components/agents/test_agent.py b/test/components/agents/test_agent.py deleted file mode 100644 index b09374e1..00000000 --- a/test/components/agents/test_agent.py +++ /dev/null @@ -1,524 +0,0 @@ -# SPDX-FileCopyrightText: 2022-present deepset GmbH -# -# SPDX-License-Identifier: Apache-2.0 - -import copy -import os -from pathlib import Path -from typing import Any, Optional - -import pytest -from haystack import Pipeline, component -from haystack.human_in_the_loop import ( - AlwaysAskPolicy, - BlockingConfirmationStrategy, - ConfirmationUIResult, - ToolExecutionDecision -) -from haystack.human_in_the_loop.types import ConfirmationUI -from haystack.components.builders import ChatPromptBuilder -from haystack.components.generators.chat import OpenAIChatGenerator -from haystack.core.errors import BreakpointException -from haystack.core.pipeline.breakpoint import load_pipeline_snapshot -from haystack.dataclasses import ChatMessage, ToolCall, PipelineSnapshot -from haystack.tools import Tool, create_tool_from_function - -from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore -from haystack_experimental.components.agents.agent import Agent -from haystack_experimental.components.agents.human_in_the_loop import BreakpointConfirmationStrategy -from haystack_experimental.components.agents.human_in_the_loop.breakpoint import ( - get_tool_calls_and_descriptions_from_snapshot, -) -from haystack_experimental.components.retrievers import ChatMessageRetriever -from haystack_experimental.components.writers import ChatMessageWriter - - -@pytest.fixture -def store(): - msg_store = InMemoryChatMessageStore() - yield msg_store - msg_store.delete_all_messages() - -@component -class MockChatGenerator: - @component.output_types(replies=list[ChatMessage]) - def run(self, messages: list[ChatMessage], tools: Any) -> dict[str, list[ChatMessage]]: - return {"replies": [ChatMessage.from_assistant("This is a mock response.")]} - - -@component -class MockChatGeneratorToolsResponse: - @component.output_types(replies=list[ChatMessage]) - def run(self, messages: list[ChatMessage], tools: Any) -> dict[str, list[ChatMessage]]: - return { - "replies": [ - ChatMessage.from_assistant( - tool_calls=[ToolCall(tool_name="addition_tool", arguments={"a": 2, "b": 3})] - ) - ] - } - - -@component -class MockAgent: - def __init__(self, system_prompt: Optional[str] = None): - self.system_prompt = system_prompt - - @component.output_types(messages=list[ChatMessage], last_message=ChatMessage) - def run(self, messages: list[ChatMessage]) -> dict[str, Any]: - if self.system_prompt: - system_msg = ChatMessage.from_system(self.system_prompt) - messages = [system_msg, *messages] - - assistant_msg = ChatMessage.from_assistant("This is a mock response.") - return {"messages": [*messages, assistant_msg], "last_message": assistant_msg} - - -class MockUserInterface(ConfirmationUI): - def __init__(self, ui_result: ConfirmationUIResult) -> None: - self.ui_result = ui_result - - def get_user_confirmation( - self, tool_name: str, tool_description: str, tool_params: dict[str, Any] - ) -> ConfirmationUIResult: - return self.ui_result - - -def frontend_simulate_tool_decision( - tool_calls: list[dict[str, Any]], - tool_descriptions: dict[str, str], - confirmation_ui_result: ConfirmationUIResult, -) -> list[dict]: - confirmation_strategy = BlockingConfirmationStrategy( - confirmation_policy=AlwaysAskPolicy(), - confirmation_ui=MockUserInterface(ui_result=confirmation_ui_result), - ) - - tool_execution_decisions = [] - for tc in tool_calls: - tool_execution_decisions.append( - confirmation_strategy.run( - tool_name=tc["tool_name"], - tool_description=tool_descriptions[tc["tool_name"]], - tool_call_id=tc["id"], - tool_params=tc["arguments"], - ) - ) - return [ted.to_dict() for ted in tool_execution_decisions] - - -def get_latest_snapshot(snapshot_file_path: str) -> PipelineSnapshot: - snapshot_dir = Path(snapshot_file_path) - possible_snapshots = [snapshot_dir / f for f in os.listdir(snapshot_dir)] - latest_snapshot_file = str(max(possible_snapshots, key=os.path.getctime)) - return load_pipeline_snapshot(latest_snapshot_file) - - -def run_agent( - agent: Agent, - messages: list[ChatMessage], - snapshot_file_path: Optional[str] = None, - tool_execution_decisions: Optional[list[dict[str, Any]]] = None, -) -> Optional[dict[str, Any]]: - # Load the latest snapshot if a path is provided - snapshot = None - if snapshot_file_path: - snapshot = get_latest_snapshot(snapshot_file_path=snapshot_file_path) - - # Add any new tool execution decisions to the snapshot - if tool_execution_decisions: - teds = [ToolExecutionDecision.from_dict(ted) for ted in tool_execution_decisions] - existing_decisions = snapshot.agent_snapshot.tool_execution_decisions or [] - snapshot.agent_snapshot.tool_execution_decisions = existing_decisions + teds - - try: - return agent.run(messages=messages, snapshot=snapshot.agent_snapshot if snapshot else None) - except BreakpointException: - return None - - -def run_pipeline_with_agent( - pipeline: Pipeline, - messages: list[ChatMessage], - snapshot_file_path: Optional[str] = None, - tool_execution_decisions: Optional[list[dict[str, Any]]] = None, -) -> Optional[dict[str, Any]]: - # Load the latest snapshot if a path is provided - snapshot = None - if snapshot_file_path: - snapshot = get_latest_snapshot(snapshot_file_path=snapshot_file_path) - - # Add any new tool execution decisions to the snapshot - if tool_execution_decisions: - teds = [ToolExecutionDecision.from_dict(ted) for ted in tool_execution_decisions] - existing_decisions = snapshot.agent_snapshot.tool_execution_decisions or [] - snapshot.agent_snapshot.tool_execution_decisions = existing_decisions + teds - - try: - return pipeline.run({"agent": {"messages": messages}}, pipeline_snapshot=snapshot) - except BreakpointException: - return None - - -async def run_agent_async( - agent: Agent, - messages: list[ChatMessage], - snapshot_file_path: Optional[str] = None, - tool_execution_decisions: Optional[list[dict[str, Any]]] = None, -) -> Optional[dict[str, Any]]: - # Load the latest snapshot if a path is provided - snapshot = None - if snapshot_file_path: - snapshot = get_latest_snapshot(snapshot_file_path=snapshot_file_path) - - # Add any new tool execution decisions to the snapshot - if tool_execution_decisions: - teds = [ToolExecutionDecision.from_dict(ted) for ted in tool_execution_decisions] - existing_decisions = snapshot.agent_snapshot.tool_execution_decisions or [] - snapshot.agent_snapshot.tool_execution_decisions = existing_decisions + teds - - try: - return await agent.run_async(messages=messages, snapshot=snapshot.agent_snapshot if snapshot else None) - except BreakpointException: - return None - - -def addition_tool(a: int, b: int) -> int: - return a + b - - -@pytest.fixture -def tools() -> list[Tool]: - tool = create_tool_from_function( - function=addition_tool, name="addition_tool", description="A tool that adds two integers together." - ) - return [tool] - - -class TestAgent: - def test_to_dict(self, tools, monkeypatch): - monkeypatch.setenv("OPENAI_API_KEY", "test") - agent = Agent( - chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), tools=tools, chat_message_store=InMemoryChatMessageStore(), - ) - agent_dict = agent.to_dict() - assert agent_dict == { - "type": "haystack_experimental.components.agents.agent.Agent", - "init_parameters": { - "chat_generator": { - "type": "haystack.components.generators.chat.openai.OpenAIChatGenerator", - "init_parameters": { - "model": "gpt-4o-mini", - "streaming_callback": None, - "api_base_url": None, - "organization": None, - "generation_kwargs": {}, - "api_key": {"type": "env_var", "env_vars": ["OPENAI_API_KEY"], "strict": True}, - "timeout": None, - "max_retries": None, - "tools": None, - "tools_strict": False, - "http_client_kwargs": None, - }, - }, - "chat_message_store": { - "type": "haystack_experimental.chat_message_stores.in_memory.InMemoryChatMessageStore", - "init_parameters": { - "last_k": 10, - "skip_system_messages": True, - }, - }, - "tools": [ - { - "type": "haystack.tools.tool.Tool", - "data": { - "name": "addition_tool", - "description": "A tool that adds two integers together.", - "parameters": { - "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, - "required": ["a", "b"], - "type": "object", - }, - "function": "test.components.agents.test_agent.addition_tool", - "outputs_to_string": None, - "inputs_from_state": None, - "outputs_to_state": None, - }, - } - ], - "system_prompt": None, - "user_prompt": None, - "required_variables": None, - "exit_conditions": ["text"], - "state_schema": {}, - "max_agent_steps": 100, - "streaming_callback": None, - "raise_on_tool_invocation_failure": False, - "tool_invoker_kwargs": None, - "confirmation_strategies": None, - }, - } - - def test_from_dict(self, tools, monkeypatch): - monkeypatch.setenv("OPENAI_API_KEY", "test") - agent = Agent( - chat_generator=OpenAIChatGenerator(), tools=tools, chat_message_store=InMemoryChatMessageStore(), - ) - deserialized_agent = Agent.from_dict(agent.to_dict()) - assert deserialized_agent.to_dict() == agent.to_dict() - assert isinstance(deserialized_agent.chat_generator, OpenAIChatGenerator) - assert len(deserialized_agent.tools) == 1 - assert deserialized_agent.tools[0].name == "addition_tool" - assert isinstance(deserialized_agent._tool_invoker, type(agent._tool_invoker)) - assert isinstance(deserialized_agent._chat_message_store, InMemoryChatMessageStore) - - -class TestAgentConfirmationStrategy: - def test_get_tool_calls_and_descriptions_from_snapshot_no_mutation_of_snapshot( - self, tools, tmp_path, monkeypatch - ): - monkeypatch.setenv("HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED", "true") - agent = Agent( - chat_generator=MockChatGeneratorToolsResponse(), - tools=tools, - confirmation_strategies={ - "addition_tool": BreakpointConfirmationStrategy(snapshot_file_path=str(tmp_path)), - }, - ) - agent.warm_up() - - # Run the agent to create a snapshot with a breakpoint - try: - agent.run([ChatMessage.from_user("What is 2+2?")]) - except BreakpointException: - pass - - # Load the latest snapshot from disk - loaded_snapshot = get_latest_snapshot(snapshot_file_path=str(tmp_path)) - - original_snapshot = copy.deepcopy(loaded_snapshot) - - # Extract tool calls and descriptions - _ = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=loaded_snapshot.agent_snapshot, breakpoint_tool_only=True - ) - - # Verify that the original snapshot has not been mutated - assert loaded_snapshot == original_snapshot - - @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set") - @pytest.mark.integration - def test_run_breakpoint_confirmation_strategy_modify(self, tools, tmp_path, monkeypatch): - monkeypatch.setenv("HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED", "true") - agent = Agent( - chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), - tools=tools, - confirmation_strategies={ - "addition_tool": BreakpointConfirmationStrategy(snapshot_file_path=str(tmp_path)), - }, - ) - agent.warm_up() - - # Step 1: Initial run - result = run_agent(agent, [ChatMessage.from_user("What is 2+2?")]) - - # Step 2: Loop to handle break point confirmation strategy until agent completes - while result is None: - # Load the latest snapshot from disk and prep data for front-end - loaded_snapshot = get_latest_snapshot(snapshot_file_path=str(tmp_path)) - serialized_tool_calls, tool_descripts = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=loaded_snapshot.agent_snapshot, breakpoint_tool_only=True - ) - - # Simulate front-end interaction - serialized_teds = frontend_simulate_tool_decision( - serialized_tool_calls, - tool_descripts, - ConfirmationUIResult(action="modify", new_tool_params={"a": 2, "b": 3}), - ) - - # Re-run the agent with the new tool execution decisions - result = run_agent(agent, [], str(tmp_path), serialized_teds) - - # Step 3: Final result - last_message = result["last_message"] - assert isinstance(last_message, ChatMessage) - assert "5" in last_message.text - - @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set") - @pytest.mark.integration - def test_run_in_pipeline_breakpoint_confirmation_strategy_modify(self, tools, tmp_path, monkeypatch): - monkeypatch.setenv("HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED", "true") - agent = Agent( - chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), - tools=tools, - confirmation_strategies={ - "addition_tool": BreakpointConfirmationStrategy(snapshot_file_path=str(tmp_path)), - }, - ) - - pipeline = Pipeline() - pipeline.add_component("agent", agent) - - # Step 1: Initial run - result = run_pipeline_with_agent(pipeline, [ChatMessage.from_user("What is 2+2?")]) - - # Step 2: Loop to handle break point confirmation strategy until pipeline with agent completes - while result is None: - # Load the latest snapshot from disk and prep data for front-end - loaded_snapshot = get_latest_snapshot(snapshot_file_path=str(tmp_path)) - serialized_tool_calls, tool_descripts = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=loaded_snapshot.agent_snapshot, breakpoint_tool_only=True - ) - - # Simulate front-end interaction - serialized_teds = frontend_simulate_tool_decision( - serialized_tool_calls, - tool_descripts, - ConfirmationUIResult(action="modify", new_tool_params={"a": 2, "b": 3}), - ) - - # Re-run the agent with the new tool execution decisions - result = run_pipeline_with_agent(pipeline, [], str(tmp_path), serialized_teds) - - # Step 3: Final result - last_message = result["agent"]["last_message"] - assert isinstance(last_message, ChatMessage) - assert "5" in last_message.text - - @pytest.mark.skipif(not os.environ.get("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set") - @pytest.mark.integration - @pytest.mark.asyncio - async def test_run_async_breakpoint_confirmation_strategy_modify(self, tools, tmp_path, monkeypatch): - monkeypatch.setenv("HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED", "true") - agent = Agent( - chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), - tools=tools, - confirmation_strategies={ - "addition_tool": BreakpointConfirmationStrategy(snapshot_file_path=str(tmp_path)), - }, - ) - agent.warm_up() - - # Step 1: Initial run - result = await run_agent_async(agent, [ChatMessage.from_user("What is 2+2?")]) - - # Step 2: Loop to handle break point confirmation strategy until agent completes - while result is None: - # Load the latest snapshot from disk and prep data for front-end - loaded_snapshot = get_latest_snapshot(snapshot_file_path=str(tmp_path)) - serialized_tool_calls, tool_descripts = get_tool_calls_and_descriptions_from_snapshot( - agent_snapshot=loaded_snapshot.agent_snapshot, breakpoint_tool_only=True - ) - - # Simulate front-end interaction - serialized_teds = frontend_simulate_tool_decision( - serialized_tool_calls, - tool_descripts, - ConfirmationUIResult(action="modify", new_tool_params={"a": 2, "b": 3}), - ) - - # Re-run the agent with the new tool execution decisions - result = await run_agent_async(agent, [], str(tmp_path), serialized_teds) - - # Step 3: Final result - last_message = result["last_message"] - assert isinstance(last_message, ChatMessage) - assert "5" in last_message.text - - -class TestAgentWithChatMessageStore: - def test_external_chat_message_store_with_agent(self, store): - pipe = Pipeline() - pipe.add_component( - "prompt_builder", - ChatPromptBuilder(template=[ChatMessage.from_user("{{ query }}")], required_variables=["query"]), - ) - pipe.add_component("message_retriever", ChatMessageRetriever(store)) - pipe.add_component("agent", MockAgent(system_prompt="This is a system prompt.")) - pipe.add_component("message_writer", ChatMessageWriter(store)) - - pipe.connect("prompt_builder.prompt", "message_retriever.current_messages") - pipe.connect("message_retriever.messages", "agent.messages") - pipe.connect("agent.messages", "message_writer.messages") - - chat_history_id = "user_123_session_456" - result = pipe.run( - data={ - "prompt_builder": {"query": "What is the capital of Germany?"}, - "message_retriever": {"chat_history_id": chat_history_id}, - "message_writer": {"chat_history_id": chat_history_id}, - }, - include_outputs_from={"agent"}, - ) - assert result["agent"]["messages"] == [ - ChatMessage.from_system("This is a system prompt."), - ChatMessage.from_user("What is the capital of Germany?"), - ChatMessage.from_assistant("This is a mock response."), - ] - assert store.retrieve_messages(chat_history_id) == [ - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ] - - # Second run - result = pipe.run( - data={ - "prompt_builder": {"query": "What is the capital of Italy?"}, - "message_retriever": {"chat_history_id": chat_history_id}, - "message_writer": {"chat_history_id": chat_history_id}, - }, - include_outputs_from={"agent"}, - ) - assert result["agent"]["messages"] == [ - ChatMessage.from_system("This is a system prompt."), - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ChatMessage.from_user("What is the capital of Italy?"), - ChatMessage.from_assistant("This is a mock response."), - ] - assert store.retrieve_messages(chat_history_id) == [ - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ChatMessage.from_user("What is the capital of Italy?", meta={"chat_message_id": "2"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "3"}), - ] - - def test_internal_chat_message_store_with_agent(self, store): - agent = Agent( - chat_generator=MockChatGenerator(), system_prompt="This is a system prompt.", chat_message_store=store - ) - - chat_history_id = "user_123_session_456" - result = agent.run( - messages=[ChatMessage.from_user("What is the capital of Germany?")], - chat_message_store_kwargs={"chat_history_id": chat_history_id, "last_k": None}, - ) - assert result["messages"] == [ - ChatMessage.from_system("This is a system prompt."), - ChatMessage.from_user("What is the capital of Germany?"), - ChatMessage.from_assistant("This is a mock response."), - ] - assert store.retrieve_messages(chat_history_id) == [ - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ] - - # Second run - result = agent.run( - messages=[ChatMessage.from_user("What is the capital of Italy?")], - chat_message_store_kwargs={"chat_history_id": chat_history_id, "last_k": None}, - ) - assert result["messages"] == [ - ChatMessage.from_system("This is a system prompt."), - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ChatMessage.from_user("What is the capital of Italy?"), - ChatMessage.from_assistant("This is a mock response."), - ] - assert store.retrieve_messages(chat_history_id) == [ - ChatMessage.from_user("What is the capital of Germany?", meta={"chat_message_id": "0"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "1"}), - ChatMessage.from_user("What is the capital of Italy?", meta={"chat_message_id": "2"}), - ChatMessage.from_assistant("This is a mock response.", meta={"chat_message_id": "3"}), - ]