Base

Agent

Bases: Node

Base class for an AI Agent that interacts with a Language Model and tools.

Source code in dynamiq/nodes/agents/base.py

class Agent(Node):
    """Base class for an AI Agent that interacts with a Language Model and tools."""

    AGENT_PROMPT_TEMPLATE: ClassVar[str] = AGENT_PROMPT_TEMPLATE

    llm: Node = Field(..., description="LLM used by the agent.")
    group: NodeGroup = NodeGroup.AGENTS
    error_handling: ErrorHandling = Field(default_factory=lambda: ErrorHandling(timeout_seconds=600))
    tools: list[Node] = []
    files: list[io.BytesIO | bytes] | None = None
    images: list[str | bytes | io.BytesIO] = None
    name: str = "Agent"
    max_loops: int = 1
    tool_output_max_length: int = TOOL_MAX_TOKENS
    tool_output_truncate_enabled: bool = True
    memory: Memory | None = Field(None, description="Memory node for the agent.")
    memory_limit: int = Field(100, description="Maximum number of messages to retrieve from memory")
    memory_retrieval_strategy: MemoryRetrievalStrategy | None = MemoryRetrievalStrategy.ALL
    verbose: bool = Field(False, description="Whether to print verbose logs.")

    input_message: Message | VisionMessage = Message(role=MessageRole.USER, content="{{input}}")
    role: str | None = ""
    _prompt_blocks: dict[str, str] = PrivateAttr(default_factory=dict)
    _prompt_variables: dict[str, Any] = PrivateAttr(default_factory=dict)

    model_config = ConfigDict(arbitrary_types_allowed=True)
    input_schema: ClassVar[type[AgentInputSchema]] = AgentInputSchema

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._intermediate_steps: dict[int, dict] = {}
        self._run_depends: list[dict] = []
        self._prompt = Prompt(messages=[])
        self._init_prompt_blocks()

    @model_validator(mode="after")
    def validate_input_fields(self):
        if self.input_message:
            self.input_message.role = MessageRole.USER

        return self

    def get_context_for_input_schema(self) -> dict:
        """Provides context for input schema that is required for proper validation."""
        return {"input_message": self.input_message, "role": self.role}

    @property
    def to_dict_exclude_params(self):
        return super().to_dict_exclude_params | {
            "llm": True,
            "tools": True,
            "memory": True,
            "files": True,
            "images": True,
        }

    def to_dict(self, **kwargs) -> dict:
        """Converts the instance to a dictionary."""
        data = super().to_dict(**kwargs)
        data["llm"] = self.llm.to_dict(**kwargs)
        data["tools"] = [tool.to_dict(**kwargs) for tool in self.tools]
        data["memory"] = self.memory.to_dict(**kwargs) if self.memory else None
        if self.files:
            data["files"] = [{"name": getattr(f, "name", f"file_{i}")} for i, f in enumerate(self.files)]
        if self.images:
            data["images"] = [{"name": getattr(f, "name", f"image_{i}")} for i, f in enumerate(self.images)]
        return data

    def init_components(self, connection_manager: ConnectionManager | None = None):
        """
        Initialize components for the manager and agents.

        Args:
            connection_manager (ConnectionManager, optional): The connection manager. Defaults to ConnectionManager.
        """
        connection_manager = connection_manager or ConnectionManager()
        super().init_components(connection_manager)
        if self.llm.is_postponed_component_init:
            self.llm.init_components(connection_manager)

        for tool in self.tools:
            if tool.is_postponed_component_init:
                tool.init_components(connection_manager)
            tool.is_optimized_for_agents = True

    def sanitize_tool_name(self, s: str):
        """Sanitize tool name to follow [^a-zA-Z0-9_-]."""
        s = s.replace(" ", "-")
        sanitized = re.sub(r"[^a-zA-Z0-9_-]", "", s)
        return sanitized

    def _init_prompt_blocks(self):
        """Initializes default prompt blocks and variables."""
        self._prompt_blocks = {
            "date": "{date}",
            "tools": "{tool_description}",
            "files": "{file_description}",
            "instructions": "",
        }
        self._prompt_variables = {
            "tool_description": self.tool_description,
            "file_description": self.file_description,
            "date": datetime.now().strftime("%d %B %Y"),
        }

    def set_block(self, block_name: str, content: str):
        """Adds or updates a prompt block."""
        self._prompt_blocks[block_name] = content

    def set_prompt_variable(self, variable_name: str, value: Any):
        """Sets or updates a prompt variable."""
        self._prompt_variables[variable_name] = value

    def _prepare_metadata(self, input_data: dict) -> dict:
        """
        Prepare metadata from input data.

        Args:
            input_data (dict): Input data containing user information

        Returns:
            dict: Processed metadata
        """
        EXCLUDED_KEYS = {"user_id", "session_id", "input", "metadata", "files", "images", "tool_params"}
        custom_metadata = input_data.get("metadata", {}).copy()
        custom_metadata.update({k: v for k, v in input_data.items() if k not in EXCLUDED_KEYS})

        if "files" in custom_metadata:
            del custom_metadata["files"]
        if "images" in custom_metadata:
            del custom_metadata["images"]
        if "tool_params" in custom_metadata:
            del custom_metadata["tool_params"]

        user_id = input_data.get("user_id")
        session_id = input_data.get("session_id")

        if user_id:
            custom_metadata["user_id"] = user_id
        if session_id:
            custom_metadata["session_id"] = session_id

        return custom_metadata

    def execute(
        self,
        input_data: AgentInputSchema,
        input_message: Message | VisionMessage | None = None,
        config: RunnableConfig | None = None,
        **kwargs,
    ) -> dict[str, Any]:
        """
        Executes the agent with the given input data.
        """
        logger.info(f"Agent {self.name} - {self.id}: started with input {dict(input_data)}")
        self.reset_run_state()
        config = ensure_config(config)
        self.run_on_node_execute_run(config.callbacks, **kwargs)

        custom_metadata = self._prepare_metadata(dict(input_data))

        input_message = create_message_from_input(dict(input_data))

        input_message = input_message or self.input_message
        input_message = input_message.format_message(**dict(input_data))

        use_memory = self.memory and (dict(input_data).get("user_id") or dict(input_data).get("session_id"))

        if use_memory:
            history_messages = self._retrieve_memory(dict(input_data))
            if len(history_messages) > 0:
                history_messages.insert(
                    0,
                    Message(
                        role=MessageRole.SYSTEM,
                        content="Below is the previous conversation history. "
                        "Use this context to inform your response.",
                    ),
                )
            if isinstance(input_message, Message):
                memory_content = input_message.content
            else:
                text_parts = [
                    content.text for content in input_message.content if isinstance(content, VisionMessageTextContent)
                ]
                memory_content = " ".join(text_parts) if text_parts else "Image input"
            self.memory.add(role=MessageRole.USER, content=memory_content, metadata=custom_metadata)
        else:
            history_messages = None

        if self.role:
            self._prompt_blocks["context"] = Template(self.role).render(**dict(input_data))

        files = input_data.files
        if files:
            self.files = files
            self._prompt_variables["file_description"] = self.file_description

        if input_data.tool_params:
            kwargs["tool_params"] = input_data.tool_params

        self._prompt_variables.update(dict(input_data))
        kwargs = kwargs | {"parent_run_id": kwargs.get("run_id")}
        kwargs.pop("run_depends", None)

        result = self._run_agent(input_message, history_messages, config=config, **kwargs)

        if use_memory:
            self.memory.add(role=MessageRole.ASSISTANT, content=result, metadata=custom_metadata)

        execution_result = {
            "content": result,
            "intermediate_steps": self._intermediate_steps,
        }
        logger.info(f"Node {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")

        return execution_result

    def retrieve_conversation_history(
        self,
        user_query: str = None,
        user_id: str = None,
        session_id: str = None,
        limit: int = None,
        strategy: MemoryRetrievalStrategy = MemoryRetrievalStrategy.ALL,
    ) -> list[Message]:
        """
        Retrieves conversation history for the agent using the specified strategy.

        Args:
            user_query: Current user input to find relevant context (for RELEVANT/HYBRID strategies)
            user_id: Optional user identifier
            session_id: Optional session identifier
            limit: Maximum number of messages to return (defaults to memory_limit)
            strategy: Which retrieval strategy to use (ALL, RELEVANT, or HYBRID)

        Returns:
            List of messages forming a valid conversation context
        """
        if not self.memory or not (user_id or session_id):
            return []

        filters = {}
        if user_id:
            filters["user_id"] = user_id
        if session_id:
            filters["session_id"] = session_id

        limit = limit or self.memory_limit

        if strategy == MemoryRetrievalStrategy.RELEVANT and not user_query:
            logger.warning("RELEVANT strategy selected but no user_query provided - falling back to ALL")
            strategy = MemoryRetrievalStrategy.ALL

        conversation = self.memory.get_agent_conversation(
            query=user_query,  # Pass the user query for relevance search
            limit=limit,
            filters=filters,
            strategy=strategy,
        )

        if self.verbose:
            logger.debug(
                f"Agent {self.name} retrieved {len(conversation)} messages using {strategy.value} strategy. "
                f"First message role: {conversation[0].role.value if conversation else 'None'}"
            )

        return conversation

    def _retrieve_memory(self, input_data: dict) -> list[Message]:
        """
        Retrieves memory messages when user_id and/or session_id are provided.
        """
        user_id = input_data.get("user_id")
        session_id = input_data.get("session_id")

        user_query = input_data.get("input", "")

        return self.retrieve_conversation_history(
            user_query=user_query,
            user_id=user_id,
            session_id=session_id,
            strategy=self.memory_retrieval_strategy,
        )

    def _run_llm(self, messages: list[Message | VisionMessage], config: RunnableConfig | None = None, **kwargs) -> str:
        """Runs the LLM with a given prompt and handles streaming or full responses."""
        try:
            llm_result = self.llm.run(
                input_data={},
                config=config,
                prompt=Prompt(messages=messages),
                run_depends=self._run_depends,
                **kwargs,
            )
            self._run_depends = [NodeDependency(node=self.llm).to_dict()]
            if llm_result.status != RunnableStatus.SUCCESS:
                error_message = f"LLM '{self.llm.name}' failed: {llm_result.output.get('content')}"
                raise ValueError({error_message})

            return llm_result

        except Exception as e:
            raise e

    def stream_content(
        self,
        content: str | dict,
        source: str,
        step: str,
        config: RunnableConfig | None = None,
        by_tokens: bool | None = None,
        **kwargs,
    ) -> str | dict:
        """
        Streams data.

        Args:
            content (str | dict): Data that will be streamed.
            source (str): Source of the content.
            step (str): Description of the step.
            by_tokens (Optional[bool]): Determines whether to stream content by tokens or not.
                If None it is determined based on StreamingConfig. Defaults to None.
            config (Optional[RunnableConfig]): Configuration for the runnable.
            **kwargs: Additional keyword arguments.

        Returns:
            str | dict: Streamed data.
        """
        if (by_tokens is None and self.streaming.by_tokens) or by_tokens:
            return self.stream_by_tokens(content=content, source=source, step=step, config=config, **kwargs)
        return self.stream_response(content=content, source=source, step=step, config=config, **kwargs)

    def stream_by_tokens(self, content: str, source: str, step: str, config: RunnableConfig | None = None, **kwargs):
        """Streams the input content to the callbacks."""
        if isinstance(content, dict):
            return self.stream_response(content, source, step, config, **kwargs)
        tokens = content.split(" ")
        final_response = []
        for token in tokens:
            final_response.append(token)
            token_with_prefix = " " + token
            token_for_stream = StreamChunk(
                choices=[
                    StreamChunkChoice(delta=StreamChunkChoiceDelta(content=token_with_prefix, source=source, step=step))
                ]
            )
            self.run_on_node_execute_stream(
                callbacks=config.callbacks,
                chunk=token_for_stream.model_dump(),
                **kwargs,
            )
        return " ".join(final_response)

    def stream_response(
        self, content: str | dict, source: str, step: str, config: RunnableConfig | None = None, **kwargs
    ):
        response_for_stream = StreamChunk(
            choices=[StreamChunkChoice(delta=StreamChunkChoiceDelta(content=content, source=source, step=step))]
        )

        self.run_on_node_execute_stream(
            callbacks=config.callbacks,
            chunk=response_for_stream.model_dump(),
            **kwargs,
        )
        return content

    def _run_agent(
        self,
        input_message: Message | VisionMessage,
        history_messages: list[Message] | None = None,
        config: RunnableConfig | None = None,
        **kwargs,
    ) -> str:
        """Runs the agent with the generated prompt and handles exceptions."""
        formatted_prompt = self.generate_prompt()
        system_message = Message(role=MessageRole.SYSTEM, content=formatted_prompt)
        if history_messages:
            self._prompt.messages = [system_message, *history_messages, input_message]
        else:
            self._prompt.messages = [system_message, input_message]

        try:
            llm_result = self._run_llm(self._prompt.messages, config=config, **kwargs).output["content"]
            self._prompt.messages.append(Message(role=MessageRole.ASSISTANT, content=llm_result))

            if self.streaming.enabled:
                return self.stream_content(
                    content=llm_result,
                    source=self.name,
                    step="answer",
                    config=config,
                    **kwargs,
                )
            return llm_result

        except Exception as e:
            raise e

    def _extract_final_answer(self, output: str) -> str:
        """Extracts the final answer from the output string."""
        match = re.search(r"Answer:\s*(.*)", output, re.DOTALL)
        return match.group(1).strip() if match else ""

    def _get_tool(self, action: str) -> Node:
        """Retrieves the tool corresponding to the given action."""
        tool = self.tool_by_names.get(self.sanitize_tool_name(action))
        if not tool:
            raise AgentUnknownToolException(
                f"Unknown tool: {action}."
                "Use only available tools and provide only the tool's name in the action field. "
                "Do not include any additional reasoning. "
                "Please correct the action field or state that you cannot answer the question."
            )
        return tool

    def _apply_parameters(self, merged_input: dict, params: dict, source: str, debug_info: list = None):
        """Apply parameters from the specified source to the merged input."""
        if debug_info is None:
            debug_info = []
        for key, value in params.items():
            if key in merged_input and isinstance(value, dict) and isinstance(merged_input[key], dict):
                merged_nested = merged_input[key].copy()
                merged_input[key] = deep_merge(value, merged_nested)
                debug_info.append(f"  - From {source}: Merged nested {key}")
            else:
                merged_input[key] = value
                debug_info.append(f"  - From {source}: Set {key}={value}")

    def _run_tool(self, tool: Node, tool_input: dict, config, **kwargs) -> Any:
        """Runs a specific tool with the given input."""
        if self.files:
            if tool.is_files_allowed is True:
                tool_input["files"] = self.files

        merged_input = tool_input.copy() if isinstance(tool_input, dict) else {"input": tool_input}
        raw_tool_params = kwargs.get("tool_params", ToolParams())
        tool_params = (
            ToolParams.model_validate(raw_tool_params) if isinstance(raw_tool_params, dict) else raw_tool_params
        )

        if tool_params:
            debug_info = []
            if self.verbose:
                debug_info.append(f"Tool parameter merging for {tool.name} (ID: {tool.id}):")
                debug_info.append(f"Starting with input: {merged_input}")

            # 1. Apply global parameters (lowest priority)
            global_params = tool_params.global_params
            if global_params:
                self._apply_parameters(merged_input, global_params, "global", debug_info)

            # 2. Apply parameters by tool name (medium priority)
            name_params = tool_params.by_name_params.get(tool.name, {}) or tool_params.by_name_params.get(
                self.sanitize_tool_name(tool.name), {}
            )
            if name_params:
                self._apply_parameters(merged_input, name_params, f"name:{tool.name}", debug_info)

            # 3. Apply parameters by tool ID (highest priority)
            id_params = tool_params.by_id_params.get(tool.id, {})
            if id_params:
                self._apply_parameters(merged_input, id_params, f"id:{tool.id}", debug_info)

            if self.verbose and debug_info:
                logger.debug("\n".join(debug_info))

        tool_result = tool.run(
            input_data=merged_input,
            config=config,
            run_depends=self._run_depends,
            **(kwargs | {"recoverable_error": True}),
        )
        self._run_depends = [NodeDependency(node=tool).to_dict()]
        if tool_result.status != RunnableStatus.SUCCESS:
            error_message = f"Tool '{tool.name}' failed: {tool_result.output}"
            if tool_result.output["recoverable"]:
                raise ToolExecutionException({error_message})
            else:
                raise ValueError({error_message})
        tool_result_content = tool_result.output.get("content")
        tool_result_content_processed = process_tool_output_for_agent(
            content=tool_result_content,
            max_tokens=self.tool_output_max_length,
            truncate=self.tool_output_truncate_enabled,
        )
        return tool_result_content_processed

    @property
    def tool_description(self) -> str:
        """Returns a description of the tools available to the agent."""
        return (
            "\n".join(
                [
                    f"{tool.name}:\n <{tool.name}_description>\n{tool.description.strip()}\n<\\{tool.name}_description>"
                    for tool in self.tools
                ]
            )
            if self.tools
            else ""
        )

    @property
    def file_description(self) -> str:
        """Returns a description of the files available to the agent."""
        if self.files:
            file_description = "You can work with the following files:\n"
            for file in self.files:
                name = getattr(file, "name", "Unnamed file")
                description = getattr(file, "description", "No description")
                file_description += f"<file>: {name} - {description} <\\file>\n"
            return file_description
        return ""

    @property
    def tool_names(self) -> str:
        """Returns a comma-separated list of tool names available to the agent."""
        return ",".join([self.sanitize_tool_name(tool.name) for tool in self.tools])

    @property
    def tool_by_names(self) -> dict[str, Node]:
        """Returns a dictionary mapping tool names to their corresponding Node objects."""
        return {self.sanitize_tool_name(tool.name): tool for tool in self.tools}

    def reset_run_state(self):
        """Resets the agent's run state."""
        self._intermediate_steps = {}
        self._run_depends = []

    def generate_prompt(self, block_names: list[str] | None = None, **kwargs) -> str:
        """Generates the prompt using specified blocks and variables."""
        temp_variables = self._prompt_variables.copy()
        temp_variables.update(kwargs)

        formatted_prompt_blocks = {}
        for block, content in self._prompt_blocks.items():
            if block_names is None or block in block_names:

                formatted_content = content.format(**temp_variables)
                if content:
                    formatted_prompt_blocks[block] = formatted_content

        prompt = Template(self.AGENT_PROMPT_TEMPLATE).render(formatted_prompt_blocks).strip()
        prompt = self._clean_prompt(prompt)
        return textwrap.dedent(prompt)

    def _clean_prompt(self, prompt_text):
        cleaned = re.sub(r"\n{3,}", "\n\n", prompt_text)
        return cleaned.strip()

