エージェントは入力イベントを処理し、出力イベントを yield して会話を制御します。
エージェントとは
エージェントは入出力イベントループを制御します。process メソッドはイベント(ユーザーの発話、通話開始など)を受け取り、レスポンスを yield します。
エージェントには次の 2 つの形式があります:
process メソッドを持つ クラス
- 同じシグネチャ
(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()
サポートされているモデル
| プロバイダー | モデル例 |
|---|
| Anthropic | anthropic/claude-haiku-4-5-20251001、anthropic/claude-sonnet-4-5 |
| OpenAI | gpt-5.4、gpt-5.2 |
| Google | gemini/gemini-2.5-flash-preview-09-2025、gemini/gemini-3.0-preview |
| その他、LiteLLM 経由で 100 以上のプロバイダー | |
LlmConfig オプション
| オプション | 型 | 説明 |
|---|
system_prompt | str | エージェントの動作を定義するシステムプロンプト |
introduction | Optional[str] | 通話開始時に送信されるメッセージ。None または "" の場合はユーザーの発話を待つ |
temperature | Optional[float] | サンプリング温度 |
max_tokens | Optional[int] | レスポンスごとの最大トークン数 |
top_p | Optional[float] | Nucleus サンプリングの閾値 |
stop | Optional[List[str]] | ストップシーケンス |
seed | Optional[int] | 再現性のための乱数シード |
presence_penalty | Optional[float] | トークン生成のプレゼンスペナルティ |
frequency_penalty | Optional[float] | トークン生成の頻度ペナルティ |
num_retries | int | 失敗時のリトライ回数(デフォルト: 2) |
fallbacks | Optional[List[str]] | プライマリが失敗した場合のフォールバックモデル |
timeout | Optional[float] | リクエストタイムアウト(秒) |
reasoning_effort | Optional[str] | none、low、medium、または high。プロバイダーに依存。 |
extra | Dict[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])
ガードレール、ルーティング、エージェントラッパーなどの高度なパターンについては 高度なパターン を参照してください。
着信通話の処理
通話が着信すると、発信者情報を確認したり、エージェントが起動する前に応答方法を設定したりできます。
- Web クライアントまたはテレフォニープロバイダーから通話が着信
pre_call_handler が発信者の詳細を含む CallRequest を受け取る
- 構成(ボイス、言語)を返すか、通話を拒否する
get_agent 関数が、強化されたリクエストを使ってエージェントを作成
CallRequest の解析
着信通話に関する情報が含まれます:
| フィールド | 型 | 説明 |
|---|
call_id | str | 通話の一意の識別子 |
from_ | str | 発信者識別子(電話番号またはクライアント ID) |
to | str | 着信番号またはエージェント ID |
agent_call_id | str | ロギング/相関用のエージェント call ID |
metadata | Optional[dict] | クライアントアプリケーションから渡されたカスタムデータ |
agent | AgentConfig | プレイグラウンドまたは API で構成されたプロンプト |
agent フィールドには次の項目を持つ AgentConfig が含まれます:
| フィールド | 型 | 説明 |
|---|
system_prompt | Optional[str] | プレイグラウンドまたは WebSocket API で構成されたシステムプロンプト |
introduction | Optional[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_id | string | ボイス識別子(UUID) |
model | string | TTS モデル(sonic-3.5、sonic-3、sonic-turbo) |
language | string | 言語コード(en、es、hi など) |
pronunciation_dict_id | string | カスタム発音辞書 の ID |
STT オプション:
| オプション | 型 | 説明 |
|---|
language | string | 音声認識の言語コード |
通話の拒否
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",
}
}
)
エージェントロジックで通話メタデータにアクセスする
CallRequest は get_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() は優先順位を自動的に処理します:
CallRequest.agent.system_prompt の値(設定されている場合)
- ユーザーが指定したフォールバック値(指定されている場合)
- 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 の推論なしでイベント処理ロジックを完全に制御したい場合に強力です。