文章介绍了Qwen-Omni全模态大模型的快速上手方法，重点展示了如何使用Python实现音频输入、文本识别、回答生成和语音输出的完整流程。提供了详细的代码示例，包括单个音频查询处理和批量处理功能，并展示了实际运行结果。开发者可以通过这些代码快速构建基于Qwen-Omni的音频问答系统。

---

音频+文本输入

• 仅支持输入一个音频文件；
• 文件大小

• Qwen3-Omni-Flash：不能超过 100MB，时长最长 20 分钟。
• Qwen-Omni-Turbo：不能超过 10MB，时长最长 3 分钟。

以下示例代码以传入为例，当前只支持以流式输出的方式进行调用。

下面是一个端到端的音频问答处理函数，能够：

1. 识别音频内容（语音转文本）
2. 生成文本回答
3. 生成语音回复（文本转语音）
4. 保存语音文件

def process_audio_query(audio_url, query_text, output_audio_path="response_audio.wav", voice="Cherry", model="qwen3-omni-flash"):    """    处理音频查询并返回文本回复和保存音频文件    Args:        audio_url (str): 音频文件的URL        query_text (str): 查询文本        output_audio_path (str): 输出音频文件的路径        voice (str): 语音音色        model (str): 使用的模型    Returns:        dict: 包含文本回复和音频文件路径的字典    """    try:        # 创建聊天补全请求        completion = client.chat.completions.create(            model=model,            messages=[                {                    "role": "user",                    "content": [                        {                            "type": "input_audio",                            "input_audio": {                                "data": audio_url,                                "format": "wav",                            },                        },                        {"type": "text", "text": query_text},                    ],                },            ],            modalities=["text", "audio"],            audio={"voice": voice, "format": "wav"},            stream=True,            stream_options={"include_usage": True},        )        # 初始化变量来收集响应        collected_text = ""        audio_base64_string = ""  # 用于累积音频的base64数据        usage_info = None        print("=== 开始接收响应 ===")        # 处理流式响应        for chunk in completion:            # 处理文本内容            if (chunk.choices and                 chunk.choices[0].delta and                 chunk.choices[0].delta.content):                text_content = chunk.choices[0].delta.content                collected_text += text_content                print(text_content, end="", flush=True)            # 处理音频数据 - 使用您提供的代码逻辑            if (chunk.choices and                 hasattr(chunk.choices[0].delta, "audio") and                 chunk.choices[0].delta.audio):                # 从音频数据中获取base64编码的字符串并拼接                audio_data = chunk.choices[0].delta.audio.get("data", "")                if audio_data:                    audio_base64_string += audio_data            # 保存使用量信息            if hasattr(chunk, 'usage') and chunk.usage:                usage_info = chunk.usage        # 打印完整的文本回复        print(f"\n\n=== 完整文本回复 ===")        print(collected_text)        # 保存音频文件 - 使用您提供的优化方法        if audio_base64_string:            try:                # 将base64字符串解码为字节数据                wav_bytes = base64.b64decode(audio_base64_string)                # 将字节数据转换为numpy数组，使用16位整数格式（WAV标准格式）                audio_np = np.frombuffer(wav_bytes, dtype=np.int16)                # 使用soundfile库将音频数据保存为WAV文件                # 参数：文件名、音频数据、采样率（24kHz）                sf.write(output_audio_path, audio_np, samplerate=24000)                print(f"\n=== 音频回复已保存 ===")                print(f"音频文件: {output_audio_path}")                print(f"文件大小: {len(wav_bytes)} 字节")                print(f"音频采样点数: {len(audio_np)}")                # 验证文件是否成功创建                if os.path.exists(output_audio_path) and os.path.getsize(output_audio_path) > 0:                    print("✓ 音频文件保存成功")                else:                    print("✗ 音频文件保存失败")                    output_audio_path = None            except Exception as audio_error:                print(f"\n音频文件保存失败: {audio_error}")                output_audio_path = None        else:            print(f"\n=== 未接收到音频数据 ===")            output_audio_path = None        # 打印使用量统计信息        if usage_info:            print(f"\n=== 使用量统计 ===")            print(f"Token使用量: {usage_info}")        # 返回结果        return {            "text_response": collected_text,            "audio_file_path": output_audio_path,            "audio_data_size": len(audio_base64_string) if audio_base64_string else 0,            "usage_info": usage_info        }    except Exception as e:        print(f"\n=== 发生错误 ===")        print(f"错误信息: {e}")        return {            "text_response": "",            "audio_file_path": None,            "audio_data_size": 0,            "usage_info": None,            "error": str(e)        }