file_description: str property

Returns a description of the files available to the agent.

tool_by_names: dict[str, Node] property

Returns a dictionary mapping tool names to their corresponding Node objects.

tool_description: str property

Returns a description of the tools available to the agent.

tool_names: str property

Returns a comma-separated list of tool names available to the agent.

execute(input_data, input_message=None, config=None, **kwargs)

Executes the agent with the given input data.

Source code in dynamiq/nodes/agents/base.py

def execute(
    self,
    input_data: AgentInputSchema,
    input_message: Message | VisionMessage | None = None,
    config: RunnableConfig | None = None,
    **kwargs,
) -> dict[str, Any]:
    """
    Executes the agent with the given input data.
    """
    logger.info(f"Agent {self.name} - {self.id}: started with input {dict(input_data)}")
    self.reset_run_state()
    config = ensure_config(config)
    self.run_on_node_execute_run(config.callbacks, **kwargs)

    custom_metadata = self._prepare_metadata(dict(input_data))

    input_message = create_message_from_input(dict(input_data))

    input_message = input_message or self.input_message
    input_message = input_message.format_message(**dict(input_data))

    use_memory = self.memory and (dict(input_data).get("user_id") or dict(input_data).get("session_id"))

    if use_memory:
        history_messages = self._retrieve_memory(dict(input_data))
        if len(history_messages) > 0:
            history_messages.insert(
                0,
                Message(
                    role=MessageRole.SYSTEM,
                    content="Below is the previous conversation history. "
                    "Use this context to inform your response.",
                ),
            )
        if isinstance(input_message, Message):
            memory_content = input_message.content
        else:
            text_parts = [
                content.text for content in input_message.content if isinstance(content, VisionMessageTextContent)
            ]
            memory_content = " ".join(text_parts) if text_parts else "Image input"
        self.memory.add(role=MessageRole.USER, content=memory_content, metadata=custom_metadata)
    else:
        history_messages = None

    if self.role:
        self._prompt_blocks["context"] = Template(self.role).render(**dict(input_data))

    files = input_data.files
    if files:
        self.files = files
        self._prompt_variables["file_description"] = self.file_description

    if input_data.tool_params:
        kwargs["tool_params"] = input_data.tool_params

    self._prompt_variables.update(dict(input_data))
    kwargs = kwargs | {"parent_run_id": kwargs.get("run_id")}
    kwargs.pop("run_depends", None)

    result = self._run_agent(input_message, history_messages, config=config, **kwargs)

    if use_memory:
        self.memory.add(role=MessageRole.ASSISTANT, content=result, metadata=custom_metadata)

    execution_result = {
        "content": result,
        "intermediate_steps": self._intermediate_steps,
    }
    logger.info(f"Node {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")

    return execution_result

