基于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的亮点在于其自研的LPU（Language 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 系统架构设计图（概念层）

虽然不能画图，但我们可以用文字清晰地描述数据流：

1. 用户交互层（Streamlit App）：用户打开浏览器，访问Streamlit应用。点击“开始录音”按钮，通过浏览器API录制语音。录音结束后，前端将音频数据（通常是WAV格式）通过HTTP POST请求发送到后端。
2. API服务层（FastAPI Server）：FastAPI接收到音频数据后，首先调用**语音转文本（Speech-to-Text, STT）**服务（如OpenAI Whisper API，或本地部署的faster-whisper）。将音频转换为文字指令。
3. 智能体核心层（Agent Core）：转换后的文字指令被送入智能体（Agent）逻辑模块。这个模块的核心是一个基于Groq LLM的“大脑”。我们采用类似ReAct或LangChain的思维框架，让LLM根据指令决定是否需要调用工具（如网络搜索、计算器、查数据库），并规划执行步骤。LLM与Groq API通信。
4. 工具执行层（Tools）：智能体调用相应的工具函数执行具体任务（如用requests库爬取天气信息，用wolfram-alpha库进行计算）。
5. 响应生成层：工具执行的结果返回给智能体，智能体再次通过Groq LLM组织成一段通顺、友好的自然语言回复。
6. 文本转语音层（Text-to-Speech, TTS）：FastAPI将最终的文本回复，调用文本转语音服务（如Google TTS, pyttsx3本地引擎，或Edge TTS）转换为音频文件。
7. 响应返回：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/ML（Groq、Whisper、TTS）、前端（Streamlit）。创建一个requirements.txt文件：

# 后端核心 fastapi==0.104.1 uvicorn[standard]==0.24.0 # ASGI服务器，用于运行FastAPI httpx==0.25.1 # 异步HTTP客户端，用于调用外部API python-multipart==0.0.6 # 处理文件上传（音频） # AI与语音处理 groq==0.3.0 # Groq官方SDK openai-whisper==20231117 # OpenAI Whisper语音识别 # 或者使用更快的 faster-whisper (需要额外安装CTranslate2) # faster-whisper==0.9.0 pyttsx3==2.90 # 离线文本转语音（跨平台） # 或者使用在线TTS，如 edge-tts # edge-tts==6.1.9 langchain==0.0.340 # 可选，用于构建智能体框架 langchain-groq==0.0.2 # 可选，LangChain的Groq集成 # 前端 streamlit==1.28.0 streamlit-webrtc==0.44.2 # 用于处理前端音频流（高级选项，基础版可先不用） # 工具与工具 requests==2.31.0 # 用于工具函数，如网络搜索 python-dotenv==1.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密钥。

1. 访问 Groq Cloud 注册账号。
2. 在控制台找到API Keys部分，创建一个新的密钥。
3. 在项目根目录创建一个名为.env的文件（注意前面的点），将密钥写入：

   GROQ_API_KEY=your_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(level=logging.INFO) logger = logging.getLogger(__name__) # 初始化FastAPI应用 app = FastAPI(title="Voice-Controlled AI Agent API", version="1.0.0") # 添加CORS中间件，允许Streamlit前端（默认端口8501）跨域访问 app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:8501"], # 生产环境需替换为实际域名 allow_credentials=True, 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_model=TextQuery) async def transcribe_audio(file: UploadFile = File(...)): """ 接收音频文件（WAV/MP3等），使用Whisper转换为文字。 """ if not file.content_type.startswith("audio/"): raise HTTPException(status_code=400, detail="File must be an audio file.") try: # 创建一个临时文件保存上传的音频 suffix = Path(file.filename).suffix with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) 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(f"Transcribed text: {transcribed_text}") return TextQuery(text=transcribed_text) except Exception as e: logger.error(f"Transcription failed: {e}") raise HTTPException(status_code=500, detail=f"Transcription error: {str(e)}")

实操心得：

1. 模型选择：base模型在大多数场景下准确度和速度都不错。如果你的应用场景对精度要求极高（如专业术语），可以用small或medium，但推理时间会显著增加。tiny最快，但精度有损失。
2. 异步处理：Whisper推理是CPU/GPU密集型同步操作，直接调用会阻塞FastAPI的异步事件循环，影响其他请求。使用asyncio.run_in_executor将其放到线程池中执行是关键。
3. 临时文件：一定要记得删除临时文件，否则服务器磁盘很快会被撑满。