示例用法

公网URL的音频：

audio_url = "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250211/tixcef/cherry.wav"query_text = "这段音频在说什么"output_path = "audio_response.wav"print("开始处理音频查询...")result = process_audio_query(    audio_url=audio_url,    query_text=query_text,    output_audio_path=output_path,    voice="Cherry")print(f"\n=== 处理结果 ===")print(f"文本回复: {result['text_response']}")print(f"音频文件: {result['audio_file_path']}")print(f"音频大小: {result['audio_data_size']} 字节")

输出为：

开始处理音频查询...=== 开始接收响应 ===这段音频描述了一幅海滩上的画面：一个女人和一只狗在玩耍，背景是海浪拍打沙滩，天空发白，可能是日出或日落时分。说话者还邀请对方继续提问。=== 完整文本回复 ===这段音频描述了一幅海滩上的画面：一个女人和一只狗在玩耍，背景是海浪拍打沙滩，天空发白，可能是日出或日落时分。说话者还邀请对方继续提问。=== 音频回复已保存 ===音频文件: audio_response.wav文件大小: 645120 字节音频采样点数: 322560✓ 音频文件保存成功=== 使用量统计 ===Token使用量: CompletionUsage(completion_tokens=214, prompt_tokens=722, total_tokens=936, completion_tokens_details=CompletionTokensDetails(accepted_prediction_tokens=None, audio_tokens=168, reasoning_tokens=None, rejected_prediction_tokens=None, text_tokens=46), prompt_tokens_details=PromptTokensDetails(audio_tokens=415, cached_tokens=None, text_tokens=307))=== 处理结果 ===文本回复: 这段音频描述了一幅海滩上的画面：一个女人和一只狗在玩耍，背景是海浪拍打沙滩，天空发白，可能是日出或日落时分。说话者还邀请对方继续提问。音频文件: audio_response.wav音频大小: 860160 字节

批量处理多个音频查询

def batch_process_audio_queries(queries_list, output_dir="audio_outputs"):    """    批量处理多个音频查询    Args:        queries_list (list): 包含多个查询的列表，每个查询是字典格式        output_dir (str): 输出目录    Returns:        list: 处理结果列表    """    os.makedirs(output_dir, exist_ok=True)    results = []    for i, query in enumerate(queries_list):        print(f"\n处理第 {i+1} 个查询...")        output_path = os.path.join(output_dir, f"response_{i+1}.wav")        result = process_audio_query(            audio_url=query.get("audio_url"),            query_text=query.get("query_text"),            output_audio_path=output_path,            voice=query.get("voice", "Cherry")        )        results.append(result)    return results

示例

queries = [    {        "audio_url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250211/tixcef/cherry.wav",        "query_text": "这段音频的内容是什么？",        "voice": "Cherry"    },    {        "audio_url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250214/pijhos/welcome.mp3?spm=a2c4g.11186623.0.0.65d33748bfR1N4&file=welcome.mp3",         "query_text": "请总结这段音频的要点",        "voice": "Cherry"    }]batch_results = batch_process_audio_queries(queries)

输出为：

