Koei's Digital Garden

Tasks and Task Groups

7 min read

Tasks and Task Groups

参照元: LiveKit Agents Documentation ロードマップ: 学習ロードマップ

What（何についてか）

Task は、エージェント内で特定の目的を達成して結果を返す、最小の実行単位である。1つの Task は明確な責務を持ち、完了時に typed result を返して制御を呼び出し元へ戻す。

TaskGroup は、複数 Task を順序付きで実行するための実行コンテナである。単純な直列処理だけではなく、ユーザー都合による「前ステップへの戻り（regress）」を許容しながら、会話文脈を保ったまま最終結果へ収束させる。

この2つは階層が異なる概念であり、Workflow（会話アプリ全体の進行設計）に対して、Task/TaskGroup はその実装部品として機能する。

Why（なぜ必要か）

音声対話の実運用では、ユーザーの発話が常に順序どおりに進むとは限らない。住所確認後に名前を訂正したくなる、最終確認フェーズで連絡先を修正したくなる、といった非線形な進行が頻発する。

Task を単発で積み上げるだけでは、この「戻り」に対する設計が散らばりやすく、状態不整合や実装の重複が発生しやすい。TaskGroup を使うことで、順序・戻り・結果統合を 1 つの枠組みで扱えるため、会話フローの保守性と回復性を同時に確保できる。

How（どう動くか）

Task は AgentTask[T] を継承して定義する。開始処理は on_enter() に記述し、完了時に complete(result) を呼ぶ。

TaskGroup は add(task_factory, id, description) で Task を順序登録する。task_factory に lambda を使うのは、回帰時に同一条件で Task を再生成するためである。description は LLM が「どのステップへ戻るべきか」を判断するヒントになり、曖昧だと回帰精度が下がる。

Group 内 Task は同じ chat_ctx を共有するため、途中で戻っても文脈断絶が起きにくい。完了時は summarize_chat_ctx=True で Group 内対話を要約し、親 Agent 側の文脈に戻せる。return_exceptions=True にすると、途中エラーを結果に保持したまま後続処理を継続できる。

graph TD
    A[Parent Agent] --> B[TaskGroup Start]
    B --> C[Task 1: Consent]
    C --> D[Task 2: Contact Collection]
    D --> E[Task 3: Confirmation]
    E --> F{User correction?}
    F -- Yes --> D
    F -- No --> G[Group Result Summary]
    G --> H[Return control to Parent Agent]

sequenceDiagram
    participant U as User
    participant PA as Parent Agent
    participant TG as TaskGroup
    participant T1 as ConsentTask
    participant T2 as ContactTask
    participant T3 as ConfirmTask

    PA->>TG: start(chat_ctx)
    TG->>T1: run
    T1-->>TG: consent=true
    TG->>T2: run
    T2-->>TG: email/address collected
    TG->>T3: run
    U->>T3: "住所を訂正したい"
    T3->>TG: regress to T2
    TG->>T2: rerun with shared context
    T2-->>TG: corrected address
    TG->>T3: rerun
    T3-->>TG: confirmed
    TG-->>PA: summarized result

コードベースでの解説

以下は、TaskGroup を使って「同意取得 → 連絡先収集 → 最終確認」を組み立てる最小構成である。実装の要点は、Task の責務分離と TaskGroup の回帰制御にある。

from dataclasses import dataclass
from livekit.agents import AgentTask
from livekit.agents.beta.workflows import TaskGroup
 
@dataclass
class ConsentResult:
    accepted: bool
 
@dataclass
class ContactResult:
    email: str
    address: str
 
@dataclass
class ConfirmResult:
    confirmed: bool
 
class ConsentTask(AgentTask[ConsentResult]):
    async def on_enter(self) -> None:
        # ユーザーへ説明し、同意が得られたら complete
        await self.say("この通話は品質向上のため記録されます。同意いただけますか？")
        # ここでは簡略化して true を返す
        self.complete(ConsentResult(accepted=True))
 
class ContactTask(AgentTask[ContactResult]):
    async def on_enter(self) -> None:
        await self.say("連絡先のメールアドレスと住所を教えてください。")
        # STT/validation を経て値を構築する想定
        self.complete(ContactResult(email="user@example.com", address="Tokyo"))
 
class ConfirmTask(AgentTask[ConfirmResult]):
    async def on_enter(self) -> None:
        await self.say("内容を確定してよいですか？ 修正があれば言ってください。")
        # ユーザーが修正要求した場合、TaskGroup が前段へ regress 可能
        self.complete(ConfirmResult(confirmed=True))
 
async def run_group(active_agent):
    group = TaskGroup(
        chat_ctx=active_agent.session.chat_ctx,
        summarize_chat_ctx=True,
        return_exceptions=False,
    )
 
    group.add(
        task_factory=lambda: ConsentTask(),
        id="consent",
        description="録音同意の取得",
    )
    group.add(
        task_factory=lambda: ContactTask(),
        id="contact",
        description="メールアドレスと住所の収集・確認",
    )
    group.add(
        task_factory=lambda: ConfirmTask(),
        id="confirm",
        description="最終確認。必要なら contact に戻る",
    )
 
    results = await group
    return results

この構造の利点は以下の3点に整理できる。第1に、Task を単責務化できるため、検証ロジックや例外処理の再利用が容易になる。第2に、戻り操作が Group 側に集約されるため、Task 自体が状態遷移を抱え込まない。第3に、親 Agent へ戻す際の文脈圧縮（要約）を明示設定でき、長時間セッションでもコンテキスト肥大化を抑制できる。

Key Concepts

用語説明
Task1目的を達成して typed result を返す最小単位。
AgentTask[T]Task 実装の基底クラス。on_enter() と complete() が基本。
TaskGroup複数 Task の順序実行・回帰・結果統合を担う実行コンテナ。
regress後段から前段 Task へ戻して再実行する操作。
task_factory回帰時の再生成を可能にする Task 生成 callable。
summarize_chat_ctxGroup内会話を要約して親文脈へ戻す設定。
return_exceptions例外を結果へ保持して続行するかどうかの設定。

一言まとめ

Task は「1工程の実行単位」、TaskGroup は「多段会話を壊れにくく進める進行制御レイヤー」であり、Workflow を実装するための中核コンポーネントである。

Graph View

Backlinks