#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 多进程音频录音系统 基于进程隔离的音频处理架构 """ import os import sys import argparse import json import time from typing import Dict, Any def check_dependencies(): """检查系统依赖""" missing_deps = [] try: import pyaudio except ImportError: missing_deps.append("pyaudio") try: import numpy except ImportError: missing_deps.append("numpy") try: import requests except ImportError: missing_deps.append("requests") try: import websockets except ImportError: missing_deps.append("websockets") if missing_deps: print("❌ 缺少以下依赖库:") for dep in missing_deps: print(f" - {dep}") print("\n请运行以下命令安装:") print(f"pip install {' '.join(missing_deps)}") return False return True def check_environment(): """检查运行环境""" print("🔍 检查运行环境...") # 检查Python版本 python_version = sys.version_info if python_version.major < 3 or (python_version.major == 3 and python_version.minor < 7): print(f"❌ Python版本过低: {python_version.major}.{python_version.minor}") print("需要Python 3.7或更高版本") return False print(f"✅ Python版本: {python_version.major}.{python_version.minor}.{python_version.micro}") # 检查操作系统 import platform system = platform.system().lower() print(f"✅ 操作系统: {system}") # 检查音频设备 try: import pyaudio audio = pyaudio.PyAudio() device_count = audio.get_device_count() print(f"✅ 音频设备数量: {device_count}") if device_count == 0: print("❌ 未检测到音频设备") return False audio.terminate() except Exception as e: print(f"❌ 音频设备检查失败: {e}") return False # 检查网络连接 try: import requests response = requests.get("https://www.baidu.com", timeout=5) print("✅ 网络连接正常") except: print("⚠️ 网络连接可能有问题，会影响在线AI功能") # 检查API密钥 api_key = os.environ.get("ARK_API_KEY") if api_key: print("✅ ARK_API_KEY 已设置") else: print("⚠️ ARK_API_KEY 未设置，大语言模型功能将被禁用") print(" 请运行: export ARK_API_KEY='your_api_key_here'") return True def list_characters(): """列出可用角色""" characters_dir = os.path.join(os.path.dirname(__file__), "characters") if not os.path.exists(characters_dir): print("❌ 角色目录不存在") return characters = [] for file in os.listdir(characters_dir): if file.endswith('.json'): character_name = file[:-5] config_file = os.path.join(characters_dir, file) try: with open(config_file, 'r', encoding='utf-8') as f: config = json.load(f) name = config.get('name', character_name) desc = config.get('description', '无描述') characters.append(f"{character_name}: {name} - {desc}") except: characters.append(f"{character_name}: 配置文件读取失败") if characters: print("🎭 可用角色列表:") for char in characters: print(f" - {char}") else: print("❌ 未找到任何角色配置文件") def create_sample_config(): """创建示例配置文件""" config = { "system": { "max_queue_size": 1000, "process_timeout": 30, "heartbeat_interval": 1.0, "log_level": "INFO" }, "audio": { "sample_rate": 16000, "channels": 1, "chunk_size": 1024, "format": "paInt16" }, "recording": { "min_duration": 2.0, "max_duration": 30.0, "silence_threshold": 3.0, "pre_record_duration": 2.0 }, "processing": { "enable_asr": True, "enable_llm": True, "enable_tts": True, "character": "libai", "max_tokens": 50 }, "detection": { "zcr_min": 2400, "zcr_max": 12000, "consecutive_silence_count": 30, "max_zcr_history": 30 }, "playback": { "buffer_size": 1000, "show_progress": True, "progress_interval": 100, "chunk_size": 512 } } config_file = "config.json" try: with open(config_file, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) print(f"✅ 示例配置文件已创建: {config_file}") except Exception as e: print(f"❌ 创建配置文件失败: {e}") def main(): """主函数""" parser = argparse.ArgumentParser( description='多进程音频录音系统', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: python multiprocess_recorder.py # 使用默认角色 python multiprocess_recorder.py -c zhubajie # 指定角色 python multiprocess_recorder.py -l # 列出角色 python multiprocess_recorder.py --create-config # 创建配置文件 """ ) parser.add_argument('--character', '-c', type=str, default='libai', help='选择角色 (默认: libai)') parser.add_argument('--list-characters', '-l', action='store_true', help='列出所有可用角色') parser.add_argument('--config', type=str, help='配置文件路径') parser.add_argument('--create-config', action='store_true', help='创建示例配置文件') parser.add_argument('--check-env', action='store_true', help='检查运行环境') parser.add_argument('--verbose', '-v', action='store_true', help='详细输出') args = parser.parse_args() # 显示欢迎信息 print("🚀 多进程音频录音系统") print("=" * 60) # 检查依赖 if not check_dependencies(): sys.exit(1) # 创建配置文件 if args.create_config: create_sample_config() return # 检查环境 if args.check_env: check_environment() return # 列出角色 if args.list_characters: list_characters() return # 检查characters目录 characters_dir = os.path.join(os.path.dirname(__file__), "characters") if not os.path.exists(characters_dir): print(f"⚠️ 角色目录不存在: {characters_dir}") print("请确保characters目录存在并包含角色配置文件") # 检查指定角色 character_file = os.path.join(characters_dir, f"{args.character}.json") if not os.path.exists(character_file): print(f"⚠️ 角色文件不存在: {character_file}") print(f"可用角色:") list_characters() return print(f"🎭 当前角色: {args.character}") print("🎯 系统特点:") print(" - 多进程架构：输入输出完全隔离") print(" - 零切换延迟：无需音频设备重置") print(" - 实时响应：并行处理录音和播放") print(" - 智能检测：基于ZCR的语音识别") print(" - 流式TTS：实时音频生成和播放") print(" - 角色扮演：支持多种AI角色") print("=" * 60) # 显示使用说明 print("📖 使用说明:") print(" - 检测到语音自动开始录音") print(" - 持续静音3秒自动结束录音") print(" - 录音完成后自动处理和播放") print(" - 按 Ctrl+C 退出") print("=" * 60) # 加载配置 config = None if args.config: try: with open(args.config, 'r', encoding='utf-8') as f: config = json.load(f) print(f"📋 加载配置文件: {args.config}") except Exception as e: print(f"⚠️ 配置文件加载失败: {e}") print("使用默认配置") try: # 导入控制系统 from control_system import ControlSystem # 创建控制系统 control_system = ControlSystem(config) # 设置角色 control_system.config['processing']['character'] = args.character # 设置日志级别 if args.verbose: control_system.config['system']['log_level'] = "DEBUG" # 启动系统（自动校准和监听） print("🚀 启动系统（包含自动校准和监听）...") control_system.start(auto_calibration=True, auto_monitoring=True) except KeyboardInterrupt: print("\n👋 用户中断") except Exception as e: print(f"❌ 系统启动失败: {e}") if args.verbose: import traceback traceback.print_exc() finally: print("👋 系统退出") if __name__ == "__main__": main()