Method execute_llm_step

src/harness/execution/llm_executor.py:194–257

Execute a single LLM step with tool calling loop. Prepares messages and tools, then delegates to multi_step_tool_call_loop for the actual LLM interaction. Handles system prompt resolution and effective tool configuration. Args: instruction: The prompt/in

(
        self,
        instruction: str,
        prev_output: str,  # noqa: ARG002 - kept for API compatibility, use context["prev_output"] instead
        step_id: Union[int, str],
        args: EffectiveArgs,
        logger: Logger,
        context: Optional[Dict[str, Any]] = None,
        system_prompt: Optional[str] = None,
        step_config: Optional[Dict[str, Any]] = None,
    )

        return os.path.join(host_ws, "sessions", sess_key, vpath.lstrip("/"))

    def execute_llm_step(
        self,
        instruction: str,
        prev_output: str,  # noqa: ARG002 - kept for API compatibility, use context["prev_output"] instead
        step_id: Union[int, str],
        args: EffectiveArgs,
        logger: Logger,
        context: Optional[Dict[str, Any]] = None,
        system_prompt: Optional[str] = None,
        step_config: Optional[Dict[str, Any]] = None,
    ) -> Tuple[Any, int]:
        """Execute a single LLM step with tool calling loop.

        Prepares messages and tools, then delegates to multi_step_tool_call_loop
        for the actual LLM interaction. Handles system prompt resolution and
        effective tool configuration.

        Args:
            instruction: The prompt/instruction for this step.
            prev_output: Output from the previous step (for context).
            step_id: Identifier for this step in the workflow.
            args: Effective arguments with model configuration.
            logger: Logger instance for output.
            context: Optional workflow context dictionary.
            system_prompt: Optional custom system prompt (overrides default).
            step_config: Optional step-specific configuration for tool filtering.

        Returns:
            Tuple of (output, tokens) where output is the final response
            and tokens is the total number of tokens used.
        """
        if context is None:
            context = {}
        if system_prompt is not None:
            system_prompt = resolve_system_prompt(system_prompt, context)
        else:
            system_prompt = get_yaml_system_prompt(
                context.get("task_name", "unknown"),
                context.get("goal", ""),
                args,
                context,
            )

        prompt = instruction

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ]

        tool_config = EffectiveToolConfig.compute(self.tools, context, step_config)
        expose_submodules = context.get("expose_submodules_as_tools", True)
        combined_tools = tool_config.tools + (
            tool_config.submodule_tools if expose_submodules else []
        )
        return self.multi_step_tool_call_loop(
            step_id,
            messages,