处理第 1 个查询...=== 开始接收响应 ===这段音频描述了一幅海滩上的画面：一个女人和一只狗在玩耍，背景是海浪拍打沙滩，天空发白，可能是日出或日落时分。说话者还表示，如果还有其他想知道的内容，可以继续提问。=== 完整文本回复 ===这段音频描述了一幅海滩上的画面：一个女人和一只狗在玩耍，背景是海浪拍打沙滩，天空发白，可能是日出或日落时分。说话者还表示，如果还有其他想知道的内容，可以继续提问。=== 音频回复已保存 ===音频文件: audio_outputs\response_1.wav文件大小: 771840 字节音频采样点数: 385920✓ 音频文件保存成功=== 使用量统计 ===Token使用量: CompletionUsage(completion_tokens=254, prompt_tokens=723, total_tokens=977, completion_tokens_details=CompletionTokensDetails(accepted_prediction_tokens=None, audio_tokens=201, reasoning_tokens=None, rejected_prediction_tokens=None, text_tokens=53), prompt_tokens_details=PromptTokensDetails(audio_tokens=415, cached_tokens=None, text_tokens=308))处理第 2 个查询...=== 开始接收响应 ===这段音频是阿里云的欢迎语，没有其他内容。=== 完整文本回复 ===这段音频是阿里云的欢迎语，没有其他内容。=== 音频回复已保存 ===音频文件: audio_outputs\response_2.wav文件大小: 165120 字节音频采样点数: 82560✓ 音频文件保存成功=== 使用量统计 ===Token使用量: CompletionUsage(completion_tokens=56, prompt_tokens=353, total_tokens=409, completion_tokens_details=CompletionTokensDetails(accepted_prediction_tokens=None, audio_tokens=43, reasoning_tokens=None, rejected_prediction_tokens=None, text_tokens=13), prompt_tokens_details=PromptTokensDetails(audio_tokens=44, cached_tokens=None, text_tokens=309))

​最后

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。

我整理出这套 AI 大模型突围资料包：

• ✅AI大模型学习路线图
• ✅Agent行业报告
• ✅100集大模型视频教程
• ✅大模型书籍PDF
• ✅DeepSeek教程
• ✅AI产品经理入门资料

完整的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
​​

为什么说现在普通人就业/升职加薪的首选是AI大模型？

人工智能技术的爆发式增长，正以不可逆转之势重塑就业市场版图。从DeepSeek等国产大模型引发的科技圈热议，到全国两会关于AI产业发展的政策聚焦，再到招聘会上排起的长队，AI的热度已从技术领域渗透到就业市场的每一个角落。

智联招聘的最新数据给出了最直观的印证：2025年2月，AI领域求职人数同比增幅突破200% ，远超其他行业平均水平；整个人工智能行业的求职增速达到33.4%，位居各行业榜首，其中人工智能工程师岗位的求职热度更是飙升69.6%。

AI产业的快速扩张，也让人才供需矛盾愈发突出。麦肯锡报告明确预测，到2030年中国AI专业人才需求将达600万人，人才缺口可能高达400万人，这一缺口不仅存在于核心技术领域，更蔓延至产业应用的各个环节。

​​

资料包有什么？

①从入门到精通的全套视频教程⑤⑥

包含提示词工程、RAG、Agent等技术点

② AI大模型学习路线图（还有视频解说）

全过程AI大模型学习路线

③学习电子书籍和技术文档

市面上的大模型书籍确实太多了，这些是我精选出来的

④各大厂大模型面试题目详解

⑤ 这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

所有的视频教程由智泊AI老师录制，且资料与智泊AI共享，相互补充。这份学习大礼包应该算是现在最全面的大模型学习资料了。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

智泊AI始终秉持着“让每个人平等享受到优质教育资源”的育人理念‌，通过动态追踪大模型开发、数据标注伦理等前沿技术趋势‌，构建起"前沿课程+智能实训+精准就业"的高效培养体系。

课堂上不光教理论，还带着学员做了十多个真实项目。学员要亲自上手搞数据清洗、模型调优这些硬核操作，把课本知识变成真本事‌！

​​​​

如果说你是以下人群中的其中一类，都可以来智泊AI学习人工智能，找到高薪工作，一次小小的“投资”换来的是终身受益！

应届毕业生‌：无工作经验但想要系统学习AI大模型技术，期待通过实战项目掌握核心技术。

零基础转型‌：非技术背景但关注AI应用场景，计划通过低代码工具实现“AI+行业”跨界‌。

业务赋能 ‌突破瓶颈：传统开发者（Java/前端等）学习Transformer架构与LangChain框架，向AI全栈工程师转型‌。

👉获取方式：

😝有需要的小伙伴，可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓**

​