for agent asks, what is the best way to ensure that tasks are complete before proceeding to next task?

Thanks for reaching out, Gage. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

what do you think of this guardrail concept outlined below: 1. Goodbye Detection Hook (in

_BaseTask

) python

class _BaseTask(AgentTask[ResultT], ABC):
    GOODBYE_PATTERNS = frozenset([
        "have a good", "take care", "looking forward", "you've got this",
        "rooting for you", "good luck", "bye", "talk soon", "catch you later",
        "i'll check in", "until next time"
    ])
    
    async def on_agent_speech_committed(self, content: str) -> None:
        """Called when agent speech is committed to context."""
        if self.done():
            return
        
        content_lower = content.lower()
        if any(pattern in content_lower for pattern in self.GOODBYE_PATTERNS):
            logger.warning(
                "[PipelineV3] Goodbye detected in incomplete task",
                extra={"task_id": self.TASK_ID, "content_preview": content[:100]}
            )
            
            # Inject correction
            remaining = self._get_remaining_tasks()
            await self.session.generate_reply(
                instructions=f"""
STOP — you just said goodbye but this task isn't complete.
You haven't called mark_complete yet.

Current task: {self.TASK_NAME}
Still needed: {', '.join(remaining) if remaining else 'complete this task'}

Continue naturally without apologizing. Ask about: {self._get_missing_info_hint()}
"""
            )
    
    def _get_remaining_tasks(self) -> list[str]:
        """Get remaining task names after current."""
        order = TASK_ORDER_BY_CALL_TYPE.get(self._session_data.call_type, [])
        try:
            idx = order.index(self.TASK_ID)
            return order[idx + 1:]
        except ValueError:
            return []
    
    def _get_missing_info_hint(self) -> str:
        """Get hint about what info is still needed."""
        missing = [f for f in self.MIN_REQUIRED_FIELDS if not self._partial_data.get(f)]
        return ", ".join(missing) if missing else "the required information"

2. Task Awareness Injection (in instructions) python

def _build_persistent_instructions(self) -> str:
    remaining = self._get_remaining_tasks()
    
    remaining_block = ""
    if remaining:
        remaining_block = f"""
⚠️ WORKFLOW POSITION: You are on task {len(TASK_ORDER_BY_CALL_TYPE[self._session_data.call_type]) - len(remaining)} of {len(TASK_ORDER_BY_CALL_TYPE[self._session_data.call_type])}.
After this: {', '.join(remaining[:2])}{'...' if len(remaining) > 2 else ''}
DO NOT say goodbye or wrap up until ALL tasks complete.
"""
    
    return render_task_instructions(
        f"{GLOBAL_INSTRUCTION_PREFIX}\n\n{remaining_block}\n\n{self.TASK_INSTRUCTIONS}".strip(),
        self._brief,
        self._session_data,
    )

3. Active Timeout (Reduce from 150s) python

class ExploreGoalsTask(_BaseIntroTask[ExploreGoalsResult]):
    TIMEOUT_SECONDS = 90.0  # Reduced from 150
    IDLE_TIMEOUT_SECONDS = 30.0  # New: timeout after 30s of no user speech

4. Goodbye Prevention in GLOBAL_INSTRUCTION_PREFIX python

GLOBAL_INSTRUCTION_PREFIX = """
=== CORE BEHAVIORAL RULES ===

CALL COMPLETION PROTOCOL:
- You MUST call mark_complete for each task before the call ends
- NEVER say goodbye phrases ("have a good one", "take care", etc.) 
  until mark_complete confirms "all objectives complete"
- If you feel ready to wrap up: STOP and check — did you call mark_complete?

TRANSITIONS: Never announce the next stage...
"""

