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的杀手锏则是一个字快。它使用了独特的LPULanguage Processing Unit推理引擎专门为自回归文本生成就是LLM那种一个词一个词往外蹦的生成方式做了极致优化。用Groq的API调用像Llama、Mixtral这样的开源大模型速度通常是传统GPU云服务的数倍甚至数十倍且响应时间极其稳定。对于AI Agent来说速度就是生命。一个Agent接到指令后往往需要经过多轮“思考”Chain-of-Thought理解指令、规划步骤、调用工具、处理结果、生成回复。每一步都需要调用LLM。如果每次调用都要等上好几秒整个交互体验就会变得卡顿、令人沮丧。Groq的极速推理能力使得复杂的多步任务能在眨眼间完成让语音控制的AI Agent真正有了“实时”交互的可能。注意Groq本身不提供独家模型它提供的是针对开源模型如Llama 3、Mixtral、Gemma的高速推理服务。你需要根据任务复杂度、成本和效果来选择合适的模型。例如对于需要较强推理能力的AgentLlama 3 70B Instruct是很好的选择若追求极致的速度与成本平衡Mixtral 8x7B Instruct可能更合适。2.2 AI Agent的核心工作流设计一个能听、会想、能干的语音AI Agent其工作流是一个清晰的闭环。下图展示了从你开口说话到AI执行任务并反馈的核心步骤flowchart TD A[用户语音指令br“今天天气如何”] -- B[AssemblyAIbr实时语音转文本] 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[文本转语音 TTSbr可选如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_openon_open, on_dataon_data, on_erroron_error, on_closeon_close, sample_rate16000, ) self.is_running True # 连接并开始转录 self.transcriber.connect() # 在一个单独的线程中处理队列中的文本避免阻塞音频流 threading.Thread(targetself._process_queue, daemonTrue).start() # 开始监听麦克风 self.transcriber.stream() def _process_queue(self): 处理转写文本队列的线程函数 while self.is_running: try: text self.text_queue.get(timeout1) 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_keyGROQ_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( modelllama3-70b-8192, # 使用Groq上的Llama 3 70B模型 messagesself.conversation_history, toolsself.tools, tool_choiceauto, # 让模型自动决定是否调用工具 ) 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( modelllama3-70b-8192, messagesself.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_choiceauto是关键参数它将是否调用工具、调用哪个工具的决定权交给了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_callbackself.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(texttext, voiceRachel) # play(audio) pass def run(self): 启动应用 print(启动语音控制AI Agent...) print(请开始说话确保麦克风正常。说退出或按CtrlC结束程序。) 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, headersheaders, jsonpayload) 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(modelllama3-70b-8192, groq_api_keyGROQ_API_KEY, temperature0) # 2. 将函数包装成LangChain Tool对象 tools [ Tool( nameWeather, funcget_current_weather, description获取某个城市的当前天气。输入应为一个城市名称。 ), Tool( nameCalculator, funccalculator, description执行数学计算。输入应为一个有效的数学表达式如 3 5 * 2。 ), Tool( nameWebSearch, funcsearch_web, description搜索互联网获取最新信息。输入应为一个搜索查询词。 ) ] # 3. 创建提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的语音控制AI助手。请根据用户需求使用可用工具来回答问题。如果不需要工具请直接回答。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 创建Agent agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 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_rate16000可以减少网络延迟。Agent层面的优化缓存Caching对于相同或相似的查询例如短时间内多次询问同一城市天气可以将LLM的回复或工具调用结果缓存起来避免重复计算和API调用。限制工具调用次数在AgentExecutor中设置max_iterations参数防止Agent陷入无限循环的“思考”中。清晰的工具描述这是提升工具调用准确率最有效且零成本的方法。描述要精确说明工具的用途、输入格式和输出含义。5. 常见问题与实战排坑指南在实际开发和测试中你几乎一定会遇到下面这些问题。这里是我踩过坑后的经验总结。5.1 语音转写准确率不理想问题在嘈杂环境或带有口音时AssemblyAI转写错误较多。排查与解决检查音频质量确保麦克风硬件正常尝试使用外接麦克风。在代码中可以尝试提高sample_rate如44100但需确认AssemblyAI支持。启用增强模型AssemblyAI有专门的word_boost参数可以传入你领域内的专业词汇如产品名、人名来提高识别率。预处理音频在音频流送入AssemblyAI之前可以先用本地库如pydub进行简单的降噪、增益标准化等预处理。使用“流式”而非“实时”如果对延迟要求不是毫秒级可以考虑使用普通的异步转录API它可能使用了更复杂的模型准确率更高。5.2 Agent错误调用工具或拒绝调用工具问题用户说“上海天气”Agent却调用了计算器或者直接回答“我不知道怎么查天气”。排查与解决优化工具描述这是最常见的原因。检查description和parameters的描述是否足够清晰、无歧义。例如“获取天气”不如“获取指定城市名称的当前天气情况包括温度、湿度和天气状况”来得明确。调整系统提示词System Prompt在给LLM的指令中明确要求它“积极使用提供的工具来回答问题”。可以强调“如果你需要实时信息或进行计算请务必使用工具”。提供少量示例Few-shot在对话历史messages的开头插入一两个用户使用工具的成功示例让LLM学会模式。调整温度Temperature参数过高的temperature如0.8会增加随机性可能导致工具调用不稳定。对于需要精确工具调用的Agent建议设置在0.1到0.3之间。5.3 响应延迟过高体验卡顿问题从说完话到听到回复等待时间过长。排查与解决网络延迟检查你的服务器或本地网络到Groq和AssemblyAI服务器的延迟。可以使用ping或traceroute简单测试。串行调用确保你的代码没有不必要地串行等待。例如语音转写、LLM思考、工具调用、TTS生成在可能的情况下应使用异步编程asyncio来优化。Groq模型选择如果当前使用70B模型感觉慢可以降级到8x7B或7B模型测试速度提升是否在可接受的性能损失范围内。流式输出Streaming对于较长的LLM回复可以启用Groq的流式响应让回复内容逐词返回并立即开始TTS的流式合成如果TTS支持实现“边想边说”的效果极大提升感知速度。5.4 上下文长度与记忆管理问题对话进行一段时间后Agent似乎“忘记”了之前聊过的内容。排查与解决了解模型上下文窗口Llama 3 70B的上下文窗口是8192个token。你需要管理conversation_history的总长度避免超出。实现摘要式记忆当对话历史过长时可以调用一次LLM让它对之前的对话进行摘要然后用摘要替换掉旧的历史记录只保留最近几条原始对话。这是一种平衡记忆和成本/长度的经典方法。使用向量数据库进行长期记忆对于需要记忆跨会话信息如用户偏好的场景可以将重要的对话片段转换成向量存入像Chroma、Pinecone这样的向量数据库。当新对话开始时先检索相关的历史记忆作为上下文提供给LLM。这个项目就像一个乐高积木AssemblyAI和Groq是两个核心的高质量部件。围绕它们你可以根据需求添加各种工具天气、日历、邮件、智能家居控制、集成不同的TTS服务甚至结合本地模型来保护隐私。我个人的体会是构建过程本身就是一个对AI Agent技术栈的深度理解之旅从语音接口到推理引擎再到工具编排每一步的优化都能带来体验上的显著提升。最关键的是从一个小而具体的功能开始比如先做好“查天气”跑通整个流程然后再逐步添加新工具这样能持续获得正反馈避免一开始就陷入过于复杂的泥潭。
基于AssemblyAI与Groq构建实时语音AI智能体:从语音识别到工具调用的全栈实践
发布时间:2026/5/27 12:26:21
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的杀手锏则是一个字快。它使用了独特的LPULanguage Processing Unit推理引擎专门为自回归文本生成就是LLM那种一个词一个词往外蹦的生成方式做了极致优化。用Groq的API调用像Llama、Mixtral这样的开源大模型速度通常是传统GPU云服务的数倍甚至数十倍且响应时间极其稳定。对于AI Agent来说速度就是生命。一个Agent接到指令后往往需要经过多轮“思考”Chain-of-Thought理解指令、规划步骤、调用工具、处理结果、生成回复。每一步都需要调用LLM。如果每次调用都要等上好几秒整个交互体验就会变得卡顿、令人沮丧。Groq的极速推理能力使得复杂的多步任务能在眨眼间完成让语音控制的AI Agent真正有了“实时”交互的可能。注意Groq本身不提供独家模型它提供的是针对开源模型如Llama 3、Mixtral、Gemma的高速推理服务。你需要根据任务复杂度、成本和效果来选择合适的模型。例如对于需要较强推理能力的AgentLlama 3 70B Instruct是很好的选择若追求极致的速度与成本平衡Mixtral 8x7B Instruct可能更合适。2.2 AI Agent的核心工作流设计一个能听、会想、能干的语音AI Agent其工作流是一个清晰的闭环。下图展示了从你开口说话到AI执行任务并反馈的核心步骤flowchart TD A[用户语音指令br“今天天气如何”] -- B[AssemblyAIbr实时语音转文本] 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[文本转语音 TTSbr可选如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_openon_open, on_dataon_data, on_erroron_error, on_closeon_close, sample_rate16000, ) self.is_running True # 连接并开始转录 self.transcriber.connect() # 在一个单独的线程中处理队列中的文本避免阻塞音频流 threading.Thread(targetself._process_queue, daemonTrue).start() # 开始监听麦克风 self.transcriber.stream() def _process_queue(self): 处理转写文本队列的线程函数 while self.is_running: try: text self.text_queue.get(timeout1) 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_keyGROQ_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( modelllama3-70b-8192, # 使用Groq上的Llama 3 70B模型 messagesself.conversation_history, toolsself.tools, tool_choiceauto, # 让模型自动决定是否调用工具 ) 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( modelllama3-70b-8192, messagesself.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_choiceauto是关键参数它将是否调用工具、调用哪个工具的决定权交给了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_callbackself.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(texttext, voiceRachel) # play(audio) pass def run(self): 启动应用 print(启动语音控制AI Agent...) print(请开始说话确保麦克风正常。说退出或按CtrlC结束程序。) 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, headersheaders, jsonpayload) 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(modelllama3-70b-8192, groq_api_keyGROQ_API_KEY, temperature0) # 2. 将函数包装成LangChain Tool对象 tools [ Tool( nameWeather, funcget_current_weather, description获取某个城市的当前天气。输入应为一个城市名称。 ), Tool( nameCalculator, funccalculator, description执行数学计算。输入应为一个有效的数学表达式如 3 5 * 2。 ), Tool( nameWebSearch, funcsearch_web, description搜索互联网获取最新信息。输入应为一个搜索查询词。 ) ] # 3. 创建提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的语音控制AI助手。请根据用户需求使用可用工具来回答问题。如果不需要工具请直接回答。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 创建Agent agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 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_rate16000可以减少网络延迟。Agent层面的优化缓存Caching对于相同或相似的查询例如短时间内多次询问同一城市天气可以将LLM的回复或工具调用结果缓存起来避免重复计算和API调用。限制工具调用次数在AgentExecutor中设置max_iterations参数防止Agent陷入无限循环的“思考”中。清晰的工具描述这是提升工具调用准确率最有效且零成本的方法。描述要精确说明工具的用途、输入格式和输出含义。5. 常见问题与实战排坑指南在实际开发和测试中你几乎一定会遇到下面这些问题。这里是我踩过坑后的经验总结。5.1 语音转写准确率不理想问题在嘈杂环境或带有口音时AssemblyAI转写错误较多。排查与解决检查音频质量确保麦克风硬件正常尝试使用外接麦克风。在代码中可以尝试提高sample_rate如44100但需确认AssemblyAI支持。启用增强模型AssemblyAI有专门的word_boost参数可以传入你领域内的专业词汇如产品名、人名来提高识别率。预处理音频在音频流送入AssemblyAI之前可以先用本地库如pydub进行简单的降噪、增益标准化等预处理。使用“流式”而非“实时”如果对延迟要求不是毫秒级可以考虑使用普通的异步转录API它可能使用了更复杂的模型准确率更高。5.2 Agent错误调用工具或拒绝调用工具问题用户说“上海天气”Agent却调用了计算器或者直接回答“我不知道怎么查天气”。排查与解决优化工具描述这是最常见的原因。检查description和parameters的描述是否足够清晰、无歧义。例如“获取天气”不如“获取指定城市名称的当前天气情况包括温度、湿度和天气状况”来得明确。调整系统提示词System Prompt在给LLM的指令中明确要求它“积极使用提供的工具来回答问题”。可以强调“如果你需要实时信息或进行计算请务必使用工具”。提供少量示例Few-shot在对话历史messages的开头插入一两个用户使用工具的成功示例让LLM学会模式。调整温度Temperature参数过高的temperature如0.8会增加随机性可能导致工具调用不稳定。对于需要精确工具调用的Agent建议设置在0.1到0.3之间。5.3 响应延迟过高体验卡顿问题从说完话到听到回复等待时间过长。排查与解决网络延迟检查你的服务器或本地网络到Groq和AssemblyAI服务器的延迟。可以使用ping或traceroute简单测试。串行调用确保你的代码没有不必要地串行等待。例如语音转写、LLM思考、工具调用、TTS生成在可能的情况下应使用异步编程asyncio来优化。Groq模型选择如果当前使用70B模型感觉慢可以降级到8x7B或7B模型测试速度提升是否在可接受的性能损失范围内。流式输出Streaming对于较长的LLM回复可以启用Groq的流式响应让回复内容逐词返回并立即开始TTS的流式合成如果TTS支持实现“边想边说”的效果极大提升感知速度。5.4 上下文长度与记忆管理问题对话进行一段时间后Agent似乎“忘记”了之前聊过的内容。排查与解决了解模型上下文窗口Llama 3 70B的上下文窗口是8192个token。你需要管理conversation_history的总长度避免超出。实现摘要式记忆当对话历史过长时可以调用一次LLM让它对之前的对话进行摘要然后用摘要替换掉旧的历史记录只保留最近几条原始对话。这是一种平衡记忆和成本/长度的经典方法。使用向量数据库进行长期记忆对于需要记忆跨会话信息如用户偏好的场景可以将重要的对话片段转换成向量存入像Chroma、Pinecone这样的向量数据库。当新对话开始时先检索相关的历史记忆作为上下文提供给LLM。这个项目就像一个乐高积木AssemblyAI和Groq是两个核心的高质量部件。围绕它们你可以根据需求添加各种工具天气、日历、邮件、智能家居控制、集成不同的TTS服务甚至结合本地模型来保护隐私。我个人的体会是构建过程本身就是一个对AI Agent技术栈的深度理解之旅从语音接口到推理引擎再到工具编排每一步的优化都能带来体验上的显著提升。最关键的是从一个小而具体的功能开始比如先做好“查天气”跑通整个流程然后再逐步添加新工具这样能持续获得正反馈避免一开始就陷入过于复杂的泥潭。