Голосовой AI-ассистент на Python: от Telegram до автокоммитов в Git

Зачем разработчику голосовой ассистент в 2026 году

Представьте: вы едете в метро, смотрите на код проекта и замечаете баг. Руки заняты. Вы говорите в телефон: "Создай issue 'Утечка памяти в кэше Redis', добавь лейблы bug, high-priority, назначь на меня". Через секунду задача уже в Jira. Или другой сценарий: вы настраиваете сервер, но не хотите отвлекаться от консоли. Голосовая команда: "Запусти деплой staging-окружения, проверь health-check". Система выполняет.

Это не фантастика. Это production-система, которую можно собрать сегодня. Причем без облачных API с ежемесячными счетами в тысячи долларов. Локально или на дешевом VPS.

Архитектура: что скрывается за простой командой

Когда вы говорите "Создай коммит с сообщением 'фикс бага в аутентификации'", система проходит 7 этапов:

1. Telegram бот принимает голосовое сообщение
2. GigaASR преобразует аудио в текст (с точностью 96% для русского)
3. Gemini 2.5 Flash анализирует интент: понимает, что это команда для Git
4. Claude Code 3.5 генерирует конкретные команды: git add, git commit -m
5. Система выполняет команды в указанном репозитории
6. Silero TTS синтезирует голосовой ответ
7. Telegram отправляет аудио-подтверждение и текстовый лог

Каждый компонент выбран не случайно. GigaASR — потому что русский язык. Gemini Flash — потому что скорость (обработка за 200-300мс). Claude Code — потому что качество генерации команд. Silero TTS — потому что натуральность и локальность.

Стек технологий 2026 года: что действительно работает

Внимание на даты: это актуальные версии на февраль 2026. GigaASR v4 вышла в январе с улучшенной поддержкой технических терминов. Claude Code 3.5 научился понимать контекст git-репозитория — видит измененные файлы перед генерацией команды.

Подготовка: железо, софт и токены

Сначала плохая новость: полностью локальная система потребует RTX 4090 или эквивалента. Хорошая новость: можно разнести компоненты. STT и TTS — на свой сервер (даже на CPU). LLM — использовать облачные API (Gemini Flash стоит $0.00015 за 1K токенов).

1 Базовое окружение

# Ubuntu 22.04 или новее
sudo apt update
sudo apt install -y python3.11 python3.11-venv ffmpeg git

# Создаем виртуальное окружение
python3.11 -m venv ~/voice_assistant
source ~/voice_assistant/bin/activate

# Базовые зависимости
pip install --upgrade pip
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu121  # для CUDA 12.1
# Или для CPU: pip install torch torchaudio

2 Telegram бот: не просто приемник файлов

Большинство гайдов показывают простого бота, который только принимает аудио. Наш бот должен:

• Поддерживать сессии (чтобы помнить контекст репозитория)
• Валидировать пользователей (не все могут создавать коммиты)
• Обрабатывать параллельные запросы (async/await обязательно)
• Логировать все действия (для отладки и аудита)

# bot/core.py - основа
from telegram.ext import Application, CommandHandler, MessageHandler, filters
import asyncio
from datetime import datetime
import json

class VoiceAssistantBot:
    def __init__(self, token: str):
        self.app = Application.builder().token(token).build()
        self.user_sessions = {}  # user_id -> {current_repo: ..., last_command: ...}
        self.setup_handlers()
    
    def setup_handlers(self):
        self.app.add_handler(CommandHandler("start", self.start))
        self.app.add_handler(CommandHandler("set_repo", self.set_repo))
        # Обработчик голосовых сообщений
        self.app.add_handler(MessageHandler(
            filters.VOICE | filters.AUDIO,
            self.handle_voice
        ))
    
    async def start(self, update, context):
        user_id = update.effective_user.id
        self.user_sessions[user_id] = {
            "created_at": datetime.now(),
            "commands_count": 0,
            "current_repo": None
        }
        await update.message.reply_text(
            "Голосовой ассистент активирован. Используйте /set_repo для указания репозитория."
        )
    
    async def handle_voice(self, update, context):
        # 1. Скачать файл
        voice_file = await update.message.voice.get_file()
        audio_path = f"/tmp/voice_{update.message.id}.ogg"
        await voice_file.download_to_drive(audio_path)
        
        # 2. Отправить в очередь обработки
        task_id = await self.process_queue.put({
            "user_id": update.effective_user.id,
            "audio_path": audio_path,
            "message_id": update.message.id
        })
        
        await update.message.reply_text(
            f"Обрабатываю команду (ID: {task_id[:8]})..."
        )

