LLMAgent

Agent Module.

LLMAgent

A simple LLM Agent Class.

Attributes:

Name	Type	Description
llm	LLM	The backbone LLM.
tools_registry	dict[str, Tool]	The tools the LLM agent can equip the LLM with, represented as a dict.
templates	LLMAgentTemplates	Prompt templates for LLM Agent.
logger	Logger	LLMAgent logger.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

class LLMAgent:
    """A simple LLM Agent Class.

    Attributes:
        llm (LLM): The backbone LLM.
        tools_registry (dict[str, Tool]): The tools the LLM agent can equip
            the LLM with, represented as a dict.
        templates (LLMAgentTemplates): Prompt templates for LLM Agent.
        logger (logging.Logger): LLMAgent logger.
    """

    def __init__(
        self,
        llm: LLM,
        tools: list[Tool] | None = None,
        templates: LLMAgentTemplates = default_templates,
        # added in ch07
        memories: list[Memory] | None = None,
    ):
        """Initialize an LLMAgent.

        Args:
            llm (LLM): The backbone LLM of the LLM agent.
            tools (list[Tool], optional): The set of tools with which the
                LLM can be equipped. Defaults to None.
            templates (LLMAgentTemplates): Prompt templates for LLM Agent.
            memories (list[Memory] | None): Episodic memory backends
                to consult at task start and update at task end. Defaults
                to None (no memory). Added in Chapter 7.
        """
        self.llm = llm
        tools = tools or []
        # validate no duplications in tool names
        if len({t.name for t in tools}) < len(tools):
            raise LLMAgentError(
                "Provided tool list contains duplicate tool names.",
            )
        self.tools_registry = {t.name: t for t in tools}
        self.templates = templates
        self.logger = get_logger(self.__class__.__name__)
        # added in ch07
        self.memories = memories or []

    @property
    def tools(self) -> list[Tool]:
        """Return tools as list."""
        return list(self.tools_registry.values())

    def add_tool(self, tool: Tool) -> Self:
        """Add a tool to the agents tool set.

        NOTE: Supports fluent style for convenience.

        Args:
            tool (Tool): The tool to equip the LLM agent.

        """
        if tool.name in self.tools_registry:
            raise LLMAgentError(f"Tool with name {tool.name} already exists.")
        self.tools_registry[tool.name] = tool
        return self

    class TaskHandler(asyncio.Future):
        """Handler for processing tasks.

        Attributes:
            llm_agent (LLMAgent): The LLM agent.
            task: The task to execute.
            rollout: The execution log of the task.
            step_counter: The number of TaskSteps executed.
            logger: TaskHandler logger.
            skills (dict[str, Skill]): Skills discovered at the start of
                each run, keyed by name. Added in Chapter 6.
            _explicit_only_skills (set[str]): Skill names excluded from the
                model-visible catalog for this run. They remain loadable via
                ``run_with_skill()``. Added in Chapter 6.
            _use_skill_tool (UseSkillTool | None): Task-scoped skill
                activation tool. Set when skills are discovered; ``None``
                otherwise. Added in Chapter 6.
        """

        def __init__(
            self,
            llm_agent: "LLMAgent",
            task: Task,
            # added in ch06
            skills_scopes: list[SkillScope] | None = None,
            explicit_only_skills: set[str] | None = None,
            *args: Any,
            **kwargs: Any,
        ) -> None:
            """Initialize a TaskHandler.

            Args:
                llm_agent (LLMAgent): The LLM agent.
                task (Task): The task to process.
                skills_scopes (list[SkillScope] | None): Scopes to scan for
                    skills. Defaults to ``[USER, PROJECT]``. Added in
                    Chapter 6.
                explicit_only_skills (set[str] | None): Skill names to
                    exclude from the model catalog. Defaults to None.
                    Added in Chapter 6.
                *args: Additional positional arguments.
                **kwargs: Additional keyword arguments.
            """
            super().__init__(*args, **kwargs)
            self.llm_agent = llm_agent
            self.task = task
            self.rollout = ""
            self.step_counter = 0
            self._background_task: asyncio.Task | None = None
            self.logger = get_logger(self.__class__.__name__)
            # added in ch06
            _scopes = (
                skills_scopes
                if skills_scopes is not None
                else [SkillScope.USER, SkillScope.PROJECT]
            )
            self.skills: dict[str, Skill] = discover_skills(_scopes)
            self._explicit_only_skills: set[str] = explicit_only_skills or set()
            self._use_skill_tool: UseSkillTool | None = (
                UseSkillTool(
                    skills=self.skills,
                    explicit_only_skills=self._explicit_only_skills,
                )
                if self.skills
                else None
            )
            # added in ch07
            self._recalled_memories: str = ""

        @property
        def background_task(self) -> asyncio.Task:
            """Get the background ~asyncio.Task for the handler."""
            if not self._background_task:
                raise TaskHandlerError(
                    "No background task is running for this handler.",
                )
            return self._background_task

        @background_task.setter
        def background_task(self, asyncio_task: asyncio.Task) -> None:
            """Setter for background_task."""
            if self._background_task is not None:
                raise TaskHandlerError(
                    "A background task has already been set.",
                )
            self._background_task = asyncio_task

        @property
        def _skills_catalog(self) -> str:
            """Return formatted skills catalog, or empty string.

            Added in Chapter 6.

            Builds the ``<available_skills>`` XML block from discovered
            skills that have not opted out of model-driven activation
            (i.e. ``disable-model-invocation`` is not set). Returns an
            empty string when no visible skills remain so callers can
            append it unconditionally without adding noise.
            """
            visible = [
                skill
                for name, skill in self.skills.items()
                if name not in self._explicit_only_skills
            ]
            if not visible:
                return ""
            entries = "\n".join(skill.catalog() for skill in visible)
            return self.llm_agent.templates["skills_catalog"].format(
                skills=entries,
            )

        def _format_step_for_rollout(
            self,
            chat_history: list[ChatMessage],
        ) -> str:
            """Format a run_step's chat history as a rollout entry."""
            rollout_lines = ["=== Task Step Start ==="]
            for msg in chat_history:
                # don't include system messages in rollout
                content = msg.content
                role = msg.role

                if role == "system":
                    continue

                if role == "user":
                    # From the LLMAgent to the backbone LLM, but in a rollout
                    # we'll simplify to just LLM agent having a monologue
                    role = ChatRole.ASSISTANT
                    content = self.llm_agent.templates[
                        "step_rollout_content_instruction"
                    ].format(
                        instruction=content,
                    )

                if msg.tool_calls and msg.role == "assistant":
                    called_tools = "\n\n".join(
                        [
                            f"{t.model_dump_json(indent=4)}"
                            for t in msg.tool_calls
                        ],
                    )
                    content = self.llm_agent.templates[
                        "step_rollout_content_tool_call_request"
                    ].format(
                        called_tools=called_tools,
                    )

                rollout_lines.append(
                    self.llm_agent.templates[
                        "step_rollout_chat_message"
                    ].format(
                        actor=("🔧 " if role == ChatRole.TOOL else "💬 ")
                        + role.value,
                        content=content,
                    ),
                )

            rollout_lines.append(
                "=== Task Step End ===",
            )

            return "\n\n".join(rollout_lines)

        def _format_memories_for_system_prompt(
            self,
            memories: list[str],
        ) -> str:
            if memories:
                entries = "\n".join(memories)
                return self.llm_agent.templates["memories"].format(
                    memories=entries,
                )
            return ""

        async def get_next_step(
            self,
            previous_step_result: (
                TaskStepResult
                | RejectedTaskResult  # added in ch08
                | None
            ),
        ) -> TaskStep | TaskResult:
            """Based on previous step result, get next step or conclude task.

            Returns:
                TaskStep | TaskResult: Either the next step or the result of
                    the task.
            """
            if not previous_step_result:
                return TaskStep(
                    task_id=self.task.id_,
                    instruction=self.task.instruction,
                )
            # added in ch08: rejection bypasses LLM routing
            if isinstance(previous_step_result, RejectedTaskResult):
                self.logger.info(
                    f"🧠 New Step (rejection): {previous_step_result.feedback}",
                )
                return TaskStep(
                    task_id=self.task.id_,
                    instruction=self.llm_agent.templates[
                        "approval_rejection_feedback"
                    ].format(
                        content=previous_step_result.failed_result_content,
                        feedback=previous_step_result.feedback,
                    ),
                )
            self.logger.debug(f"🧵 Rollout: {self.rollout}")

            prompt = self.llm_agent.templates["get_next_step"].format(
                instruction=self.task.instruction,
                current_rollout=self.rollout,
                current_response=previous_step_result.content,
            )
            self.logger.debug(f"---NEXT STEP PROMPT: {prompt}")
            try:
                next_step = await self.llm_agent.llm.structured_output(
                    prompt=prompt,
                    mdl=NextStepDecision,
                )
                self.logger.debug(
                    f"---NEXT STEP: {next_step.model_dump_json()}",
                )
            except Exception as e:
                raise TaskHandlerError(
                    f"Failed to get next step: {str(e)}",
                ) from e

            if next_step.kind == "final_result":
                self.logger.info("No new step required.")
                retval = TaskResult(
                    task_id=self.task.id_,
                    content=previous_step_result.content,
                )
            else:  # next_step.kind == "next_step":
                self.logger.info(f"🧠 New Step: {next_step.content}")
                retval = TaskStep(
                    task_id=self.task.id_,
                    instruction=next_step.content,
                )

            return retval

        async def run_step(self, step: TaskStep) -> TaskStepResult:  # noqa: PLR0912
            """Run next step of a given task.

            A single step is executed through a single-turn conversation that
            the LLM agent has with itself. In other words, it is both the `user`
            providing the instruction (from `get_next_step`) as well as the
            `assistant` that provides the result.

            Args:
                step (TaskStep): The step to execute.

            Returns:
                TaskStepResult: The result of the step execution.
            """
            self.step_counter += 1
            self.logger.info(f"⚙️ Processing Step: {step.instruction}")
            self.logger.debug(f"🧵 Rollout: {self.rollout}")

            # include rollout as context in the system message
            system_message = ChatMessage(
                role=ChatRole.SYSTEM,
                content=self.llm_agent.templates[
                    "run_step_system_message"
                ].format(
                    llm_agent_system_message=self.llm_agent.templates[
                        "system_message"
                    ],
                    current_rollout=self.rollout,
                )
                if self.rollout
                else self.llm_agent.templates[
                    "run_step_system_message_without_rollout"
                ].format(
                    llm_agent_system_message=self.llm_agent.templates[
                        "system_message"
                    ],
                ),
            )

            # added in ch07: inject recalled memories
            if memories := self._recalled_memories:
                system_message = ChatMessage(
                    role=ChatRole.SYSTEM,
                    content=f"{system_message.content}\n\n{memories}",
                )

            # added in ch06: bolt on skills catalog when skills are available
            if catalog := self._skills_catalog:
                system_message = ChatMessage(
                    role=ChatRole.SYSTEM,
                    content=f"{system_message.content}\n\n{catalog}",
                )

            self.logger.debug(f"💬 SYSTEM: {system_message.content}")

            # fictitious user's input
            user_input = self.llm_agent.templates[
                "run_step_user_message"
            ].format(
                instruction=step.instruction,
            )
            self.logger.debug(f"💬 USER INPUT: {user_input}")

            # start single-turn conversation
            # added in ch06: include use_skill tool when skills are available
            all_tools = self.llm_agent.tools + (
                [self._use_skill_tool] if self._use_skill_tool else []
            )
            user_message, response_message = await self.llm_agent.llm.chat(
                input=user_input,
                chat_history=[system_message],
                tools=all_tools,
            )
            self.logger.debug(f"💬 ASSISTANT: {response_message.content}")

            # check if there are tool calls
            if response_message.tool_calls:
                tool_call_results = []
                for tool_call in response_message.tool_calls:
                    self.logger.info(
                        f"🛠️ Executing Tool Call: {tool_call.tool_name}",
                    )
                    if tool := (
                        self.llm_agent.tools_registry.get(
                            tool_call.tool_name,
                        )
                        or (
                            self._use_skill_tool
                            if self._use_skill_tool
                            and tool_call.tool_name == self._use_skill_tool.name
                            else None
                        )
                    ):
                        if isinstance(tool, AsyncBaseTool):
                            tool_call_result = await tool(tool_call=tool_call)
                        else:
                            tool_call_result = tool(tool_call=tool_call)
                        msg = (
                            "✅ Successful Tool Call: "
                            f"{tool_call_result.content}"
                        )
                        self.logger.info(msg)
                    else:
                        error_msg = (
                            f"Tool with name {tool_call.tool_name} "
                            "doesn't exist."
                        )
                        tool_call_result = ToolCallResult(
                            tool_call_id=tool_call.id_,
                            error=True,
                            content=error_msg,
                        )
                        self.logger.info(
                            f"❌ Tool Call Failure: {tool_call_result.content}",
                        )
                    tool_call_results.append(tool_call_result)

                # send tool call results back to llm to get result
                (
                    tool_messages,
                    another_response_message,
                ) = await self.llm_agent.llm.continue_chat_with_tool_results(  # noqa: E501
                    tool_call_results=tool_call_results,
                    chat_history=[
                        system_message,
                        user_message,
                        response_message,
                    ],
                )

                # get final content and update chat history
                if another_response_message.tool_calls:
                    # if has tool calls, we'll make them in the next step
                    final_content = "I need to make the following tool-calls:\n"
                    final_content += "\n".join(
                        t.model_dump_json(indent=4)
                        for t in another_response_message.tool_calls
                    )
                else:
                    final_content = another_response_message.content
                chat_history = (
                    [
                        system_message,
                        user_message,
                        response_message,
                    ]
                    + tool_messages
                    + [another_response_message]
                )
            else:
                final_content = response_message.content
                chat_history = [
                    system_message,
                    user_message,
                    response_message,
                ]

            # augment rollout from this turn
            formatted_step = self._format_step_for_rollout(
                chat_history=chat_history,
            )
            if self.rollout:
                self.rollout += "\n\n" + formatted_step

            else:
                self.rollout = formatted_step

            self.logger.info(
                f"✅ Step Result: {final_content}",
            )
            return TaskStepResult(
                task_step_id=step.id_,
                content=final_content,
            )

        async def load_memories(self) -> None:
            """Recall relevant episodes from all configured memory backends.

            Added in Chapter 7.

            Calls ``recall`` on each memory in ``self.llm_agent.memories``
            and stores the formatted string in ``self._recalled_memories``
            for prompt injection during ``run_step``. No-op when no memories
            are configured.
            """
            loaded = []
            for memory in self.llm_agent.memories:
                block = await memory.recall(self.task)
                loaded.append(block)
            self._recalled_memories = self._format_memories_for_system_prompt(
                loaded,
            )

        async def record_memory(
            self,
            result: TaskResult | None = None,
            error: Exception | None = None,
        ) -> None:
            """Build an Episode and write it to all configured memories.

            Exactly one of ``result`` or ``error`` must be provided.
            Called before ``set_result()`` / ``set_exception()`` so that
            ``await agent.run(task)`` returns only after the episode is
            written.

            Added in Chapter 7.

            Args:
                result (TaskResult | None): The successful task result.
                error (Exception | None): The exception from a failed task.

            Raises:
                RecordMemoryError: If neither ``result`` nor ``error`` is
                    provided.
            """
            if result is None and error is None:
                raise RecordMemoryError(
                    "record_memory() requires either result or error.",
                )
            episode = Episode(
                task=self.task,
                rollout=self.rollout,
                result=result,
                error=error,
            )
            for memory in self.llm_agent.memories:
                await memory.record(episode)

        async def request_approval(
            self,
            result: TaskResult,
        ) -> ApprovalResult:
            """Ask a human to approve or reject the proposed task result.

            Added in Chapter 8.

            Operator-gated human-in-the-loop pattern; unlike
            ``HumanInputTool``, the pause is not agent-initiated.
            Runs the blocking rich prompts in a thread via
            ``asyncio.to_thread``. Auto-approves on ``EOFError`` or
            ``KeyboardInterrupt`` (headless / interrupted terminal).

            Args:
                result (TaskResult): The proposed task result to review.

            Returns:
                ApprovalResult: The approval decision.
            """

            def _prompt_for_approval(
                task_result: TaskResult,
            ) -> ApprovalResult:
                console = Console()
                console.print(
                    Panel(
                        task_result.content,
                        title="Proposed Task Result",
                        border_style="cyan",
                    ),
                )
                approved = Confirm.ask("Approve this result?", console=console)
                if approved:
                    return ApprovalResult(approved=True, feedback="")
                feedback = Prompt.ask(
                    "Provide your correction rationale for the LLM agent to address",  # noqa: E501
                    console=console,
                )
                return ApprovalResult(approved=False, feedback=feedback)

            try:
                return await asyncio.to_thread(
                    _prompt_for_approval,
                    result,
                )
            except EOFError:
                self.logger.info(
                    "Approval prompt got EOF (headless); auto-approving.",
                )
                return ApprovalResult(approved=True, feedback="")
            except KeyboardInterrupt:
                self.logger.info(
                    "Approval prompt interrupted by operator; auto-approving.",
                )
                return ApprovalResult(
                    approved=True,
                    feedback="",
                )

    class SupervisedTaskHandler(TaskHandler):
        """TaskHandler for human-driven stepwise execution.

        Added in Chapter 8. Caller-driven human-in-the-loop pattern;
        unlike ``HumanInputTool`` (agent-initiated) and
        ``request_approval`` (operator-gated at result time), the human
        controls the entire execution cadence. Returned by
        ``run_supervised()``; the caller drives the loop manually via
        ``get_next_step()`` and ``run_step()`` and finalises execution
        with ``complete()`` or ``abort()``.
        """

        @property
        def background_task(self) -> asyncio.Task:
            """Not available in supervised mode."""
            raise TaskHandlerError(
                "SupervisedTaskHandler has no background task — "
                "execution is caller-driven via get_next_step() "
                "and run_step().",
            )

        @background_task.setter
        def background_task(self, asyncio_task: asyncio.Task) -> None:
            """Not available in supervised mode."""
            raise TaskHandlerError(
                "SupervisedTaskHandler has no background task — "
                "execution is caller-driven via get_next_step() "
                "and run_step().",
            )

        async def complete(self, result: TaskResult) -> None:
            """Accept the final result and resolve the handler.

            Added in Chapter 8.

            Args:
                result: The ``TaskResult`` to accept.
            """
            if not isinstance(result, TaskResult):
                raise TaskHandlerError(
                    f"complete() requires a TaskResult, "
                    f"got {type(result).__name__}.",
                )
            await self.record_memory(result=result)
            self.set_result(result)

        def reject(
            self,
            result: TaskResult,
            feedback: str,
        ) -> RejectedTaskResult:
            """Reject a proposed TaskResult and return feedback for re-routing.

            Added in Chapter 8.

            Args:
                result: The ``TaskResult`` to reject.
                feedback: Correction rationale passed back to the agent.

            Returns:
                RejectedTaskResult: Pass to ``get_next_step()`` to
                    re-enter the loop without consulting the LLM.
            """
            return RejectedTaskResult(
                failed_result_content=result.content,
                feedback=feedback,
            )

        async def abort(self, error: Exception | None = None) -> None:
            """Abort the supervised task and resolve the handler.

            Added in Chapter 8.

            Args:
                error: Exception to set. Defaults to
                    ``TaskHandlerError("Task aborted.")``.
            """
            err = error or TaskHandlerError("Task aborted.")
            await self.record_memory(error=err)
            self.set_exception(err)

    def run(
        self,
        task: Task,
        max_steps: int | None = None,
        # added in ch06
        skills_scopes: list[SkillScope] | None = None,
        explicit_only_skills: set[str] | None = None,
        # added in ch08
        with_approval: bool = False,
    ) -> TaskHandler:
        """Agent's processing loop for executing tasks.

        Args:
            task (Task): the Task to perform.
            max_steps (int | None): Maximum number of steps to run for task.
                Defaults to None.
            skills_scopes (list[SkillScope] | None): Scopes to scan for
                skills, in processing order (last wins on name collision).
                Defaults to ``[USER, PROJECT]``. Added in Chapter 6.
            explicit_only_skills (set[str] | None): Skill names to exclude
                from the model catalog for this run. They remain activatable
                via ``run_with_skill()``. Defaults to None. Added in
                Chapter 6.
            with_approval (bool): When ``True``, an end-of-loop human
                approval gate fires before each ``TaskResult`` is accepted.
                The human may approve (result is recorded and returned) or
                reject with feedback (feedback re-enters the loop as a new
                step). Rejections do not consume the step budget; pair with
                ``max_steps`` to bound repeated-rejection loops. Defaults
                to ``False``. Added in Chapter 8.

        Returns:
            TaskHandler: the TaskHandler object responsible for task execution.
        """
        task_handler = self.TaskHandler(
            llm_agent=self,
            task=task,
            skills_scopes=skills_scopes,
            explicit_only_skills=explicit_only_skills,
        )

        async def _process_loop() -> None:
            """The processing loop for the task handler execute its task.

            Cycle between get_next_step and run_step, until the task_handler
            is marked as done, either through a set result or an exception being
            set.
            """
            self.logger.info(f"🚀 Starting task: {task.instruction}")
            step_result = None

            # added in ch07
            await task_handler.load_memories()

            while not task_handler.done():
                try:
                    if task_handler.step_counter == max_steps:
                        raise MaxStepsReachedError("Max steps reached.")

                    next_step = await task_handler.get_next_step(step_result)

                    match next_step:
                        case TaskStep():
                            step_result = await task_handler.run_step(
                                next_step,
                            )
                        case TaskResult():
                            # added in ch08
                            if with_approval:
                                approval = await task_handler.request_approval(
                                    next_step,
                                )
                                if not approval.approved:
                                    step_result = RejectedTaskResult(
                                        failed_result_content=next_step.content,
                                        feedback=approval.feedback,
                                    )
                                    self.logger.info(
                                        "🔁 Task result rejected; "
                                        "re-entering loop with feedback.",
                                    )
                                    continue
                            await task_handler.record_memory(
                                result=next_step,
                            )  # added in ch07
                            task_handler.set_result(next_step)
                            self.logger.info(
                                f"🏁 Task completed: {next_step.content}",
                            )

                except Exception as e:
                    await task_handler.record_memory(error=e)  # added in ch07
                    task_handler.set_exception(e)

        task_handler.background_task = asyncio.create_task(_process_loop())

        return task_handler

    def run_with_skill(
        self,
        skill_name: str,
        prompt: str | None = None,
        max_steps: int | None = None,
        # added in ch08
        with_approval: bool = False,
    ) -> TaskHandler:
        """User-explicit skill activation: the programmatic slash command.

        Added in Chapter 6.

        Frames the task instruction to direct the model to activate the named
        skill as its first action, then runs the full agent loop. Relies on
        the model's tool-use ability to call ``use_skill`` — a fair assumption
        given the whole system depends on it. Unknown skill names are caught
        by the guard in ``UseSkillTool.__call__``.

        Args:
            skill_name (str): Name of the skill to activate.
            prompt (str | None): Optional instruction to pass alongside the
                skill activation. Defaults to None.
            max_steps (int | None): Maximum number of steps to run.
                Defaults to None.
            with_approval (bool): Passed through to ``run()``. Added in
                Chapter 8.

        Returns:
            TaskHandler: The handler responsible for task execution.
        """
        if prompt:
            instruction = EXPLICIT_SKILL_ACTIVATION_WITH_PROMPT_TEMPLATE.format(
                name=skill_name,
                prompt=prompt,
            )
        else:
            instruction = EXPLICIT_SKILL_ACTIVATION_TEMPLATE.format(
                name=skill_name,
            )
        task = Task(instruction=instruction)

        return self.run(
            task=task,
            max_steps=max_steps,
            # added in ch08
            with_approval=with_approval,
        )

    async def run_supervised(
        self,
        task: Task,
        skills_scopes: list[SkillScope] | None = None,
        explicit_only_skills: set[str] | None = None,
    ) -> SupervisedTaskHandler:
        """Human-driven stepwise task execution.

        Added in Chapter 8. Creates and returns a
        ``SupervisedTaskHandler`` with memories loaded, without starting
        the autonomous ``_process_loop``. The caller drives execution
        cell-by-cell via ``get_next_step()`` and ``run_step()``, then
        finalises with ``complete()`` or ``abort()``.

        Contrasts with ``run()``: supervised = human controls cadence;
        autonomous = agent runs to completion.

        Args:
            task: The task to perform.
            skills_scopes (list[SkillScope] | None): Scopes to scan for
                skills. Defaults to ``[USER, PROJECT]``.
            explicit_only_skills (set[str] | None): Skill names to
                exclude from the model catalog. Defaults to None.

        Returns:
            SupervisedTaskHandler: Ready for stepwise execution.
        """
        task_handler = self.SupervisedTaskHandler(
            llm_agent=self,
            task=task,
            skills_scopes=skills_scopes,
            explicit_only_skills=explicit_only_skills,
        )
        await task_handler.load_memories()
        return task_handler

    async def run_supervised_with_skill(
        self,
        skill_name: str,
        prompt: str | None = None,
        skills_scopes: list[SkillScope] | None = None,
        explicit_only_skills: set[str] | None = None,
    ) -> SupervisedTaskHandler:
        """Human-driven stepwise execution with a pre-loaded skill.

        Added in Chapter 8. Combines ``run_with_skill()`` (skill
        activation framing) with ``run_supervised()`` (caller-controlled
        cadence). The named skill is embedded in the task instruction so
        the model activates it as its first action; the caller then
        drives execution cell-by-cell via ``get_next_step()`` and
        ``run_step()``.

        Args:
            skill_name (str): Name of the skill to activate.
            prompt (str | None): Optional instruction to pass alongside
                the skill activation. Defaults to None.
            skills_scopes (list[SkillScope] | None): Scopes to scan for
                skills. Defaults to ``[USER, PROJECT]``.
            explicit_only_skills (set[str] | None): Skill names to
                exclude from the model catalog. Defaults to None.

        Returns:
            SupervisedTaskHandler: Ready for stepwise execution.
        """
        if prompt:
            instruction = EXPLICIT_SKILL_ACTIVATION_WITH_PROMPT_TEMPLATE.format(
                name=skill_name,
                prompt=prompt,
            )
        else:
            instruction = EXPLICIT_SKILL_ACTIVATION_TEMPLATE.format(
                name=skill_name,
            )
        task = Task(instruction=instruction)
        return await self.run_supervised(
            task=task,
            skills_scopes=skills_scopes,
            explicit_only_skills=explicit_only_skills,
        )