generate_prompt(block_names=None, **kwargs)

Generates the prompt using specified blocks and variables.

Source code in dynamiq/nodes/agents/base.py

def generate_prompt(self, block_names: list[str] | None = None, **kwargs) -> str:
    """Generates the prompt using specified blocks and variables."""
    temp_variables = self._prompt_variables.copy()
    temp_variables.update(kwargs)

    formatted_prompt_blocks = {}
    for block, content in self._prompt_blocks.items():
        if block_names is None or block in block_names:

            formatted_content = content.format(**temp_variables)
            if content:
                formatted_prompt_blocks[block] = formatted_content

    prompt = Template(self.AGENT_PROMPT_TEMPLATE).render(formatted_prompt_blocks).strip()
    prompt = self._clean_prompt(prompt)
    return textwrap.dedent(prompt)

get_context_for_input_schema()

Provides context for input schema that is required for proper validation.

Source code in dynamiq/nodes/agents/base.py

def get_context_for_input_schema(self) -> dict:
    """Provides context for input schema that is required for proper validation."""
    return {"input_message": self.input_message, "role": self.role}

init_components(connection_manager=None)

Initialize components for the manager and agents.

Parameters:

Name	Type	Description	Default
connection_manager	ConnectionManager	The connection manager. Defaults to ConnectionManager.	None