Ключевой момент: бот не обрабатывает аудио сам. Он только принимает файл и кладет в очередь (Redis + Celery). Обработка в отдельном воркере.

Сердце системы: пайплайн обработки голосовой команды

Вот где большинство проектов спотыкаются. Они делают линейную обработку: скачал → распознал → выполнил. А что если распознавание заняло 5 секунд? Пользователь уже ушел. Что если команда опасная (rm -rf /)? Нужна валидация.

Правильная архитектура — pipeline с контрольными точками:

# pipeline/processor.py
import torch
import torchaudio
from gigasr import GigaASRPipeline
from silero_tts import silero_tts
import subprocess
import tempfile
from typing import Dict, Optional

class VoiceCommandPipeline:
    def __init__(self):
        # Инициализация моделей
        self.stt_model = GigaASRPipeline.from_pretrained(
            "ai-forever/gigasr-v4",
            device="cuda" if torch.cuda.is_available() else "cpu"
        )
        self.tts_model = silero_tts(
            model_name="v4_ru",  # актуальная на 2026
            language="ru",
            speaker="aidar",
            device="cpu"  # TTS хорошо работает на CPU
        )
        
        # Кэш для повторяющихся команд
        self.command_cache = {}
    
    async def process(self, audio_path: str, user_context: Dict) -> Dict:
        """Основной пайплайн обработки"""
        results = {
            "original_audio": audio_path,
            "steps": {},
            "final_result": None
        }
        
        # Шаг 1: Конвертация формата (Telegram отправляет OGG)
        wav_path = await self._convert_to_wav(audio_path)
        results["steps"]["conversion"] = {"status": "success", "path": wav_path}
        
        # Шаг 2: Распознавание речи
        transcription = await self._transcribe_audio(wav_path)
        results["steps"]["transcription"] = {
            "status": "success", 
            "text": transcription,
            "confidence": 0.96  # GigaASR v4 дает метрики качества
        }
        
        # Шаг 3: Анализ интента (что хочет пользователь?)
        intent = await self._analyze_intent(transcription, user_context)
        results["steps"]["intent"] = intent
        
        # Шаг 4: Валидация команды (безопасность!)
        if not await self._validate_command(intent):
            results["final_result"] = {
                "type": "error",
                "message": "Команда не прошла валидацию безопасности"
            }
            return results
        
        # Шаг 5: Генерация конкретных команд
        commands = await self._generate_commands(intent, user_context)
        results["steps"]["commands_generated"] = commands
        
        # Шаг 6: Выполнение
        execution_results = []
        for cmd in commands["shell_commands"]:
            result = await self._execute_safe(cmd, user_context["current_repo"])
            execution_results.append(result)
        
        results["steps"]["execution"] = execution_results
        
        # Шаг 7: Генерация ответа
        tts_audio = await self._generate_response(
            execution_results, 
            intent["command_type"]
        )
        
        results["final_result"] = {
            "type": "success",
            "tts_audio": tts_audio,
            "text_summary": self._create_summary(execution_results),
            "commands_executed": [cmd["command"] for cmd in execution_results]
        }
        
        return results
    
    async def _analyze_intent(self, text: str, context: Dict) -> Dict:
        """Используем Gemini Flash для понимания, что хочет пользователь"""
        prompt = f"""
        Пользователь сказал: "{text}"
        Контекст: работает с репозиторием {context.get('current_repo', 'не указан')}
        
        Определи тип команды:
        - git_commit (создать коммит)
        - git_push (отправить изменения)
        - git_status (проверить статус)
        - jira_create (создать задачу)
        - jira_update (обновить задачу)
        - deployment (запустить деплой)
        - other (другое)
        
        Верни JSON:
        {{
            "command_type": "тип_команды",
            "parameters": {{"param": "value"}},
            "confidence": 0.95
        }}
        """
        
        # Здесь вызов Gemini 2.5 Flash API
        # response = await gemini_client.generate(prompt)
        # На практике используем библиотеку google-generativeai
        
        # Заглушка для примера
        return {
            "command_type": "git_commit",
            "parameters": {
                "message": "фикс бага в аутентификации",
                "files": "all",
                "amend": False
            },
            "confidence": 0.97
        }

