AI Agents · 课时

音频 + 文本智能体工作流

Q: 「音频 + 文本智能体工作流」这节课中我会学到什么？

端到端完成转录 → 推理 → 音频回复流程 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

端到端完成转录 → 推理 → 音频回复流程

音频 + 文本智能体工作流 是 CoddyKit 上的免费 AI Agents 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

音频-文本智能体流程概览

音频-文本智能体流程在语音世界与文字世界之间进行转换。标准流程是：输入音频 → Whisper 转写 → 文本智能体 → 输出 TTS 音频。这使语音助手、通话分析、会议摘要和免手操作界面成为可能。

支持的音频格式

OpenAI Whisper 支持以下格式：mp3、mp4、mpeg、mpga、m4a、wav 和 webm。文件大小上限为 25 MB。对于更大的文件，您必须先拆分或压缩音频，然后再发送。

为获得最佳转录质量并尽量减小文件大小，请始终以 16 kHz 单声道录制或转换音频。

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 转录音频

OpenAI 的 audio.transcriptions.create 端点封装了 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)

带时间戳的转录

对于会议分析或说话人分离，请请求 verbose_json 格式。该格式会返回单词级或片段级时间戳，从而让您能够定位音频中的确切时刻，以便引用或剪辑。

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'}]}

文本代理处理

转录完成后，文本会作为普通文本消息经过 LLM 代理处理。请加入用于设置上下文的系统提示：这些文本来自口语表达，因此应预期其中包含填充词和不完整句子。

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 进行文本转语音

使用 OpenAI 的 TTS API 将代理的文本响应转换回语音。请选择一种声音（alloy、echo、fable、onyx、nova、shimmer）和一个模型（tts-1 速度更快，tts-1-hd 质量更高）。您可以将结果保存为 mp3 文件，也可以直接进行流式传输。

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'
)

流式音频输出

对于实时语音响应，请分块传输 TTS 音频，而不是等待完整文件生成。OpenAI 的 stream_to_file 辅助工具或手动分块迭代功能，可以让您在其余音频仍在生成时就开始播放。

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

拆分长音频文件

大于 25 MB 的音频文件必须先拆分，然后才能发送给 Whisper。请使用 pydub 库按时间间隔进行拆分，分别处理每个音频块，然后连接各段转录文本。

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)

实时音频处理流程

实时处理流程会分块捕获麦克风输入，在每个音频块到达后将其发送给 Whisper，并将 TTS 输出流式传回，从而形成近实时的语音对话循环。关键挑战在于管理每个阶段的延迟。

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)

语言检测与自动路由

Whisper 会自动检测口语所使用的语言。您可以使用检测到的语言将对话路由到正确的代理角色，或设置响应所使用的 TTS 语言。

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)

处理背景噪声和低质量音频

低质量音频会降低转录准确率。实用的缓解措施包括：使用降噪功能（noisereduce 库）进行预处理，在 Whisper 的 prompt 中加入领域词汇，验证转录文本的置信度，并在置信度较低时请求澄清。

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.'
    }

知识检查

Whisper 的 prompt 参数有什么作用？

回顾：音频-文本代理工作流

很好！要点如下：

Whisper：支持 mp3/wav/m4a 等格式，最大 25 MB；使用 pydub 拆分大文件
时间戳：将 verbose_json 与 timestamp_granularities 一起使用，以获取片段时间信息
TTS：OpenAI TTS 提供多种声音选项；流式传输音频块以降低延迟
语言检测：Whisper 会自动检测语言；根据检测到的语言将请求路由到正确的声音或代理
实时处理流程：缓冲音频块 → 转录 → 代理 → 流式传输 TTS

下一步：代理中的视频理解——帧提取与时间推理。

常见问题解答

「音频 + 文本智能体工作流」课时是免费的吗？

是的 — 「音频 + 文本智能体工作流」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「音频 + 文本智能体工作流」这节课中我会学到什么？

端到端完成转录 → 推理 → 音频回复流程你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「音频 + 文本智能体工作流」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents