基于AssemblyAI与Groq构建实时语音AI智能体：从语音识别到工具调用的全栈实践-平芜编程栈

1. 项目概述：当语音指令遇见高速推理AI

最近在捣鼓一个挺有意思的东西：用语音直接控制一个能联网、能思考、能执行任务的AI智能体。听起来有点像科幻电影里的场景，对吧？但实现起来，核心就靠两个现在特别火的服务：AssemblyAI和Groq。这个项目的目标很明确，就是让你动动嘴皮子，比如问一句“今天科技圈有什么大新闻？”，AI就能自动去网上搜一圈，整理成摘要告诉你；或者说“帮我查查明天从北京飞上海的航班，下午出发的”，它就能去航司网站扒信息。这背后，AssemblyAI负责把你说的话一字不差、甚至带着情绪和意图地“听”明白，而Groq则像给AI大脑装上了火箭引擎，让它能以惊人的速度“思考”并规划行动。

我之所以折腾这个，是因为看到大语言模型（LLM）和语音技术的结合点正在爆发。单纯的语音助手只能完成预设指令，而结合了强大推理能力和工具调用（Function Calling）的AI Agent，才是真正意义上的“智能助理”。这个项目非常适合开发者、产品经理，或者任何想探索下一代人机交互可能性的朋友。无论你是想做个酷炫的个人项目，还是为公司产品探索语音交互的新形态，这里面的思路和代码都能给你直接的参考。

2. 核心架构与工具选型解析

2.1 为什么是AssemblyAI + Groq？

这个组合不是随便选的，它背后是当前技术栈在“高精度”和“高速度”两个维度上的最优解之一。

AssemblyAI的核心价值在于其超精准的语音转文本（STT）和深度理解能力。市面上STT服务很多，但AssemblyAI在准确率，尤其是在嘈杂环境、多人对话、专业术语识别上表现突出。更重要的是，它不止于转写，还提供了一系列“理解”功能：

说话人分离（Speaker Diarization）：能自动区分录音中有几个人在说话，并标记出来。这对于构建会议助手类应用至关重要。
实体检测（Entity Detection）：自动识别转写文本中的人名、地点、日期、组织等。
情感分析（Sentiment Analysis）：判断说话者的情绪是积极、消极还是中性。
内容审核（Content Moderation）：自动检测不适当内容。

在这个语音控制AI Agent的场景里，我们最看重的是其极高的转写准确率和低延迟的流式转录（Real-time Streaming）能力。你总不希望AI把“订机票”听成“顶鸡票”吧？准确的文本是后续一切智能操作的基础。而流式转录意味着你可以像和真人对话一样，边说边处理，体验更自然。

Groq的杀手锏则是一个字：快。它使用了独特的LPU（Language Processing Unit）推理引擎，专门为自回归文本生成（就是LLM那种一个词一个词往外蹦的生成方式）做了极致优化。用Groq的API调用像Llama、Mixtral这样的开源大模型，速度通常是传统GPU云服务的数倍甚至数十倍，且响应时间极其稳定。

对于AI Agent来说，速度就是生命。一个Agent接到指令后，往往需要经过多轮“思考”（Chain-of-Thought）：理解指令、规划步骤、调用工具、处理结果、生成回复。每一步都需要调用LLM。如果每次调用都要等上好几秒，整个交互体验就会变得卡顿、令人沮丧。Groq的极速推理能力，使得复杂的多步任务能在眨眼间完成，让语音控制的AI Agent真正有了“实时”交互的可能。

注意：Groq本身不提供独家模型，它提供的是针对开源模型（如Llama 3、Mixtral、Gemma）的高速推理服务。你需要根据任务复杂度、成本和效果来选择合适的模型。例如，对于需要较强推理能力的Agent，Llama 3 70B Instruct是很好的选择；若追求极致的速度与成本平衡，Mixtral 8x7B Instruct可能更合适。

2.2 AI Agent的核心工作流设计

一个能听、会想、能干的语音AI Agent，其工作流是一个清晰的闭环。下图展示了从你开口说话到AI执行任务并反馈的核心步骤：

flowchart TD A[用户语音指令<br>“今天天气如何？”] --> B[AssemblyAI<br>实时语音转文本] B --> C[获得精准文本指令] C --> D[Groq (LLM)<br>理解意图与规划] D --> E{是否需要调用工具？} E -- 是 --> F[执行工具调用<br>（如：查询天气API）] F --> G[获取工具执行结果<br>（如：JSON格式天气数据）] G --> D E -- 否 --> H[Groq (LLM)<br>组织最终回复文本] H --> I[文本转语音 TTS<br>（可选，如：ElevenLabs）] I --> J[语音播报回复<br>“今天晴，25度”]

这个工作流的核心在于Groq驱动的“思考-行动”循环。LLM首先判断用户指令是否需要调用外部工具（如查询天气、搜索网络、计算等）。如果需要，它会根据预定义的工具描述，生成结构化的调用请求（通常是JSON）。我们的程序接收到这个请求后，去执行对应的函数（比如调用一个天气API），并将结果返回给LLM。LLM再根据这个结果，决定是继续调用其他工具，还是已经收集到足够信息，可以组织成最终的自然语言回复。这个过程可能循环多次，直到任务完成为止。

3. 分步实现与核心代码拆解

3.1 环境搭建与基础配置

首先，我们需要准备好战场。假设你使用Python，项目依赖的核心库大致如下：

# 创建虚拟环境是良好的习惯 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install assemblyai groq requests python-dotenv

接下来是获取并配置API密钥，这是与两大服务通信的通行证。

AssemblyAI：去其官网注册，在控制台找到你的API Key。
Groq：同样去官网注册，获取API Key。

为了安全，绝不将密钥硬编码在代码中。我们使用.env文件来管理：

# .env 文件 ASSEMBLYAI_API_KEY=你的_assemblyai_密钥 GROQ_API_KEY=你的_groq_密钥

然后在Python代码中加载它们：

# config.py import os from dotenv import load_dotenv load_dotenv() ASSEMBLYAI_API_KEY = os.getenv("ASSEMBLYAI_API_KEY") GROQ_API_KEY = os.getenv("GROQ_API_KEY") if not ASSEMBLYAI_API_KEY or not GROQ_API_KEY: raise ValueError("请在 .env 文件中设置 ASSEMBLYAI_API_KEY 和 GROQ_API_KEY")

3.2 语音捕获与实时转写模块

这是项目的“耳朵”。我们使用AssemblyAI的实时流式转录API。这里的关键是处理音频流和实时文本回调。

# speech_to_text.py import assemblyai as aai import threading import queue from config import ASSEMBLYAI_API_KEY class SpeechTranscriber: def __init__(self, on_transcript_callback): """ 初始化语音转写器 :param on_transcript_callback: 回调函数，每当有新的转写结果时被调用 """ aai.settings.api_key = ASSEMBLYAI_API_KEY self.callback = on_transcript_callback self.text_queue = queue.Queue() # 用于存放转写结果的队列 self.transcriber = None self.is_running = False def start(self): """开始监听麦克风并进行实时转写""" def on_open(session_opened): print("麦克风已打开，开始聆听...") def on_data(transcript: aai.RealtimeTranscript): if not transcript.text: return # 只有当一句话完整结束时才处理 if transcript.message_type == aai.RealtimeMessageType.FinalTranscript: final_text = transcript.text print(f"[用户说]: {final_text}") self.text_queue.put(final_text) # 将最终文本放入队列 def on_error(error): print(f"转写错误: {error}") def on_close(): print("连接关闭") # 创建实时转录器配置 self.transcriber = aai.RealtimeTranscriber( on_open=on_open, on_data=on_data, on_error=on_error, on_close=on_close, sample_rate=16000, ) self.is_running = True # 连接并开始转录 self.transcriber.connect() # 在一个单独的线程中处理队列中的文本，避免阻塞音频流 threading.Thread(target=self._process_queue, daemon=True).start() # 开始监听麦克风 self.transcriber.stream() def _process_queue(self): """处理转写文本队列的线程函数""" while self.is_running: try: text = self.text_queue.get(timeout=1) if self.callback: self.callback(text) # 调用回调函数，将文本传递给Agent except queue.Empty: continue def stop(self): """停止转写""" self.is_running = False if self.transcriber: self.transcriber.close()

