💻 Примеры: Whisper и мультимодальный pipeline

Пример 1. Локальный Whisper: файл или микрофон

Это аккуратная версия llm-course/lesson-ai-08/ai8-1.py: зависимости ставятся заранее, скрипт не вызывает pip install внутри себя.

# whisper_local.py
import argparse
import os
import tempfile
from datetime import datetime

import sounddevice as sd
import soundfile as sf
import torch
import whisper


def get_device():
    return "cuda" if torch.cuda.is_available() else "cpu"


def record_audio(seconds, sample_rate=16000):
    print(f"Recording {seconds} seconds...")
    audio = sd.rec(int(seconds * sample_rate), samplerate=sample_rate, channels=1)
    sd.wait()
    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
        path = tmp.name
    sf.write(path, audio, sample_rate)
    return path


def transcribe(path, model_size="base", language=None):
    model = whisper.load_model(model_size, device=get_device())
    options = {}
    if language:
        options["language"] = language
    return model.transcribe(path, **options)


def save_text(text, output=None):
    if output is None:
        stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output = f"transcription_{stamp}.txt"
    with open(output, "w", encoding="utf-8") as f:
        f.write(text.strip() + "\n")
    return output


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--file")
    parser.add_argument("--record", type=int, default=0)
    parser.add_argument("--model", default="base", choices=["tiny", "base", "small", "medium", "large"])
    parser.add_argument("--language", help="Language code, for example ru or en")
    parser.add_argument("--output")
    args = parser.parse_args()

    audio_path = args.file
    temporary = False
    if not audio_path and args.record:
        audio_path = record_audio(args.record)
        temporary = True
    if not audio_path:
        raise SystemExit("Use --file audio.wav or --record 10")

    result = transcribe(audio_path, args.model, args.language)
    print(result["text"])
    print("Saved:", save_text(result["text"], args.output))

    if temporary:
        os.remove(audio_path)


if __name__ == "__main__":
    main()

# Запуск
python whisper_local.py --file lecture.wav --language ru --model base
python whisper_local.py --record 10 --language ru --output my_voice.txt

Пример 2. Speech-to-Text API

API-вариант удобен, когда не хочется держать модель локально или нужен серверный сервис с предсказуемой установкой.

# transcribe_api.py
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
client = OpenAI()


def transcribe_file(path):
    with open(path, "rb") as audio_file:
        result = client.audio.transcriptions.create(
            model="gpt-4o-transcribe",
            file=audio_file,
        )
    return result.text


print(transcribe_file("lecture.wav"))

Пример 3. Аудио → текст → изображение → CLIP

Это адаптация идеи из llm-course/lesson-ai-08/ai8-2.py. В учебной версии pipeline делится на явные функции, чтобы каждый шаг можно было проверить отдельно.

# speech_to_image_pipeline.py
import os

import clip
import torch
import whisper
from dotenv import load_dotenv
from huggingface_hub import InferenceClient

load_dotenv()


def speech_to_text(audio_path):
    model = whisper.load_model("base")
    result = model.transcribe(audio_path)
    return result["text"].strip()


def text_to_image(prompt, filename="generated_from_voice.png"):
    client = InferenceClient(provider="fal-ai", api_key=os.environ["HF_TOKEN"])
    image = client.text_to_image(prompt, model="black-forest-labs/FLUX.1-dev")
    image.save(filename)
    return image, filename


def analyze_image(image):
    device = "cuda" if torch.cuda.is_available() else "cpu"
    model, preprocess = clip.load("ViT-B/32", device=device)
    labels = [
        "a technical diagram",
        "a realistic city scene",
        "a classroom slide",
        "a digital artwork",
        "an indoor office scene",
    ]
    image_input = preprocess(image.convert("RGB")).unsqueeze(0).to(device)
    text_input = clip.tokenize(labels).to(device)

    with torch.no_grad():
        image_features = model.encode_image(image_input)
        text_features = model.encode_text(text_input)
        image_features /= image_features.norm(dim=-1, keepdim=True)
        text_features /= text_features.norm(dim=-1, keepdim=True)
        scores = (100.0 * image_features @ text_features.T).softmax(dim=-1)

    best = scores[0].argmax().item()
    return labels[best], scores[0][best].item()


def main():
    transcript = speech_to_text("voice_prompt.wav")
    print("Transcript:", transcript)

    image, filename = text_to_image(transcript)
    label, confidence = analyze_image(image)

    print("Image:", filename)
    print(f"CLIP best label: {label} ({confidence * 100:.2f}%)")


if __name__ == "__main__":
    main()

Пример 4. Та же идея как LangChain Runnable

# runnable_pipeline.py
from langchain_core.runnables import RunnableLambda

pipeline = (
    RunnableLambda(lambda path: speech_to_text(path))
    | RunnableLambda(lambda text: text_to_image(text)[0])
    | RunnableLambda(lambda image: analyze_image(image))
)

print(pipeline.invoke("voice_prompt.wav"))

Такой вариант полезен, когда нужно логировать шаги, заменять функции или встраивать цепочку в более крупное LangChain-приложение.