1. 项目概述一个能听会说的AI智能体最近在捣鼓一个挺有意思的东西一个能通过语音对话来驱动的AI智能体。想象一下你对着麦克风说“帮我查一下明天的天气”或者“总结一下我昨天会议的要点”它不仅能听懂还能调用各种工具去执行任务最后用语音把结果告诉你。这听起来像是科幻电影里的场景但现在用一些现成的开源工具和API我们自己就能在电脑上搭一个出来。这个项目的核心就是标题里提到的三个技术栈FastAPI、Groq和Streamlit。简单来说FastAPI负责构建一个高效、可靠的后端服务处理所有复杂的逻辑Groq提供了超高速的LLM推理能力让AI的“大脑”转得飞快Streamlit则用来快速搭建一个简洁美观的Web前端让我们能通过浏览器轻松地和AI对话。把它们组合起来就构成了一个从语音输入、智能理解、任务执行到语音输出的完整闭环。无论你是想做一个个人语音助手还是探索智能体Agent的落地应用这个项目都是一个绝佳的起点。接下来我会带你一步步拆解这个项目的设计思路、技术选型背后的考量并分享从零搭建到最终跑起来的完整过程以及我踩过的那些坑。2. 核心架构与工具选型解析2.1 为什么是FastAPI Groq Streamlit在开始写代码之前花点时间想清楚为什么选这三个技术比盲目开干重要得多。这决定了整个项目的性能上限和开发体验。FastAPI现代Python Web框架的首选作为后端核心我们需要一个能快速处理请求、支持异步操作、并且文档友好的框架。Django太重Flask的异步生态相对年轻。FastAPI几乎是为此场景量身定做的性能卓越基于Starlette异步和Pydantic数据验证原生支持async/await在处理大量并发的语音识别请求或LLM API调用时能极大提升吞吐量避免阻塞。开发效率高自动生成交互式API文档Swagger UI和ReDoc通过Python类型提示就能定义请求/响应模型减少了大量样板代码和调试时间。易于集成与各种数据库、消息队列、任务调度器如Celery的集成都非常顺畅为未来扩展智能体的记忆存储、任务队列等功能铺平了道路。Groq追求极致的推理速度智能体的响应速度直接影响用户体验。传统的云LLM API如OpenAI虽然强大但网络延迟和排队时间有时会成为瓶颈。Groq的亮点在于其自研的LPULanguage Processing Unit推理引擎惊人的速度对于像Llama、Mixtral这类开源模型Groq能提供每秒输出数百个token的推理速度这意味着你几乎感觉不到AI“思考”的延迟对话体验非常流畅。成本透明目前其提供的API有免费的额度对于个人项目或原型验证非常友好避免了初期就被API账单吓到。模型生态支持Llama、Mixtral、Gemma等一批优秀的开源模型我们可以根据任务复杂度代码生成、总结、对话灵活切换模型而不被单一供应商绑定。Streamlit快速原型与交互界面我们需要一个让用户能说话、看到对话历史、简单配置的前端。用传统前端框架React/Vue开发周期太长。Streamlit的优势在于极简开发用纯Python脚本就能创建交互式Web应用。几个st.开头的函数就能搞定录音按钮、聊天记录展示、参数侧边栏特别适合数据科学家和算法工程师快速验证想法。实时更新其“脚本从头到尾执行”的模型结合会话状态Session State管理能轻松实现聊天记录的累积和界面元素的动态更新。无缝对接后端通过requests或httpx库调用我们写好的FastAPI接口非常方便前后端分离清晰。这个技术组合确保了从原型到生产级应用都有良好的支撑。FastAPI保证了服务的健壮性Groq保证了智能的“快”Streamlit保证了交互的“简”。2.2 系统架构设计图概念层虽然不能画图但我们可以用文字清晰地描述数据流用户交互层Streamlit App用户打开浏览器访问Streamlit应用。点击“开始录音”按钮通过浏览器API录制语音。录音结束后前端将音频数据通常是WAV格式通过HTTP POST请求发送到后端。API服务层FastAPI ServerFastAPI接收到音频数据后首先调用**语音转文本Speech-to-Text, STT**服务如OpenAI Whisper API或本地部署的faster-whisper。将音频转换为文字指令。智能体核心层Agent Core转换后的文字指令被送入智能体Agent逻辑模块。这个模块的核心是一个基于Groq LLM的“大脑”。我们采用类似ReAct或LangChain的思维框架让LLM根据指令决定是否需要调用工具如网络搜索、计算器、查数据库并规划执行步骤。LLM与Groq API通信。工具执行层Tools智能体调用相应的工具函数执行具体任务如用requests库爬取天气信息用wolfram-alpha库进行计算。响应生成层工具执行的结果返回给智能体智能体再次通过Groq LLM组织成一段通顺、友好的自然语言回复。文本转语音层Text-to-Speech, TTSFastAPI将最终的文本回复调用文本转语音服务如Google TTS, pyttsx3本地引擎或Edge TTS转换为音频文件。响应返回FastAPI将生成的音频文件或音频流返回给Streamlit前端。前端播放该音频并在聊天界面上以文字形式展示用户的问题和AI的回答完成一次交互循环。整个架构是典型的分层设计松耦合每一层都可以独立替换或升级比如把STT从Whisper换成Azure Speech Services。3. 环境准备与核心依赖安装3.1 Python环境与虚拟环境强烈建议使用Python 3.9以上的版本以确保所有库的最佳兼容性。第一步永远是创建独立的虚拟环境避免污染系统Python。# 创建项目目录并进入 mkdir voice-ai-agent cd voice-ai-agent # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活后命令行提示符前会出现(venv)字样。3.2 依赖库清单与安装我们需要安装三大块的依赖后端FastAPI及相关、AI/MLGroq、Whisper、TTS、前端Streamlit。创建一个requirements.txt文件# 后端核心 fastapi0.104.1 uvicorn[standard]0.24.0 # ASGI服务器用于运行FastAPI httpx0.25.1 # 异步HTTP客户端用于调用外部API python-multipart0.0.6 # 处理文件上传音频 # AI与语音处理 groq0.3.0 # Groq官方SDK openai-whisper20231117 # OpenAI Whisper语音识别 # 或者使用更快的 faster-whisper (需要额外安装CTranslate2) # faster-whisper0.9.0 pyttsx32.90 # 离线文本转语音跨平台 # 或者使用在线TTS如 edge-tts # edge-tts6.1.9 langchain0.0.340 # 可选用于构建智能体框架 langchain-groq0.0.2 # 可选LangChain的Groq集成 # 前端 streamlit1.28.0 streamlit-webrtc0.44.2 # 用于处理前端音频流高级选项基础版可先不用 # 工具与工具 requests2.31.0 # 用于工具函数如网络搜索 python-dotenv1.0.0 # 管理环境变量API密钥然后使用pip安装pip install -r requirements.txt注意openai-whisper依赖ffmpeg。你需要确保系统已安装FFmpeg并将其添加到环境变量PATH中。在Ubuntu上可以sudo apt install ffmpeg在Mac上可以brew install ffmpeg在Windows上需要去官网下载可执行文件并配置。3.3 获取并配置API密钥本项目最关键的外部依赖是Groq的API密钥。访问 Groq Cloud 注册账号。在控制台找到API Keys部分创建一个新的密钥。在项目根目录创建一个名为.env的文件注意前面的点将密钥写入GROQ_API_KEYyour_groq_api_key_here永远不要将.env文件提交到Git等版本控制系统你应该在.gitignore文件中添加.env。如果你计划使用其他付费的STT或TTS服务如OpenAI的Whisper API虽然本地Whisper免费但慢也需要将它们的密钥以类似方式配置在.env文件中。4. 后端核心FastAPI服务搭建4.1 初始化FastAPI应用与路由设计我们在项目根目录创建backend文件夹并在其中创建main.py作为入口点。# backend/main.py from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import os from dotenv import load_dotenv import logging # 加载环境变量 load_dotenv() # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 初始化FastAPI应用 app FastAPI(titleVoice-Controlled AI Agent API, version1.0.0) # 添加CORS中间件允许Streamlit前端默认端口8501跨域访问 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:8501], # 生产环境需替换为实际域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 定义数据模型 class TextQuery(BaseModel): text: str class AgentResponse(BaseModel): text: str audio_url: str None # 存储生成的语音文件URL app.get(/) async def root(): return {message: Voice AI Agent API is running} app.get(/health) async def health_check(): return {status: healthy} # 核心端点将在后续步骤中添加 # /transcribe (POST) - 语音转文字 # /chat (POST) - 处理文字查询并返回AI回复 # /synthesize (POST) - 文字转语音这个骨架定义了应用的基本结构和两个简单的状态检查端点。CORSMiddleware至关重要因为我们的前端Streamlit运行在8501端口和后端FastAPI默认8000端口是不同源的。4.2 语音转文本STT端点实现我们将使用本地运行的whisper模型它足够准确且免费。对于生产环境可以考虑使用更快更准的API服务。# backend/main.py (续) import whisper import tempfile import asyncio from pathlib import Path # 加载Whisper模型小型模型平衡速度与精度 model whisper.load_model(base) # 可选tiny, base, small, medium, large app.post(/transcribe, response_modelTextQuery) async def transcribe_audio(file: UploadFile File(...)): 接收音频文件WAV/MP3等使用Whisper转换为文字。 if not file.content_type.startswith(audio/): raise HTTPException(status_code400, detailFile must be an audio file.) try: # 创建一个临时文件保存上传的音频 suffix Path(file.filename).suffix with tempfile.NamedTemporaryFile(deleteFalse, suffixsuffix) as tmp: content await file.read() tmp.write(content) tmp_path tmp.name # 使用Whisper进行转录 # 注意whisper.transcribe是同步的在异步上下文中需要用run_in_executor避免阻塞事件循环 loop asyncio.get_event_loop() result await loop.run_in_executor(None, model.transcribe, tmp_path) # 清理临时文件 os.unlink(tmp_path) transcribed_text result[text].strip() logger.info(fTranscribed text: {transcribed_text}) return TextQuery(texttranscribed_text) except Exception as e: logger.error(fTranscription failed: {e}) raise HTTPException(status_code500, detailfTranscription error: {str(e)})实操心得模型选择base模型在大多数场景下准确度和速度都不错。如果你的应用场景对精度要求极高如专业术语可以用small或medium但推理时间会显著增加。tiny最快但精度有损失。异步处理Whisper推理是CPU/GPU密集型同步操作直接调用会阻塞FastAPI的异步事件循环影响其他请求。使用asyncio.run_in_executor将其放到线程池中执行是关键。临时文件一定要记得删除临时文件否则服务器磁盘很快会被撑满。4.3 集成Groq LLM与智能体逻辑这是项目的“大脑”。我们先实现一个简单的、无工具调用的对话版本再扩展成智能体。# backend/main.py (续) from groq import Groq import json # 初始化Groq客户端 groq_client Groq(api_keyos.getenv(GROQ_API_KEY)) # 系统提示词定义AI助手的角色和能力 SYSTEM_PROMPT 你是一个有用的语音助手。请用简洁、清晰、口语化的中文回答用户的问题。如果用户的问题需要联网搜索最新信息或执行特定计算请告知用户你目前无法直接执行这些操作但可以基于已有知识进行推理和回答。回答尽量控制在2-3句话内。 async def get_llm_response(user_query: str, conversation_history: list None) - str: 调用Groq LLM生成回复。 conversation_history 格式: [{role: user, content: ...}, {role: assistant, content: ...}, ...] messages [{role: system, content: SYSTEM_PROMPT}] if conversation_history: messages.extend(conversation_history[-6:]) # 只保留最近6轮对话防止上下文过长 messages.append({role: user, content: user_query}) try: chat_completion groq_client.chat.completions.create( messagesmessages, modelmixtral-8x7b-32768, # Groq上速度很快的一个模型 temperature0.7, max_tokens500, streamFalse, # 非流式一次性返回 ) response_text chat_completion.choices[0].message.content return response_text.strip() except Exception as e: logger.error(fGroq API call failed: {e}) return 抱歉我暂时无法处理你的请求。请稍后再试。 app.post(/chat, response_modelAgentResponse) async def chat_with_agent(query: TextQuery): 接收用户文本查询调用LLM生成回复。 user_text query.text if not user_text: raise HTTPException(status_code400, detailQuery text cannot be empty.) logger.info(fProcessing query: {user_text}) ai_response_text await get_llm_response(user_text) logger.info(fAI response: {ai_response_text}) # 目前先返回文本音频URL将在/synthesize端点生成后补充 return AgentResponse(textai_response_text)现在我们已经有了一个能听懂话/transcribe并能思考回答/chat的后端了。接下来让它“开口说话”。4.4 文本转语音TTS端点实现我们使用离线的pyttsx3库它无需API密钥但声音可能比较机械。你也可以选择edge-tts微软Edge的在线语音免费且自然但需要网络或其他付费服务。# backend/main.py (续) import pyttsx3 from fastapi.responses import FileResponse import uuid # 初始化TTS引擎 tts_engine pyttsx3.init() # 配置语音属性可选 voices tts_engine.getProperty(voices) # 尝试找一个中文语音取决于系统安装的语音包 for voice in voices: if chinese in voice.name.lower() or zh in voice.id.lower(): tts_engine.setProperty(voice, voice.id) break tts_engine.setProperty(rate, 180) # 语速 tts_engine.setProperty(volume, 0.9) # 音量 # 创建一个目录来存储生成的语音文件 AUDIO_OUTPUT_DIR Path(generated_audio) AUDIO_OUTPUT_DIR.mkdir(exist_okTrue) app.post(/synthesize) async def synthesize_speech(response: AgentResponse): 将AI回复文本转换为语音文件MP3并返回文件URL。 text_to_speak response.text if not text_to_speak: raise HTTPException(status_code400, detailText for synthesis cannot be empty.) # 生成唯一文件名 filename f{uuid.uuid4().hex}.mp3 filepath AUDIO_OUTPUT_DIR / filename try: # 使用pyttsx3保存语音到文件 # 注意pyttsx3的save_to_file是同步的且在某些环境下对异步支持不好。 # 一种稳妥的做法是在线程中运行。 def _save_speech(): tts_engine.save_to_file(text_to_speak, str(filepath)) tts_engine.runAndWait() loop asyncio.get_event_loop() await loop.run_in_executor(None, _save_speech) # 假设我们的音频文件可以通过静态文件服务访问 # 在生产环境中你需要配置静态文件服务或使用CDN audio_url f/static/audio/{filename} # 暂时我们直接返回文件。实际部署时需要Nginx等提供静态服务。 return FileResponse(pathfilepath, media_typeaudio/mpeg, filenamefilename) except Exception as e: logger.error(fSpeech synthesis failed: {e}) raise HTTPException(status_code500, detailfSpeech synthesis error: {str(e)}) # 为了能直接访问生成的音频文件添加一个静态文件路由简易版 from fastapi.staticfiles import StaticFiles app.mount(/static, StaticFiles(directorygenerated_audio), namestatic)注意事项TTS引擎选择pyttsx3是离线方案方便但音质和自然度有限。edge-tts音质好是异步的但需要处理网络请求和临时文件。根据你的需求选择。这里为了简化先用pyttsx3。文件管理生成的音频文件会不断累积需要定期清理。可以在/synthesize端点中添加逻辑删除超过一定时间的旧文件或者使用内存中的字节流而不保存到磁盘更复杂。静态文件服务在开发阶段StaticFiles中间件很方便。在生产环境如使用Uvicorn反向代理通常由Nginx或Apache来提供静态文件服务性能更好也更安全。4.5 启动后端服务在backend目录下运行uvicorn main:app --reload --host 0.0.0.0 --port 8000访问http://localhost:8000/docs就能看到自动生成的API文档并可以测试/transcribe、/chat等端点。5. 前端交互Streamlit应用开发5.1 构建基础聊天界面在项目根目录创建frontend文件夹并创建app.py。# frontend/app.py import streamlit as st import requests import io import base64 import time from audio_recorder_streamlit import audio_recorder # 一个方便的录音组件 # 配置页面 st.set_page_config(page_title语音AI助手, page_icon, layoutwide) st.title( 语音控制AI智能体) # 初始化会话状态用于保存聊天历史 if messages not in st.session_state: st.session_state.messages [] if audio_bytes not in st.session_state: st.session_state.audio_bytes None # 显示聊天历史 for message in st.session_state.messages: with st.chat_message(message[role]): st.markdown(message[content]) # 如果是AI的回复并且有音频显示一个播放器 if message.get(audio): # 这里简化处理实际应该用返回的音频URL st.audio(message[audio], formataudio/mp3) # 侧边栏用于配置和说明 with st.sidebar: st.header(设置) api_base st.text_input(后端API地址, valuehttp://localhost:8000) st.markdown(---) st.markdown(### 使用说明) st.markdown( 1. 点击下方的录音按钮开始说话。 2. 松开按钮结束录音并发送。 3. AI会处理你的语音并生成文字和语音回复。 4. 对话历史会显示在上方。 ) if st.button(清空对话历史): st.session_state.messages [] st.session_state.audio_bytes None st.rerun() # 主区域录音和发送 st.markdown(---) st.subheader(对我说点什么吧) # 方法一使用 audio_recorder_streamlit 组件需安装 # 这是一个简单的录音按钮返回音频字节流 audio_bytes audio_recorder(text, icon_size2x, pause_threshold2.0) # 方法二备选使用 streamlit-audiorecorder 或其他组件 # 这里以 audio_recorder 为例 if audio_bytes: # 显示一个临时音频播放器让用户确认录音 st.audio(audio_bytes, formataudio/wav) if st.button(发送录音): with st.spinner(正在处理...): # 步骤1: 发送音频到后端进行转录 files {file: (audio.wav, audio_bytes, audio/wav)} try: transcribe_response requests.post(f{api_base}/transcribe, filesfiles) if transcribe_response.status_code 200: user_text transcribe_response.json()[text] # 将用户语音转成的文字添加到聊天记录 st.session_state.messages.append({role: user, content: user_text}) st.chat_message(user).markdown(user_text) # 步骤2: 将文字发送给AI处理 chat_response requests.post(f{api_base}/chat, json{text: user_text}) if chat_response.status_code 200: ai_response chat_response.json() ai_text ai_response[text] # 将AI的文字回复添加到聊天记录 st.session_state.messages.append({role: assistant, content: ai_text}) with st.chat_message(assistant): st.markdown(ai_text) # 步骤3: 请求AI回复的语音合成 synthesize_response requests.post(f{api_base}/synthesize, jsonai_response) if synthesize_response.status_code 200: # 假设后端直接返回音频文件内容 ai_audio_bytes synthesize_response.content st.session_state.audio_bytes ai_audio_bytes # 播放AI的语音回复 st.audio(ai_audio_bytes, formataudio/mp3) # 也可以将音频数据保存到会话状态的消息中以便历史记录回放 # 这里需要将字节转换为base64或文件路径略复杂先简化。 else: st.error(语音合成失败) else: st.error(AI处理失败) else: st.error(语音识别失败) except requests.exceptions.ConnectionError: st.error(无法连接到后端服务请检查API地址和后端是否运行。) # 处理完成后清空当前录音准备下一次 # audio_bytes 变量会在下次循环时更新这里无需手动清空 st.rerun() # 使用rerun刷新界面显示新消息这个前端界面已经具备了核心功能录音、发送、显示对话历史、播放AI语音回复。它通过三个连续的HTTP请求与后端交互模拟了完整的语音对话流程。5.2 处理音频流与改进用户体验上面的实现是“录音-发送-等待-播放”的同步模式用户体验有卡顿。我们可以做一些优化实时反馈在等待后端处理时使用st.spinner和st.status显示明确的进度“识别中...”、“思考中...”、“合成语音...”。错误处理对每个网络请求都添加更细致的try...except并给用户友好的错误提示。音频自动播放合成语音后使用JavaScript自动播放Streamlit原生支持有限可能需要自定义组件。对话历史持久化将st.session_state.messages保存到本地文件或数据库刷新页面不丢失。一个改进的发送逻辑片段如下# 在 frontend/app.py 的发送按钮逻辑中改进 if st.button(发送录音, keysend_audio): progress_bar st.progress(0) status_text st.empty() status_text.text( 正在识别语音...) progress_bar.progress(25) # ... 调用 /transcribe ... if transcribe_ok: status_text.text( AI正在思考...) progress_bar.progress(50) # ... 调用 /chat ... if chat_ok: status_text.text(️ 正在生成语音...) progress_bar.progress(75) # ... 调用 /synthesize ... if synthesize_ok: status_text.text(✅ 完成) progress_bar.progress(100) time.sleep(0.5) status_text.empty() progress_bar.empty() else: status_text.text(❌ 语音生成失败) else: status_text.text(❌ AI处理失败) else: status_text.text(❌ 语音识别失败)5.3 启动前端应用在frontend目录下运行streamlit run app.py浏览器会自动打开http://localhost:8501。确保后端服务http://localhost:8000也在运行。6. 进阶从简单聊天到智能体Agent目前我们的AI只是一个“聊天机器人”它不会主动调用工具。要升级为“智能体”我们需要赋予它使用工具的能力。这里我们引入简单的ReActReasoning Acting模式而不必一开始就上完整的LangChain。6.1 定义工具函数在后端backend/main.py中定义一些工具例如# backend/tools.py (新建一个文件或在main.py中定义) import requests from datetime import datetime import math def get_current_time(query: str) - str: 获取当前日期和时间。 now datetime.now() return f当前时间是{now.strftime(%Y年%m月%d日 %H:%M:%S)} def search_web(query: str) - str: 模拟网络搜索。实际应用中可集成Serper API、Google Custom Search等。 # 这里是模拟真实情况需要调用搜索API return f关于{query}的搜索结果模拟这是一个模拟的搜索结果摘要。在实际项目中你需要接入真实的搜索API。 def calculate(expression: str) - str: 计算数学表达式。警告使用eval有安全风险仅作演示。生产环境应用安全库如asteval。 try: # 极度危险仅用于演示。永远不要在生产中直接用eval处理用户输入。 # 应使用安全的数学表达式求值库。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果{expression} {result} except Exception as e: return f计算表达式{expression}时出错{e} # 工具列表供LLM选择 AVAILABLE_TOOLS [ { name: get_current_time, description: 当用户询问当前时间、日期、今天星期几时使用此工具。, parameters: {type: object, properties: {query: {type: string}}} }, { name: search_web, description: 当用户询问需要最新信息、新闻、事实核查或你不知道的知识时使用此工具。, parameters: {type: object, properties: {query: {type: string}}} }, { name: calculate, description: 当用户需要进行数学计算、算术、单位换算时使用此工具。, parameters: {type: object, properties: {expression: {type: string}}} } ]6.2 实现智能体推理循环修改后端的/chat端点或创建一个新的/agent_chat端点实现一个简单的ReAct循环# backend/main.py (新增函数和端点) import json import re def parse_llm_for_action(llm_output: str): 解析LLM的输出看是否包含工具调用指令。 我们约定一个简单的格式 THOUGHT: ... ACTION: tool_name ARGS: {...} ANSWER: ... thought action None args {} answer thought_match re.search(rTHOUGHT:(.*?)(?ACTION:|ANSWER:|$), llm_output, re.DOTALL) action_match re.search(rACTION:(\w), llm_output) args_match re.search(rARGS:(\{.*?\}), llm_output, re.DOTALL) answer_match re.search(rANSWER:(.*), llm_output, re.DOTALL) if thought_match: thought thought_match.group(1).strip() if action_match: action action_match.group(1).strip() if args_match: try: args json.loads(args_match.group(1).strip()) except json.JSONDecodeError: args {} if answer_match: answer answer_match.group(1).strip() return thought, action, args, answer async def run_agent_loop(user_query: str, max_steps: int 3) - str: 运行一个简单的ReAct循环。 conversation_context f用户提问{user_query}\n\n你可以使用的工具{json.dumps(AVAILABLE_TOOLS, ensure_asciiFalse)} full_prompt f你是一个AI助手可以调用工具来帮助用户。请遵循以下格式思考 THOUGHT: 你需要思考用户的问题决定是否需要使用工具以及使用哪个工具。 ACTION: 如果需要使用工具在这里写出工具名。如果不需要写NONE。 ARGS: 如果需要使用工具在这里以JSON格式写出调用参数。如果不需要写{{}}。 ANSWER: 最终给用户的回答。如果你使用了工具请基于工具返回的结果进行总结。 {conversation_context} 请开始 history [] # 简化不携带长历史 steps 0 final_answer while steps max_steps: steps 1 llm_raw_response await get_llm_response(full_prompt, history) thought, action, args, answer parse_llm_for_action(llm_raw_response) logger.info(fStep {steps} - Thought: {thought}, Action: {action}, Args: {args}) if action and action ! NONE: # 执行工具调用 tool_result if action get_current_time: tool_result get_current_time(args.get(query, )) elif action search_web: tool_result search_web(args.get(query, user_query)) elif action calculate: tool_result calculate(args.get(expression, )) else: tool_result f未知工具{action} # 将工具结果加入到上下文中进行下一轮思考 full_prompt f\n\n工具 {action} 返回的结果是{tool_result} # 也可以将本轮交互加入history让LLM知道上下文 history.append({role: assistant, content: llm_raw_response}) # 模拟一个系统消息告知工具结果 history.append({role: system, content: fTool Result: {tool_result}}) else: # 不需要或无法使用工具直接返回答案 final_answer answer if answer else llm_raw_response break if steps max_steps: final_answer 经过多次尝试我仍然无法完美解决你的问题。请尝试更清晰地描述你的需求。 break return final_answer app.post(/agent_chat, response_modelAgentResponse) async def chat_with_agent_advanced(query: TextQuery): 使用智能体带工具调用处理用户查询。 user_text query.text if not user_text: raise HTTPException(status_code400, detailQuery text cannot be empty.) logger.info(fAgent processing query: {user_text}) ai_response_text await run_agent_loop(user_text) logger.info(fAgent final response: {ai_response_text}) return AgentResponse(textai_response_text)前端只需要将请求从/chat改为/agent_chat就能体验到智能体调用工具的能力如问“现在几点”或“计算一下125的平方根”。踩坑实录提示工程Prompt Engineering让LLM稳定地输出THOUGHT/ACTION/ARGS/ANSWER格式需要精心设计提示词。多轮调试是必不可少的。工具安全性calculate工具中直接使用eval()是极其危险的绝对不能用于生产环境。这里仅作演示真实项目必须使用安全的表达式求值库或沙箱。循环控制必须设置最大步数max_steps以防止LLM陷入无限循环。同时要处理工具调用失败的情况让智能体能够优雅降级。7. 部署、优化与常见问题7.1 本地与生产环境部署本地运行终端1在backend目录下uvicorn main:app --reload --host 0.0.0.0 --port 8000终端2在frontend目录下streamlit run app.py --server.port 8501浏览器访问http://localhost:8501生产部署考虑后端FastAPI使用uvicorn配合gunicornwith Uvicorn workers部署在Linux服务器上。用Nginx作为反向代理处理静态文件、SSL加密和负载均衡。# 使用gunicorn启动在backend目录 gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000前端StreamlitStreamlit应用本身可以作为服务运行。对于更正式的生产环境可以考虑将Streamlit前端“静态化”比较困难或者将其视为一个独立的服务并通过Nginx代理。Streamlit也支持配置server.address和server.port。环境变量所有API密钥、数据库连接字符串等敏感信息必须通过环境变量或安全的密钥管理服务传入绝不可写在代码中。进程管理使用systemd或supervisor来管理后端和前端进程确保它们崩溃后能自动重启。7.2 性能优化点Whisper模型加速使用faster-whisper基于CTranslate2替代原版whisper推理速度可提升数倍且内存占用更低。Groq模型选择Groq提供了多个模型。mixtral-8x7b-32768在速度和能力上平衡较好。对于简单任务可以尝试更小的llama2-70b-4096或gemma-7b-it以获得更快的响应。TTS缓存对相同的文本回复可以缓存其语音文件避免重复合成。可以在后端用字典或Redis存储文本MD5 - 音频文件路径的映射。前端异步请求Streamlit的st.button会导致整个脚本重跑。对于更流畅的体验可以考虑使用st.form或自定义组件来管理状态或者用JavaScript发起异步请求但这会显著增加前端复杂度。7.3 常见问题与排查问题1Streamlit前端报错“无法连接到后端服务”。检查确保后端FastAPI服务正在运行http://localhost:8000能访问。检查前端代码中的api_base地址是否正确。检查后端FastAPI是否启用了CORS并正确配置了允许前端源http://localhost:8501。问题2Whisper转录速度非常慢或者报错关于ffmpeg。检查确认系统已安装ffmpeg并可在命令行中调用。优化换用faster-whisper库并尝试使用更小的模型如tiny或base。注意首次加载Whisper模型会下载模型文件需要网络和时间。问题3Groq API调用返回429频率限制或401认证失败。检查.env文件中的GROQ_API_KEY是否正确是否在代码中被正确加载。检查Groq控制台查看API使用量和频率限制。免费额度有每分钟请求数RPM和每天令牌数TPD的限制。优化在前端添加请求节流Throttling避免用户快速连续点击。问题4合成的语音听起来很机械或者不是中文。检查pyttsx3的语音包。在Windows上可以到系统“语音设置”里查看和下载其他语音包。在Linux上可能需要安装espeak或festival及其中文语音数据。替代方案切换到edge-tts它提供更自然的中文语音如zh-CN-XiaoxiaoNeural但需要处理网络请求和异步。问题5智能体Agent经常不调用工具或者调用格式错误。调试打印出LLM每一步的完整输出llm_raw_response检查提示词是否清晰LLM是否理解了指令格式。优化提示词在提示词中提供更清晰的工具描述和1-2个完整的示例Few-shot Learning能大幅提升工具调用的准确率。简化工具初期尽量减少工具数量并确保工具的功能描述非常精确无歧义。这个项目从零搭建了一个完整的、可交互的语音AI智能体原型。它涵盖了从语音识别、大模型推理、工具调用到语音合成的全链路。你可以在此基础上继续扩展工具集如发送邮件、控制智能家居、增加长期记忆向量数据库、优化语音交互体验流式响应、语音唤醒甚至将其部署到云服务器通过手机随时随地访问。
基于FastAPI、Groq与Streamlit构建语音交互AI智能体全栈实践
1. 项目概述一个能听会说的AI智能体最近在捣鼓一个挺有意思的东西一个能通过语音对话来驱动的AI智能体。想象一下你对着麦克风说“帮我查一下明天的天气”或者“总结一下我昨天会议的要点”它不仅能听懂还能调用各种工具去执行任务最后用语音把结果告诉你。这听起来像是科幻电影里的场景但现在用一些现成的开源工具和API我们自己就能在电脑上搭一个出来。这个项目的核心就是标题里提到的三个技术栈FastAPI、Groq和Streamlit。简单来说FastAPI负责构建一个高效、可靠的后端服务处理所有复杂的逻辑Groq提供了超高速的LLM推理能力让AI的“大脑”转得飞快Streamlit则用来快速搭建一个简洁美观的Web前端让我们能通过浏览器轻松地和AI对话。把它们组合起来就构成了一个从语音输入、智能理解、任务执行到语音输出的完整闭环。无论你是想做一个个人语音助手还是探索智能体Agent的落地应用这个项目都是一个绝佳的起点。接下来我会带你一步步拆解这个项目的设计思路、技术选型背后的考量并分享从零搭建到最终跑起来的完整过程以及我踩过的那些坑。2. 核心架构与工具选型解析2.1 为什么是FastAPI Groq Streamlit在开始写代码之前花点时间想清楚为什么选这三个技术比盲目开干重要得多。这决定了整个项目的性能上限和开发体验。FastAPI现代Python Web框架的首选作为后端核心我们需要一个能快速处理请求、支持异步操作、并且文档友好的框架。Django太重Flask的异步生态相对年轻。FastAPI几乎是为此场景量身定做的性能卓越基于Starlette异步和Pydantic数据验证原生支持async/await在处理大量并发的语音识别请求或LLM API调用时能极大提升吞吐量避免阻塞。开发效率高自动生成交互式API文档Swagger UI和ReDoc通过Python类型提示就能定义请求/响应模型减少了大量样板代码和调试时间。易于集成与各种数据库、消息队列、任务调度器如Celery的集成都非常顺畅为未来扩展智能体的记忆存储、任务队列等功能铺平了道路。Groq追求极致的推理速度智能体的响应速度直接影响用户体验。传统的云LLM API如OpenAI虽然强大但网络延迟和排队时间有时会成为瓶颈。Groq的亮点在于其自研的LPULanguage Processing Unit推理引擎惊人的速度对于像Llama、Mixtral这类开源模型Groq能提供每秒输出数百个token的推理速度这意味着你几乎感觉不到AI“思考”的延迟对话体验非常流畅。成本透明目前其提供的API有免费的额度对于个人项目或原型验证非常友好避免了初期就被API账单吓到。模型生态支持Llama、Mixtral、Gemma等一批优秀的开源模型我们可以根据任务复杂度代码生成、总结、对话灵活切换模型而不被单一供应商绑定。Streamlit快速原型与交互界面我们需要一个让用户能说话、看到对话历史、简单配置的前端。用传统前端框架React/Vue开发周期太长。Streamlit的优势在于极简开发用纯Python脚本就能创建交互式Web应用。几个st.开头的函数就能搞定录音按钮、聊天记录展示、参数侧边栏特别适合数据科学家和算法工程师快速验证想法。实时更新其“脚本从头到尾执行”的模型结合会话状态Session State管理能轻松实现聊天记录的累积和界面元素的动态更新。无缝对接后端通过requests或httpx库调用我们写好的FastAPI接口非常方便前后端分离清晰。这个技术组合确保了从原型到生产级应用都有良好的支撑。FastAPI保证了服务的健壮性Groq保证了智能的“快”Streamlit保证了交互的“简”。2.2 系统架构设计图概念层虽然不能画图但我们可以用文字清晰地描述数据流用户交互层Streamlit App用户打开浏览器访问Streamlit应用。点击“开始录音”按钮通过浏览器API录制语音。录音结束后前端将音频数据通常是WAV格式通过HTTP POST请求发送到后端。API服务层FastAPI ServerFastAPI接收到音频数据后首先调用**语音转文本Speech-to-Text, STT**服务如OpenAI Whisper API或本地部署的faster-whisper。将音频转换为文字指令。智能体核心层Agent Core转换后的文字指令被送入智能体Agent逻辑模块。这个模块的核心是一个基于Groq LLM的“大脑”。我们采用类似ReAct或LangChain的思维框架让LLM根据指令决定是否需要调用工具如网络搜索、计算器、查数据库并规划执行步骤。LLM与Groq API通信。工具执行层Tools智能体调用相应的工具函数执行具体任务如用requests库爬取天气信息用wolfram-alpha库进行计算。响应生成层工具执行的结果返回给智能体智能体再次通过Groq LLM组织成一段通顺、友好的自然语言回复。文本转语音层Text-to-Speech, TTSFastAPI将最终的文本回复调用文本转语音服务如Google TTS, pyttsx3本地引擎或Edge TTS转换为音频文件。响应返回FastAPI将生成的音频文件或音频流返回给Streamlit前端。前端播放该音频并在聊天界面上以文字形式展示用户的问题和AI的回答完成一次交互循环。整个架构是典型的分层设计松耦合每一层都可以独立替换或升级比如把STT从Whisper换成Azure Speech Services。3. 环境准备与核心依赖安装3.1 Python环境与虚拟环境强烈建议使用Python 3.9以上的版本以确保所有库的最佳兼容性。第一步永远是创建独立的虚拟环境避免污染系统Python。# 创建项目目录并进入 mkdir voice-ai-agent cd voice-ai-agent # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活后命令行提示符前会出现(venv)字样。3.2 依赖库清单与安装我们需要安装三大块的依赖后端FastAPI及相关、AI/MLGroq、Whisper、TTS、前端Streamlit。创建一个requirements.txt文件# 后端核心 fastapi0.104.1 uvicorn[standard]0.24.0 # ASGI服务器用于运行FastAPI httpx0.25.1 # 异步HTTP客户端用于调用外部API python-multipart0.0.6 # 处理文件上传音频 # AI与语音处理 groq0.3.0 # Groq官方SDK openai-whisper20231117 # OpenAI Whisper语音识别 # 或者使用更快的 faster-whisper (需要额外安装CTranslate2) # faster-whisper0.9.0 pyttsx32.90 # 离线文本转语音跨平台 # 或者使用在线TTS如 edge-tts # edge-tts6.1.9 langchain0.0.340 # 可选用于构建智能体框架 langchain-groq0.0.2 # 可选LangChain的Groq集成 # 前端 streamlit1.28.0 streamlit-webrtc0.44.2 # 用于处理前端音频流高级选项基础版可先不用 # 工具与工具 requests2.31.0 # 用于工具函数如网络搜索 python-dotenv1.0.0 # 管理环境变量API密钥然后使用pip安装pip install -r requirements.txt注意openai-whisper依赖ffmpeg。你需要确保系统已安装FFmpeg并将其添加到环境变量PATH中。在Ubuntu上可以sudo apt install ffmpeg在Mac上可以brew install ffmpeg在Windows上需要去官网下载可执行文件并配置。3.3 获取并配置API密钥本项目最关键的外部依赖是Groq的API密钥。访问 Groq Cloud 注册账号。在控制台找到API Keys部分创建一个新的密钥。在项目根目录创建一个名为.env的文件注意前面的点将密钥写入GROQ_API_KEYyour_groq_api_key_here永远不要将.env文件提交到Git等版本控制系统你应该在.gitignore文件中添加.env。如果你计划使用其他付费的STT或TTS服务如OpenAI的Whisper API虽然本地Whisper免费但慢也需要将它们的密钥以类似方式配置在.env文件中。4. 后端核心FastAPI服务搭建4.1 初始化FastAPI应用与路由设计我们在项目根目录创建backend文件夹并在其中创建main.py作为入口点。# backend/main.py from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import os from dotenv import load_dotenv import logging # 加载环境变量 load_dotenv() # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 初始化FastAPI应用 app FastAPI(titleVoice-Controlled AI Agent API, version1.0.0) # 添加CORS中间件允许Streamlit前端默认端口8501跨域访问 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:8501], # 生产环境需替换为实际域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 定义数据模型 class TextQuery(BaseModel): text: str class AgentResponse(BaseModel): text: str audio_url: str None # 存储生成的语音文件URL app.get(/) async def root(): return {message: Voice AI Agent API is running} app.get(/health) async def health_check(): return {status: healthy} # 核心端点将在后续步骤中添加 # /transcribe (POST) - 语音转文字 # /chat (POST) - 处理文字查询并返回AI回复 # /synthesize (POST) - 文字转语音这个骨架定义了应用的基本结构和两个简单的状态检查端点。CORSMiddleware至关重要因为我们的前端Streamlit运行在8501端口和后端FastAPI默认8000端口是不同源的。4.2 语音转文本STT端点实现我们将使用本地运行的whisper模型它足够准确且免费。对于生产环境可以考虑使用更快更准的API服务。# backend/main.py (续) import whisper import tempfile import asyncio from pathlib import Path # 加载Whisper模型小型模型平衡速度与精度 model whisper.load_model(base) # 可选tiny, base, small, medium, large app.post(/transcribe, response_modelTextQuery) async def transcribe_audio(file: UploadFile File(...)): 接收音频文件WAV/MP3等使用Whisper转换为文字。 if not file.content_type.startswith(audio/): raise HTTPException(status_code400, detailFile must be an audio file.) try: # 创建一个临时文件保存上传的音频 suffix Path(file.filename).suffix with tempfile.NamedTemporaryFile(deleteFalse, suffixsuffix) as tmp: content await file.read() tmp.write(content) tmp_path tmp.name # 使用Whisper进行转录 # 注意whisper.transcribe是同步的在异步上下文中需要用run_in_executor避免阻塞事件循环 loop asyncio.get_event_loop() result await loop.run_in_executor(None, model.transcribe, tmp_path) # 清理临时文件 os.unlink(tmp_path) transcribed_text result[text].strip() logger.info(fTranscribed text: {transcribed_text}) return TextQuery(texttranscribed_text) except Exception as e: logger.error(fTranscription failed: {e}) raise HTTPException(status_code500, detailfTranscription error: {str(e)})实操心得模型选择base模型在大多数场景下准确度和速度都不错。如果你的应用场景对精度要求极高如专业术语可以用small或medium但推理时间会显著增加。tiny最快但精度有损失。异步处理Whisper推理是CPU/GPU密集型同步操作直接调用会阻塞FastAPI的异步事件循环影响其他请求。使用asyncio.run_in_executor将其放到线程池中执行是关键。临时文件一定要记得删除临时文件否则服务器磁盘很快会被撑满。4.3 集成Groq LLM与智能体逻辑这是项目的“大脑”。我们先实现一个简单的、无工具调用的对话版本再扩展成智能体。# backend/main.py (续) from groq import Groq import json # 初始化Groq客户端 groq_client Groq(api_keyos.getenv(GROQ_API_KEY)) # 系统提示词定义AI助手的角色和能力 SYSTEM_PROMPT 你是一个有用的语音助手。请用简洁、清晰、口语化的中文回答用户的问题。如果用户的问题需要联网搜索最新信息或执行特定计算请告知用户你目前无法直接执行这些操作但可以基于已有知识进行推理和回答。回答尽量控制在2-3句话内。 async def get_llm_response(user_query: str, conversation_history: list None) - str: 调用Groq LLM生成回复。 conversation_history 格式: [{role: user, content: ...}, {role: assistant, content: ...}, ...] messages [{role: system, content: SYSTEM_PROMPT}] if conversation_history: messages.extend(conversation_history[-6:]) # 只保留最近6轮对话防止上下文过长 messages.append({role: user, content: user_query}) try: chat_completion groq_client.chat.completions.create( messagesmessages, modelmixtral-8x7b-32768, # Groq上速度很快的一个模型 temperature0.7, max_tokens500, streamFalse, # 非流式一次性返回 ) response_text chat_completion.choices[0].message.content return response_text.strip() except Exception as e: logger.error(fGroq API call failed: {e}) return 抱歉我暂时无法处理你的请求。请稍后再试。 app.post(/chat, response_modelAgentResponse) async def chat_with_agent(query: TextQuery): 接收用户文本查询调用LLM生成回复。 user_text query.text if not user_text: raise HTTPException(status_code400, detailQuery text cannot be empty.) logger.info(fProcessing query: {user_text}) ai_response_text await get_llm_response(user_text) logger.info(fAI response: {ai_response_text}) # 目前先返回文本音频URL将在/synthesize端点生成后补充 return AgentResponse(textai_response_text)现在我们已经有了一个能听懂话/transcribe并能思考回答/chat的后端了。接下来让它“开口说话”。4.4 文本转语音TTS端点实现我们使用离线的pyttsx3库它无需API密钥但声音可能比较机械。你也可以选择edge-tts微软Edge的在线语音免费且自然但需要网络或其他付费服务。# backend/main.py (续) import pyttsx3 from fastapi.responses import FileResponse import uuid # 初始化TTS引擎 tts_engine pyttsx3.init() # 配置语音属性可选 voices tts_engine.getProperty(voices) # 尝试找一个中文语音取决于系统安装的语音包 for voice in voices: if chinese in voice.name.lower() or zh in voice.id.lower(): tts_engine.setProperty(voice, voice.id) break tts_engine.setProperty(rate, 180) # 语速 tts_engine.setProperty(volume, 0.9) # 音量 # 创建一个目录来存储生成的语音文件 AUDIO_OUTPUT_DIR Path(generated_audio) AUDIO_OUTPUT_DIR.mkdir(exist_okTrue) app.post(/synthesize) async def synthesize_speech(response: AgentResponse): 将AI回复文本转换为语音文件MP3并返回文件URL。 text_to_speak response.text if not text_to_speak: raise HTTPException(status_code400, detailText for synthesis cannot be empty.) # 生成唯一文件名 filename f{uuid.uuid4().hex}.mp3 filepath AUDIO_OUTPUT_DIR / filename try: # 使用pyttsx3保存语音到文件 # 注意pyttsx3的save_to_file是同步的且在某些环境下对异步支持不好。 # 一种稳妥的做法是在线程中运行。 def _save_speech(): tts_engine.save_to_file(text_to_speak, str(filepath)) tts_engine.runAndWait() loop asyncio.get_event_loop() await loop.run_in_executor(None, _save_speech) # 假设我们的音频文件可以通过静态文件服务访问 # 在生产环境中你需要配置静态文件服务或使用CDN audio_url f/static/audio/{filename} # 暂时我们直接返回文件。实际部署时需要Nginx等提供静态服务。 return FileResponse(pathfilepath, media_typeaudio/mpeg, filenamefilename) except Exception as e: logger.error(fSpeech synthesis failed: {e}) raise HTTPException(status_code500, detailfSpeech synthesis error: {str(e)}) # 为了能直接访问生成的音频文件添加一个静态文件路由简易版 from fastapi.staticfiles import StaticFiles app.mount(/static, StaticFiles(directorygenerated_audio), namestatic)注意事项TTS引擎选择pyttsx3是离线方案方便但音质和自然度有限。edge-tts音质好是异步的但需要处理网络请求和临时文件。根据你的需求选择。这里为了简化先用pyttsx3。文件管理生成的音频文件会不断累积需要定期清理。可以在/synthesize端点中添加逻辑删除超过一定时间的旧文件或者使用内存中的字节流而不保存到磁盘更复杂。静态文件服务在开发阶段StaticFiles中间件很方便。在生产环境如使用Uvicorn反向代理通常由Nginx或Apache来提供静态文件服务性能更好也更安全。4.5 启动后端服务在backend目录下运行uvicorn main:app --reload --host 0.0.0.0 --port 8000访问http://localhost:8000/docs就能看到自动生成的API文档并可以测试/transcribe、/chat等端点。5. 前端交互Streamlit应用开发5.1 构建基础聊天界面在项目根目录创建frontend文件夹并创建app.py。# frontend/app.py import streamlit as st import requests import io import base64 import time from audio_recorder_streamlit import audio_recorder # 一个方便的录音组件 # 配置页面 st.set_page_config(page_title语音AI助手, page_icon, layoutwide) st.title( 语音控制AI智能体) # 初始化会话状态用于保存聊天历史 if messages not in st.session_state: st.session_state.messages [] if audio_bytes not in st.session_state: st.session_state.audio_bytes None # 显示聊天历史 for message in st.session_state.messages: with st.chat_message(message[role]): st.markdown(message[content]) # 如果是AI的回复并且有音频显示一个播放器 if message.get(audio): # 这里简化处理实际应该用返回的音频URL st.audio(message[audio], formataudio/mp3) # 侧边栏用于配置和说明 with st.sidebar: st.header(设置) api_base st.text_input(后端API地址, valuehttp://localhost:8000) st.markdown(---) st.markdown(### 使用说明) st.markdown( 1. 点击下方的录音按钮开始说话。 2. 松开按钮结束录音并发送。 3. AI会处理你的语音并生成文字和语音回复。 4. 对话历史会显示在上方。 ) if st.button(清空对话历史): st.session_state.messages [] st.session_state.audio_bytes None st.rerun() # 主区域录音和发送 st.markdown(---) st.subheader(对我说点什么吧) # 方法一使用 audio_recorder_streamlit 组件需安装 # 这是一个简单的录音按钮返回音频字节流 audio_bytes audio_recorder(text, icon_size2x, pause_threshold2.0) # 方法二备选使用 streamlit-audiorecorder 或其他组件 # 这里以 audio_recorder 为例 if audio_bytes: # 显示一个临时音频播放器让用户确认录音 st.audio(audio_bytes, formataudio/wav) if st.button(发送录音): with st.spinner(正在处理...): # 步骤1: 发送音频到后端进行转录 files {file: (audio.wav, audio_bytes, audio/wav)} try: transcribe_response requests.post(f{api_base}/transcribe, filesfiles) if transcribe_response.status_code 200: user_text transcribe_response.json()[text] # 将用户语音转成的文字添加到聊天记录 st.session_state.messages.append({role: user, content: user_text}) st.chat_message(user).markdown(user_text) # 步骤2: 将文字发送给AI处理 chat_response requests.post(f{api_base}/chat, json{text: user_text}) if chat_response.status_code 200: ai_response chat_response.json() ai_text ai_response[text] # 将AI的文字回复添加到聊天记录 st.session_state.messages.append({role: assistant, content: ai_text}) with st.chat_message(assistant): st.markdown(ai_text) # 步骤3: 请求AI回复的语音合成 synthesize_response requests.post(f{api_base}/synthesize, jsonai_response) if synthesize_response.status_code 200: # 假设后端直接返回音频文件内容 ai_audio_bytes synthesize_response.content st.session_state.audio_bytes ai_audio_bytes # 播放AI的语音回复 st.audio(ai_audio_bytes, formataudio/mp3) # 也可以将音频数据保存到会话状态的消息中以便历史记录回放 # 这里需要将字节转换为base64或文件路径略复杂先简化。 else: st.error(语音合成失败) else: st.error(AI处理失败) else: st.error(语音识别失败) except requests.exceptions.ConnectionError: st.error(无法连接到后端服务请检查API地址和后端是否运行。) # 处理完成后清空当前录音准备下一次 # audio_bytes 变量会在下次循环时更新这里无需手动清空 st.rerun() # 使用rerun刷新界面显示新消息这个前端界面已经具备了核心功能录音、发送、显示对话历史、播放AI语音回复。它通过三个连续的HTTP请求与后端交互模拟了完整的语音对话流程。5.2 处理音频流与改进用户体验上面的实现是“录音-发送-等待-播放”的同步模式用户体验有卡顿。我们可以做一些优化实时反馈在等待后端处理时使用st.spinner和st.status显示明确的进度“识别中...”、“思考中...”、“合成语音...”。错误处理对每个网络请求都添加更细致的try...except并给用户友好的错误提示。音频自动播放合成语音后使用JavaScript自动播放Streamlit原生支持有限可能需要自定义组件。对话历史持久化将st.session_state.messages保存到本地文件或数据库刷新页面不丢失。一个改进的发送逻辑片段如下# 在 frontend/app.py 的发送按钮逻辑中改进 if st.button(发送录音, keysend_audio): progress_bar st.progress(0) status_text st.empty() status_text.text( 正在识别语音...) progress_bar.progress(25) # ... 调用 /transcribe ... if transcribe_ok: status_text.text( AI正在思考...) progress_bar.progress(50) # ... 调用 /chat ... if chat_ok: status_text.text(️ 正在生成语音...) progress_bar.progress(75) # ... 调用 /synthesize ... if synthesize_ok: status_text.text(✅ 完成) progress_bar.progress(100) time.sleep(0.5) status_text.empty() progress_bar.empty() else: status_text.text(❌ 语音生成失败) else: status_text.text(❌ AI处理失败) else: status_text.text(❌ 语音识别失败)5.3 启动前端应用在frontend目录下运行streamlit run app.py浏览器会自动打开http://localhost:8501。确保后端服务http://localhost:8000也在运行。6. 进阶从简单聊天到智能体Agent目前我们的AI只是一个“聊天机器人”它不会主动调用工具。要升级为“智能体”我们需要赋予它使用工具的能力。这里我们引入简单的ReActReasoning Acting模式而不必一开始就上完整的LangChain。6.1 定义工具函数在后端backend/main.py中定义一些工具例如# backend/tools.py (新建一个文件或在main.py中定义) import requests from datetime import datetime import math def get_current_time(query: str) - str: 获取当前日期和时间。 now datetime.now() return f当前时间是{now.strftime(%Y年%m月%d日 %H:%M:%S)} def search_web(query: str) - str: 模拟网络搜索。实际应用中可集成Serper API、Google Custom Search等。 # 这里是模拟真实情况需要调用搜索API return f关于{query}的搜索结果模拟这是一个模拟的搜索结果摘要。在实际项目中你需要接入真实的搜索API。 def calculate(expression: str) - str: 计算数学表达式。警告使用eval有安全风险仅作演示。生产环境应用安全库如asteval。 try: # 极度危险仅用于演示。永远不要在生产中直接用eval处理用户输入。 # 应使用安全的数学表达式求值库。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果{expression} {result} except Exception as e: return f计算表达式{expression}时出错{e} # 工具列表供LLM选择 AVAILABLE_TOOLS [ { name: get_current_time, description: 当用户询问当前时间、日期、今天星期几时使用此工具。, parameters: {type: object, properties: {query: {type: string}}} }, { name: search_web, description: 当用户询问需要最新信息、新闻、事实核查或你不知道的知识时使用此工具。, parameters: {type: object, properties: {query: {type: string}}} }, { name: calculate, description: 当用户需要进行数学计算、算术、单位换算时使用此工具。, parameters: {type: object, properties: {expression: {type: string}}} } ]6.2 实现智能体推理循环修改后端的/chat端点或创建一个新的/agent_chat端点实现一个简单的ReAct循环# backend/main.py (新增函数和端点) import json import re def parse_llm_for_action(llm_output: str): 解析LLM的输出看是否包含工具调用指令。 我们约定一个简单的格式 THOUGHT: ... ACTION: tool_name ARGS: {...} ANSWER: ... thought action None args {} answer thought_match re.search(rTHOUGHT:(.*?)(?ACTION:|ANSWER:|$), llm_output, re.DOTALL) action_match re.search(rACTION:(\w), llm_output) args_match re.search(rARGS:(\{.*?\}), llm_output, re.DOTALL) answer_match re.search(rANSWER:(.*), llm_output, re.DOTALL) if thought_match: thought thought_match.group(1).strip() if action_match: action action_match.group(1).strip() if args_match: try: args json.loads(args_match.group(1).strip()) except json.JSONDecodeError: args {} if answer_match: answer answer_match.group(1).strip() return thought, action, args, answer async def run_agent_loop(user_query: str, max_steps: int 3) - str: 运行一个简单的ReAct循环。 conversation_context f用户提问{user_query}\n\n你可以使用的工具{json.dumps(AVAILABLE_TOOLS, ensure_asciiFalse)} full_prompt f你是一个AI助手可以调用工具来帮助用户。请遵循以下格式思考 THOUGHT: 你需要思考用户的问题决定是否需要使用工具以及使用哪个工具。 ACTION: 如果需要使用工具在这里写出工具名。如果不需要写NONE。 ARGS: 如果需要使用工具在这里以JSON格式写出调用参数。如果不需要写{{}}。 ANSWER: 最终给用户的回答。如果你使用了工具请基于工具返回的结果进行总结。 {conversation_context} 请开始 history [] # 简化不携带长历史 steps 0 final_answer while steps max_steps: steps 1 llm_raw_response await get_llm_response(full_prompt, history) thought, action, args, answer parse_llm_for_action(llm_raw_response) logger.info(fStep {steps} - Thought: {thought}, Action: {action}, Args: {args}) if action and action ! NONE: # 执行工具调用 tool_result if action get_current_time: tool_result get_current_time(args.get(query, )) elif action search_web: tool_result search_web(args.get(query, user_query)) elif action calculate: tool_result calculate(args.get(expression, )) else: tool_result f未知工具{action} # 将工具结果加入到上下文中进行下一轮思考 full_prompt f\n\n工具 {action} 返回的结果是{tool_result} # 也可以将本轮交互加入history让LLM知道上下文 history.append({role: assistant, content: llm_raw_response}) # 模拟一个系统消息告知工具结果 history.append({role: system, content: fTool Result: {tool_result}}) else: # 不需要或无法使用工具直接返回答案 final_answer answer if answer else llm_raw_response break if steps max_steps: final_answer 经过多次尝试我仍然无法完美解决你的问题。请尝试更清晰地描述你的需求。 break return final_answer app.post(/agent_chat, response_modelAgentResponse) async def chat_with_agent_advanced(query: TextQuery): 使用智能体带工具调用处理用户查询。 user_text query.text if not user_text: raise HTTPException(status_code400, detailQuery text cannot be empty.) logger.info(fAgent processing query: {user_text}) ai_response_text await run_agent_loop(user_text) logger.info(fAgent final response: {ai_response_text}) return AgentResponse(textai_response_text)前端只需要将请求从/chat改为/agent_chat就能体验到智能体调用工具的能力如问“现在几点”或“计算一下125的平方根”。踩坑实录提示工程Prompt Engineering让LLM稳定地输出THOUGHT/ACTION/ARGS/ANSWER格式需要精心设计提示词。多轮调试是必不可少的。工具安全性calculate工具中直接使用eval()是极其危险的绝对不能用于生产环境。这里仅作演示真实项目必须使用安全的表达式求值库或沙箱。循环控制必须设置最大步数max_steps以防止LLM陷入无限循环。同时要处理工具调用失败的情况让智能体能够优雅降级。7. 部署、优化与常见问题7.1 本地与生产环境部署本地运行终端1在backend目录下uvicorn main:app --reload --host 0.0.0.0 --port 8000终端2在frontend目录下streamlit run app.py --server.port 8501浏览器访问http://localhost:8501生产部署考虑后端FastAPI使用uvicorn配合gunicornwith Uvicorn workers部署在Linux服务器上。用Nginx作为反向代理处理静态文件、SSL加密和负载均衡。# 使用gunicorn启动在backend目录 gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000前端StreamlitStreamlit应用本身可以作为服务运行。对于更正式的生产环境可以考虑将Streamlit前端“静态化”比较困难或者将其视为一个独立的服务并通过Nginx代理。Streamlit也支持配置server.address和server.port。环境变量所有API密钥、数据库连接字符串等敏感信息必须通过环境变量或安全的密钥管理服务传入绝不可写在代码中。进程管理使用systemd或supervisor来管理后端和前端进程确保它们崩溃后能自动重启。7.2 性能优化点Whisper模型加速使用faster-whisper基于CTranslate2替代原版whisper推理速度可提升数倍且内存占用更低。Groq模型选择Groq提供了多个模型。mixtral-8x7b-32768在速度和能力上平衡较好。对于简单任务可以尝试更小的llama2-70b-4096或gemma-7b-it以获得更快的响应。TTS缓存对相同的文本回复可以缓存其语音文件避免重复合成。可以在后端用字典或Redis存储文本MD5 - 音频文件路径的映射。前端异步请求Streamlit的st.button会导致整个脚本重跑。对于更流畅的体验可以考虑使用st.form或自定义组件来管理状态或者用JavaScript发起异步请求但这会显著增加前端复杂度。7.3 常见问题与排查问题1Streamlit前端报错“无法连接到后端服务”。检查确保后端FastAPI服务正在运行http://localhost:8000能访问。检查前端代码中的api_base地址是否正确。检查后端FastAPI是否启用了CORS并正确配置了允许前端源http://localhost:8501。问题2Whisper转录速度非常慢或者报错关于ffmpeg。检查确认系统已安装ffmpeg并可在命令行中调用。优化换用faster-whisper库并尝试使用更小的模型如tiny或base。注意首次加载Whisper模型会下载模型文件需要网络和时间。问题3Groq API调用返回429频率限制或401认证失败。检查.env文件中的GROQ_API_KEY是否正确是否在代码中被正确加载。检查Groq控制台查看API使用量和频率限制。免费额度有每分钟请求数RPM和每天令牌数TPD的限制。优化在前端添加请求节流Throttling避免用户快速连续点击。问题4合成的语音听起来很机械或者不是中文。检查pyttsx3的语音包。在Windows上可以到系统“语音设置”里查看和下载其他语音包。在Linux上可能需要安装espeak或festival及其中文语音数据。替代方案切换到edge-tts它提供更自然的中文语音如zh-CN-XiaoxiaoNeural但需要处理网络请求和异步。问题5智能体Agent经常不调用工具或者调用格式错误。调试打印出LLM每一步的完整输出llm_raw_response检查提示词是否清晰LLM是否理解了指令格式。优化提示词在提示词中提供更清晰的工具描述和1-2个完整的示例Few-shot Learning能大幅提升工具调用的准确率。简化工具初期尽量减少工具数量并确保工具的功能描述非常精确无歧义。这个项目从零搭建了一个完整的、可交互的语音AI智能体原型。它涵盖了从语音识别、大模型推理、工具调用到语音合成的全链路。你可以在此基础上继续扩展工具集如发送邮件、控制智能家居、增加长期记忆向量数据库、优化语音交互体验流式响应、语音唤醒甚至将其部署到云服务器通过手机随时随地访问。