Thanks — this is a thoughtful, concrete guardrail design. I’ll walk through what’s strong about it, where it could fail, and practical suggestions to make it more robust in production. What’s good • Clear multi-layer defense: combining detection (Goodbye Hook), prevention (GLOBAL_INSTRUCTION_PREFIX), visibility (Task Awareness Injection), and operational limits (Active Timeout) is the right pattern — multiple layers reduce false positives and missed cases. • Use of explicit control handoff: the hook relies on the fact that tasks must call mark_complete; that aligns with the task handoff model and guarantees a definite signal when a task truly finishes. • Context-aware messaging: injecting the remaining tasks and missing fields into instructions helps the model produce targeted corrections (better than a generic “don’t say goodbye”). • Shorter timeouts and idle-time detection: reducing TIMEOUT_SECONDS and adding IDLE_TIMEOUT_SECONDS helps avoid the agent drifting into a wrap-up state after long silence. Potential issues and edge cases • False positives from natural language: Goodbye phrases can appear inside other content (e.g., “I’ll check in about that after you have a good look”). Your substring matching may incorrectly flag those. • Timing of injection vs. user flow: If injection triggers right after the agent says a short “thanks” while the user is mid-response, it could confuse the user or interrupt a natural exchange. • Repetition and model compliance: If the model receives the injected correction mid-response, it may either ignore it, apologize repeatedly, or produce awkward transitions unless the instruction is very concise and concrete. • Instruction length and competing instructions: The GLOBAL_INSTRUCTION_PREFIX + persistent task instructions + injected correction could become long and conflicting. That may reduce model adherence or increase latency. • Edge case when task is logically complete but mark_complete wasn’t called due to a bug: the guardrail would keep intervening; need safe escape or fallback. • Multilingual / phrasing variations: current GOODBYE_PATTERNS are English and literal substrings; they miss synonyms, contractions, punctuation variants, and other locales. Suggestions to strengthen the design 1. Improve goodbye detection quality ◦ Use a lightweight intent classifier (few-shot or a small separate model) to detect “wrap-up/bye” intent rather than pure substring matching. That reduces false positives from incidental phrases. ◦ Add word-boundary checks and punctuation normalization to avoid matching inside other words. ◦ Expand with common contractions and polite closers (e.g., “thanks, bye”, “take care —”). 1. Make injections safe and minimal ◦ Keep injected correction short and actionable. Example: “Hold — this task is incomplete. Ask about X and call mark_complete when done.” Avoid admonishing language; prefer a single corrective sentence. ◦ Prefer adding a short system instruction (or a high-priority context token) rather than generating a whole reply when possible, so the model updates behavior without interrupting flow. 1. Rate-limit or debounce the hook ◦ Only inject once per task (or once per N minutes) to avoid repeated interruptions. ◦ If a correction was already injected and ignored, escalate to a different mitigation (e.g., log diagnostic info, set a flag to force the agent into an explicit completion flow). 1. Provide an automated fallback / developer-visible signal ◦ If goodbye detection happens, log a structured event with context (task_id, content_preview, missing_fields) so developers can debug false positives or missed mark_complete calls. ◦ Offer a safe fallback: if the task appears complete to the model but mark_complete wasn’t called after two corrections, move the session into a verification step that asks the user a simple confirm question (“Have I completed everything you wanted for this task?”). 1. Tighten task-awareness instructions ◦ Instead of stating “DO NOT say goodbye,” phrase as positive directives: “Do X to finish this task, then call mark_complete. Only after mark_complete may you end the call.” ◦ Show only the nearest 1–2 remaining tasks and the concrete missing fields — too much future context can overload the policy. 1. Timeouts and idle behavior ◦ Use both absolute task timeout and an idle timer, as you suggested. For idle timeout, consider prompting the user before timing out (e.g., “Are you still there? I’ll pause if I don’t hear from you.”). ◦ On TIMEOUT, ensure a deterministic recovery: either auto-call mark_complete with the best-effort state and flag it, or move to a human-review queue (if applicable). 1. Tests and monitoring ◦ Run A/B tests comparing substring detection vs. classifier-based detection to measure false positive/negative rates. ◦ Monitor metrics: frequency of injected corrections, user confusion/abandonment after injection, and cases where mark_complete remains uncalled despite corrections. Concrete micro-updates you can apply quickly • Debounce injections: set a boolean like

self._injection_sent

and skip sending multiple corrections for the same task. • Replace raw substring match with a regex using word boundaries and lowercasing. • Limit injected instruction length to one or two sentences and include a one-line developer log with task id. • Add a confirmation step after timeout: “I’m pausing this task. Did we finish X?” before closing. If you’d like, I can: • Draft a minimal rewrite of the injected correction text to be concise and model-friendly. • Propose a small regex-based goodbye detector to replace the substring check. • Sketch a short fallback flow to run after two ignored injections. Which of these would be most helpful next — the concise injection rewrite, a regex/intent detector example, or the fallback flow sketch?

Say I use taskGroups instead, would I still depend on task.complete() being called by llm? which could fail to execute if there is no goodbye protection in place? Effort: Medium-High python

from livekit.agents.beta.workflows import TaskGroup

class V3WorkflowAgent(Agent):
    async def on_enter(self) -> None:
        task_group = TaskGroup(chat_ctx=self.chat_ctx)
        
        task_group.add(
            lambda: OpenAndConnectTask(...),
            id="open_and_connect",
            description="Greet user, establish rapport, discover underlying why"
        )
        task_group.add(
            lambda: ExploreGoalsTask(...),
            id="explore_goals", 
            description="Sharpen goal into something specific and measurable"
        )
        # ... more tasks
        
        results = await task_group

