> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cartesia.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 高度なパターン

本番運用される音声エージェント向けのパターン：オブザーバビリティ、ツール設計、マルチエージェントシステム、ガードレール。

## 完全な例：マルチエージェントのカスタマーサービス

この例ではプロンプト、3 種類すべてのツール、マルチエージェントのハンドオフを組み合わせています：

```python theme={null}
import os
from typing import Annotated
from line import CallRequest
from line.llm_agent import (
    LlmAgent, LlmConfig, loopback_tool, passthrough_tool,
    agent_as_handoff, end_call
)
from line.events import AgentSendText, AgentTransferCall
from line.voice_agent_app import AgentEnv, VoiceAgentApp

# Loopback tool: Fetch order info for LLM to contextualize
@loopback_tool
async def get_order_status(ctx, order_id: Annotated[str, "The order ID"]):
    """Look up order status by ID."""
    order = await db.get_order(order_id)
    return f"Order {order_id}: {order.status}, delivers {order.delivery_date}"

# Passthrough tool: Deterministic transfer action
@passthrough_tool
async def transfer_to_human(ctx):
    """Transfer to a human agent."""
    yield AgentSendText(text="Let me connect you with a team member who can help further.")
    yield AgentTransferCall(target_phone_number="+18005551234")

SYSTEM_PROMPT = """You are a friendly customer service agent for Acme Corp.

You can:
- Look up order status using get_order_status
- Transfer to a human agent using transfer_to_human
- Transfer to Spanish support using transfer_to_spanish
- End calls politely using end_call

Rules:
- Always confirm the order ID before looking it up
- Offer to transfer to a human if you can't resolve the issue
- Transfer to Spanish support if the user speaks Spanish or requests it
- Be empathetic and professional
"""

async def get_agent(env: AgentEnv, call_request: CallRequest):
    # Spanish-speaking specialist agent
    spanish_agent = LlmAgent(
        model="gpt-5-nano",
        api_key=os.getenv("OPENAI_API_KEY"),
        tools=[get_order_status, transfer_to_human, end_call],
        config=LlmConfig(
            system_prompt="Eres un agente de servicio al cliente amigable para Acme Corp. Habla solo en español.",
            introduction="¡Hola! Gracias por llamar a Acme Corp. ¿Cómo puedo ayudarte hoy?",
        ),
    )

    # Main English-speaking agent with handoff capability
    return LlmAgent(
        model="anthropic/claude-haiku-4-5-20251001",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        tools=[
            get_order_status,
            transfer_to_human,
            agent_as_handoff(
                spanish_agent,
                handoff_message="Transferring you to our Spanish-speaking team...",
                name="transfer_to_spanish",
                description="Transfer to Spanish support when user speaks Spanish or requests it.",
            ),
            end_call,
        ],
        config=LlmConfig(
            system_prompt=SYSTEM_PROMPT,
            introduction="Hi! Thanks for calling Acme Corp. How can I help you today?",
        ),
    )

app = VoiceAgentApp(get_agent=get_agent)

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

***

## オブザーバビリティ（可観測性）

### メトリクスの記録

パフォーマンスやビジネスメトリクスを追跡します：

```python theme={null}
from line.events import LogMetric, LogMessage

@loopback_tool
async def process_order(ctx, order_id: Annotated[str, "Order ID"]):
    """Process a customer order."""
    import time
    start = time.time()

    result = await api.process_order(order_id)

    # Log timing metric
    yield LogMetric(name="order_processing_ms", value=(time.time() - start) * 1000)

    # Log business event
    yield LogMessage(
        name="order_processed",
        level="info",
        message=f"Processed order {order_id}",
        metadata={"status": result.status}
    )

    return f"Order {order_id} processed: {result.status}"
```

### 組み込みの LLM エージェントメトリクス

`LlmAgent` は、コードなしで毎ターン 3 つのタイミングメトリクスを自動的に発行します：

| メトリクス                | 説明                                            |
| -------------------- | --------------------------------------------- |
| `llm_first_chunk_ms` | 応答生成の開始から、LLM からの最初のチャンク（テキストまたはツール呼び出し）までの時間 |
| `llm_first_text_ms`  | 応答生成の開始から、最初のテキストチャンクまでの時間                    |
| `agent_turn_ms`      | そのターンのエージェント処理時間の合計                           |

***

## ツールパターン

### ツール内でのバリデーション

処理する前に入力をバリデートします：

```python theme={null}
@loopback_tool
async def book_appointment(
    ctx,
    date: Annotated[str, "Date in YYYY-MM-DD format"],
    time: Annotated[str, "Time in HH:MM format"]
):
    """Book an appointment."""
    from datetime import datetime

    try:
        dt = datetime.strptime(f"{date} {time}", "%Y-%m-%d %H:%M")
    except ValueError:
        return "Invalid date or time format. Please use YYYY-MM-DD and HH:MM."

    if dt < datetime.now():
        return "Cannot book appointments in the past."

    # Proceed with booking
    return f"Appointment booked for {dt.strftime('%B %d at %I:%M %p')}"
```

### ツール内の非同期処理

適切なタイムアウト処理を伴う長時間実行の処理：

```python theme={null}
import asyncio