Source code in dynamiq/nodes/agents/base.py

def init_components(self, connection_manager: ConnectionManager | None = None):
    """
    Initialize components for the manager and agents.

    Args:
        connection_manager (ConnectionManager, optional): The connection manager. Defaults to ConnectionManager.
    """
    connection_manager = connection_manager or ConnectionManager()
    super().init_components(connection_manager)
    if self.llm.is_postponed_component_init:
        self.llm.init_components(connection_manager)

    for tool in self.tools:
        if tool.is_postponed_component_init:
            tool.init_components(connection_manager)
        tool.is_optimized_for_agents = True

reset_run_state()

Resets the agent's run state.

Source code in dynamiq/nodes/agents/base.py

def reset_run_state(self):
    """Resets the agent's run state."""
    self._intermediate_steps = {}
    self._run_depends = []

retrieve_conversation_history(user_query=None, user_id=None, session_id=None, limit=None, strategy=MemoryRetrievalStrategy.ALL)

Retrieves conversation history for the agent using the specified strategy.

Parameters:

Name	Type	Description	Default
user_query	str	Current user input to find relevant context (for RELEVANT/HYBRID strategies)	None
user_id	str	Optional user identifier	None
session_id	str	Optional session identifier	None
limit	int	Maximum number of messages to return (defaults to memory_limit)	None
strategy	MemoryRetrievalStrategy	Which retrieval strategy to use (ALL, RELEVANT, or HYBRID)	ALL

