Voice Live エージェントを作成する

5 分

Microsoft Foundry エージェントで Voice Live を使用するアプリケーションを作成して実行できます。 Voice Live でエージェントを使用すると、モデルに直接接続するよりも次の利点があります。

エージェントは、セッションコードで命令を指定するのではなく、エージェント自体に命令と構成をカプセル化します。
エージェントは複雑なロジックと動作をサポートしているため、クライアントコードを変更せずに会話フローを簡単に管理および更新できます。
エージェントのアプローチにより、統合が合理化されます。エージェント ID は接続に使用され、必要なすべての設定が内部的に処理されるため、コードで手動で構成する必要が減ります。
エージェントロジックを音声実装から分離すると、複数の会話エクスペリエンスやビジネスロジックのバリエーションが必要なシナリオで、保守性とスケーラビリティが向上します。

エージェントプレイグラウンドで音声エージェントを作成する

Microsoft Foundry ポータルでエージェントを開発するときに、 音声モード を有効にして Voice Live をエージェントに簡単に統合し、プレイグラウンドでテストできます。

音声モードが有効になっているエージェントプレイグラウンドのスクリーンショット。

音声モードを有効にした後、[ 構成 ] ウィンドウを使用して、次のような Voice Live 設定を有効にすることができます。

言語: エージェントが話し、理解する言語。
詳細設定:
- 中断と音声の終了を検出するための音声アクティビティ検出 (VAD) 設定。
- バックグラウンドノイズとオーディオ品質を軽減するためのオーディオの強化。
音声: エージェントによって使用される特定の音声と、トーンと話す速度を制御するための高度な音声設定。
中間応答: エージェントは、モデルの応答を待機している間に音声を自動的に生成できます。
アバター: エージェントを表すビジュアルアバターを含めます。

コードを使用して音声エージェントを作成する

コードを使用してエージェントを作成する場合は、適切な Foundry Agent SDK (Foundry SDK for Python など) を使用してエージェントを作成し、その定義に Voice Live メタデータを追加できます。

import os
import json
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition

load_dotenv()

# Setup client
project_client = AIProjectClient(
    "PROJECT_ENDPOINT",
    credential=DefaultAzureCredential(),
)

# Define Voice Live session settings
voice_live_config = {
    "session": {
        "voice": {
            "name": "en-US-Ava:DragonHDLatestNeural",
            "type": "azure-standard",
            "temperature": 0.8
        },
        "input_audio_transcription": {
            "model": "azure-speech"
        },
        "turn_detection": {
            "type": "azure_semantic_vad",
            "end_of_utterance_detection": {
                "model": "semantic_detection_v1_multilingual"
            }
        },
        "input_audio_noise_reduction": {"type": "azure_deep_noise_suppression"},
        "input_audio_echo_cancellation": {"type": "server_echo_cancellation"}
    }
}

# Create agent with Voice Live configuration in metadata
agent = project_client.agents.create_version(
    agent_name="AGENT_NAME",
    definition=PromptAgentDefinition(
        model="MODEL_DEPLOYMENT_NAME",
        instructions="You are a helpful assistant.",
    ),
    metadata=chunk_config(json.dumps(voice_live_config))
)
print(f"Agent created: {agent.name} (version {agent.version})")


# Helper function for Voice Live configuration chunking (to handle 512-char metadata limit)
def chunk_config(config_json: str, limit: int = 512) -> dict:
    """Split config into chunked metadata entries."""
    metadata = {"microsoft.voice-live.configuration": config_json[:limit]}
    remaining = config_json[limit:]
    chunk_num = 1
    while remaining:
        metadata[f"microsoft.voice-live.configuration.{chunk_num}"] = remaining[:limit]
        remaining = remaining[limit:]
        chunk_num += 1
    return metadata

クライアントアプリケーションでエージェントを使用する

エージェントを使用するには、次のクライアントアプリケーションを構築する必要があります。

エージェントに接続します
オーディオハードウェアの入力と出力を構成します
音声ライブセッションを開始します
オーディオシステムのアクティビティを監視します
イベント (ユーザーの音声入力やエージェントからの応答など) を処理します。

API で使用できる機能を使用してこれらのタスクを実装できますが、Voice Live クライアントアプリケーションに推奨されるパターンは次のとおりです。

