メインコンテンツへスキップ
エージェントは入力イベントを処理し、出力イベントを yield して会話を制御します。

エージェントとは

エージェントは入出力イベントループを制御します。process メソッドはイベント(ユーザーの発話、通話開始など)を受け取り、レスポンスを yield します。 エージェントには次の 2 つの形式があります:
  1. process メソッドを持つ クラス
  2. 同じシグネチャ (env, event) -> AsyncIterable[OutputEvent] を持つ 関数
from line.events import CallStarted, UserTurnEnded, AgentSendText

class HelloAgent:
    async def process(self, env, event):
        if isinstance(event, CallStarted):
            yield AgentSendText(text="Hello!")
        elif isinstance(event, UserTurnEnded):
            yield AgentSendText(text="I heard you!")
エージェントの仕組み:
  • イベントが到着(ユーザーの発話、通話開始、ボタン押下)
  • SDK が agent.process(env, event) を呼び出す
  • エージェントが出力イベント(発話、ツール呼び出し、ハンドオフ)を yield する
  • SDK が音声、LLM 呼び出し、状態管理を担当

LlmAgent

LiteLLM を介して 100 以上の LLM プロバイダーをラップする組み込みの LlmAgent を使用します:
from line.llm_agent import LlmAgent, LlmConfig

agent = LlmAgent(
    model="anthropic/claude-haiku-4-5-20251001",  # Or "gpt-5.2", "gemini/gemini-2.5-flash", etc.
    api_key="your-api-key",
    tools=[...],  # Optional list of tools
    config=LlmConfig(
        system_prompt="You are a helpful assistant...",
        introduction="Hello! How can I help you today?",
    ),
)

プロンプト

system_prompt でエージェントのパーソナリティを定義し、introduction で挨拶を指定します:
import os
from line import CallRequest
from line.llm_agent import LlmAgent, LlmConfig, end_call
from line.voice_agent_app import AgentEnv, VoiceAgentApp

SYSTEM_PROMPT = """You are a friendly customer service agent.

Rules:
- Be polite and empathetic
- Confirm understanding before taking action
-  end_call to gracefully end conversations
"""

async def get_agent(env: AgentEnv, call_request: CallRequest):
    return LlmAgent(
        model="anthropic/claude-haiku-4-5-20251001",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        tools=[end_call],
        config=LlmConfig(
            system_prompt=SYSTEM_PROMPT,
            introduction="Hello! How can I help you today?",
        ),
    )

app = VoiceAgentApp(get_agent=get_agent)

if __name__ == "__main__":
    app.run()

サポートされているモデル

プロバイダーモデル例
Anthropicanthropic/claude-haiku-4-5-20251001anthropic/claude-sonnet-4-5
OpenAIgpt-5.4gpt-5.2
Googlegemini/gemini-2.5-flash-preview-09-2025gemini/gemini-3.0-preview
その他、LiteLLM 経由で 100 以上のプロバイダー

LlmConfig オプション

オプション説明
system_promptstrエージェントの動作を定義するシステムプロンプト
introductionOptional[str]通話開始時に送信されるメッセージ。None または "" の場合はユーザーの発話を待つ
temperatureOptional[float]サンプリング温度
max_tokensOptional[int]レスポンスごとの最大トークン数
top_pOptional[float]Nucleus サンプリングの閾値
stopOptional[List[str]]ストップシーケンス
seedOptional[int]再現性のための乱数シード
presence_penaltyOptional[float]トークン生成のプレゼンスペナルティ
frequency_penaltyOptional[float]トークン生成の頻度ペナルティ
num_retriesint失敗時のリトライ回数(デフォルト: 2)
fallbacksOptional[List[str]]プライマリが失敗した場合のフォールバックモデル
timeoutOptional[float]リクエストタイムアウト(秒)
reasoning_effortOptional[str]nonelowmedium、または high。プロバイダーに依存。
extraDict[str, Any]LiteLLM に渡されるプロバイダー固有のオプション

履歴管理

LlmAgent は、LLM が参照する会話履歴を構造化された形で制御するための history 属性を公開しています。 エントリの追加:
# Append a user note (role="user" is the default)
agent.history.add_entry("The user prefers formal language.")

# Insert before a specific event
agent.history.add_entry("Context about the caller.", before=some_event)
履歴セグメントの置換:
# Replace the entire history
agent.history.update(new_events)

# Replace everything from `start` onward
agent.history.update(new_events, start=some_event)

# Replace a specific segment
agent.history.update(new_events, start=start_event, end=end_event)