实操心得：AssemblyAI的流式API有两种消息类型：PartialTranscript（中间结果）和FinalTranscript（最终结果）。在语音控制场景中，我们通常只处理FinalTranscript，以避免AI在用户还没说完话时就仓促响应，造成误触发。sample_rate设置为16000Hz是语音识别的常用采样率，兼容大多数麦克风。

3.3 AI Agent大脑：Groq与工具调用集成

这是项目的“大脑”。我们使用Groq提供的Llama 3 70B Instruct模型，并为其装备“工具”（函数调用能力）。

# ai_agent.py import json from groq import Groq from config import GROQ_API_KEY # 示例工具函数：获取天气 def get_current_weather(location: str, unit: str = "celsius"): """获取指定城市的当前天气情况。""" # 这里应该调用真实的天气API，如OpenWeatherMap # 为示例，我们返回模拟数据 print(f"[工具调用] 查询 {location} 的天气，单位: {unit}") weather_data = { "location": location, "temperature": 22, "unit": unit, "condition": "晴朗", "humidity": 65 } return json.dumps(weather_data) # 示例工具函数：执行计算 def calculator(expression: str): """执行一个数学计算表达式。""" print(f"[工具调用] 计算: {expression}") try: # 警告：在生产环境中使用eval有安全风险，此处仅作演示。 # 应使用更安全的表达式求值库，如 `ast.literal_eval` 或自定义解析器。 result = eval(expression) return json.dumps({"result": result, "expression": expression}) except Exception as e: return json.dumps({"error": str(e), "expression": expression}) class VoiceControlledAgent: def __init__(self): self.client = Groq(api_key=GROQ_API_KEY) # 定义可供LLM调用的工具列表 self.tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "获取某个城市的当前天气", "parameters": { "type": "object", "properties": { "location": {"type": "string", "description": "城市名称，例如：北京，上海"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位"} }, "required": ["location"], }, }, }, { "type": "function", "function": { "name": "calculator", "description": "执行一个数学计算", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式，例如：3 + 5 * 2"} }, "required": ["expression"], }, }, } ] # 工具名称到实际函数的映射 self.tool_functions = { "get_current_weather": get_current_weather, "calculator": calculator, } self.conversation_history = [] # 维护对话历史，使Agent有上下文记忆 def process_text(self, user_input: str): """处理用户输入文本，驱动Agent思考并行动""" print(f"\n[Agent处理] 用户指令: {user_input}") self.conversation_history.append({"role": "user", "content": user_input}) # 第一步：将用户输入和历史交给LLM，看它是否决定调用工具 response = self.client.chat.completions.create( model="llama3-70b-8192", # 使用Groq上的Llama 3 70B模型 messages=self.conversation_history, tools=self.tools, tool_choice="auto", # 让模型自动决定是否调用工具 ) response_message = response.choices[0].message self.conversation_history.append(response_message) # 记录LLM的响应 # 第二步：检查LLM是否要求调用工具 tool_calls = response_message.tool_calls if tool_calls: # 处理每一个工具调用 for tool_call in tool_calls: function_name = tool_call.function.name function_args = json.loads(tool_call.function.arguments) print(f"[Agent决策] 决定调用工具: {function_name}, 参数: {function_args}") # 找到对应的函数并执行 if function_name in self.tool_functions: function_to_call = self.tool_functions[function_name] function_response = function_to_call(**function_args) # 第三步：将工具执行结果返回给LLM，让它生成最终回复 self.conversation_history.append({ "role": "tool", "tool_call_id": tool_call.id, "name": function_name, "content": function_response, }) # 再次调用LLM，让它基于工具结果生成回复 second_response = self.client.chat.completions.create( model="llama3-70b-8192", messages=self.conversation_history, ) final_message = second_response.choices[0].message.content self.conversation_history.append({"role": "assistant", "content": final_message}) return final_message else: return f"错误：未知的工具 '{function_name}'" else: # 如果LLM没有调用工具，直接返回它的回复 final_message = response_message.content self.conversation_history.append({"role": "assistant", "content": final_message}) return final_message

