Class BasicStepsMixin

src/harness/execution/step_handlers/basic.py:23–172

Mixin providing basic step and task step execution. These are the most common step types, used for direct LLM interactions with optional tool usage and conversation history management. Required Attributes: llm_executor: LLMExecutor for running LLM completions. tools: Li

class BasicStepsMixin:
    """Mixin providing basic step and task step execution.

    These are the most common step types, used for direct LLM interactions
    with optional tool usage and conversation history management.

    Required Attributes:
        llm_executor: LLMExecutor for running LLM completions.
        tools: List of available tool definitions.
    """

    def execute_basic_step(
        self,
        step_data: Dict[str, Any],
        context: Dict[str, Any],
        step_number: int,
        args: EffectiveArgs,
        logger: Logger,
    ) -> Tuple[Any, int]:
        """Execute a step with persistent conversation history.

        Steps maintain conversation history across iterations, enabling
        multi-turn interactions with tool use. Each step appends a new
        user message to the workflow-level conversation history and runs
        the model against that shared history.

        Args:
            step_data: Step configuration containing 'step' key with:
                - instruction: The prompt to send to the LLM (required)
                - name: Optional step name for logging
                - system_prompt: Ignored (use workflow-level instead)
                - save_as: Optional variable name to save output to context
            context: Current workflow context dictionary.
            step_number: Identifier for this step in the workflow.
            args: Effective arguments for LLM configuration.
            logger: Logger instance for output.

        Returns:
            Tuple of (output, tokens) where output is the final LLM response
            and tokens is the total number of tokens used across all iterations.
        """
        step = step_data["step"]
        raw_name = step.get("name", f"step_{step_number}")
        name = expand_template_variables(raw_name, context)
        instruction = step["instruction"]
        expanded_instruction = expand_template_variables(instruction, context)
        logger(f"\n==== Executing step {step_number}: {name} ====")
        logger(f"Instruction: {expanded_instruction}")

        system_prompt = step.get("system_prompt")
        history = ensure_conversation_history(context, args)
        if system_prompt is not None:
            logger(
                "Step system_prompt is ignored; set system_prompt at the workflow level instead."
            )

        history.append({"role": "user", "content": expanded_instruction})
        working_messages = list(history)