Benefits: • Built-in regression support (user can say "go back to the goals part") • Automatic context summarization • Task descriptions passed to LLM for awareness •

on_task_completed

callback for checkpoint persistence Drawbacks: • Still relies on

complete()

being called — no goodbye protection

Thanks — this is a great, focused question. I’ll respond to the core: if you move to TaskGroup, do you still depend on the LLM calling

complete()

and what happens if it doesn’t (for example, because the agent said goodbye early)? Short answer • Yes — TaskGroup still depends on each task invoking

complete()

(or otherwise returning a result) to mark that task finished. If a task fails to call

complete()

(for example, the LLM emits a goodbye and the task never triggers completion), the TaskGroup will not advance past that task and your workflow will stall unless you add guardrails or fallbacks. Why (brief) • TaskGroup coordinates task execution by awaiting each task’s completion and collecting typed results. The framework’s control handoff model requires the running task to explicitly hand control back (via

self.complete(result)

or an equivalent return). Without that signal, the TaskGroup will wait indefinitely (or until a timeout/abort occurs). Practical implications and failure modes • Premature wrap-ups: If the model issues goodbye language instead of completing the task, the task stays incomplete and the TaskGroup blocks subsequent tasks. • Silent failures: Bugs that prevent calling

complete()

(logic errors, exceptions, or token-limited responses) prevent progress in the TaskGroup. • User-initiated regression: If user asks to “go back,” TaskGroup can support it, but you still need a way to recover if a prior task didn’t complete cleanly. • Observability: Without logging/metrics, it can be hard to detect tasks stuck waiting for

complete()

. Recommended mitigations (practical, ordered) 1. Keep the Goodbye Guardrail you outlined ◦ Use the Goodbye Detection Hook + brief injected correction to catch early wrap-ups and prompt the model to finish the task. This directly reduces the most common cause of missing

complete()

. 1. Debounce & rate-limit injections ◦ Inject once per task (e.g., set

self._injection_sent = True

) to avoid spamming the model and confusing users. 1. Add task-level timeouts with deterministic behavior ◦ Use both a shorter idle timeout and an absolute task timeout. ◦ On timeout, take a deterministic fallback action: ▪︎ Option A: Auto-call

complete()

with a best-effort partial result plus a

partial=true

flag. ▪︎ Option B: Transition to a verification task that asks the user a single confirmatory question (e.g., “Did we complete X?”), then call

complete()

based on the reply. ◦ TaskGroup can then proceed based on the returned typed result or partial flag. 1. Add an explicit “force-complete” / recovery route inside TaskGroup ◦ Allow TaskGroup to detect stuck tasks (e.g., task not completed after N seconds or after M injections) and run a small recovery handler that: ▪︎ Persists the partial state (via

on_task_completed

or a checkpoint), ▪︎ Advances the group with a "timed_out" sentinel result, and ▪︎ Flags the session for later review. 1. Improve goodbye detection quality (reduce false positives) ◦ Replace raw substring matching with: ▪︎ A regex with word boundaries and punctuation normalization, or ▪︎ A lightweight intent classifier (few-shot prompt or small model) that returns “wrap-up intent” vs “incidental phrase.” ◦ This reduces unnecessary injections that could harm flow. 1. Short, prioritized correction messages for TaskGroup ◦ When injecting corrections during a TaskGroup run, keep them extremely concise and task-specific: one sentence that instructs the model to finish required fields and call

mark_complete

. ◦ Prefer adding a high-priority system instruction (if your runtime supports it) rather than a user-facing interrupt. 1. Instrumentation & monitoring ◦ Emit structured logs whenever goodbye detection fires, when a correction is injected, when timeouts occur, and when a task completes with

partial=true

. ◦ Track metrics: stuck-task rate, injection frequency, and how often timeouts force completion. 1. Make TaskGroup results expressive ◦ Return structured results with fields like

status: completed|partial|timed_out

,

partial=true/false

, and

missing_fields

. ◦ Downstream tasks can then adapt based on that status (e.g., re-run a short subtask to collect missing info). Small code-level patterns you can add quickly • Debounce injection:

if not getattr(self, "_injection_sent", False):
    self._injection_sent = True
    await self.session.generate_reply(instructions="Hold — this task is incomplete. Ask about X and call mark_complete.")

• Timeout fallback sketch:

try:
    result = await asyncio.wait_for(self._run_task(), timeout=self.TIMEOUT_SECONDS)