@loopback_tool
async def search_inventory(ctx, query: Annotated[str, "Search query"]):
    """Search inventory with timeout protection."""
    try:
        result = await asyncio.wait_for(
            inventory_api.search(query),
            timeout=5.0
        )
        return f"Found {len(result.items)} items matching '{query}'"
    except asyncio.TimeoutError:
        return "Search is taking longer than expected. Please try a more specific query."
```

### エラー処理

ツール内でエラーを適切に処理します：

```python theme={null}
@loopback_tool
async def get_account_info(ctx, account_id: Annotated[str, "Account ID"]):
    """Look up account information."""
    try:
        account = await api.get_account(account_id)
        return f"Account {account_id}: Balance ${account.balance:.2f}"
    except AccountNotFoundError:
        return f"Account {account_id} not found."
    except Exception as e:
        logger.error(f"Error fetching account: {e}")
        return "Sorry, I couldn't retrieve that account information right now."
```

***

## エージェントラッパー

エージェントラッパーは、ベースとなるエージェントを変更せずに横断的な動作（ロギング、バリデーション、ルーティング）を追加できます。

### ガードレール: 安全性とコンテンツフィルタリング

ラッパーは、双方向の不適切なコンテンツをフィルタリングするガードレールの実装に最適です：

```python theme={null}
class GuardrailsAgent:
    def __init__(self, inner_agent, safety_api):
        self.inner = inner_agent
        self.safety_api = safety_api

    async def process(self, env, event):
        # Pre-processing: Check user input for unsafe content
        if isinstance(event, UserTurnEnded):
            user_text = event.content[0].content if event.content else ""

            if await self.safety_api.is_unsafe(user_text):
                yield AgentSendText(text="I'm here to help with appropriate requests. Let's keep our conversation respectful.")
                return

        # Post-processing: Check agent output for safety issues
        async for output in self.inner.process(env, event):
            if isinstance(output, AgentSendText):
                if await self.safety_api.is_unsafe(output.text):
                    yield LogMessage(
                        name="safety_violation",
                        level="warning",
                        message=f"Blocked unsafe output: {output.text[:100]}..."
                    )
                    yield AgentSendText(text="I apologize, but I can't provide that information.")
                    continue

            yield output
```

よくあるガードレールパターン：

* コンテンツ安全性フィルタリング（トキシシティ、ヘイトスピーチ、PII）
* レート制限と乱用防止
* コンプライアンスチェック（HIPAA、金融規制）
* ブランドセーフティ（ブランドにそぐわない応答）

### 複数エージェント間のルーティング

会話のコンテキストに応じて、特化型エージェントを動的に切り替えます：

```python theme={null}
class RouterAgent:
    def __init__(self, default_agent, specialists: dict):
        self.default = default_agent
        self.specialists = specialists
        self.current = default_agent

    async def process(self, env, event):
        # Switch agent based on user input
        if isinstance(event, UserTurnEnded):
            user_text = event.content[0].content if event.content else ""

            if "billing" in user_text.lower():
                self.current = self.specialists.get("billing", self.default)
            elif "technical" in user_text.lower():
                self.current = self.specialists.get("technical", self.default)

        async for output in self.current.process(env, event):
            yield output
```

`LlmAgent` と組み合わせて使用：

```python theme={null}
async def get_agent(env, call_request):
    return RouterAgent(
        default_agent=LlmAgent(
            model="gpt-5-nano",
            api_key=os.getenv("OPENAI_API_KEY"),
            config=LlmConfig(system_prompt="You are a helpful assistant..."),
        ),
        specialists={
            "billing": LlmAgent(
                model="gpt-5-nano",
                api_key=os.getenv("OPENAI_API_KEY"),
                config=LlmConfig(system_prompt="You are a billing specialist..."),
            ),
            "technical": LlmAgent(
                model="anthropic/claude-haiku-4-5-20251001",
                api_key=os.getenv("ANTHROPIC_API_KEY"),
                config=LlmConfig(system_prompt="You are a technical support specialist..."),
            ),
        }
    )
```

### ベストプラクティス

ラッパーは単一の責務に集中させてください。ストリーミングを保つために `async for` と `yield` を使用してください。1 つの複雑なラッパーを作るより、シンプルなラッパーを重ねてください。

```python theme={null}
# Composable wrappers
agent = LoggingWrapper(
    ValidationWrapper(
        LlmAgent(...)
    )
)
```

***

## 実装例

これらのパターンを示す動作可能な完全な例：

| 例                                                                                             | パターン               | 説明                             |
| --------------------------------------------------------------------------------------------- | ------------------ | ------------------------------ |
| [Form Filler](https://github.com/cartesia-ai/line/tree/main/examples/form_filler)             | ステートフルツール          | YAML 定義のフォームをバリデーション付きでユーザーに案内 |
| [Multi-Agent Transfer](https://github.com/cartesia-ai/line/tree/main/examples/transfer_agent) | agent\_as\_handoff | 英語／スペイン語エージェントのハンドオフ           |
| [Chat Supervisor](https://github.com/cartesia-ai/line/tree/main/examples/chat_supervisor)     | バックグラウンドリサーチ       | 会話用と長時間思考用で別エージェント             |