ターンごとのオーバーライド

process() はキーワード引数を受け付け、エージェントを変更せずにそのターンだけに適用します:
# Higher temperature for just this turn
await agent.process(env, event, config=LlmConfig(temperature=0.9))

# Swap a specific tool for one turn
await agent.process(env, event, tools=[custom_lookup_tool])

# Inject ephemeral context
await agent.process(env, event, context="The user is a VIP customer.")

# Completely override history for one turn
await agent.process(env, event, history=custom_history_list)
明示的に設定された LlmConfig のフィールドのみが反映され、未設定のフィールドはエージェントに保存されている設定にフォールスルーします。 ツールを永続的に変更する場合(例:特定時点以降に end_call を有効化)、ターンごとのオーバーライドではなく agent.tools を直接変更してください。

会話ループの制御

イベントフィルター を使用して、エージェントの process メソッドが実行されるタイミング、そしてそれを中断できるイベントを制御します。

デフォルトの動作

# Agent processes these events:
run_filter = [CallStarted, UserTurnEnded, CallEnded]

# These events interrupt the agent:
cancel_filter = [UserTurnStarted]
つまり、エージェントは通話開始時に挨拶し、ユーザーが発話を終えると応答し、ユーザーから割り込まれることができます。

フィルターのカスタマイズ

デフォルトを上書きするには get_agent からタプルを返します:
from line.events import CallStarted, UserTurnEnded, UserTurnStarted, CallEnded

async def get_agent(env, call_request):
    agent = LlmAgent(...)
    
    # Customize behavior
    run_filter = [CallStarted, UserTurnEnded, CallEnded]
    cancel_filter = [UserTurnStarted]
    
    return (agent, run_filter, cancel_filter)

よくあるカスタマイズ

応答性を高める(部分的な書き起こしを処理):
from line.events import CallStarted, UserTurnEnded, UserTextSent, CallEnded

run_filter = [CallStarted, UserTurnEnded, UserTextSent, CallEnded]
cancel_filter = [UserTurnStarted]
これによりエージェントは、ユーザーの発話が終わる前に処理を開始するため、よりレスポンシブな体験を実現できます。 割り込み不可能なターン: 特定のメッセージをユーザーに中断されずに完了させたい場合は、AgentSendText で送信するときに出力を interruptible=False としてマークします。
from line.events import AgentSendText

yield AgentSendText(
    text="Before we continue, I need to share a quick disclaimer.",
    interruptible=False,
)
関数を使ったカスタムロジック:
def business_hours_only(event):
    hour = datetime.now().hour
    if isinstance(event, (CallStarted, CallEnded)):
        return True
    return isinstance(event, UserTurnEnded) and 9 <= hour < 17

return (agent, business_hours_only, [UserTurnStarted])
ガードレール、ルーティング、エージェントラッパーなどの高度なパターンについては 高度なパターン を参照してください。

着信通話の処理

通話が着信すると、発信者情報を確認したり、エージェントが起動する前に応答方法を設定したりできます。
  1. Web クライアントまたはテレフォニープロバイダーから通話が着信
  2. pre_call_handler が発信者の詳細を含む CallRequest を受け取る
  3. 構成(ボイス、言語)を返すか、通話を拒否する
  4. get_agent 関数が、強化されたリクエストを使ってエージェントを作成

CallRequest の解析

着信通話に関する情報が含まれます:
フィールド説明
call_idstr通話の一意の識別子
from_str発信者識別子(電話番号またはクライアント ID)
tostr着信番号またはエージェント ID
agent_call_idstrロギング/相関用のエージェント call ID
metadataOptional[dict]クライアントアプリケーションから渡されたカスタムデータ
agentAgentConfigプレイグラウンドまたは API で構成されたプロンプト
agent フィールドには次の項目を持つ AgentConfig が含まれます:
フィールド説明
system_promptOptional[str]プレイグラウンドまたは WebSocket API で構成されたシステムプロンプト
introductionOptional[str]プレイグラウンドまたは WebSocket API で構成された導入メッセージ

PreCallResult を返す

エージェントが起動する前に、pre_call_handler でボイスや言語を設定するか、通話を拒否します:
from line.voice_agent_app import CallRequest, PreCallResult, VoiceAgentApp

async def pre_call_handler(call_request: CallRequest):
    return PreCallResult(
        metadata={"tier": "premium"},  # Merged into call_request.metadata
        config={
            "tts": {
                "voice_id": "a0e99841-438c-4a64-b679-ae501e7d6091",
                "model": "sonic-3.5",
                "language": "en",
            }
        }
    )