核心解析：tool_choice="auto"是关键参数，它将是否调用工具、调用哪个工具的决定权交给了LLM。LLM会根据我们对工具的描述（description和parameters）来判断用户意图是否匹配某个工具。工具描述写得越清晰准确，LLM调用工具的准确率就越高。conversation_history维护了完整的对话上下文，使得Agent能记住之前的对话，实现多轮交互。

3.4 主程序循环与交互逻辑

最后，我们把“耳朵”和“大脑”连接起来，形成一个完整的交互循环。

# main.py from speech_to_text import SpeechTranscriber from ai_agent import VoiceControlledAgent import threading import time class VoiceAIAgentApp: def __init__(self): self.agent = VoiceControlledAgent() self.transcriber = SpeechTranscriber(on_transcript_callback=self.on_user_speech) self.is_active = True def on_user_speech(self, text: str): """接收到用户语音转写的文本后的回调函数""" if not text.strip(): return # 可以在这里添加一个唤醒词检测，例如只有以“小助手”开头才处理 # if text.startswith("小助手"): # text = text.replace("小助手", "").strip() print(f"\n{'='*40}") print(f"接收到指令: {text}") print(f"{'='*40}") # 调用Agent处理文本 try: response = self.agent.process_text(text) print(f"[Agent回复]: {response}") # 在这里可以集成TTS（如pyttsx3, ElevenLabs API）将回复语音播报出来 # self.speak(response) except Exception as e: print(f"处理指令时出错: {e}") response = "抱歉，我处理你的请求时遇到了点问题。" # self.speak(response) def speak(self, text: str): """文本转语音（示例，需安装相应库）""" # 示例1：使用系统自带TTS (pyttsx3) # import pyttsx3 # engine = pyttsx3.init() # engine.say(text) # engine.runAndWait() # 示例2：使用高质量TTS API (如ElevenLabs) # 需要安装 elevenlabs 库并配置API KEY # from elevenlabs import generate, play # audio = generate(text=text, voice="Rachel") # play(audio) pass def run(self): """启动应用""" print("启动语音控制AI Agent...") print("请开始说话（确保麦克风正常）。说'退出'或按Ctrl+C结束程序。") self.transcriber.start() # 主线程保持运行，监听退出命令 try: while self.is_active: # 这里可以监听其他命令，例如从控制台输入‘exit’退出 time.sleep(0.5) except KeyboardInterrupt: print("\n接收到中断信号，正在关闭...") finally: self.shutdown() def shutdown(self): """关闭应用""" self.is_active = False self.transcriber.stop() print("应用已关闭。") if __name__ == "__main__": app = VoiceAIAgentApp() app.run()

4. 高级功能扩展与优化实践

4.1 为Agent装备更多“技能”（工具）

一个强大的Agent取决于它有多少可用的工具。除了天气和计算，我们可以轻松扩展：

# tools.py import requests from datetime import datetime def search_web(query: str, max_results: int = 5): """使用搜索引擎API（如Serper Dev）搜索网络信息。""" # 需要注册Serper Dev等服务获取API Key url = "https://google.serper.dev/search" headers = {'X-API-KEY': 'YOUR_SERPER_API_KEY'} payload = {"q": query, "num": max_results} response = requests.post(url, headers=headers, json=payload) if response.status_code == 200: results = response.json().get('organic', []) summaries = [f"{r['title']}: {r['snippet']}" for r in results[:max_results]] return json.dumps({"query": query, "results": summaries}) return json.dumps({"error": "搜索失败"}) def get_current_time(location: str = None): """获取当前时间。如果提供地点，可尝试返回该时区时间（需时区API）。""" now = datetime.now() if location: # 此处简化处理，实际应调用时区API将地点转换为时区 return json.dumps({"location": location, "time": now.strftime("%Y-%m-%d %H:%M:%S"), "note": "地点功能待实现"}) return json.dumps({"time": now.strftime("%Y-%m-%d %H:%M:%S")}) def send_email(to: str, subject: str, body: str): """发送电子邮件（需要配置SMTP或邮件服务API）。""" # 使用smtplib或Resend等API实现 print(f"[工具调用] 模拟发送邮件给 {to}, 主题: {subject}") # ... 实际发送逻辑 return json.dumps({"status": "success", "message": f"邮件已发送至 {to}"}) # 在ai_agent.py的__init__中，将新工具添加到self.tools和self.tool_functions中