Returns:

Type	Description
list[Message]	List of messages forming a valid conversation context

Source code in dynamiq/nodes/agents/base.py

def retrieve_conversation_history(
    self,
    user_query: str = None,
    user_id: str = None,
    session_id: str = None,
    limit: int = None,
    strategy: MemoryRetrievalStrategy = MemoryRetrievalStrategy.ALL,
) -> list[Message]:
    """
    Retrieves conversation history for the agent using the specified strategy.

    Args:
        user_query: Current user input to find relevant context (for RELEVANT/HYBRID strategies)
        user_id: Optional user identifier
        session_id: Optional session identifier
        limit: Maximum number of messages to return (defaults to memory_limit)
        strategy: Which retrieval strategy to use (ALL, RELEVANT, or HYBRID)

    Returns:
        List of messages forming a valid conversation context
    """
    if not self.memory or not (user_id or session_id):
        return []

    filters = {}
    if user_id:
        filters["user_id"] = user_id
    if session_id:
        filters["session_id"] = session_id

    limit = limit or self.memory_limit

    if strategy == MemoryRetrievalStrategy.RELEVANT and not user_query:
        logger.warning("RELEVANT strategy selected but no user_query provided - falling back to ALL")
        strategy = MemoryRetrievalStrategy.ALL

    conversation = self.memory.get_agent_conversation(
        query=user_query,  # Pass the user query for relevance search
        limit=limit,
        filters=filters,
        strategy=strategy,
    )

    if self.verbose:
        logger.debug(
            f"Agent {self.name} retrieved {len(conversation)} messages using {strategy.value} strategy. "
            f"First message role: {conversation[0].role.value if conversation else 'None'}"
        )

    return conversation