app = VoiceAgentApp(get_agent=get_agent, pre_call_handler=pre_call_handler)
クライアントアプリケーションは、コールリクエストでメタデータ(ユーザー ID、言語設定、アカウントティアなど)を渡せます。pre_call_handler はこれを読み取り、それに応じて TTS/STT を構成します。

構成オプション

TTS オプション:
オプション説明
voice_idstringボイス識別子(UUID)
modelstringTTS モデル(sonic-3.5sonic-3sonic-turbo
languagestring言語コード(eneshi など)
pronunciation_dict_idstringカスタム発音辞書 の ID
STT オプション:
オプション説明
languagestring音声認識の言語コード

通話の拒否

None を返して、403 ステータスで通話を拒否します:
async def pre_call_handler(call_request: CallRequest):
    if is_blocked(call_request.from_):
        return None  # Rejects with 403
    return PreCallResult()

カスタム発音

特定の単語の発音方法を制御するには 発音辞書 を使用します:
async def pre_call_handler(call_request: CallRequest):
    return PreCallResult(
        config={
            "tts": {
                "voice_id": "a0e99841-438c-4a64-b679-ae501e7d6091",
                "model": "sonic-3.5",
                "pronunciation_dict_id": "your-dict-id",
            }
        }
    )

エージェントロジックで通話メタデータにアクセスする

CallRequestget_agent で利用できます:
async def get_agent(env, call_request):
    # Log call information
    logger.info(f"Call {call_request.call_id} from {call_request.from_}")

    # Access metadata passed from your application (or added in pre_call_handler)
    customer_id = call_request.metadata.get("customer_id") if call_request.metadata else None
    customer_name = call_request.metadata.get("customer_name") if call_request.metadata else None

    # Build a personalized system prompt using metadata
    base_prompt = call_request.agent.system_prompt or "You are a helpful customer service agent."

    if customer_id:
        base_prompt += f"\n\nCurrent customer ID: {customer_id}"
    if customer_name:
        base_prompt += f"\nCustomer name: {customer_name}"

    return LlmAgent(
        model="gpt-5-nano",
        api_key=os.getenv("OPENAI_API_KEY"),
        config=LlmConfig(
            system_prompt=base_prompt,
            introduction=call_request.agent.introduction,
        ),
    )
LlmConfig.from_call_request() は優先順位を自動的に処理します:
  1. CallRequest.agent.system_prompt の値(設定されている場合)
  2. ユーザーが指定したフォールバック値(指定されている場合)
  3. SDK のデフォルト
async def get_agent(env, call_request):
    return LlmAgent(
        model="anthropic/claude-haiku-4-5-20251001",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        tools=[end_call],
        config=LlmConfig.from_call_request(
            call_request,
            fallback_system_prompt="You are a sales assistant.",
            fallback_introduction="Hi! How can I help with your purchase?",
            temperature=0.7,  # Additional LlmConfig options
        ),
    )
CallRequest を使うと、プレイグラウンドからシステムプロンプトを瞬時にイテレーションでき、コード側では技術的な構成とフォールバックのデフォルトを担当できます。

ユーザーに先に話させる

ユーザーが先に話すのを待つには、introduction を空文字列に設定します:
config=LlmConfig.from_call_request(
    call_request,
    fallback_system_prompt=SYSTEM_PROMPT,
    fallback_introduction="",
)

カスタムエージェント関数

高度なユースケースでは、エージェントを関数としてゼロから構築できます:
from line.events import UserTurnEnded, AgentSendText, CallStarted

async def my_agent(env, event):
    if isinstance(event, CallStarted):
        yield AgentSendText(text="Hello! How can I help?")
    elif isinstance(event, UserTurnEnded):
        user_text = event.content[0].content if event.content else ""
        yield AgentSendText(text=f"You said: {user_text}")

カスタムエージェントクラス

または、状態を持つクラスとしても構築できます:
class GreetingAgent:
    def __init__(self, greeting: str):
        self.greeting = greeting
        self.greeted = False

    async def process(self, env, event):
        if isinstance(event, CallStarted) and not self.greeted:
            yield AgentSendText(text=self.greeting)
            self.greeted = True
ほとんどの開発者は、カスタムエージェントをゼロから構築するのではなく、LlmAgent をツールと組み合わせて使うのがおすすめです。カスタムエージェントは、LLM の推論なしでイベント処理ロジックを完全に制御したい場合に強力です。