except asyncio.TimeoutError:
    # checkpoint partial state
    partial_result = self._best_effort_result()
    partial_result["status"] = "timed_out"
    await self.complete(partial_result)

• TaskGroup-aware sentinel result:

# in on_task_completed callback
if result.get("status") == "timed_out":
    # persist checkpoint and let TaskGroup continue with sentinel

When to prefer TaskGroup vs. per-task management • Use TaskGroup if you want built-in regression, summaries, and a single coordinated flow — but still implement timeouts, guardrails, and recovery as above. • If your workflows are fragile and you expect frequent LLM misbehaviors, start with per-task management and explicit state persistence, then move to TaskGroup once you have robust guardrails and monitoring. If you want, I can: • Draft a concise injected-correction message optimized for TaskGroup usage. • Provide a regex-based goodbye detector example to replace substring matching. • Sketch a complete timeout+fallback flow integrated into your TaskGroup and on_task_completed callback. Which would be most helpful next?

is this as effective as multiagent workflow?

Thanks — great question. I’ll compare effectiveness directly and give practical guidance so you can pick the right approach for your workflow. Short answer • TaskGroup is very effective for structured, linear, stateful workflows (data collection, step-by-step onboarding). It’s simpler and more reliable for those cases. • Multi-agent workflows are more effective when you need role separation, independent agents with different capabilities/permissions, or asynchronous/parallel work. They’re complementary patterns, not one-size-fits-all. Why TaskGroup is effective • Sequential control: TaskGroup awaits each task’s completion, so the orchestrating agent won’t proceed until a task calls

complete()

or returns — this makes linear flows predictable. • Shared context and summarization: Tasks inherit the same chat context and TaskGroup provides automatic summaries, which reduces drift and maintains continuity. • Built-in regression and checkpoint hooks: TaskGroup supports going back to earlier tasks, persists checkpoints via

on_task_completed

, and collects typed results that downstream steps can rely on. • Lower operational complexity: Everything lives inside one agent, so there’s less overhead coordinating multiple agents, fewer context-switching concerns, and simpler deployment. When TaskGroup can fail (and how to mitigate) • It still depends on

complete()

being called. If an LLM prematurely says goodbye or fails to call

complete()

, TaskGroup will stall on that task. • Mitigations you should add: ◦ Goodbye guardrails (your detection + short correction injection). ◦ Debounce injections and improve detection quality (regex or a lightweight intent classifier). ◦ Task-level timeouts with deterministic fallbacks (auto-complete with

partial=true

or prompt a single confirmation). ◦ Recovery handler in TaskGroup that advances with a sentinel result after N failed attempts/timeouts. ◦ Instrumentation (logs, metrics) to surface stuck tasks. When multi-agent workflows are more effective • Role/permission separation: If different steps require different access, tools, or privacy boundaries (e.g., intake → billing → legal), separate agents help enforce those boundaries. • Independent scaling and specialization: Teams can develop and tune agents independently; agents can run different prompts, models, or toolsets. • Asynchronous or parallel flows: If parts of the workflow can run in parallel or need to wait for external processes, multiple agents can coordinate better. • Clear handoffs: If you intentionally want an agent handoff (explicitly pass control to another agent), a multi-agent design is a better fit. Trade-offs at a glance • Reliability for linear flows: TaskGroup wins (less coordination overhead, consistent context). • Flexibility & isolation: Multi-agent wins (separation of concerns, security/perms). • Complexity: TaskGroup is simpler; multi-agent requires more orchestration and state passing. • Failure modes: TaskGroup needs guardrails to avoid stalls from missing `complete()`; multi-agent needs careful state transfer and routing logic. Practical recommendation • Start with TaskGroup for structured, single-session, linear workflows (consent → goals → details). Add the guardrails you designed: goodbye detection, short injections, timeouts, and an auto-complete or verification fallback. • Move to a multi-agent architecture when: ◦ Steps require distinct permissions or tool access, or ◦ You need independent scaling, or ◦ Workflow complexity grows (parallel tasks, long-running handoffs). • Hybrid approach: use TaskGroup within each agent for internal sequencing, and use multi-agent handoffs between major stages (agent per role). This gives the best of both worlds. If you want, I can: • Draft a concise goodbye-correction message optimized for TaskGroup. • Sketch a timeout+fallback pattern that auto-completes with

partial=true

and how TaskGroup should interpret that. • Outline a hybrid architecture example (TaskGroup inside agents + agent handoffs between stages). Which follow-up would you like first? Sources: Tasks and task groups | LiveKit Documentation | Building multi-agent architectures with LiveKit agents | LiveKit | Collect recording consent with tasks | LiveKit Documentation