4.3 集成Groq LLM与智能体逻辑

这是项目的“大脑”。我们先实现一个简单的、无工具调用的对话版本，再扩展成智能体。

# backend/main.py (续) from groq import Groq import json # 初始化Groq客户端 groq_client = Groq(api_key=os.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( messages=messages, model="mixtral-8x7b-32768", # Groq上速度很快的一个模型 temperature=0.7, max_tokens=500, stream=False, # 非流式，一次性返回 ) response_text = chat_completion.choices[0].message.content return response_text.strip() except Exception as e: logger.error(f"Groq API call failed: {e}") return "抱歉，我暂时无法处理你的请求。请稍后再试。" @app.post("/chat", response_model=AgentResponse) async def chat_with_agent(query: TextQuery): """ 接收用户文本查询，调用LLM生成回复。 """ user_text = query.text if not user_text: raise HTTPException(status_code=400, detail="Query text cannot be empty.") logger.info(f"Processing query: {user_text}") ai_response_text = await get_llm_response(user_text) logger.info(f"AI response: {ai_response_text}") # 目前先返回文本，音频URL将在/synthesize端点生成后补充 return AgentResponse(text=ai_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_ok=True) @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_code=400, detail="Text 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(path=filepath, media_type="audio/mpeg", filename=filename) except Exception as e: logger.error(f"Speech synthesis failed: {e}") raise HTTPException(status_code=500, detail=f"Speech synthesis error: {str(e)}") # 为了能直接访问生成的音频文件，添加一个静态文件路由（简易版） from fastapi.staticfiles import StaticFiles app.mount("/static", StaticFiles(directory="generated_audio"), name="static")

注意事项：

1. TTS引擎选择：pyttsx3是离线方案，方便但音质和自然度有限。edge-tts音质好，是异步的，但需要处理网络请求和临时文件。根据你的需求选择。这里为了简化，先用pyttsx3。
2. 文件管理：生成的音频文件会不断累积，需要定期清理。可以在/synthesize端点中添加逻辑，删除超过一定时间的旧文件，或者使用内存中的字节流而不保存到磁盘（更复杂）。
3. 静态文件服务：在开发阶段，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="🤖", layout="wide") 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"], format="audio/mp3") # 侧边栏用于配置和说明 with st.sidebar: st.header("设置") api_base = st.text_input("后端API地址", value="http://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_size="2x", pause_threshold=2.0) # 方法二（备选）：使用 streamlit-audiorecorder 或其他组件 # 这里以 audio_recorder 为例 if audio_bytes: # 显示一个临时音频播放器，让用户确认录音 st.audio(audio_bytes, format="audio/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", files=files) 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", json=ai_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, format="audio/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 处理音频流与改进用户体验

上面的实现是“录音-发送-等待-播放”的同步模式，用户体验有卡顿。我们可以做一些优化：

1. 实时反馈：在等待后端处理时，使用st.spinner和st.status显示明确的进度（“识别中...”、“思考中...”、“合成语音...”）。
2. 错误处理：对每个网络请求都添加更细致的try...except，并给用户友好的错误提示。
3. 音频自动播放：合成语音后，使用JavaScript自动播放（Streamlit原生支持有限，可能需要自定义组件）。
4. 对话历史持久化：将st.session_state.messages保存到本地文件或数据库，刷新页面不丢失。

一个改进的发送逻辑片段如下：

# 在 frontend/app.py 的发送按钮逻辑中改进 if st.button("发送录音", key="send_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只是一个“聊天机器人”，它不会主动调用工具。要升级为“智能体”，我们需要赋予它使用工具的能力。这里我们引入简单的ReAct（Reasoning + 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(r"THOUGHT:(.*?)(?=ACTION:|ANSWER:|$)", llm_output, re.DOTALL) action_match = re.search(r"ACTION:(\w+)", llm_output) args_match = re.search(r"ARGS:(\{.*?\})", llm_output, re.DOTALL) answer_match = re.search(r"ANSWER:(.*)", 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_ascii=False)}" 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(f"Step {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": f"Tool 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_model=AgentResponse) async def chat_with_agent_advanced(query: TextQuery): """ 使用智能体（带工具调用）处理用户查询。 """ user_text = query.text if not user_text: raise HTTPException(status_code=400, detail="Query text cannot be empty.") logger.info(f"Agent processing query: {user_text}") ai_response_text = await run_agent_loop(user_text) logger.info(f"Agent final response: {ai_response_text}") return AgentResponse(text=ai_response_text)