sanitize_tool_name(s)

Sanitize tool name to follow [^a-zA-Z0-9_-].

Source code in dynamiq/nodes/agents/base.py

def sanitize_tool_name(self, s: str):
    """Sanitize tool name to follow [^a-zA-Z0-9_-]."""
    s = s.replace(" ", "-")
    sanitized = re.sub(r"[^a-zA-Z0-9_-]", "", s)
    return sanitized

set_block(block_name, content)

Adds or updates a prompt block.

Source code in dynamiq/nodes/agents/base.py

def set_block(self, block_name: str, content: str):
    """Adds or updates a prompt block."""
    self._prompt_blocks[block_name] = content

set_prompt_variable(variable_name, value)

Sets or updates a prompt variable.

Source code in dynamiq/nodes/agents/base.py

def set_prompt_variable(self, variable_name: str, value: Any):
    """Sets or updates a prompt variable."""
    self._prompt_variables[variable_name] = value

stream_by_tokens(content, source, step, config=None, **kwargs)

Streams the input content to the callbacks.

Source code in dynamiq/nodes/agents/base.py

def stream_by_tokens(self, content: str, source: str, step: str, config: RunnableConfig | None = None, **kwargs):
    """Streams the input content to the callbacks."""
    if isinstance(content, dict):
        return self.stream_response(content, source, step, config, **kwargs)
    tokens = content.split(" ")
    final_response = []
    for token in tokens:
        final_response.append(token)
        token_with_prefix = " " + token
        token_for_stream = StreamChunk(
            choices=[
                StreamChunkChoice(delta=StreamChunkChoiceDelta(content=token_with_prefix, source=source, step=step))
            ]
        )
        self.run_on_node_execute_stream(
            callbacks=config.callbacks,
            chunk=token_for_stream.model_dump(),
            **kwargs,
        )
    return " ".join(final_response)

