音声・テキストエージェントのワークフロー

音声・テキストエージェントパイプラインの概要

対応音声形式

import os

SUPPORTED_FORMATS = {'.mp3', '.mp4', '.mpeg', '.mpga', '.m4a', '.wav', '.webm'}
MAX_FILE_SIZE_BYTES = 25 * 1024 * 1024  # 25 MB

def validate_audio_file(path: str) -> dict:
    ext = os.path.splitext(path)[1].lower()
    size = os.path.getsize(path) if os.path.exists(path) else 0
    return {
        'path': path,
        'format_ok': ext in SUPPORTED_FORMATS,
        'size_ok': size <= MAX_FILE_SIZE_BYTES,
        'size_mb': round(size / 1024 / 1024, 2),
        'extension': ext
    }

# Usage:
info = validate_audio_file('meeting.wav')
print(info)

Whisperによる音声の文字起こし

from openai import OpenAI

client = OpenAI(api_key='YOUR_OPENAI_API_KEY')

def transcribe_audio(
    audio_path: str,
    language: str = 'en',
    prompt: str = ''
) -> str:
    with open(audio_path, 'rb') as audio_file:
        transcript = client.audio.transcriptions.create(
            model='whisper-1',
            file=audio_file,
            language=language,
            prompt=prompt,  # e.g. 'Python, NestJS, TypeScript, API'
            response_format='text'
        )
    return transcript

text = transcribe_audio('user_query.mp3', language='en',
                        prompt='Kubernetes, microservices, Docker')
print('Transcription:', text)

タイムスタンプ付き文字起こし

def transcribe_with_timestamps(audio_path: str) -> dict:
    from openai import OpenAI
    client = OpenAI(api_key='YOUR_OPENAI_API_KEY')
    with open(audio_path, 'rb') as f:
        result = client.audio.transcriptions.create(
            model='whisper-1',
            file=f,
            response_format='verbose_json',
            timestamp_granularities=['segment']
        )
    # result.segments: list of {start, end, text}
    segments = [
        {'start': s.start, 'end': s.end, 'text': s.text}
        for s in result.segments
    ]
    return {'full_text': result.text, 'segments': segments}

# Example output structure:
# {'full_text': 'Hello team...', 'segments': [{'start': 0.0, 'end': 2.3, 'text': 'Hello team'}]}

テキストエージェントによる処理

import anthropic

VOICE_AGENT_SYSTEM = (
    'You are a helpful voice assistant. The user\'s message was transcribed from speech '
    'and may contain filler words ("um", "uh"), false starts, or incomplete sentences. '
    'Interpret the intent charitably and respond concisely in 1-3 sentences. '
    'Your response will be converted to speech, so avoid markdown, lists, or code blocks.'
)

def voice_agent_respond(transcription: str) -> str:
    client = anthropic.Anthropic(api_key='YOUR_ANTHROPIC_API_KEY')
    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=256,
        system=VOICE_AGENT_SYSTEM,
        messages=[{'role': 'user', 'content': transcription}]
    )
    return response.content[0].text

OpenAI TTSによるテキスト読み上げ

from openai import OpenAI
from pathlib import Path

client = OpenAI(api_key='YOUR_OPENAI_API_KEY')

def text_to_speech(
    text: str,
    output_path: str = 'response.mp3',
    voice: str = 'nova',
    model: str = 'tts-1'
) -> str:
    response = client.audio.speech.create(
        model=model,
        voice=voice,
        input=text,
        response_format='mp3'
    )
    Path(output_path).write_bytes(response.content)
    print(f'TTS audio saved to {output_path}')
    return output_path

# Full pipeline:
tts_file = text_to_speech(
    'The weather in London is currently 15 degrees and partly cloudy.',
    voice='nova'
)

音声出力のストリーミング

from openai import OpenAI

client = OpenAI(api_key='YOUR_OPENAI_API_KEY')

def stream_tts_to_file(text: str, output_path: str):
    # OpenAI SDK streams the audio response
    with client.audio.speech.with_streaming_response.create(
        model='tts-1',
        voice='nova',
        input=text,
        response_format='mp3'
    ) as response:
        response.stream_to_file(output_path)
    print(f'Streamed to {output_path}')