4.2 引入智能体框架（LangChain）进行专业化管理

当工具越来越多，工作流越来越复杂时，手动管理对话历史、工具调用链会变得繁琐。此时，引入像LangChain或LlamaIndex这样的框架是更优选择。它们提供了成熟的Agent、Chain、Memory等抽象，让构建复杂应用更规范。

# 使用LangChain重构Agent核心 (示例片段) from langchain_groq import ChatGroq from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.tools import Tool from langchain.prompts import ChatPromptTemplate from tools import get_current_weather, calculator, search_web # 导入工具函数 # 1. 初始化Groq LLM llm = ChatGroq(model="llama3-70b-8192", groq_api_key=GROQ_API_KEY, temperature=0) # 2. 将函数包装成LangChain Tool对象 tools = [ Tool( name="Weather", func=get_current_weather, description="获取某个城市的当前天气。输入应为一个城市名称。" ), Tool( name="Calculator", func=calculator, description="执行数学计算。输入应为一个有效的数学表达式，如 '3 + 5 * 2'。" ), Tool( name="WebSearch", func=search_web, description="搜索互联网获取最新信息。输入应为一个搜索查询词。" ) ] # 3. 创建提示词模板 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有帮助的语音控制AI助手。请根据用户需求，使用可用工具来回答问题。如果不需要工具，请直接回答。"), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) # 4. 创建Agent agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) # 5. 使用Agent执行 result = agent_executor.invoke({"input": "北京今天天气怎么样？然后计算一下华氏度是多少。"}) print(result["output"])

使用LangChain后，工具调用、错误处理、中间步骤（agent_scratchpad）都被框架优雅地管理起来，代码更清晰，也更容易扩展复杂的工作流（如多个Agent协作）。

4.3 性能优化与成本控制

在实际部署中，性能和成本是需要重点考虑的。

Groq模型选择：
- Llama 3 70B：能力最强，适合复杂推理和规划，但单次调用成本相对最高，速度在Groq上依然很快。
- Mixtral 8x7B：在速度、成本和能力上取得了很好的平衡，是许多生产应用的性价比之选。
- Gemma 7B：轻量级，速度极快，成本最低，适合对响应速度要求极高、但任务相对简单的场景。可以通过在提示词（Prompt）上下更多功夫来弥补其能力上的些许不足。
AssemblyAI优化：
- 使用流式端点（Realtime）：对于实时交互，必须使用流式API以获得最低延迟。
- 选择性开启增强功能：说话人分离、情感分析等功能会增加处理时间和成本。如果应用场景不需要，可以在创建RealtimeTranscriber时关闭它们。
- 设置合理的端点（Endpoint）：如果你的用户主要在某个区域，选择地理上最近的API端点（如wss://api.assemblyai.com/v2/realtime/ws?sample_rate=16000）可以减少网络延迟。
Agent层面的优化：
- 缓存（Caching）：对于相同或相似的查询（例如，短时间内多次询问同一城市天气），可以将LLM的回复或工具调用结果缓存起来，避免重复计算和API调用。
- 限制工具调用次数：在AgentExecutor中设置max_iterations参数，防止Agent陷入无限循环的“思考”中。
- 清晰的工具描述：这是提升工具调用准确率最有效且零成本的方法。描述要精确说明工具的用途、输入格式和输出含义。

5. 常见问题与实战排坑指南

在实际开发和测试中，你几乎一定会遇到下面这些问题。这里是我踩过坑后的经验总结。