stream_content(content, source, step, config=None, by_tokens=None, **kwargs)

Streams data.

Parameters:

Name	Type	Description	Default
content	str | dict	Data that will be streamed.	required
source	str	Source of the content.	required
step	str	Description of the step.	required
by_tokens	Optional[bool]	Determines whether to stream content by tokens or not. If None it is determined based on StreamingConfig. Defaults to None.	None
config	Optional[RunnableConfig]	Configuration for the runnable.	None
**kwargs		Additional keyword arguments.	{}

Returns:

Type	Description
str | dict	str | dict: Streamed data.

Source code in dynamiq/nodes/agents/base.py

def stream_content(
    self,
    content: str | dict,
    source: str,
    step: str,
    config: RunnableConfig | None = None,
    by_tokens: bool | None = None,
    **kwargs,
) -> str | dict:
    """
    Streams data.

    Args:
        content (str | dict): Data that will be streamed.
        source (str): Source of the content.
        step (str): Description of the step.
        by_tokens (Optional[bool]): Determines whether to stream content by tokens or not.
            If None it is determined based on StreamingConfig. Defaults to None.
        config (Optional[RunnableConfig]): Configuration for the runnable.
        **kwargs: Additional keyword arguments.

    Returns:
        str | dict: Streamed data.
    """
    if (by_tokens is None and self.streaming.by_tokens) or by_tokens:
        return self.stream_by_tokens(content=content, source=source, step=step, config=config, **kwargs)
    return self.stream_response(content=content, source=source, step=step, config=config, **kwargs)

to_dict(**kwargs)

Converts the instance to a dictionary.

Source code in dynamiq/nodes/agents/base.py

def to_dict(self, **kwargs) -> dict:
    """Converts the instance to a dictionary."""
    data = super().to_dict(**kwargs)
    data["llm"] = self.llm.to_dict(**kwargs)
    data["tools"] = [tool.to_dict(**kwargs) for tool in self.tools]
    data["memory"] = self.memory.to_dict(**kwargs) if self.memory else None
    if self.files:
        data["files"] = [{"name": getattr(f, "name", f"file_{i}")} for i, f in enumerate(self.files)]
    if self.images:
        data["images"] = [{"name": getattr(f, "name", f"image_{i}")} for i, f in enumerate(self.images)]
    return data

AgentManager

Bases: Agent

Manager class that extends the Agent class to include specific actions.

Source code in dynamiq/nodes/agents/base.py

class AgentManager(Agent):
    """Manager class that extends the Agent class to include specific actions."""

    _actions: dict[str, Callable] = PrivateAttr(default_factory=dict)
    name: str = "Agent Manager"
    input_schema: ClassVar[type[AgentManagerInputSchema]] = AgentManagerInputSchema

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._init_actions()

    def to_dict(self, **kwargs) -> dict:
        """Converts the instance to a dictionary."""
        data = super().to_dict(**kwargs)
        data["_actions"] = {
            k: getattr(action, "__name__", str(action))
            for k, action in self._actions.items()
        }
        return data

    def _init_actions(self):
        """Initializes the default actions for the manager."""
        self._actions = {"plan": self._plan, "assign": self._assign, "final": self._final}

    def add_action(self, name: str, action: Callable):
        """Adds a custom action to the manager."""
        self._actions[name] = action

    def get_context_for_input_schema(self) -> dict:
        """Provides context for input schema that is required for proper validation."""
        return {"actions": list(self._actions.keys())}

    def execute(
        self, input_data: AgentManagerInputSchema, config: RunnableConfig | None = None, **kwargs
    ) -> dict[str, Any]:
        """Executes the manager agent with the given input data and action."""
        logger.info(f"Agent {self.name} - {self.id}: started with INPUT DATA:\n{input_data}")
        self.reset_run_state()
        config = config or RunnableConfig()
        self.run_on_node_execute_run(config.callbacks, **kwargs)

        action = input_data.action

        self._prompt_variables.update(dict(input_data))

        kwargs = kwargs | {"parent_run_id": kwargs.get("run_id")}
        kwargs.pop("run_depends", None)
        _result_llm = self._actions[action](config=config, **kwargs)
        result = {"action": action, "result": _result_llm}

        execution_result = {
            "content": result,
            "intermediate_steps": self._intermediate_steps,
        }
        logger.info(f"Agent {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")

        return execution_result

    def _plan(self, config: RunnableConfig, **kwargs) -> str:
        """Executes the 'plan' action."""
        prompt = self._prompt_blocks.get("plan").format(**self._prompt_variables, **kwargs)

        llm_result = self._run_llm([Message(role=MessageRole.USER, content=prompt)], config, **kwargs).output["content"]
        if self.streaming.enabled and self.streaming.mode == StreamingMode.ALL:
            return self.stream_content(
                content=llm_result, step="manager_planning", source=self.name, config=config, by_tokens=False, **kwargs
            )

        return llm_result

    def _assign(self, config: RunnableConfig, **kwargs) -> str:
        """Executes the 'assign' action."""
        prompt = self._prompt_blocks.get("assign").format(**self._prompt_variables, **kwargs)
        llm_result = self._run_llm([Message(role=MessageRole.USER, content=prompt)], config, **kwargs).output["content"]
        if self.streaming.enabled and self.streaming.mode == StreamingMode.ALL:
            return self.stream_content(
                content=llm_result, step="manager_assigning", source=self.name, config=config, by_tokens=False, **kwargs
            )
        return llm_result

    def _final(self, config: RunnableConfig, **kwargs) -> str:
        """Executes the 'final' action."""
        prompt = self._prompt_blocks.get("final").format(**self._prompt_variables, **kwargs)
        llm_result = self._run_llm(
            [Message(role=MessageRole.USER, content=prompt)], config, by_tokens=False, **kwargs
        ).output["content"]
        if self.streaming.enabled:
            return self.stream_content(
                content=llm_result,
                step="manager_final_output",
                source=self.name,
                config=config,
                by_tokens=False,
                **kwargs,
            )
        return llm_result