Автокоммиты в Git: магия или опасность?

Самая спорная часть системы. Автоматические коммиты в Git могут испортить историю, создать конфликты, закоммитить чувствительные данные. Но при правильной настройке — это суперсила.

# git/autocommit.py
import subprocess
import os
from pathlib import Path
import re

class SafeGitAutocommit:
    def __init__(self, repo_path: str):
        self.repo_path = Path(repo_path).absolute()
        self.forbidden_patterns = [
            r'\.env$',
            r'secret',
            r'password',
            r'private_key',
            r'\.pem$',
            r'config/local\.',
            r'node_modules/'  # случайно не закоммитить
        ]
    
    async def create_commit(self, commit_message: str, author: str) -> Dict:
        """Безопасное создание коммита"""
        
        # 1. Проверить, есть ли изменения
        status_result = await self._run_git("status --porcelain")
        if not status_result["stdout"]:
            return {"success": False, "reason": "Нет изменений для коммита"}
        
        # 2. Проверить каждое измененное файл на запрещенные паттерны
        changed_files = status_result["stdout"].split('\n')
        for file_status in changed_files:
            if not file_status:
                continue
            # Формат: " M file.txt"
            filename = file_status[3:].strip()
            if self._is_forbidden(filename):
                return {
                    "success": False,
                    "reason": f"Файл {filename} содержит запрещенный паттерн"
                }
        
        # 3. Посмотреть diff (что именно меняем)
        diff_result = await self._run_git("diff --staged")
        if self._contains_sensitive_data(diff_result["stdout"]):
            return {"success": False, "reason": "Diff содержит чувствительные данные"}
        
        # 4. Добавить все файлы (кроме .gitignore)
        await self._run_git("add .")
        
        # 5. Создать коммит
        commit_cmd = f"commit -m '{commit_message}' --author='{author}'"
        commit_result = await self._run_git(commit_cmd)
        
        # 6. Вернуть хеш коммита
        hash_result = await self._run_git("log -1 --format=%H")
        
        return {
            "success": True,
            "commit_hash": hash_result["stdout"].strip(),
            "files_changed": len([f for f in changed_files if f]),
            "message": commit_message
        }
    
    def _is_forbidden(self, filename: str) -> bool:
        """Проверка файла на запрещенные паттерны"""
        for pattern in self.forbidden_patterns:
            if re.search(pattern, filename, re.IGNORECASE):
                return True
        return False
    
    async def _run_git(self, command: str) -> Dict:
        """Безопасный запуск git команды"""
        full_cmd = f"git {command}"
        try:
            process = await asyncio.create_subprocess_shell(
                full_cmd,
                cwd=self.repo_path,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            stdout, stderr = await process.communicate()
            
            return {
                "success": process.returncode == 0,
                "stdout": stdout.decode().strip(),
                "stderr": stderr.decode().strip(),
                "returncode": process.returncode
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

Интеграция с таск-трекером: Jira, Linear, ClickUp

Голосовые команды для управления задачами — это следующий уровень. "Перенеси задачу PROJ-123 в статус In Review, добавь комментарий 'Код готов к ревью'". Система должна:

• Распознать номер задачи (PROJ-123)
• Понять целевой статус (In Review)
• Извлечь текст комментария
• Выполнить через API трекера

Код для Jira (аналогично для других систем):

# integrations/jira_manager.py
from jira import JIRA
import re

class VoiceJiraManager:
    def __init__(self, url: str, email: str, api_token: str):
        self.client = JIRA(
            server=url,
            basic_auth=(email, api_token)
        )
        self.status_map = {
            "в работу": "In Progress",
            "на ревью": "In Review",
            "готово": "Done",
            "открыт": "To Do",
            "блокирует": "Blocked"
        }
    
    async def handle_voice_command(self, command_text: str) -> Dict:
        """Обработка голосовой команды для Jira"""
        # 1. Извлечь номер задачи
        task_match = re.search(r'([A-Z]+-\d+)', command_text)
        if not task_match:
            return {"success": False, "reason": "Не найден номер задачи"}
        
        task_key = task_match.group(1)
        
        # 2. Определить действие
        action = None
        if "перенеси" in command_text or "перемести" in command_text:
            action = "transition"
        elif "создай" in command_text:
            action = "create"
        elif "обнови" in command_text:
            action = "update"
        
        # 3. Для перехода — найти целевой статус
        if action == "transition":
            target_status = None
            for ru_status, en_status in self.status_map.items():
                if ru_status in command_text:
                    target_status = en_status
                    break
            
            if not target_status:
                return {"success": False, "reason": "Не понятен целевой статус"}
            
            # Выполнить переход
            return await self._transition_task(task_key, target_status)
        
        # 4. Для создания задачи — извлечь параметры
        if action == "create":
            # Используем LLM для извлечения полей
            extracted = await self._extract_task_fields(command_text)
            return await self._create_task(extracted)
        
        return {"success": False, "reason": "Неизвестное действие"}
    
    async def _transition_task(self, task_key: str, target_status: str):
        """Перевести задачу в другой статус"""
        try:
            issue = self.client.issue(task_key)
            transitions = self.client.transitions(issue)
            
            # Найти ID перехода для целевого статуса
            transition_id = None
            for t in transitions:
                if t['to']['name'].lower() == target_status.lower():
                    transition_id = t['id']
                    break
            
            if transition_id:
                self.client.transition_issue(issue, transition_id)
                return {
                    "success": True,
                    "task": task_key,
                    "new_status": target_status
                }
            else:
                return {"success": False, "reason": f"Невозможно перевести в {target_status}"}
        except Exception as e:
            return {"success": False, "reason": str(e)}

Типичные ошибки и как их избежать

Я видел десятки попыток сделать подобные системы. Вот что ломается чаще всего:

Деплой: куда и как размещать

Варианты от дешевого к мощному:

1. VPS за $10/мес: Только бот + очередь. STT и TTS через облачные API (Yandex SpeechKit, Google Speech-to-Text). LLM — Gemini Flash API. Минус: задержки из-за сетевых вызовов.
2. Сервер с GPU (от $100/мес): Локальный GigaASR и Silero TTS. LLM остаются облачными. Оптимальный баланс.
3. Полностью локальный (от $300/мес): Добавляем локальную LLM (Qwen2.5-Coder-32B или CodeLlama 34B). Нулевая зависимость от интернета, полная конфиденциальность.

Мой выбор на 2026: Hetzner AX161 с RTX 4090. $250/мес. Помещаются все модели. Задержка обработки команды: 1.5-2 секунды от голоса до выполнения.

Что дальше? Эволюция голосовых агентов

Система, которую мы построили, — это фундамент. Дальше можно добавлять:

• Мультиагентность: Разные агенты для разных типов команд (Git-агент, Jira-агент, Deployment-агент). Как в архитектуре для локальных голосовых агентов.
• Контекстную память: Система помнит, что вы делали вчера, и предлагает продолжение. "Вчера ты начал фиксить баг в auth-мидлваре — продолжить?"
• Визуальную обратную связь: Не только голос, но и скриншоты, графики, диаграммы в Telegram. "Покажи граф зависимостей этого модуля".
• Интеграцию с n8n: Для сложных workflow, как в локальном голосовом ассистенте с n8n.

Главный тренд 2026-2027: голосовые агенты перестают быть отдельными системами. Они становятся интерфейсом ко всей инфраструктуре разработки. Один голосовой агент управляет Git, CI/CD, мониторингом, развертыванием, трекерами задач.

Совет напоследок: начните с простого. Не пытайтесь сразу сделать систему с 20 интеграциями. Сначала Telegram бот + Git коммиты. Когда это работает стабильно — добавляйте Jira. Потом — деплой. Постепенно, но уверенно.

И помните: самая опасная команда в такой системе — не "rm -rf", а "задеплой в прод без тестов". Всегда оставляйте человеческое подтверждение для критических действий. Автоматизация должна помогать, а не заменять мозги.