tools property

tools

Return tools as list.

TaskHandler

Bases: Future

Handler for processing tasks.

Attributes:

Name	Type	Description
llm_agent	LLMAgent	The LLM agent.
task		The task to execute.
rollout		The execution log of the task.
step_counter		The number of TaskSteps executed.
logger		TaskHandler logger.
skills	dict[str, Skill]	Skills discovered at the start of each run, keyed by name. Added in Chapter 6.
_explicit_only_skills	set[str]	Skill names excluded from the model-visible catalog for this run. They remain loadable via run_with_skill(). Added in Chapter 6.
_use_skill_tool	UseSkillTool | None	Task-scoped skill activation tool. Set when skills are discovered; None otherwise. Added in Chapter 6.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

class TaskHandler(asyncio.Future):
    """Handler for processing tasks.

    Attributes:
        llm_agent (LLMAgent): The LLM agent.
        task: The task to execute.
        rollout: The execution log of the task.
        step_counter: The number of TaskSteps executed.
        logger: TaskHandler logger.
        skills (dict[str, Skill]): Skills discovered at the start of
            each run, keyed by name. Added in Chapter 6.
        _explicit_only_skills (set[str]): Skill names excluded from the
            model-visible catalog for this run. They remain loadable via
            ``run_with_skill()``. Added in Chapter 6.
        _use_skill_tool (UseSkillTool | None): Task-scoped skill
            activation tool. Set when skills are discovered; ``None``
            otherwise. Added in Chapter 6.
    """

    def __init__(
        self,
        llm_agent: "LLMAgent",
        task: Task,
        # added in ch06
        skills_scopes: list[SkillScope] | None = None,
        explicit_only_skills: set[str] | None = None,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        """Initialize a TaskHandler.

        Args:
            llm_agent (LLMAgent): The LLM agent.
            task (Task): The task to process.
            skills_scopes (list[SkillScope] | None): Scopes to scan for
                skills. Defaults to ``[USER, PROJECT]``. Added in
                Chapter 6.
            explicit_only_skills (set[str] | None): Skill names to
                exclude from the model catalog. Defaults to None.
                Added in Chapter 6.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.
        """
        super().__init__(*args, **kwargs)
        self.llm_agent = llm_agent
        self.task = task
        self.rollout = ""
        self.step_counter = 0
        self._background_task: asyncio.Task | None = None
        self.logger = get_logger(self.__class__.__name__)
        # added in ch06
        _scopes = (
            skills_scopes
            if skills_scopes is not None
            else [SkillScope.USER, SkillScope.PROJECT]
        )
        self.skills: dict[str, Skill] = discover_skills(_scopes)
        self._explicit_only_skills: set[str] = explicit_only_skills or set()
        self._use_skill_tool: UseSkillTool | None = (
            UseSkillTool(
                skills=self.skills,
                explicit_only_skills=self._explicit_only_skills,
            )
            if self.skills
            else None
        )
        # added in ch07
        self._recalled_memories: str = ""

    @property
    def background_task(self) -> asyncio.Task:
        """Get the background ~asyncio.Task for the handler."""
        if not self._background_task:
            raise TaskHandlerError(
                "No background task is running for this handler.",
            )
        return self._background_task

    @background_task.setter
    def background_task(self, asyncio_task: asyncio.Task) -> None:
        """Setter for background_task."""
        if self._background_task is not None:
            raise TaskHandlerError(
                "A background task has already been set.",
            )
        self._background_task = asyncio_task

    @property
    def _skills_catalog(self) -> str:
        """Return formatted skills catalog, or empty string.

        Added in Chapter 6.

        Builds the ``<available_skills>`` XML block from discovered
        skills that have not opted out of model-driven activation
        (i.e. ``disable-model-invocation`` is not set). Returns an
        empty string when no visible skills remain so callers can
        append it unconditionally without adding noise.
        """
        visible = [
            skill
            for name, skill in self.skills.items()
            if name not in self._explicit_only_skills
        ]
        if not visible:
            return ""
        entries = "\n".join(skill.catalog() for skill in visible)
        return self.llm_agent.templates["skills_catalog"].format(
            skills=entries,
        )

    def _format_step_for_rollout(
        self,
        chat_history: list[ChatMessage],
    ) -> str:
        """Format a run_step's chat history as a rollout entry."""
        rollout_lines = ["=== Task Step Start ==="]
        for msg in chat_history:
            # don't include system messages in rollout
            content = msg.content
            role = msg.role

            if role == "system":
                continue

            if role == "user":
                # From the LLMAgent to the backbone LLM, but in a rollout
                # we'll simplify to just LLM agent having a monologue
                role = ChatRole.ASSISTANT
                content = self.llm_agent.templates[
                    "step_rollout_content_instruction"
                ].format(
                    instruction=content,
                )

            if msg.tool_calls and msg.role == "assistant":
                called_tools = "\n\n".join(
                    [
                        f"{t.model_dump_json(indent=4)}"
                        for t in msg.tool_calls
                    ],
                )
                content = self.llm_agent.templates[
                    "step_rollout_content_tool_call_request"
                ].format(
                    called_tools=called_tools,
                )

            rollout_lines.append(
                self.llm_agent.templates[
                    "step_rollout_chat_message"
                ].format(
                    actor=("🔧 " if role == ChatRole.TOOL else "💬 ")
                    + role.value,
                    content=content,
                ),
            )

        rollout_lines.append(
            "=== Task Step End ===",
        )

        return "\n\n".join(rollout_lines)

    def _format_memories_for_system_prompt(
        self,
        memories: list[str],
    ) -> str:
        if memories:
            entries = "\n".join(memories)
            return self.llm_agent.templates["memories"].format(
                memories=entries,
            )
        return ""

    async def get_next_step(
        self,
        previous_step_result: (
            TaskStepResult
            | RejectedTaskResult  # added in ch08
            | None
        ),
    ) -> TaskStep | TaskResult:
        """Based on previous step result, get next step or conclude task.

        Returns:
            TaskStep | TaskResult: Either the next step or the result of
                the task.
        """
        if not previous_step_result:
            return TaskStep(
                task_id=self.task.id_,
                instruction=self.task.instruction,
            )
        # added in ch08: rejection bypasses LLM routing
        if isinstance(previous_step_result, RejectedTaskResult):
            self.logger.info(
                f"🧠 New Step (rejection): {previous_step_result.feedback}",
            )
            return TaskStep(
                task_id=self.task.id_,
                instruction=self.llm_agent.templates[
                    "approval_rejection_feedback"
                ].format(
                    content=previous_step_result.failed_result_content,
                    feedback=previous_step_result.feedback,
                ),
            )
        self.logger.debug(f"🧵 Rollout: {self.rollout}")

        prompt = self.llm_agent.templates["get_next_step"].format(
            instruction=self.task.instruction,
            current_rollout=self.rollout,
            current_response=previous_step_result.content,
        )
        self.logger.debug(f"---NEXT STEP PROMPT: {prompt}")
        try:
            next_step = await self.llm_agent.llm.structured_output(
                prompt=prompt,
                mdl=NextStepDecision,
            )
            self.logger.debug(
                f"---NEXT STEP: {next_step.model_dump_json()}",
            )
        except Exception as e:
            raise TaskHandlerError(
                f"Failed to get next step: {str(e)}",
            ) from e

        if next_step.kind == "final_result":
            self.logger.info("No new step required.")
            retval = TaskResult(
                task_id=self.task.id_,
                content=previous_step_result.content,
            )
        else:  # next_step.kind == "next_step":
            self.logger.info(f"🧠 New Step: {next_step.content}")
            retval = TaskStep(
                task_id=self.task.id_,
                instruction=next_step.content,
            )

        return retval

    async def run_step(self, step: TaskStep) -> TaskStepResult:  # noqa: PLR0912
        """Run next step of a given task.

        A single step is executed through a single-turn conversation that
        the LLM agent has with itself. In other words, it is both the `user`
        providing the instruction (from `get_next_step`) as well as the
        `assistant` that provides the result.

        Args:
            step (TaskStep): The step to execute.

        Returns:
            TaskStepResult: The result of the step execution.
        """
        self.step_counter += 1
        self.logger.info(f"⚙️ Processing Step: {step.instruction}")
        self.logger.debug(f"🧵 Rollout: {self.rollout}")

        # include rollout as context in the system message
        system_message = ChatMessage(
            role=ChatRole.SYSTEM,
            content=self.llm_agent.templates[
                "run_step_system_message"
            ].format(
                llm_agent_system_message=self.llm_agent.templates[
                    "system_message"
                ],
                current_rollout=self.rollout,
            )
            if self.rollout
            else self.llm_agent.templates[
                "run_step_system_message_without_rollout"
            ].format(
                llm_agent_system_message=self.llm_agent.templates[
                    "system_message"
                ],
            ),
        )

        # added in ch07: inject recalled memories
        if memories := self._recalled_memories:
            system_message = ChatMessage(
                role=ChatRole.SYSTEM,
                content=f"{system_message.content}\n\n{memories}",
            )

        # added in ch06: bolt on skills catalog when skills are available
        if catalog := self._skills_catalog:
            system_message = ChatMessage(
                role=ChatRole.SYSTEM,
                content=f"{system_message.content}\n\n{catalog}",
            )

        self.logger.debug(f"💬 SYSTEM: {system_message.content}")

        # fictitious user's input
        user_input = self.llm_agent.templates[
            "run_step_user_message"
        ].format(
            instruction=step.instruction,
        )
        self.logger.debug(f"💬 USER INPUT: {user_input}")

        # start single-turn conversation
        # added in ch06: include use_skill tool when skills are available
        all_tools = self.llm_agent.tools + (
            [self._use_skill_tool] if self._use_skill_tool else []
        )
        user_message, response_message = await self.llm_agent.llm.chat(
            input=user_input,
            chat_history=[system_message],
            tools=all_tools,
        )
        self.logger.debug(f"💬 ASSISTANT: {response_message.content}")

        # check if there are tool calls
        if response_message.tool_calls:
            tool_call_results = []
            for tool_call in response_message.tool_calls:
                self.logger.info(
                    f"🛠️ Executing Tool Call: {tool_call.tool_name}",
                )
                if tool := (
                    self.llm_agent.tools_registry.get(
                        tool_call.tool_name,
                    )
                    or (
                        self._use_skill_tool
                        if self._use_skill_tool
                        and tool_call.tool_name == self._use_skill_tool.name
                        else None
                    )
                ):
                    if isinstance(tool, AsyncBaseTool):
                        tool_call_result = await tool(tool_call=tool_call)
                    else:
                        tool_call_result = tool(tool_call=tool_call)
                    msg = (
                        "✅ Successful Tool Call: "
                        f"{tool_call_result.content}"
                    )
                    self.logger.info(msg)
                else:
                    error_msg = (
                        f"Tool with name {tool_call.tool_name} "
                        "doesn't exist."
                    )
                    tool_call_result = ToolCallResult(
                        tool_call_id=tool_call.id_,
                        error=True,
                        content=error_msg,
                    )
                    self.logger.info(
                        f"❌ Tool Call Failure: {tool_call_result.content}",
                    )
                tool_call_results.append(tool_call_result)

            # send tool call results back to llm to get result
            (
                tool_messages,
                another_response_message,
            ) = await self.llm_agent.llm.continue_chat_with_tool_results(  # noqa: E501
                tool_call_results=tool_call_results,
                chat_history=[
                    system_message,
                    user_message,
                    response_message,
                ],
            )

            # get final content and update chat history
            if another_response_message.tool_calls:
                # if has tool calls, we'll make them in the next step
                final_content = "I need to make the following tool-calls:\n"
                final_content += "\n".join(
                    t.model_dump_json(indent=4)
                    for t in another_response_message.tool_calls
                )
            else:
                final_content = another_response_message.content
            chat_history = (
                [
                    system_message,
                    user_message,
                    response_message,
                ]
                + tool_messages
                + [another_response_message]
            )
        else:
            final_content = response_message.content
            chat_history = [
                system_message,
                user_message,
                response_message,
            ]

        # augment rollout from this turn
        formatted_step = self._format_step_for_rollout(
            chat_history=chat_history,
        )
        if self.rollout:
            self.rollout += "\n\n" + formatted_step

        else:
            self.rollout = formatted_step

        self.logger.info(
            f"✅ Step Result: {final_content}",
        )
        return TaskStepResult(
            task_step_id=step.id_,
            content=final_content,
        )

    async def load_memories(self) -> None:
        """Recall relevant episodes from all configured memory backends.

        Added in Chapter 7.

        Calls ``recall`` on each memory in ``self.llm_agent.memories``
        and stores the formatted string in ``self._recalled_memories``
        for prompt injection during ``run_step``. No-op when no memories
        are configured.
        """
        loaded = []
        for memory in self.llm_agent.memories:
            block = await memory.recall(self.task)
            loaded.append(block)
        self._recalled_memories = self._format_memories_for_system_prompt(
            loaded,
        )

    async def record_memory(
        self,
        result: TaskResult | None = None,
        error: Exception | None = None,
    ) -> None:
        """Build an Episode and write it to all configured memories.

        Exactly one of ``result`` or ``error`` must be provided.
        Called before ``set_result()`` / ``set_exception()`` so that
        ``await agent.run(task)`` returns only after the episode is
        written.

        Added in Chapter 7.

        Args:
            result (TaskResult | None): The successful task result.
            error (Exception | None): The exception from a failed task.

        Raises:
            RecordMemoryError: If neither ``result`` nor ``error`` is
                provided.
        """
        if result is None and error is None:
            raise RecordMemoryError(
                "record_memory() requires either result or error.",
            )
        episode = Episode(
            task=self.task,
            rollout=self.rollout,
            result=result,
            error=error,
        )
        for memory in self.llm_agent.memories:
            await memory.record(episode)

    async def request_approval(
        self,
        result: TaskResult,
    ) -> ApprovalResult:
        """Ask a human to approve or reject the proposed task result.

        Added in Chapter 8.

        Operator-gated human-in-the-loop pattern; unlike
        ``HumanInputTool``, the pause is not agent-initiated.
        Runs the blocking rich prompts in a thread via
        ``asyncio.to_thread``. Auto-approves on ``EOFError`` or
        ``KeyboardInterrupt`` (headless / interrupted terminal).

        Args:
            result (TaskResult): The proposed task result to review.

        Returns:
            ApprovalResult: The approval decision.
        """

        def _prompt_for_approval(
            task_result: TaskResult,
        ) -> ApprovalResult:
            console = Console()
            console.print(
                Panel(
                    task_result.content,
                    title="Proposed Task Result",
                    border_style="cyan",
                ),
            )
            approved = Confirm.ask("Approve this result?", console=console)
            if approved:
                return ApprovalResult(approved=True, feedback="")
            feedback = Prompt.ask(
                "Provide your correction rationale for the LLM agent to address",  # noqa: E501
                console=console,
            )
            return ApprovalResult(approved=False, feedback=feedback)

        try:
            return await asyncio.to_thread(
                _prompt_for_approval,
                result,
            )
        except EOFError:
            self.logger.info(
                "Approval prompt got EOF (headless); auto-approving.",
            )
            return ApprovalResult(approved=True, feedback="")
        except KeyboardInterrupt:
            self.logger.info(
                "Approval prompt interrupted by operator; auto-approving.",
            )
            return ApprovalResult(
                approved=True,
                feedback="",
            )

background_task property writable

background_task

Get the background ~asyncio.Task for the handler.

__init__

__init__(
    llm_agent,
    task,
    skills_scopes=None,
    explicit_only_skills=None,
    *args,
    **kwargs,
)

Initialize a TaskHandler.

Parameters:

Name	Type	Description	Default
llm_agent	LLMAgent	The LLM agent.	required
task	Task	The task to process.	required
skills_scopes	list[SkillScope] | None	Scopes to scan for skills. Defaults to [USER, PROJECT]. Added in Chapter 6.	None
explicit_only_skills	set[str] | None	Skill names to exclude from the model catalog. Defaults to None. Added in Chapter 6.	None
*args	Any	Additional positional arguments.	()
**kwargs	Any	Additional keyword arguments.	{}

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def __init__(
    self,
    llm_agent: "LLMAgent",
    task: Task,
    # added in ch06
    skills_scopes: list[SkillScope] | None = None,
    explicit_only_skills: set[str] | None = None,
    *args: Any,
    **kwargs: Any,
) -> None:
    """Initialize a TaskHandler.

    Args:
        llm_agent (LLMAgent): The LLM agent.
        task (Task): The task to process.
        skills_scopes (list[SkillScope] | None): Scopes to scan for
            skills. Defaults to ``[USER, PROJECT]``. Added in
            Chapter 6.
        explicit_only_skills (set[str] | None): Skill names to
            exclude from the model catalog. Defaults to None.
            Added in Chapter 6.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.
    """
    super().__init__(*args, **kwargs)
    self.llm_agent = llm_agent
    self.task = task
    self.rollout = ""
    self.step_counter = 0
    self._background_task: asyncio.Task | None = None
    self.logger = get_logger(self.__class__.__name__)
    # added in ch06
    _scopes = (
        skills_scopes
        if skills_scopes is not None
        else [SkillScope.USER, SkillScope.PROJECT]
    )
    self.skills: dict[str, Skill] = discover_skills(_scopes)
    self._explicit_only_skills: set[str] = explicit_only_skills or set()
    self._use_skill_tool: UseSkillTool | None = (
        UseSkillTool(
            skills=self.skills,
            explicit_only_skills=self._explicit_only_skills,
        )
        if self.skills
        else None
    )
    # added in ch07
    self._recalled_memories: str = ""

get_next_step async

get_next_step(previous_step_result)

Based on previous step result, get next step or conclude task.

Returns:

Type	Description
TaskStep | TaskResult	TaskStep | TaskResult: Either the next step or the result of the task.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def get_next_step(
    self,
    previous_step_result: (
        TaskStepResult
        | RejectedTaskResult  # added in ch08
        | None
    ),
) -> TaskStep | TaskResult:
    """Based on previous step result, get next step or conclude task.

    Returns:
        TaskStep | TaskResult: Either the next step or the result of
            the task.
    """
    if not previous_step_result:
        return TaskStep(
            task_id=self.task.id_,
            instruction=self.task.instruction,
        )
    # added in ch08: rejection bypasses LLM routing
    if isinstance(previous_step_result, RejectedTaskResult):
        self.logger.info(
            f"🧠 New Step (rejection): {previous_step_result.feedback}",
        )
        return TaskStep(
            task_id=self.task.id_,
            instruction=self.llm_agent.templates[
                "approval_rejection_feedback"
            ].format(
                content=previous_step_result.failed_result_content,
                feedback=previous_step_result.feedback,
            ),
        )
    self.logger.debug(f"🧵 Rollout: {self.rollout}")

    prompt = self.llm_agent.templates["get_next_step"].format(
        instruction=self.task.instruction,
        current_rollout=self.rollout,
        current_response=previous_step_result.content,
    )
    self.logger.debug(f"---NEXT STEP PROMPT: {prompt}")
    try:
        next_step = await self.llm_agent.llm.structured_output(
            prompt=prompt,
            mdl=NextStepDecision,
        )
        self.logger.debug(
            f"---NEXT STEP: {next_step.model_dump_json()}",
        )
    except Exception as e:
        raise TaskHandlerError(
            f"Failed to get next step: {str(e)}",
        ) from e

    if next_step.kind == "final_result":
        self.logger.info("No new step required.")
        retval = TaskResult(
            task_id=self.task.id_,
            content=previous_step_result.content,
        )
    else:  # next_step.kind == "next_step":
        self.logger.info(f"🧠 New Step: {next_step.content}")
        retval = TaskStep(
            task_id=self.task.id_,
            instruction=next_step.content,
        )

    return retval

run_step async

run_step(step)

Run next step of a given task.

A single step is executed through a single-turn conversation that the LLM agent has with itself. In other words, it is both the user providing the instruction (from get_next_step) as well as the assistant that provides the result.

Parameters:

Name	Type	Description	Default
step	TaskStep	The step to execute.	required

Returns:

Name	Type	Description
TaskStepResult	TaskStepResult	The result of the step execution.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def run_step(self, step: TaskStep) -> TaskStepResult:  # noqa: PLR0912
    """Run next step of a given task.

    A single step is executed through a single-turn conversation that
    the LLM agent has with itself. In other words, it is both the `user`
    providing the instruction (from `get_next_step`) as well as the
    `assistant` that provides the result.

    Args:
        step (TaskStep): The step to execute.

    Returns:
        TaskStepResult: The result of the step execution.
    """
    self.step_counter += 1
    self.logger.info(f"⚙️ Processing Step: {step.instruction}")
    self.logger.debug(f"🧵 Rollout: {self.rollout}")

    # include rollout as context in the system message
    system_message = ChatMessage(
        role=ChatRole.SYSTEM,
        content=self.llm_agent.templates[
            "run_step_system_message"
        ].format(
            llm_agent_system_message=self.llm_agent.templates[
                "system_message"
            ],
            current_rollout=self.rollout,
        )
        if self.rollout
        else self.llm_agent.templates[
            "run_step_system_message_without_rollout"
        ].format(
            llm_agent_system_message=self.llm_agent.templates[
                "system_message"
            ],
        ),
    )

    # added in ch07: inject recalled memories
    if memories := self._recalled_memories:
        system_message = ChatMessage(
            role=ChatRole.SYSTEM,
            content=f"{system_message.content}\n\n{memories}",
        )

    # added in ch06: bolt on skills catalog when skills are available
    if catalog := self._skills_catalog:
        system_message = ChatMessage(
            role=ChatRole.SYSTEM,
            content=f"{system_message.content}\n\n{catalog}",
        )

    self.logger.debug(f"💬 SYSTEM: {system_message.content}")

    # fictitious user's input
    user_input = self.llm_agent.templates[
        "run_step_user_message"
    ].format(
        instruction=step.instruction,
    )
    self.logger.debug(f"💬 USER INPUT: {user_input}")

    # start single-turn conversation
    # added in ch06: include use_skill tool when skills are available
    all_tools = self.llm_agent.tools + (
        [self._use_skill_tool] if self._use_skill_tool else []
    )
    user_message, response_message = await self.llm_agent.llm.chat(
        input=user_input,
        chat_history=[system_message],
        tools=all_tools,
    )
    self.logger.debug(f"💬 ASSISTANT: {response_message.content}")

    # check if there are tool calls
    if response_message.tool_calls:
        tool_call_results = []
        for tool_call in response_message.tool_calls:
            self.logger.info(
                f"🛠️ Executing Tool Call: {tool_call.tool_name}",
            )
            if tool := (
                self.llm_agent.tools_registry.get(
                    tool_call.tool_name,
                )
                or (
                    self._use_skill_tool
                    if self._use_skill_tool
                    and tool_call.tool_name == self._use_skill_tool.name
                    else None
                )
            ):
                if isinstance(tool, AsyncBaseTool):
                    tool_call_result = await tool(tool_call=tool_call)
                else:
                    tool_call_result = tool(tool_call=tool_call)
                msg = (
                    "✅ Successful Tool Call: "
                    f"{tool_call_result.content}"
                )
                self.logger.info(msg)
            else:
                error_msg = (
                    f"Tool with name {tool_call.tool_name} "
                    "doesn't exist."
                )
                tool_call_result = ToolCallResult(
                    tool_call_id=tool_call.id_,
                    error=True,
                    content=error_msg,
                )
                self.logger.info(
                    f"❌ Tool Call Failure: {tool_call_result.content}",
                )
            tool_call_results.append(tool_call_result)

        # send tool call results back to llm to get result
        (
            tool_messages,
            another_response_message,
        ) = await self.llm_agent.llm.continue_chat_with_tool_results(  # noqa: E501
            tool_call_results=tool_call_results,
            chat_history=[
                system_message,
                user_message,
                response_message,
            ],
        )

        # get final content and update chat history
        if another_response_message.tool_calls:
            # if has tool calls, we'll make them in the next step
            final_content = "I need to make the following tool-calls:\n"
            final_content += "\n".join(
                t.model_dump_json(indent=4)
                for t in another_response_message.tool_calls
            )
        else:
            final_content = another_response_message.content
        chat_history = (
            [
                system_message,
                user_message,
                response_message,
            ]
            + tool_messages
            + [another_response_message]
        )
    else:
        final_content = response_message.content
        chat_history = [
            system_message,
            user_message,
            response_message,
        ]

    # augment rollout from this turn
    formatted_step = self._format_step_for_rollout(
        chat_history=chat_history,
    )
    if self.rollout:
        self.rollout += "\n\n" + formatted_step

    else:
        self.rollout = formatted_step

    self.logger.info(
        f"✅ Step Result: {final_content}",
    )
    return TaskStepResult(
        task_step_id=step.id_,
        content=final_content,
    )

load_memories async

load_memories()

Recall relevant episodes from all configured memory backends.

Added in Chapter 7.

Calls recall on each memory in self.llm_agent.memories and stores the formatted string in self._recalled_memories for prompt injection during run_step. No-op when no memories are configured.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def load_memories(self) -> None:
    """Recall relevant episodes from all configured memory backends.

    Added in Chapter 7.

    Calls ``recall`` on each memory in ``self.llm_agent.memories``
    and stores the formatted string in ``self._recalled_memories``
    for prompt injection during ``run_step``. No-op when no memories
    are configured.
    """
    loaded = []
    for memory in self.llm_agent.memories:
        block = await memory.recall(self.task)
        loaded.append(block)
    self._recalled_memories = self._format_memories_for_system_prompt(
        loaded,
    )

record_memory async

record_memory(result=None, error=None)

Build an Episode and write it to all configured memories.

Exactly one of result or error must be provided. Called before set_result() / set_exception() so that await agent.run(task) returns only after the episode is written.

Added in Chapter 7.

Parameters:

Name	Type	Description	Default
result	TaskResult | None	The successful task result.	None
error	Exception | None	The exception from a failed task.	None

Raises:

Type	Description
RecordMemoryError	If neither result nor error is provided.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def record_memory(
    self,
    result: TaskResult | None = None,
    error: Exception | None = None,
) -> None:
    """Build an Episode and write it to all configured memories.

    Exactly one of ``result`` or ``error`` must be provided.
    Called before ``set_result()`` / ``set_exception()`` so that
    ``await agent.run(task)`` returns only after the episode is
    written.

    Added in Chapter 7.

    Args:
        result (TaskResult | None): The successful task result.
        error (Exception | None): The exception from a failed task.

    Raises:
        RecordMemoryError: If neither ``result`` nor ``error`` is
            provided.
    """
    if result is None and error is None:
        raise RecordMemoryError(
            "record_memory() requires either result or error.",
        )
    episode = Episode(
        task=self.task,
        rollout=self.rollout,
        result=result,
        error=error,
    )
    for memory in self.llm_agent.memories:
        await memory.record(episode)

request_approval async

request_approval(result)

Ask a human to approve or reject the proposed task result.

Added in Chapter 8.

Operator-gated human-in-the-loop pattern; unlike HumanInputTool, the pause is not agent-initiated. Runs the blocking rich prompts in a thread via asyncio.to_thread. Auto-approves on EOFError or KeyboardInterrupt (headless / interrupted terminal).

Parameters:

Name	Type	Description	Default
result	TaskResult	The proposed task result to review.	required

Returns:

Name	Type	Description
ApprovalResult	ApprovalResult	The approval decision.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def request_approval(
    self,
    result: TaskResult,
) -> ApprovalResult:
    """Ask a human to approve or reject the proposed task result.

    Added in Chapter 8.

    Operator-gated human-in-the-loop pattern; unlike
    ``HumanInputTool``, the pause is not agent-initiated.
    Runs the blocking rich prompts in a thread via
    ``asyncio.to_thread``. Auto-approves on ``EOFError`` or
    ``KeyboardInterrupt`` (headless / interrupted terminal).

    Args:
        result (TaskResult): The proposed task result to review.

    Returns:
        ApprovalResult: The approval decision.
    """

    def _prompt_for_approval(
        task_result: TaskResult,
    ) -> ApprovalResult:
        console = Console()
        console.print(
            Panel(
                task_result.content,
                title="Proposed Task Result",
                border_style="cyan",
            ),
        )
        approved = Confirm.ask("Approve this result?", console=console)
        if approved:
            return ApprovalResult(approved=True, feedback="")
        feedback = Prompt.ask(
            "Provide your correction rationale for the LLM agent to address",  # noqa: E501
            console=console,
        )
        return ApprovalResult(approved=False, feedback=feedback)

    try:
        return await asyncio.to_thread(
            _prompt_for_approval,
            result,
        )
    except EOFError:
        self.logger.info(
            "Approval prompt got EOF (headless); auto-approving.",
        )
        return ApprovalResult(approved=True, feedback="")
    except KeyboardInterrupt:
        self.logger.info(
            "Approval prompt interrupted by operator; auto-approving.",
        )
        return ApprovalResult(
            approved=True,
            feedback="",
        )

SupervisedTaskHandler

Bases: TaskHandler

TaskHandler for human-driven stepwise execution.

Added in Chapter 8. Caller-driven human-in-the-loop pattern; unlike HumanInputTool (agent-initiated) and request_approval (operator-gated at result time), the human controls the entire execution cadence. Returned by run_supervised(); the caller drives the loop manually via get_next_step() and run_step() and finalises execution with complete() or abort().

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

class SupervisedTaskHandler(TaskHandler):
    """TaskHandler for human-driven stepwise execution.

    Added in Chapter 8. Caller-driven human-in-the-loop pattern;
    unlike ``HumanInputTool`` (agent-initiated) and
    ``request_approval`` (operator-gated at result time), the human
    controls the entire execution cadence. Returned by
    ``run_supervised()``; the caller drives the loop manually via
    ``get_next_step()`` and ``run_step()`` and finalises execution
    with ``complete()`` or ``abort()``.
    """

    @property
    def background_task(self) -> asyncio.Task:
        """Not available in supervised mode."""
        raise TaskHandlerError(
            "SupervisedTaskHandler has no background task — "
            "execution is caller-driven via get_next_step() "
            "and run_step().",
        )

    @background_task.setter
    def background_task(self, asyncio_task: asyncio.Task) -> None:
        """Not available in supervised mode."""
        raise TaskHandlerError(
            "SupervisedTaskHandler has no background task — "
            "execution is caller-driven via get_next_step() "
            "and run_step().",
        )

    async def complete(self, result: TaskResult) -> None:
        """Accept the final result and resolve the handler.

        Added in Chapter 8.

        Args:
            result: The ``TaskResult`` to accept.
        """
        if not isinstance(result, TaskResult):
            raise TaskHandlerError(
                f"complete() requires a TaskResult, "
                f"got {type(result).__name__}.",
            )
        await self.record_memory(result=result)
        self.set_result(result)

    def reject(
        self,
        result: TaskResult,
        feedback: str,
    ) -> RejectedTaskResult:
        """Reject a proposed TaskResult and return feedback for re-routing.

        Added in Chapter 8.

        Args:
            result: The ``TaskResult`` to reject.
            feedback: Correction rationale passed back to the agent.

        Returns:
            RejectedTaskResult: Pass to ``get_next_step()`` to
                re-enter the loop without consulting the LLM.
        """
        return RejectedTaskResult(
            failed_result_content=result.content,
            feedback=feedback,
        )

    async def abort(self, error: Exception | None = None) -> None:
        """Abort the supervised task and resolve the handler.

        Added in Chapter 8.

        Args:
            error: Exception to set. Defaults to
                ``TaskHandlerError("Task aborted.")``.
        """
        err = error or TaskHandlerError("Task aborted.")
        await self.record_memory(error=err)
        self.set_exception(err)

background_task property writable

background_task

Not available in supervised mode.

complete async

complete(result)

Accept the final result and resolve the handler.

Added in Chapter 8.

Parameters:

Name	Type	Description	Default
result	TaskResult	The TaskResult to accept.	required

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def complete(self, result: TaskResult) -> None:
    """Accept the final result and resolve the handler.

    Added in Chapter 8.

    Args:
        result: The ``TaskResult`` to accept.
    """
    if not isinstance(result, TaskResult):
        raise TaskHandlerError(
            f"complete() requires a TaskResult, "
            f"got {type(result).__name__}.",
        )
    await self.record_memory(result=result)
    self.set_result(result)

reject

reject(result, feedback)

Reject a proposed TaskResult and return feedback for re-routing.

Added in Chapter 8.

Parameters:

Name	Type	Description	Default
result	TaskResult	The TaskResult to reject.	required
feedback	str	Correction rationale passed back to the agent.	required

Returns:

Name	Type	Description
RejectedTaskResult	RejectedTaskResult	Pass to get_next_step() to re-enter the loop without consulting the LLM.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def reject(
    self,
    result: TaskResult,
    feedback: str,
) -> RejectedTaskResult:
    """Reject a proposed TaskResult and return feedback for re-routing.

    Added in Chapter 8.

    Args:
        result: The ``TaskResult`` to reject.
        feedback: Correction rationale passed back to the agent.

    Returns:
        RejectedTaskResult: Pass to ``get_next_step()`` to
            re-enter the loop without consulting the LLM.
    """
    return RejectedTaskResult(
        failed_result_content=result.content,
        feedback=feedback,
    )

abort async

abort(error=None)

Abort the supervised task and resolve the handler.

Added in Chapter 8.

Parameters:

Name	Type	Description	Default
error	Exception | None	Exception to set. Defaults to TaskHandlerError("Task aborted.").	None

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def abort(self, error: Exception | None = None) -> None:
    """Abort the supervised task and resolve the handler.

    Added in Chapter 8.

    Args:
        error: Exception to set. Defaults to
            ``TaskHandlerError("Task aborted.")``.
    """
    err = error or TaskHandlerError("Task aborted.")
    await self.record_memory(error=err)
    self.set_exception(err)

__init__

__init__(
    llm,
    tools=None,
    templates=default_templates,
    memories=None,
)

Initialize an LLMAgent.

Parameters:

Name	Type	Description	Default
llm	LLM	The backbone LLM of the LLM agent.	required
tools	list[Tool]	The set of tools with which the LLM can be equipped. Defaults to None.	None
templates	LLMAgentTemplates	Prompt templates for LLM Agent.	default_templates
memories	list[Memory] | None	Episodic memory backends to consult at task start and update at task end. Defaults to None (no memory). Added in Chapter 7.	None

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def __init__(
    self,
    llm: LLM,
    tools: list[Tool] | None = None,
    templates: LLMAgentTemplates = default_templates,
    # added in ch07
    memories: list[Memory] | None = None,
):
    """Initialize an LLMAgent.

    Args:
        llm (LLM): The backbone LLM of the LLM agent.
        tools (list[Tool], optional): The set of tools with which the
            LLM can be equipped. Defaults to None.
        templates (LLMAgentTemplates): Prompt templates for LLM Agent.
        memories (list[Memory] | None): Episodic memory backends
            to consult at task start and update at task end. Defaults
            to None (no memory). Added in Chapter 7.
    """
    self.llm = llm
    tools = tools or []
    # validate no duplications in tool names
    if len({t.name for t in tools}) < len(tools):
        raise LLMAgentError(
            "Provided tool list contains duplicate tool names.",
        )
    self.tools_registry = {t.name: t for t in tools}
    self.templates = templates
    self.logger = get_logger(self.__class__.__name__)
    # added in ch07
    self.memories = memories or []

add_tool

add_tool(tool)

Add a tool to the agents tool set.

NOTE: Supports fluent style for convenience.

Parameters:

Name	Type	Description	Default
tool	Tool	The tool to equip the LLM agent.	required

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def add_tool(self, tool: Tool) -> Self:
    """Add a tool to the agents tool set.

    NOTE: Supports fluent style for convenience.

    Args:
        tool (Tool): The tool to equip the LLM agent.

    """
    if tool.name in self.tools_registry:
        raise LLMAgentError(f"Tool with name {tool.name} already exists.")
    self.tools_registry[tool.name] = tool
    return self

run

run(
    task,
    max_steps=None,
    skills_scopes=None,
    explicit_only_skills=None,
    with_approval=False,
)

Agent's processing loop for executing tasks.

Parameters:

Name	Type	Description	Default
task	Task	the Task to perform.	required
max_steps	int | None	Maximum number of steps to run for task. Defaults to None.	None
skills_scopes	list[SkillScope] | None	Scopes to scan for skills, in processing order (last wins on name collision). Defaults to [USER, PROJECT]. Added in Chapter 6.	None
explicit_only_skills	set[str] | None	Skill names to exclude from the model catalog for this run. They remain activatable via run_with_skill(). Defaults to None. Added in Chapter 6.	None
with_approval	bool	When True, an end-of-loop human approval gate fires before each TaskResult is accepted. The human may approve (result is recorded and returned) or reject with feedback (feedback re-enters the loop as a new step). Rejections do not consume the step budget; pair with max_steps to bound repeated-rejection loops. Defaults to False. Added in Chapter 8.	False

Returns:

Name	Type	Description
TaskHandler	TaskHandler	the TaskHandler object responsible for task execution.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def run(
    self,
    task: Task,
    max_steps: int | None = None,
    # added in ch06
    skills_scopes: list[SkillScope] | None = None,
    explicit_only_skills: set[str] | None = None,
    # added in ch08
    with_approval: bool = False,
) -> TaskHandler:
    """Agent's processing loop for executing tasks.

    Args:
        task (Task): the Task to perform.
        max_steps (int | None): Maximum number of steps to run for task.
            Defaults to None.
        skills_scopes (list[SkillScope] | None): Scopes to scan for
            skills, in processing order (last wins on name collision).
            Defaults to ``[USER, PROJECT]``. Added in Chapter 6.
        explicit_only_skills (set[str] | None): Skill names to exclude
            from the model catalog for this run. They remain activatable
            via ``run_with_skill()``. Defaults to None. Added in
            Chapter 6.
        with_approval (bool): When ``True``, an end-of-loop human
            approval gate fires before each ``TaskResult`` is accepted.
            The human may approve (result is recorded and returned) or
            reject with feedback (feedback re-enters the loop as a new
            step). Rejections do not consume the step budget; pair with
            ``max_steps`` to bound repeated-rejection loops. Defaults
            to ``False``. Added in Chapter 8.

    Returns:
        TaskHandler: the TaskHandler object responsible for task execution.
    """
    task_handler = self.TaskHandler(
        llm_agent=self,
        task=task,
        skills_scopes=skills_scopes,
        explicit_only_skills=explicit_only_skills,
    )

    async def _process_loop() -> None:
        """The processing loop for the task handler execute its task.

        Cycle between get_next_step and run_step, until the task_handler
        is marked as done, either through a set result or an exception being
        set.
        """
        self.logger.info(f"🚀 Starting task: {task.instruction}")
        step_result = None

        # added in ch07
        await task_handler.load_memories()

        while not task_handler.done():
            try:
                if task_handler.step_counter == max_steps:
                    raise MaxStepsReachedError("Max steps reached.")

                next_step = await task_handler.get_next_step(step_result)

                match next_step:
                    case TaskStep():
                        step_result = await task_handler.run_step(
                            next_step,
                        )
                    case TaskResult():
                        # added in ch08
                        if with_approval:
                            approval = await task_handler.request_approval(
                                next_step,
                            )
                            if not approval.approved:
                                step_result = RejectedTaskResult(
                                    failed_result_content=next_step.content,
                                    feedback=approval.feedback,
                                )
                                self.logger.info(
                                    "🔁 Task result rejected; "
                                    "re-entering loop with feedback.",
                                )
                                continue
                        await task_handler.record_memory(
                            result=next_step,
                        )  # added in ch07
                        task_handler.set_result(next_step)
                        self.logger.info(
                            f"🏁 Task completed: {next_step.content}",
                        )

            except Exception as e:
                await task_handler.record_memory(error=e)  # added in ch07
                task_handler.set_exception(e)

    task_handler.background_task = asyncio.create_task(_process_loop())

    return task_handler

run_with_skill

run_with_skill(
    skill_name,
    prompt=None,
    max_steps=None,
    with_approval=False,
)

User-explicit skill activation: the programmatic slash command.

Added in Chapter 6.

Frames the task instruction to direct the model to activate the named skill as its first action, then runs the full agent loop. Relies on the model's tool-use ability to call use_skill — a fair assumption given the whole system depends on it. Unknown skill names are caught by the guard in UseSkillTool.__call__.

Parameters:

Name	Type	Description	Default
skill_name	str	Name of the skill to activate.	required
prompt	str | None	Optional instruction to pass alongside the skill activation. Defaults to None.	None
max_steps	int | None	Maximum number of steps to run. Defaults to None.	None
with_approval	bool	Passed through to run(). Added in Chapter 8.	False

Returns:

Name	Type	Description
TaskHandler	TaskHandler	The handler responsible for task execution.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

def run_with_skill(
    self,
    skill_name: str,
    prompt: str | None = None,
    max_steps: int | None = None,
    # added in ch08
    with_approval: bool = False,
) -> TaskHandler:
    """User-explicit skill activation: the programmatic slash command.

    Added in Chapter 6.

    Frames the task instruction to direct the model to activate the named
    skill as its first action, then runs the full agent loop. Relies on
    the model's tool-use ability to call ``use_skill`` — a fair assumption
    given the whole system depends on it. Unknown skill names are caught
    by the guard in ``UseSkillTool.__call__``.

    Args:
        skill_name (str): Name of the skill to activate.
        prompt (str | None): Optional instruction to pass alongside the
            skill activation. Defaults to None.
        max_steps (int | None): Maximum number of steps to run.
            Defaults to None.
        with_approval (bool): Passed through to ``run()``. Added in
            Chapter 8.

    Returns:
        TaskHandler: The handler responsible for task execution.
    """
    if prompt:
        instruction = EXPLICIT_SKILL_ACTIVATION_WITH_PROMPT_TEMPLATE.format(
            name=skill_name,
            prompt=prompt,
        )
    else:
        instruction = EXPLICIT_SKILL_ACTIVATION_TEMPLATE.format(
            name=skill_name,
        )
    task = Task(instruction=instruction)

    return self.run(
        task=task,
        max_steps=max_steps,
        # added in ch08
        with_approval=with_approval,
    )

run_supervised async

run_supervised(
    task, skills_scopes=None, explicit_only_skills=None
)

Human-driven stepwise task execution.

Added in Chapter 8. Creates and returns a SupervisedTaskHandler with memories loaded, without starting the autonomous _process_loop. The caller drives execution cell-by-cell via get_next_step() and run_step(), then finalises with complete() or abort().

Contrasts with run(): supervised = human controls cadence; autonomous = agent runs to completion.

Parameters:

Name	Type	Description	Default
task	Task	The task to perform.	required
skills_scopes	list[SkillScope] | None	Scopes to scan for skills. Defaults to [USER, PROJECT].	None
explicit_only_skills	set[str] | None	Skill names to exclude from the model catalog. Defaults to None.	None

Returns:

Name	Type	Description
SupervisedTaskHandler	SupervisedTaskHandler	Ready for stepwise execution.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def run_supervised(
    self,
    task: Task,
    skills_scopes: list[SkillScope] | None = None,
    explicit_only_skills: set[str] | None = None,
) -> SupervisedTaskHandler:
    """Human-driven stepwise task execution.

    Added in Chapter 8. Creates and returns a
    ``SupervisedTaskHandler`` with memories loaded, without starting
    the autonomous ``_process_loop``. The caller drives execution
    cell-by-cell via ``get_next_step()`` and ``run_step()``, then
    finalises with ``complete()`` or ``abort()``.

    Contrasts with ``run()``: supervised = human controls cadence;
    autonomous = agent runs to completion.

    Args:
        task: The task to perform.
        skills_scopes (list[SkillScope] | None): Scopes to scan for
            skills. Defaults to ``[USER, PROJECT]``.
        explicit_only_skills (set[str] | None): Skill names to
            exclude from the model catalog. Defaults to None.

    Returns:
        SupervisedTaskHandler: Ready for stepwise execution.
    """
    task_handler = self.SupervisedTaskHandler(
        llm_agent=self,
        task=task,
        skills_scopes=skills_scopes,
        explicit_only_skills=explicit_only_skills,
    )
    await task_handler.load_memories()
    return task_handler

run_supervised_with_skill async

run_supervised_with_skill(
    skill_name,
    prompt=None,
    skills_scopes=None,
    explicit_only_skills=None,
)

Human-driven stepwise execution with a pre-loaded skill.

Added in Chapter 8. Combines run_with_skill() (skill activation framing) with run_supervised() (caller-controlled cadence). The named skill is embedded in the task instruction so the model activates it as its first action; the caller then drives execution cell-by-cell via get_next_step() and run_step().

Parameters:

Name	Type	Description	Default
skill_name	str	Name of the skill to activate.	required
prompt	str | None	Optional instruction to pass alongside the skill activation. Defaults to None.	None
skills_scopes	list[SkillScope] | None	Scopes to scan for skills. Defaults to [USER, PROJECT].	None
explicit_only_skills	set[str] | None	Skill names to exclude from the model catalog. Defaults to None.	None

Returns:

Name	Type	Description
SupervisedTaskHandler	SupervisedTaskHandler	Ready for stepwise execution.

Source code in src/llm_agents_from_scratch/agent/llm_agent.py

async def run_supervised_with_skill(
    self,
    skill_name: str,
    prompt: str | None = None,
    skills_scopes: list[SkillScope] | None = None,
    explicit_only_skills: set[str] | None = None,
) -> SupervisedTaskHandler:
    """Human-driven stepwise execution with a pre-loaded skill.

    Added in Chapter 8. Combines ``run_with_skill()`` (skill
    activation framing) with ``run_supervised()`` (caller-controlled
    cadence). The named skill is embedded in the task instruction so
    the model activates it as its first action; the caller then
    drives execution cell-by-cell via ``get_next_step()`` and
    ``run_step()``.

    Args:
        skill_name (str): Name of the skill to activate.
        prompt (str | None): Optional instruction to pass alongside
            the skill activation. Defaults to None.
        skills_scopes (list[SkillScope] | None): Scopes to scan for
            skills. Defaults to ``[USER, PROJECT]``.
        explicit_only_skills (set[str] | None): Skill names to
            exclude from the model catalog. Defaults to None.

    Returns:
        SupervisedTaskHandler: Ready for stepwise execution.
    """
    if prompt:
        instruction = EXPLICIT_SKILL_ACTIVATION_WITH_PROMPT_TEMPLATE.format(
            name=skill_name,
            prompt=prompt,
        )
    else:
        instruction = EXPLICIT_SKILL_ACTIVATION_TEMPLATE.format(
            name=skill_name,
        )
    task = Task(instruction=instruction)
    return await self.run_supervised(
        task=task,
        skills_scopes=skills_scopes,
        explicit_only_skills=explicit_only_skills,
    )