Repository: livekit/agents
File: livekit-agents/livekit/agents/beta/workflows/task_group.py (Python)

Code before:
        try:
            if self._summarize_chat_ctx:
                # when a task is done, the chat_ctx is going to be merged with the "caller" chat_ctx
                # enabling summarization will result on only one ChatMessage added.
                summarized_chat_ctx = await self.chat_ctx.copy(exclude_instructions=True).summarize(
                    keep_last_turns=0
                )
                await self.update_chat_ctx(summarized_chat_ctx)
        except Exception as e:
            self.complete(RuntimeError(f"failed to summarize the chat_ctx: {e}"))

        self.complete(TaskGroupResult(task_results=task_results))

    def _build_out_of_scope_tool(self, *, active_task_id: str) -> FunctionTool | None:
        if not self._visited_tasks:
            return None

        # Only allow to regress to already visited tasks
        task_ids = self._visited_tasks.copy()
        task_ids.discard(active_task_id)
        task_repr = {f.id: f.description for f in self._registered_factories if f.id in task_ids}

        description = (
            "Call to regress to another task according to what the user requested to modify, return the corresponding task id, "
            'For example, if the user wants to change their email and there is a task with id "email_task" with a description of "Collect the user\'s email", return the id ("get_email_task").'
            f"The following are the IDs and their corresponding task description. {json.dumps(task_repr)}"
        )

        @function_tool(description=description, flags=ToolFlag.IGNORE_ON_ENTER)
        async def out_of_scope(
            task_id: Annotated[
                str,
                Field(
                    description="The ID of the task requested",
                    json_schema_extra={"enum": list(task_ids)},
                ),
            ],
        ):
            if (
                task_id not in self._registered_factories
                or task_id not in self._visited_tasks.keys()
            ):
                raise ToolError("unable to regress, invalid task id")

            if not self._current_task.done():
                self._current_task.complete(_OutOfScopeError(target_task_id=task_id))

        return out_of_scope


Review comment (suggestion):
this description seems very specific to the style of agent where it's modifying information in another task.. is it robust enough for other use cases as a task?

should we also allow users to skip around? currently it's only allowing you to go back to previous tasks. I'm wondering if this should be specified on each task itself either:
* can be revisited
* can be skipped ahead

Code after:

        try:
            if self._summarize_chat_ctx:
                # when a task is done, the chat_ctx is going to be merged with the "caller" chat_ctx
                # enabling summarization will result on only one ChatMessage added.
                summarized_chat_ctx = await self.chat_ctx.copy(exclude_instructions=True).summarize(
                    llm_v=self.session.llm, keep_last_turns=0
                )
                await self.update_chat_ctx(summarized_chat_ctx)
        except Exception as e:
            self.complete(RuntimeError(f"failed to summarize the chat_ctx: {e}"))

        self.complete(TaskGroupResult(task_results=task_results))

    def _build_out_of_scope_tool(self, *, active_task_id: str) -> FunctionTool | None:
        if not self._visited_tasks:
            return None

        # Only allow to regress to already visited tasks
        task_ids = self._visited_tasks.copy()
        task_ids.discard(active_task_id)
        task_repr = {
            f.id: f.description for f in self._registered_factories.values() if f.id in task_ids
        }

        description = (
            "Call to regress to another task according to what the user requested to modify, return the corresponding task id, "
            'For example, if the user wants to change their email and there is a task with id "email_task" with a description of "Collect the user\'s email", return the id ("get_email_task").'
            f"The following are the IDs and their corresponding task description. {json.dumps(task_repr)}"
        )

        @function_tool(description=description, flags=ToolFlag.IGNORE_ON_ENTER)
        async def out_of_scope(
            task_id: Annotated[
                str,
                Field(
                    description="The ID of the task requested",
                    json_schema_extra={"enum": list(task_ids)},
                ),
            ],
        ):
            if (
                task_id not in self._registered_factories
                or task_id not in self._visited_tasks.keys()
            ):
                raise ToolError("unable to regress, invalid task id")

            if not self._current_task.done():
                self._current_task.complete(_OutOfScopeError(target_task_id=task_id))

        return out_of_scope