Microsoft Entra ID 認証を使用して、Microsoft Foundry プロジェクトのエージェントに接続します。
厳密に型指定されたエージェント構成をカプセル化し、音声ライブセッションを構成して開始する関数を定義し、音声イベントを処理するカスタム VoiceAssistant クラスを実装します。
オーディオデバイスを介して入力と出力をカプセル化するカスタム AudioProcessor クラスを実装します。

次の例は、Python でのこのパターンの最小限の実装を示しています (オーディオ入力と出力に PyAudio ライブラリを使用)。

import os
import asyncio
import base64
import queue
from dotenv import load_dotenv
import pyaudio
from azure.identity.aio import AzureCliCredential
from azure.ai.voicelive.aio import connect
from azure.ai.voicelive.models import (
    InputAudioFormat,
    Modality,
    OutputAudioFormat,
    RequestSession,
    ServerEventType,
    AudioNoiseReduction,
    AudioEchoCancellation,
    AzureSemanticVadMultilingual
) 


# Main program entry point
def main():
    """Main entry point."""

    try:
        # Get required configuration from environment variables
        load_dotenv()
        endpoint = os.environ.get("AZURE_VOICELIVE_ENDPOINT")
        agent_name = os.environ.get("AZURE_VOICELIVE_AGENT_ID")
        project_name = os.environ.get("AZURE_VOICELIVE_PROJECT_NAME")
        
        # Create credential for authentication
        credential = AzureCliCredential()
        
        # Create and start the voice assistant
        assistant = VoiceAssistant(
            endpoint=endpoint,
            credential=credential,
            agent_name=agent_name,
            project_name=project_name
        )
        
        # Run the assistant
        try:
            asyncio.run(assistant.start())
        except KeyboardInterrupt:
            # Exit if the user enters CTRL+C
            print("\nGoodbye!")


    except Exception as e:
        print(f"An error occurred: {e}")


# VoiceAssistant class
class VoiceAssistant:
    """Main voice assistant - coordinates the conversation"""
    
    def __init__(self, endpoint, credential, agent_name, project_name):
        self.endpoint = endpoint
        self.credential = credential
        
        # Agent configuration
        self.agent_config = {
            "agent_name": agent_name,
            "project_name": project_name
        }
    
    async def start(self):
        """Start the voice assistant."""
      
        try:
            # Connect the agent
            async with connect(
                endpoint=self.endpoint,
                credential=self.credential,
                api_version="2026-01-01-preview",
                agent_config=self.agent_config
            ) as connection:
                self.connection = connection

                # Initialize audio processor
                self.audio_processor = AudioProcessor(connection)
                
                # Configure the session
                await self.setup_session()
                
                # Start audio I/O 
                self.audio_processor.start_playback()
                print("\nVoice session started...")
                print("Press Ctrl+C to exit\n")
                
                # Process events
                await self.process_events()

        
        finally:
            if hasattr(self, 'audio_processor'):
                self.audio_processor.shutdown()
    
    async def setup_session(self):
        """Configure the session with audio settings."""
        
        session_config = RequestSession(
            # Enable both text and audio
            modalities=[Modality.TEXT, Modality.AUDIO],
            
            # Audio format (16-bit PCM at 24kHz)
            input_audio_format=InputAudioFormat.PCM16,
            output_audio_format=OutputAudioFormat.PCM16,
            
            # Voice activity detection (when to detect speech)
            turn_detection=AzureSemanticVadMultilingual(),
            
            # Prevent echo from speaker feedback
            input_audio_echo_cancellation=AudioEchoCancellation(),
            
            # Reduce background noise
            input_audio_noise_reduction=AudioNoiseReduction(type="azure_deep_noise_suppression")
        )
        
        await self.connection.session.update(session=session_config)
        print("Session configured")
    
    async def process_events(self):
        """Process events from the VoiceLive service."""
        
        # Listen for events from the service
        async for event in self.connection:
            await self.handle_event(event)
    
    async def handle_event(self, event):
        """Handle different event types from the service."""
        
        # Session is ready - start capturing audio
        if event.type == ServerEventType.SESSION_UPDATED:
            print(f"Connected to agent: {event.session.agent.name}")
            self.audio_processor.start_capture()
        
        # User speech was transcribed
        elif event.type == ServerEventType.CONVERSATION_ITEM_INPUT_AUDIO_TRANSCRIPTION_COMPLETED:
            print(f'You: {event.get("transcript", "")}')
        
        # Agent is responding with audio transcript
        elif event.type == ServerEventType.RESPONSE_AUDIO_TRANSCRIPT_DONE:
            print(f'Agent: {event.get("transcript", "")}')
        
        # User started speaking (interrupt any playing audio)
        elif event.type == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STARTED:
            self.audio_processor.clear_playback_queue()
            print("Listening...")
        
        # User stopped speaking
        elif event.type == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STOPPED:
            print("Thinking...")
        
        # Receiving audio response chunks
        elif event.type == ServerEventType.RESPONSE_AUDIO_DELTA:
            self.audio_processor.queue_audio(event.delta)
        
        # Audio response complete
        elif event.type == ServerEventType.RESPONSE_AUDIO_DONE:
            print("Response complete\n")
        
        # Handle errors
        elif event.type == ServerEventType.ERROR:
            print(f"Error: {event.error.message}")