def stream_tts_chunks(text: str):
    """Yield raw audio bytes for piping to an audio player."""
    with client.audio.speech.with_streaming_response.create(
        model='tts-1',
        voice='nova',
        input=text
    ) as response:
        for chunk in response.iter_bytes(chunk_size=4096):
            yield chunk  # pipe to audio player or WebSocket

長い音声ファイルの分割

from pydub import AudioSegment
import os

def split_audio(
    path: str,
    chunk_minutes: int = 10
) -> list:
    audio = AudioSegment.from_file(path)
    chunk_ms = chunk_minutes * 60 * 1000
    chunks = []
    os.makedirs('audio_chunks', exist_ok=True)
    for i, start in enumerate(range(0, len(audio), chunk_ms)):
        chunk = audio[start:start + chunk_ms]
        chunk_path = f'audio_chunks/chunk_{i:03d}.mp3'
        chunk.export(chunk_path, format='mp3')
        chunks.append(chunk_path)
    return chunks

def transcribe_long_audio(path: str) -> str:
    chunks = split_audio(path, chunk_minutes=10)
    transcripts = []
    for chunk_path in chunks:
        text = transcribe_audio(chunk_path)
        transcripts.append(text)
    # Clean up chunks
    for chunk_path in chunks:
        os.remove(chunk_path)
    return ' '.join(transcripts)

リアルタイム音声パイプライン

import asyncio
import time

async def realtime_voice_loop(mic_stream, speaker_stream, client):
    """
    mic_stream: async generator yielding audio bytes
    speaker_stream: async callable accepting audio bytes
    """
    buffer = b''
    BUFFER_THRESHOLD = 50 * 1024  # ~3 seconds at 16kHz mono mp3

    async for audio_chunk in mic_stream:
        buffer += audio_chunk

        if len(buffer) >= BUFFER_THRESHOLD:
            t0 = time.time()
            # Save buffer to temp file
            with open('/tmp/voice_chunk.mp3', 'wb') as f:
                f.write(buffer)
            buffer = b''

            # Transcribe
            text = transcribe_audio('/tmp/voice_chunk.mp3')
            print(f'Transcribed ({time.time()-t0:.1f}s): {text}')

            # Agent response
            reply = voice_agent_respond(text)

            # TTS and stream to speaker
            for audio_bytes in stream_tts_chunks(reply):
                await speaker_stream(audio_bytes)

言語検出と自動ルーティング

def transcribe_and_detect_language(audio_path: str) -> dict:
    from openai import OpenAI
    client = OpenAI(api_key='YOUR_OPENAI_API_KEY')
    with open(audio_path, 'rb') as f:
        result = client.audio.transcriptions.create(
            model='whisper-1',
            file=f,
            response_format='verbose_json'
        )
    return {
        'text': result.text,
        'language': result.language,  # e.g. 'english', 'spanish'
        'duration': result.duration
    }

LANGUAGE_VOICE_MAP = {
    'english': 'nova',
    'spanish': 'alloy',
    'french': 'echo',
    'german': 'fable'
}

def respond_in_detected_language(audio_path: str) -> str:
    info = transcribe_and_detect_language(audio_path)
    voice = LANGUAGE_VOICE_MAP.get(info['language'], 'nova')
    reply = voice_agent_respond(info['text'])
    return text_to_speech(reply, voice=voice)

背景ノイズと低品質音声への対処

def transcribe_with_quality_check(
    audio_path: str,
    min_confidence_words: int = 3
) -> dict:
    from openai import OpenAI
    import string
    client = OpenAI(api_key='YOUR_OPENAI_API_KEY')
    with open(audio_path, 'rb') as f:
        result = client.audio.transcriptions.create(
            model='whisper-1',
            file=f,
            response_format='verbose_json',
            timestamp_granularities=['word']
        )
    # Word count as a basic quality proxy
    words = result.text.translate(
        str.maketrans('', '', string.punctuation)
    ).split()
    quality_ok = len(words) >= min_confidence_words
    return {
        'text': result.text,
        'word_count': len(words),
        'quality_ok': quality_ok,
        'fallback_message': None if quality_ok else 'I could not hear you clearly. Please repeat.'
    }

知識チェック

振り返り：音声・テキストエージェントのワークフロー