Zenn
Open3

LiveKit Agent v1.0 色々お試しメモ

voice
WebRTC
LiveKit
LLM
Agent
kun432kun432

ルーム入室時に挨拶させる

Voice Agent Quickstartだとこうなっている。抜粋。

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="あなたは親切な日本語のAI音声アシスタントです。")


async def entrypoint(ctx: agents.JobContext):
    await ctx.connect()

    session = AgentSession(
        stt=openai.STT(model="whisper-1", language="ja"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(model="tts-1", voice="coral"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
    )

    await session.generate_reply(
        instructions="ユーザーに挨拶し、支援を申し出てください。"
    )

entrypointの中で、エージェントセッション開始後に手動で発話させている。このときの発話はgenerate_replyを使って、LLMに生成させている。

Pipelineノードを使うとエージェントそのものに処理を追加できる。on_enterをエージェント側に追加することもできる。上と挙動としては同じになる。

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="あなたは親切な日本語のAI音声アシスタントです。")

    async def on_enter(self):
        await session.generate_reply(
            instructions="ユーザーに挨拶し、支援を申し出てください。"
        )


async def entrypoint(ctx: agents.JobContext):
    await ctx.connect()

    session = AgentSession(
        stt=openai.STT(model="whisper-1", language="ja"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(model="tts-1", voice="coral"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
    )

session.say()を使えば、LLMを使わずに、テキストでそのままSTTに発話させる。

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="あなたは親切な日本語のAI音声アシスタントです。")

    async def on_enter(self) -> None:
        await self.session.say("こんにちは！今日はどのようなご用件ですか？")

entrypointの中と、エージェントのon_enterで両方指定した場合はどうなるか？

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="あなたは親切な日本語のAI音声アシスタントです。")

    async def on_enter(self) -> None:
        await self.session.say("on_enterで呼ばれました。")


async def entrypoint(ctx: agents.JobContext):
    await ctx.connect()

    session = AgentSession(
        stt=openai.STT(model="whisper-1", language="ja"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(model="tts-1", voice="coral"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
    )

    await session.say("entrypointで呼ばれました。")

「entrypointで呼ばれました。」→「on_enterで呼ばれました。」となる。

マルチエージェント構成などで、エージェントが切り替わって、切り替わり後のエージェントに発話させる場合などはエージェントクラス側で持たせたほうがいいと思う。接続直後に必ず固定で発話させたいとかならentrypointでやるのが良さそう。

kun432kun432

イベント周りの変更

v0.xのVoicePipelineAgentでは、ユーザ・エージェントの各アクションや処理ステップで「イベント」が発行されていた。

https://docs.livekit.io/agents/v0/voice-agent/voice-pipeline/#emitted-events

イベント 説明
user_started_speaking ユーザーが発話を開始した時。
user_stopped_speaking ユーザーが発話を停止した時。
agent_started_speaking エージェントが発話を開始した時。
agent_stopped_speaking エージェントが発話を停止した時。
user_speech_committed ユーザーの発話内容がチャットコンテキストにコミットされた時。
agent_speech_committed エージェントの発話内容がチャットコンテキストにコミットされた時。
agent_speech_interrupted エージェントが会話中に中断された時。
function_calls_collected 実行されるべき関数の完全なセットが受信された時。
function_calls_finished すべての関数呼び出しが実行された時。
metrics_collected メトリクスが収集された時。

上記は、VoicePipelineAgentのパイプラインの各コンポーネント、STT・LLM・TTSが行われたそれぞれのタイミングの前後で発火する。またv0.Xではこれらは全てVoicePipelineAgentのデコレータメソッドon()で全てのイベントを取得できた。

    agent = VoicePipelineAgent(...)

    @agent.on("metrics_collected")
    def _on_metrics_collected(mtrcs: metrics.AgentMetrics):
        metrics.log_metrics(mtrcs)

v1.0ではこれらがPipelineノードを使ってカスタマイズできるようになっている。

https://docs.livekit.io/agents/build/nodes/

ノード 説明
on_enter() エージェントがセッションに入った時。
on_exit() エージェントがセッションから退出した時。
on_user_turn_completed() ユーザーのターンが完了した時。
transcription_node() エージェントのLLMの出力の書き起こし処理時。
stt_node() エージェントのSTT処理時（パイプラインのみ）。
llm_node() エージェントの LLM 処理時ステップ（パイプラインのみ）。
tts_node() エージェントの TTS 処理時ステップ（パイプラインのみ）。
realtime_audio_output_node() エージェントの音声出力ステップ時（Realtimeのみ）。

v0.Xからのマイグレガイドにもあるように、

https://docs.livekit.io/agents/start/v0-migration/

user_started_speaking and user_stopped_speaking events are no longer emitted. They've been combined into a single user_state_changed event.

となっていて、v0.Xでは各ステップの「前後」でイベント発生していたものが、v1.0では全体パイプラインの前後などは増えているが、個々のステップの前後は取れなくなっているように思える。

また、v1.0の各Pipelineノードは、Agentクラスのメソッドとして実装することで、挙動をオーバーライドすることが目的になっているため、v0.Xのような「イベント」とはやや異なる使い方になる。

class MyAgent(Agent):
    async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
        # デフォルトのTTSをオーバーライドして、テキストの前処理を行う
        return Agent.default.tts_node(self, tokenize.utils.replace_words(text), model_settings)

従来のイベントに相当するものは、AgentSession側にデコレータメソッドon()があった。

    session = AgentSession(...)

    @session.on("agent_state_changed")
    ...
    
    @session.on("user_state_changed")    
    ...

    @session.on("metrics_collected")
    ...

    @session.on("speech_created")
    ...

    @session.on("close")
    ...

    @session.on("error")
    ...

    @session.on("conversation_item_added")
    ...

    @session.on("user_input_transcribed")
    ...

    @session.on("function_tools_executed")
    ...

それぞれのタイミングまでは確認していない。

イベントとPipelineノードでそれぞれ目的が異なる上、v0.Xとはイベントの内容も異なる、という点に注意が必要。

あとLiveKitサーバ側のルームで発生するイベントも @ctx.room.onみたいな感じで拾えるようだが、これはv0.Xもv1.0も同じだし、ちょっとエージェントのレイヤーとは異なると思う。

https://docs.livekit.io/home/client/events/

ログインするとコメントできます