前端只需要将请求从/chat改为/agent_chat，就能体验到智能体调用工具的能力（如问“现在几点？”或“计算一下125的平方根”）。

踩坑实录：

1. 提示工程（Prompt Engineering）：让LLM稳定地输出THOUGHT/ACTION/ARGS/ANSWER格式需要精心设计提示词。多轮调试是必不可少的。
2. 工具安全性：calculate工具中直接使用eval()是极其危险的，绝对不能用于生产环境。这里仅作演示，真实项目必须使用安全的表达式求值库或沙箱。
3. 循环控制：必须设置最大步数（max_steps）以防止LLM陷入无限循环。同时，要处理工具调用失败的情况，让智能体能够优雅降级。

7. 部署、优化与常见问题

7.1 本地与生产环境部署

本地运行：

1. 终端1：在backend目录下uvicorn main:app --reload --host 0.0.0.0 --port 8000
2. 终端2：在frontend目录下streamlit run app.py --server.port 8501
3. 浏览器访问http://localhost:8501

生产部署考虑：

• 后端（FastAPI）：使用uvicorn配合gunicorn（with Uvicorn workers）部署在Linux服务器上。用Nginx作为反向代理，处理静态文件、SSL加密和负载均衡。

  # 使用gunicorn启动（在backend目录） gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000

• 前端（Streamlit）：Streamlit应用本身可以作为服务运行。对于更正式的生产环境，可以考虑将Streamlit前端“静态化”（比较困难），或者将其视为一个独立的服务，并通过Nginx代理。Streamlit也支持配置server.address和server.port。
• 环境变量：所有API密钥、数据库连接字符串等敏感信息必须通过环境变量或安全的密钥管理服务传入，绝不可写在代码中。
• 进程管理：使用systemd或supervisor来管理后端和前端进程，确保它们崩溃后能自动重启。

7.2 性能优化点

1. Whisper模型加速：使用faster-whisper（基于CTranslate2）替代原版whisper，推理速度可提升数倍，且内存占用更低。
2. Groq模型选择：Groq提供了多个模型。mixtral-8x7b-32768在速度和能力上平衡较好。对于简单任务，可以尝试更小的llama2-70b-4096或gemma-7b-it以获得更快的响应。
3. TTS缓存：对相同的文本回复，可以缓存其语音文件，避免重复合成。可以在后端用字典或Redis存储文本MD5 -> 音频文件路径的映射。
4. 前端异步请求：Streamlit的st.button会导致整个脚本重跑。对于更流畅的体验，可以考虑使用st.form或自定义组件来管理状态，或者用JavaScript发起异步请求，但这会显著增加前端复杂度。

7.3 常见问题与排查

问题1：Streamlit前端报错“无法连接到后端服务”。

• 检查：确保后端FastAPI服务正在运行（http://localhost:8000能访问）。
• 检查：前端代码中的api_base地址是否正确。
• 检查：后端FastAPI是否启用了CORS，并正确配置了允许前端源（http://localhost:8501）。

问题2：Whisper转录速度非常慢，或者报错关于ffmpeg。

• 检查：确认系统已安装ffmpeg并可在命令行中调用。
• 优化：换用faster-whisper库，并尝试使用更小的模型（如tiny或base）。
• 注意：首次加载Whisper模型会下载模型文件，需要网络和时间。

问题3：Groq 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智能体原型。它涵盖了从语音识别、大模型推理、工具调用到语音合成的全链路。你可以在此基础上，继续扩展工具集（如发送邮件、控制智能家居）、增加长期记忆（向量数据库）、优化语音交互体验（流式响应、语音唤醒），甚至将其部署到云服务器，通过手机随时随地访问。