Tasks and task groups

from livekit.agents import AgentTask, function_tool

class CollectConsent(AgentTask[bool]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="""
            Ask for recording consent and get a clear yes or no answer.
            Be polite and professional.
            """,
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="""
            Briefly introduce yourself, then ask for permission to record the call for quality assurance and training purposes.
            Make it clear that they can decline.
            """
        )

    @function_tool
    async def consent_given(self) -> None:
        """Use this when the user gives consent to record."""
        self.complete(True)

    @function_tool
    async def consent_denied(self) -> None:
        """Use this when the user denies consent to record."""
        self.complete(False)

from livekit.agents import Agent, function_tool, get_job_context

class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a friendly customer service representative.")

    async def on_enter(self) -> None:
        if await CollectConsent(chat_ctx=self.chat_ctx):
            await self.session.generate_reply(instructions="Offer your assistance to the user.")
        else:
            await self.session.generate_reply(instructions="Inform the user that you are unable to proceed and will end the call.")
            job_ctx = get_job_context()
            await job_ctx.api.room.delete_room(api.DeleteRoomRequest(room=job_ctx.room.name))

from dataclasses import dataclass

@dataclass
class ContactInfoResult:
    name: str
    email_address: str
    phone_number: str

class GetContactInfoTask(AgentTask[ContactInfoResult]):
    # ....

@dataclass
class BehavioralResults:
    strengths: str
    weaknesses: str
    work_style: str

class BehavioralTask(AgentTask[BehavioralResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Collect strengths, weaknesses, and work style in any order."
        )
        self._results = {}
    
    @function_tool()
    async def record_strengths(self, strengths_summary: str):
        """Record candidate's strengths"""
        self._results["strengths"] = strengths_summary
        self._check_completion()
    
    @function_tool()
    async def record_weaknesses(self, weaknesses_summary: str):
        """Record candidate's weaknesses"""
        self._results["weaknesses"] = weaknesses_summary
        self._check_completion()
    
    @function_tool()
    async def record_work_style(self, work_style: str):
        """Record candidate's work style"""
        self._results["work_style"] = work_style
        self._check_completion()
    
    def _check_completion(self):
        required_keys = {"strengths", "weaknesses", "work_style"}
        if self._results.keys() == required_keys:
            results = BehavioralResults(
                strengths=self._results["strengths"],
                weaknesses=self._results["weaknesses"],
                work_style=self._results["work_style"]
            )
            self.complete(results)
        else:
            self.session.generate_reply(
                instructions="Continue collecting remaining information."
            )

from livekit.agents.beta.workflows import GetEmailTask

# ... within your agent ...
email_result = await GetEmailTask(chat_ctx=self.chat_ctx)
print(f"Collected email: {email_result.email_address}")

from livekit.agents.beta.workflows import GetEmailTask
from livekit.agents import function_tool, RunContext
    
@function_tool()
async def get_alternate_contact_info(context: RunContext, contact_method: str, contact_value: str) -> None:
    """Collect alternative contact information when email isn't available"""
    # Store the alternative contact info
    context.session.userdata.alternate_contact_method = contact_method
    context.session.userdata.alternate_contact_value = contact_value
    
    await context.session.generate_reply(
        instructions=f"Acknowledge that you've recorded their {contact_method}: {contact_value}. Let them know this will be used for communication instead of email."
    )

# Customize GetEmailTask with extra instructions and tools
# ... within your agent ...
@function_tool()
async def collect_contact_info(context: RunContext) -> None:
    """Collect email or alternative contact information"""
    email_result = await GetEmailTask(
        chat_ctx=self.chat_ctx,
        extra_instructions="If the user cannot provide an email, call get_alternate_contact_info() instead of decline_email_capture().",
        tools=[get_alternate_contact_info]
    )

    return f"Collected email: {email_result.email_address}"

from livekit.agents.beta.workflows import GetAddressTask
from livekit.agents import Agent, function_tool
    
@function_tool()
async def collect_shipping_address(self) -> str:
    """Collect the user's shipping address"""
    address_result = await GetAddressTask(
        chat_ctx=self.chat_ctx,
        extra_instructions="Emphasize that this is for shipping purposes and accuracy is important."
    )
    
    return f"Shipping address recorded: {address_result.address}"

from livekit.agents.beta.workflows import GetEmailTask, TaskGroup

# Create and configure TaskGroup
task_group = TaskGroup()

# Add tasks using lambda factories
task_group.add(
    lambda: GetEmailTask(), 
    id="get_email_task", 
    description="Collects the user's email"
)
task_group.add(
    lambda: GetCommuteTask(), 
    id="get_commute_task", 
    description="Records the user's commute flexibility"
)

# Execute the task group
results = await task_group  # Returns TaskGroupResult object
task_results = results.task_results

# Access results by task ID
print(task_results)
# Output: {
#   "get_email_task": GetEmailResult(email="john.doe@gmail.com"), 
#   "get_commute_task": CommuteResult(can_commute=True, commute_method="subway")
# }

# Disable context summarization
task_group = TaskGroup(summarize_chat_ctx=False)

from livekit.agents import AgentTask, function_tool, RunContext
from livekit.agents.beta.workflows import TaskGroup
from dataclasses import dataclass

@dataclass
class IntroResults:
    name: str
    intro: str

@dataclass 
class CommuteResults:
    can_commute: bool
    commute_method: str

class IntroTask(AgentTask[IntroResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Welcome the candidate and collect their name and introduction."
        )
    
    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="Welcome the candidate and gather their name."
        )
    
    @function_tool()
    async def record_intro(self, context: RunContext, name: str, intro_notes: str) -> None:
        """Record the candidate's name and introduction"""
        context.session.userdata.candidate_name = name
        results = IntroResults(name=name, intro=intro_notes)
        self.complete(results)

class CommuteTask(AgentTask[CommuteResults]):
    def __init__(self) -> None:
        super().__init__(
            instructions="Ask about the candidate's ability to commute to the office."
        )
    
    @function_tool()
    async def record_commute_flexibility(
        self, 
        context: RunContext, 
        can_commute: bool, 
        commute_method: str
    ) -> None:
        """Record commute flexibility and transportation method"""
        results = CommuteResults(can_commute=can_commute, commute_method=commute_method)
        self.complete(results)

# Set up the workflow
task_group = TaskGroup()
task_group.add(
    lambda: IntroTask(), 
    id="intro_task", 
    description="Collects name and introduction"
)
task_group.add(
    lambda: CommuteTask(), 
    id="commute_task", 
    description="Asks about commute flexibility"
)

# Execute and get results
results = await task_group
task_results = results.task_results