5.1 语音转写准确率不理想

问题：在嘈杂环境或带有口音时，AssemblyAI转写错误较多。
排查与解决：
1. 检查音频质量：确保麦克风硬件正常，尝试使用外接麦克风。在代码中，可以尝试提高sample_rate（如44100），但需确认AssemblyAI支持。
2. 启用增强模型：AssemblyAI有专门的word_boost参数，可以传入你领域内的专业词汇（如产品名、人名）来提高识别率。
3. 预处理音频：在音频流送入AssemblyAI之前，可以先用本地库（如pydub）进行简单的降噪、增益标准化等预处理。
4. 使用“流式”而非“实时”：如果对延迟要求不是毫秒级，可以考虑使用普通的异步转录API，它可能使用了更复杂的模型，准确率更高。

5.2 Agent错误调用工具或拒绝调用工具

问题：用户说“上海天气”，Agent却调用了计算器，或者直接回答“我不知道怎么查天气”。
排查与解决：
1. 优化工具描述：这是最常见的原因。检查description和parameters的描述是否足够清晰、无歧义。例如，“获取天气”不如“获取指定城市名称的当前天气情况，包括温度、湿度和天气状况”来得明确。
2. 调整系统提示词（System Prompt）：在给LLM的指令中，明确要求它“积极使用提供的工具来回答问题”。可以强调“如果你需要实时信息或进行计算，请务必使用工具”。
3. 提供少量示例（Few-shot）：在对话历史（messages）的开头，插入一两个用户使用工具的成功示例，让LLM学会模式。
4. 调整温度（Temperature）参数：过高的temperature（如>0.8）会增加随机性，可能导致工具调用不稳定。对于需要精确工具调用的Agent，建议设置在0.1到0.3之间。

5.3 响应延迟过高，体验卡顿

问题：从说完话到听到回复，等待时间过长。
排查与解决：
1. 网络延迟：检查你的服务器或本地网络到Groq和AssemblyAI服务器的延迟。可以使用ping或traceroute简单测试。
2. 串行调用：确保你的代码没有不必要地串行等待。例如，语音转写、LLM思考、工具调用、TTS生成，在可能的情况下应使用异步编程（asyncio）来优化。
3. Groq模型选择：如果当前使用70B模型感觉慢，可以降级到8x7B或7B模型测试速度提升是否在可接受的性能损失范围内。
4. 流式输出（Streaming）：对于较长的LLM回复，可以启用Groq的流式响应，让回复内容逐词返回，并立即开始TTS的流式合成（如果TTS支持），实现“边想边说”的效果，极大提升感知速度。

5.4 上下文长度与记忆管理

问题：对话进行一段时间后，Agent似乎“忘记”了之前聊过的内容。
排查与解决：
1. 了解模型上下文窗口：Llama 3 70B的上下文窗口是8192个token。你需要管理conversation_history的总长度，避免超出。
2. 实现摘要式记忆：当对话历史过长时，可以调用一次LLM，让它对之前的对话进行摘要，然后用摘要替换掉旧的历史记录，只保留最近几条原始对话。这是一种平衡记忆和成本/长度的经典方法。
3. 使用向量数据库进行长期记忆：对于需要记忆跨会话信息（如用户偏好）的场景，可以将重要的对话片段转换成向量，存入像Chroma、Pinecone这样的向量数据库。当新对话开始时，先检索相关的历史记忆，作为上下文提供给LLM。

这个项目就像一个乐高积木，AssemblyAI和Groq是两个核心的高质量部件。围绕它们，你可以根据需求添加各种工具（天气、日历、邮件、智能家居控制）、集成不同的TTS服务，甚至结合本地模型来保护隐私。我个人的体会是，构建过程本身就是一个对AI Agent技术栈的深度理解之旅，从语音接口到推理引擎，再到工具编排，每一步的优化都能带来体验上的显著提升。最关键的是，从一个小而具体的功能开始，比如先做好“查天气”，跑通整个流程，然后再逐步添加新工具，这样能持续获得正反馈，避免一开始就陷入过于复杂的泥潭。