# AudioProcessor class
class AudioProcessor:
    """Handles audio input (microphone) and output (speakers) """
    
    def __init__(self, connection):
        self.connection = connection
        self.audio = pyaudio.PyAudio()
        
        # Audio settings: 24kHz, 16-bit PCM, mono
        self.format = pyaudio.paInt16
        self.channels = 1
        self.rate = 24000
        self.chunk_size = 1200  # 50ms chunks
        
        # Streams for input and output
        self.input_stream = None
        self.output_stream = None
        self.playback_queue = queue.Queue()
    
    def start_capture(self):
        """Start capturing audio from the microphone."""
        
        def capture_callback(in_data, frame_count, time_info, status):
            # Convert audio to base64 and send to VoiceLive
            audio_base64 = base64.b64encode(in_data).decode("utf-8")
            asyncio.run_coroutine_threadsafe(
                self.connection.input_audio_buffer.append(audio=audio_base64),
                self.loop
            )
            return (None, pyaudio.paContinue)
        
        # Store event loop for use in callback thread
        self.loop = asyncio.get_event_loop()
        
        self.input_stream = self.audio.open(
            format=self.format,
            channels=self.channels,
            rate=self.rate,
            input=True,
            frames_per_buffer=self.chunk_size,
            stream_callback=capture_callback
        )
        print("Microphone started")
    
    def start_playback(self):
        """Start audio playback system."""
        
        remaining = bytes()
        
        def playback_callback(in_data, frame_count, time_info, status):
            nonlocal remaining
            
            # Calculate bytes needed
            bytes_needed = frame_count * pyaudio.get_sample_size(pyaudio.paInt16)
            output = remaining[:bytes_needed]
            remaining = remaining[bytes_needed:]
            
            # Get more audio from queue if needed
            while len(output) < bytes_needed:
                try:
                    audio_data = self.playback_queue.get_nowait()
                    if audio_data is None:  # End signal
                        break
                    output += audio_data
                except queue.Empty:
                    # Pad with silence if no audio available
                    output += bytes(bytes_needed - len(output))
                    break
            
            # Keep any extra for next callback
            if len(output) > bytes_needed:
                remaining = output[bytes_needed:]
                output = output[:bytes_needed]
            
            return (output, pyaudio.paContinue)
        
        self.output_stream = self.audio.open(
            format=self.format,
            channels=self.channels,
            rate=self.rate,
            output=True,
            frames_per_buffer=self.chunk_size,
            stream_callback=playback_callback
        )
        print("Speakers ready")
    
    def queue_audio(self, audio_data):
        """Add audio data to the playback queue."""
        self.playback_queue.put(audio_data)
    
    def clear_playback_queue(self):
        """Clear any pending audio (used when user interrupts)."""
        while not self.playback_queue.empty():
            try:
                self.playback_queue.get_nowait()
            except queue.Empty:
                break
    
    def shutdown(self):
        """Clean up audio resources."""
        if self.input_stream:
            self.input_stream.stop_stream()
            self.input_stream.close()
        
        if self.output_stream:
            self.playback_queue.put(None)  # Signal end
            self.output_stream.stop_stream()
            self.output_stream.close()
        
        self.audio.terminate()
        print("Audio stopped")


if __name__ == "__main__":
    main()

フィードバック

このページはお役に立ちましたか?

Voice Live エージェントを作成する

エージェントプレイグラウンドで音声エージェントを作成する

コードを使用して音声エージェントを作成する

クライアント アプリケーションでエージェントを使用する

フィードバック

クライアントアプリケーションでエージェントを使用する