add_action(name, action)

Adds a custom action to the manager.

Source code in dynamiq/nodes/agents/base.py

def add_action(self, name: str, action: Callable):
    """Adds a custom action to the manager."""
    self._actions[name] = action

execute(input_data, config=None, **kwargs)

Executes the manager agent with the given input data and action.

Source code in dynamiq/nodes/agents/base.py

def execute(
    self, input_data: AgentManagerInputSchema, config: RunnableConfig | None = None, **kwargs
) -> dict[str, Any]:
    """Executes the manager agent with the given input data and action."""
    logger.info(f"Agent {self.name} - {self.id}: started with INPUT DATA:\n{input_data}")
    self.reset_run_state()
    config = config or RunnableConfig()
    self.run_on_node_execute_run(config.callbacks, **kwargs)

    action = input_data.action

    self._prompt_variables.update(dict(input_data))

    kwargs = kwargs | {"parent_run_id": kwargs.get("run_id")}
    kwargs.pop("run_depends", None)
    _result_llm = self._actions[action](config=config, **kwargs)
    result = {"action": action, "result": _result_llm}

    execution_result = {
        "content": result,
        "intermediate_steps": self._intermediate_steps,
    }
    logger.info(f"Agent {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")

    return execution_result

get_context_for_input_schema()

Provides context for input schema that is required for proper validation.

Source code in dynamiq/nodes/agents/base.py

def get_context_for_input_schema(self) -> dict:
    """Provides context for input schema that is required for proper validation."""
    return {"actions": list(self._actions.keys())}

to_dict(**kwargs)

Converts the instance to a dictionary.

Source code in dynamiq/nodes/agents/base.py

def to_dict(self, **kwargs) -> dict:
    """Converts the instance to a dictionary."""
    data = super().to_dict(**kwargs)
    data["_actions"] = {
        k: getattr(action, "__name__", str(action))
        for k, action in self._actions.items()
    }
    return data

AgentStatus

Bases: str, Enum

Represents the status of an agent's execution.

Source code in dynamiq/nodes/agents/base.py

class AgentStatus(str, Enum):
    """Represents the status of an agent's execution."""

    SUCCESS = "success"
    FAIL = "fail"

StreamChunk

Bases: BaseModel

Model for streaming chunks with choices containing delta updates.

Source code in dynamiq/nodes/agents/base.py

class StreamChunk(BaseModel):
    """Model for streaming chunks with choices containing delta updates."""

    choices: list[StreamChunkChoice]

StreamChunkChoice

Bases: BaseModel

Stream chunk choice model.

Source code in dynamiq/nodes/agents/base.py

class StreamChunkChoice(BaseModel):
    """Stream chunk choice model."""

    delta: StreamChunkChoiceDelta

StreamChunkChoiceDelta

Bases: BaseModel

Delta model for content chunks.

Source code in dynamiq/nodes/agents/base.py

class StreamChunkChoiceDelta(BaseModel):
    """Delta model for content chunks."""
    content: str | dict
    source: str
    step: str