1. 项目概述从零构建你的AI个人助手Rika大家好我是Krish一个喜欢捣鼓代码和DIY项目的学生。今天我想和大家分享一个我最近完成的项目——Rika一个基于Groq API和LLaMA模型的多功能AI个人助手。这个项目源于一个简单的想法能不能有一个助手不仅能像ChatGPT一样聊天还能主动告诉我时间、天气、新闻甚至能一边听我说话一边执行任务感觉就像有个真正的伙伴在身边经过一段时间的摸索和调试Rika从概念变成了现实。她不仅能处理多轮对话还能整合外部信息并以一种更自然、更“人性化”的方式与你互动。如果你对Python编程、API调用以及如何将前沿的大语言模型LLM应用到实际项目中感兴趣那么这篇分享或许能给你带来一些直接的启发和可复现的步骤。Rika的核心是让AI能力变得触手可及且实用。我们不再需要从零开始训练一个庞大的模型而是通过Groq这样的高速推理API直接调用像LLaMA 3.2这样成熟的模型再结合Python的灵活性将天气、新闻等实时数据“喂”给AI让它生成符合上下文的、有用的回复。整个项目涉及逻辑设计、API集成、用户界面UI构建以及一些提升体验的“小魔法”。无论你是想学习如何将多个API服务串联起来还是想打造一个属于自己的桌面助手我相信下面的内容都能提供一个清晰的路线图。2. 核心架构与设计思路拆解在动手写代码之前理清整个系统的架构至关重要。Rika不是一个单一功能的脚本而是一个需要协调多个组件协同工作的系统。我的设计目标是响应快、功能多、体验自然。基于此我选择了客户端-服务器模型与事件驱动编程相结合的架构。2.1 技术栈选型背后的考量为什么选择Groq API和LLaMA模型这背后有几个实际的考虑。首先Groq API以其惊人的推理速度著称。对于个人助手这类需要实时交互的应用响应延迟是用户体验的杀手。Groq基于其自研的LPU语言处理单元推理引擎能在毫秒级返回LLaMA模型的生成结果这远比使用一些云端服务通过传统GPU推理要快得多。这意味着当用户问“今天天气如何”时Rika几乎可以瞬间开始组织回答而不是让用户等待数秒。其次LLaMA系列模型特别是LLaMA 3.2在开源模型中达到了一个很好的平衡点。它在保持较强对话和推理能力的同时模型尺寸相对可控API调用成本也更为合理。相比于动辄上千亿参数的闭源模型LLaMA 3.2 90B或更小的70B版本已经能够出色地完成日常问答、信息归纳和指令跟随任务。对于个人项目来说性能和成本的平衡是关键。最后整个项目用Python实现这是最自然的选择。Python拥有极其丰富的库生态无论是处理HTTP请求的requests库还是构建图形界面的tkinter/PyQt或是处理音频的pyttsx3、SpeechRecognition都能找到成熟、易用的解决方案。它允许我们快速搭建原型并迭代。注意Groq API目前提供免费的额度供开发者使用但对于高频调用需要留意其定价策略。同时LLaMA模型的知识截止日期是固定的它不知道这之后的事件这就是为什么我们需要集成实时新闻API来弥补这一缺陷。2.2 系统模块化设计我将Rika分解为几个核心模块每个模块负责单一职责通过清晰的接口进行通信用户交互模块负责所有输入输出。包括图形用户界面GUI用于显示对话和按钮操作语音输入模块用于接收语音指令以及文本转语音TTS模块用于语音回复。这个模块是用户感知Rika的窗口。核心逻辑控制模块这是Rika的“大脑”。它监听来自交互模块的事件如用户发送了一条文本消息或语音指令然后协调其他模块工作。它决定何时调用AI何时获取外部数据并处理多任务队列。AI服务模块封装了与Groq API的通信。它接收来自控制模块的、包含上下文和指令的提示词Prompt发送请求解析返回的AI回复并处理可能出现的错误如网络超时、额度不足。外部数据服务模块这是一个“信息收集器”。它独立地、按需或定时从外部API获取数据如从OpenWeatherMap获取天气从NewsAPI获取头条新闻以及从系统获取当前时间日期。这些数据会被格式化后存入一个共享的“上下文池”。上下文管理与提示工程模块这是提升AI表现的核心。它维护一个动态的对话历史窗口并将外部数据模块提供的实时信息与预先定义好的“系统指令”即Rika的“人设”进行组合生成最终发送给AI的提示词。好的提示词能极大地引导AI的行为符合预期。这种模块化设计的好处是显而易见的高内聚、低耦合。例如如果我想把天气API从A家换成B家我只需要修改外部数据服务模块中的相应函数其他模块完全不受影响。如果想增加新的功能如查股票也只需新增一个数据服务并更新提示词即可。3. 核心细节解析与实操要点理解了整体架构我们深入到几个关键环节看看具体是如何实现的以及其中有哪些容易踩坑的地方。3.1 Groq API的集成与提示词工程与Groq API交互本身很简单但其效果的优劣几乎完全取决于你发送给它的“提示词”Prompt。这不仅仅是问一个问题那么简单。API调用基础 首先你需要在Groq官网注册并获取API密钥。调用其Chat Completion接口通常使用requests库。一个最基础的请求示例如下import requests import json def ask_groq(api_key, messages, modelllama3-70b-8192): url https://api.groq.com/openai/v1/chat/completions headers { Authorization: fBearer {api_key}, Content-Type: application/json } data { model: model, # 例如 llama3-70b-8192, llama3.1-70b-versatile messages: messages, # 这是一个消息列表 temperature: 0.7, # 控制创造性越高回答越随机 max_tokens: 1024 } response requests.post(url, headersheaders, jsondata) return response.json()这里的messages参数是关键。它通常是一个列表包含多个字典每个字典有role角色和content内容两个字段。角色可以是system、user、assistant。构建Rika的“人设”提示词 对于Rika我构建了一个复杂的系统指令systemmessage这是塑造其行为的基础。system_prompt 你是一个名为Rika的AI个人助手。你的性格友好、热情且乐于助人对话风格自然像朋友一样。 你拥有以下实时信息可以在用户询问时提供 - 当前时间{current_time} - 当前日期{current_date} - 本周天气概况{weather_summary} - 今日新闻头条{news_headlines} **重要行为准则** 1. 除非用户明确询问否则不要主动提及时间、日期、天气或新闻。这些信息是供你调用的背景知识不是开场白。 2. 你可以同时处理多个任务。当用户提出一个复合请求时例如“告诉我天气并写个备忘录”请先清晰回复然后在后台执行写入操作。 3. 你的回复应当简洁、直接避免冗长的开场白和结束语。 4. 如果用户要求你记录内容到记事本请在回复中确认例如“已为您记录……”。 请注意{current_time}、{weather_summary}这些是占位符在实际调用前需要用真实数据从“上下文池”中替换掉。这个系统指令做了几件重要的事定义身份和风格让AI知道自己是谁该怎么说话。注入实时知识弥补了LLM的静态知识缺陷。设定行为边界明确告诉AI什么该做什么不该做如不主动播报信息这是避免AI“胡言乱语”的关键。支持多任务预先告知AI需要具备这种能力。在实际对话中messages列表会像这样增长messages [ {role: system, content: system_prompt_with_real_data}, {role: user, content: 你好Rika今天上海天气怎么样}, {role: assistant, content: 根据实时信息上海今天多云气温在22到28度之间微风。是个不错的日子}, {role: user, content: 不错。顺便帮我记一下下午三点开会。} # 这是最新的用户输入 ]每次新的用户输入到来我们都会将整个对话历史或最近N轮以防超出token限制连同最新的系统指令一起发送。这保证了AI拥有完整的对话上下文。实操心得temperature参数对助手类应用至关重要。我建议设置在0.5到0.8之间。太低如0.1会让回答过于死板和重复太高如1.2则可能让回答变得天马行空不符合助手身份。多试几次找到最适合你“人设”的值。3.2 实现“边听边说”与多任务处理这是让Rika感觉更“智能”和“自然”的两个特性。“边听边说”的异步处理 传统的语音助手流程是听完→思考→回答→结束。Rika则希望在回答的同时就能开始监听用户的下一条指令。这在Python中可以通过多线程或异步编程来实现。我选择使用threading模块因为它概念相对简单。基本逻辑如下当用户按下语音输入按钮或说出唤醒词时主线程启动语音识别。识别出文本后主线程将其放入一个任务队列并立即启动语音合成TTS来播报“我正在处理”之类的反馈同时启动一个新的监听线程。另一个工作线程或主线程从任务队列中取出任务调用AI和外部服务生成最终答复。当最终答复生成后如果当前没有在播报其他内容则立即播报它如果新的监听线程收到了用户的下一条指令则根据优先级决定是打断当前播报还是加入队列。import threading import queue import time class RikaAssistant: def __init__(self): self.task_queue queue.Queue() self.is_speaking False self.listening_thread None def listen_and_queue(self): # 这里是语音识别代码 user_input speech_recognizer.listen() if user_input: self.task_queue.put(user_input) print(f任务已排队: {user_input}) # 立即给出语音反馈并开始新的监听 self.speak_async(收到) self.start_listening() # 非阻塞地启动新的监听线程 def start_listening(self): if not self.is_speaking: # 如果没在说话可以安全开始监听 self.listening_thread threading.Thread(targetself.listen_and_queue, daemonTrue) self.listening_thread.start() def process_queue(self): while True: task self.task_queue.get() # 处理任务调用AI执行命令 result self.process_task(task) self.speak_async(result) # 播报结果 self.task_queue.task_done() def speak_async(self, text): # 在另一个线程中播报不阻塞主流程 def _speak(): self.is_speaking True tts_engine.say(text) tts_engine.runAndWait() self.is_speaking False threading.Thread(target_speak, daemonTrue).start()多任务处理的逻辑 多任务分为两个层面AI理解层面的多任务用户说“查天气并写备忘录”。在提示词中我们已经要求AI能处理复合请求。AI的回复应该是结构化的例如“上海今天晴26度。好的已为您将‘下午三点开会’记录到备忘录。” 这需要AI在单次回复中完成信息查询和任务确认。系统执行层面的多任务对于AI回复中确认要执行的任务如“写备忘录”控制模块需要解析出这个意图并触发对应的执行函数如一个写入记事本的函数write_to_notepad(memo_content)。这个执行过程应该是异步的不能阻塞对话流。实现时我让AI在回复中对于需要执行的动作使用特定的关键词或格式标记例如[ACTION:write_notepad]内容下午三点开会[/ACTION]。控制模块在收到AI回复后先提取文本部分播报给用户同时用正则表达式或简单字符串匹配找出动作标记在后台线程中执行相应操作。注意事项多线程编程需要小心处理共享资源如is_speaking标志和线程安全。使用queue.Queue是线程安全的是个好选择。另外语音识别和合成库可能对线程有特定要求需要查阅其文档。3.3 外部数据API的集成Rika的“实时性”依赖于外部API。这里以天气和新闻为例。天气API以OpenWeatherMap为例 你需要注册获取API Key。调用其current weather data接口。import requests import json from datetime import datetime def get_weather(api_key, cityShanghai): base_url http://api.openweathermap.org/data/2.5/weather? complete_url f{base_url}q{city}appid{api_key}unitsmetric # unitsmetric 得到摄氏度 try: response requests.get(complete_url, timeout5) response.raise_for_status() # 检查HTTP错误 data response.json() if data[cod] ! 200: return 无法获取天气信息。 main data[main] weather_desc data[weather][0][description] temp main[temp] feels_like main[feels_like] humidity main[humidity] # 格式化成一句友好的话 summary f{city}当前天气{weather_desc}气温{temp}度体感{feels_like}度湿度{humidity}%。 return summary except requests.exceptions.RequestException as e: print(f获取天气请求失败: {e}) return 天气服务暂时不可用。 except (KeyError, json.JSONDecodeError) as e: print(f解析天气数据失败: {e}) return 天气信息解析错误。新闻API以NewsAPI为例 同样需要注册获取Key。通常有免费额度限制调用次数。def get_news_headlines(api_key, countryus, categorygeneral, num_headlines3): url fhttps://newsapi.org/v2/top-headlines?country{country}category{category}apiKey{api_key} try: response requests.get(url, timeout5) response.raise_for_status() data response.json() if data[status] ! ok: return 暂无新闻。 articles data[articles][:num_headlines] headlines [art[title] for art in articles if art[title]] # 合并成一段文本用分号隔开 summary .join(headlines) return summary if summary else 暂无重要新闻。 except requests.exceptions.RequestException as e: print(f获取新闻请求失败: {e}) return 新闻服务暂时不可用。数据更新策略 不应该每次用户提问都去调用这些API这会造成延迟和额度浪费。我采用了一个简单的缓存策略在程序启动时获取一次然后启动一个后台定时线程每15-30分钟更新一次天气和新闻数据并将其存储在全局变量或一个小的缓存类中。当构造系统提示词时直接从缓存中读取这些信息。import threading import time class DataCache: def __init__(self): self.weather 正在获取天气... self.news 正在获取新闻... self.last_updated None def update_all(self, weather_api_key, news_api_key): self.weather get_weather(weather_api_key) self.news get_news_headlines(news_api_key) self.last_updated time.time() print(数据缓存已更新。) def background_update_task(cache, weather_key, news_key, interval1800): # 1800秒30分钟 while True: time.sleep(interval) cache.update_all(weather_key, news_key) # 在主程序中 cache DataCache() # 立即更新一次 cache.update_all(WEATHER_API_KEY, NEWS_API_KEY) # 启动后台更新线程 update_thread threading.Thread(targetbackground_update_task, args(cache, WEATHER_API_KEY, NEWS_API_KEY), daemonTrue) update_thread.start()4. 实操过程与核心环节实现现在让我们把上述模块组装起来看看一个完整的用户交互周期是如何运行的。我将以一次典型的语音交互为例拆解从用户开口到Rika执行完毕的全过程。4.1 环境准备与依赖安装首先确保你的Python环境建议3.8以上已经就绪。创建一个新的项目目录并初始化一个虚拟环境是个好习惯。mkdir rika-assistant cd rika-assistant python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心依赖库。创建一个requirements.txt文件内容如下requests2.28.0 # 用于HTTP请求Groq API 天气/新闻API pyttsx32.90 # 跨平台的文本转语音库 SpeechRecognition3.10.0 # 语音识别库默认使用Google Web API离线需配置 pyaudio0.2.11 # 语音识别所需的音频输入库安装可能需额外步骤 python-dotenv1.0.0 # 用于从.env文件加载API密钥等敏感信息然后使用pip安装pip install -r requirements.txt注意pyaudio在某些系统上可能需要先安装系统级的音频开发包。例如在Ubuntu上你可能需要先运行sudo apt-get install portaudio19-dev python3-pyaudio。在Windows上通常可以直接通过pip安装预编译的wheel文件。4.2 核心代码结构与主循环项目的主要代码可以组织在一个主文件如rika.py和几个模块文件中。这里为了演示我将核心逻辑浓缩在一个类中。# rika_core.py import threading import queue import time import json import re from datetime import datetime import requests import pyttsx3 import speech_recognition as sr from dotenv import load_dotenv import os load_dotenv() # 从.env文件加载环境变量 class RikaAssistant: def __init__(self): # 1. 初始化配置和状态 self.groq_api_key os.getenv(GROQ_API_KEY) self.weather_api_key os.getenv(WEATHER_API_KEY) self.news_api_key os.getenv(NEWS_API_KEY) self.model llama3-70b-8192 # 2. 初始化组件 self.tts_engine pyttsx3.init() self.recognizer sr.Recognizer() self.microphone sr.Microphone() # 3. 初始化任务队列和缓存 self.task_queue queue.Queue() self.cache { weather: 正在初始化..., news: 正在初始化..., time: } self.is_speaking False self.conversation_history [] # 保存最近的对话 # 4. 启动后台服务 self._update_cache() # 首次更新数据 self._start_background_updater() self._start_queue_processor() def _update_cache(self): 更新天气、新闻和时间缓存 self.cache[time] datetime.now().strftime(%Y年%m月%d日 %H:%M:%S) self.cache[weather] self._fetch_weather() self.cache[news] self._fetch_news_headlines() def _fetch_weather(self): # ... 同前文的get_weather函数使用self.weather_api_key pass def _fetch_news_headlines(self): # ... 同前文的get_news_headlines函数使用self.news_api_key pass def _start_background_updater(self): 启动后台数据更新线程 def updater(): while True: time.sleep(1800) # 30分钟 self._update_cache() print([后台] 数据缓存已更新。) thread threading.Thread(targetupdater, daemonTrue) thread.start() def _build_messages(self, user_input): 构建发送给Groq API的messages列表 system_prompt f 你是Rika一个友好、高效的AI助手。当前信息如下 - 时间{self.cache[time]} - 天气{self.cache[weather]} - 新闻{self.cache[news]} 请根据对话历史和当前信息以自然的口语化风格回答用户。如果用户要求执行操作如记录到记事本请在回复中明确确认并使用[ACTION:xxx]标签标明要执行的动作。 # 构建历史消息只保留最近5轮对话以防token超限 history_messages [] for role, content in self.conversation_history[-10:]: # 保留最多5轮每轮2条消息 history_messages.append({role: role, content: content}) messages [{role: system, content: system_prompt}] messages.extend(history_messages) messages.append({role: user, content: user_input}) return messages def _call_groq_api(self, messages): 调用Groq API并返回回复文本 url https://api.groq.com/openai/v1/chat/completions headers { Authorization: fBearer {self.groq_api_key}, Content-Type: application/json } data { model: self.model, messages: messages, temperature: 0.7, max_tokens: 1024 } try: response requests.post(url, headersheaders, jsondata, timeout10) response.raise_for_status() result response.json() return result[choices][0][message][content] except requests.exceptions.RequestException as e: return f抱歉网络或服务出现异常{e} except (KeyError, IndexError, json.JSONDecodeError) as e: return 抱歉处理AI回复时出现错误。 def _execute_action(self, action_text): 解析并执行AI回复中标记的动作 # 简单正则匹配 [ACTION:write_notepad]内容[/ACTION] pattern r\[ACTION:(\w)\](.*?)\[/ACTION\] matches re.findall(pattern, action_text, re.DOTALL) for action_type, content in matches: content content.strip() if action_type write_notepad: self._write_to_notepad(content) print(f[动作执行] 已写入记事本: {content}) # 可以扩展其他动作类型如 open_browser, set_reminder 等 def _write_to_notepad(self, content): 将内容写入到本地文本文件 filename rika_notes.txt with open(filename, a, encodingutf-8) as f: timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S) f.write(f[{timestamp}] {content}\n) def _speak_async(self, text): 异步语音播报 if not text: return def speak(): self.is_speaking True self.tts_engine.say(text) self.tts_engine.runAndWait() self.is_speaking False threading.Thread(targetspeak, daemonTrue).start() def _start_queue_processor(self): 启动任务队列处理线程 def processor(): while True: user_input self.task_queue.get() print(f[处理队列] 开始处理: {user_input}) # 1. 构建消息并调用AI messages self._build_messages(user_input) ai_response self._call_groq_api(messages) # 2. 更新对话历史 self.conversation_history.append((user, user_input)) self.conversation_history.append((assistant, ai_response)) # 3. 提取纯文本回复去除动作标签并播报 # 先移除动作标签得到纯文本 pure_response re.sub(r\[ACTION:\w\].*?\[/ACTION\], , ai_response, flagsre.DOTALL).strip() if pure_response: self._speak_async(pure_response) else: self._speak_async(操作已完成。) # 4. 执行动作如果有 self._execute_action(ai_response) self.task_queue.task_done() print(f[处理队列] 处理完成。) thread threading.Thread(targetprocessor, daemonTrue) thread.start() def listen_and_queue(self): 监听语音输入并将识别文本加入队列 print([监听] 请说话...) try: with self.microphone as source: self.recognizer.adjust_for_ambient_noise(source, duration0.5) audio self.recognizer.listen(source, timeout5, phrase_time_limit10) text self.recognizer.recognize_google(audio, languagezh-CN) # 中文识别 print(f[识别结果] {text}) if text: self.task_queue.put(text) # 立即给出一个简短的语音反馈同时允许新的监听 self._speak_async(嗯) # 如果当前没在说长句子可以开始新的监听这里简化处理实际可更复杂 if not self.is_speaking: threading.Thread(targetself.listen_and_queue, daemonTrue).start() except sr.WaitTimeoutError: print([监听] 超时未检测到语音。) except sr.UnknownValueError: print([监听] 无法理解音频。) self._speak_async(我没听清) except sr.RequestError as e: print(f[监听] 语音识别服务错误: {e}) self._speak_async(语音服务出错) def run_cli(self): 运行命令行交互界面 print(Rika助手已启动输入文字与我对话或输入 listen 开始语音输入输入 quit 退出。) while True: user_input input(\n你: ).strip() if user_input.lower() quit: break elif user_input.lower() listen: self.listen_and_queue() else: # 将文本输入也加入队列处理 self.task_queue.put(user_input) # 简单等待一下避免输入和输出混杂 time.sleep(0.5) if __name__ __main__: assistant RikaAssistant() assistant.run_cli()这个RikaAssistant类集成了我们讨论的大部分核心功能。run_cli方法提供了一个简单的命令行界面进行测试。你可以输入文字或者输入listen来触发语音输入。4.3 图形用户界面GUI构建示例命令行适合测试但一个桌面助手需要友好的界面。这里使用Python内置的tkinter库创建一个简单的GUI。# rika_gui.py import tkinter as tk from tkinter import scrolledtext, ttk import threading from rika_core import RikaAssistant # 导入我们之前写的核心类 class RikaGUI: def __init__(self, master): self.master master master.title(Rika - 个人助手) master.geometry(600x700) # 初始化核心助手 self.assistant RikaAssistant() # 创建UI组件 self.create_widgets() # 绑定关闭事件 master.protocol(WM_DELETE_WINDOW, self.on_closing) def create_widgets(self): # 对话显示区域 self.conversation_text scrolledtext.ScrolledText(self.master, wraptk.WORD, statedisabled, height20) self.conversation_text.pack(padx10, pady10, filltk.BOTH, expandTrue) # 输入区域框架 input_frame tk.Frame(self.master) input_frame.pack(padx10, pady5, filltk.X) self.user_input tk.Entry(input_frame) self.user_input.pack(sidetk.LEFT, filltk.X, expandTrue) self.user_input.bind(Return, self.send_text) # 按回车发送 send_btn tk.Button(input_frame, text发送, commandself.send_text) send_btn.pack(sidetk.RIGHT, padx(5,0)) # 按钮区域框架 button_frame tk.Frame(self.master) button_frame.pack(padx10, pady5) listen_btn tk.Button(button_frame, text 语音输入, commandself.start_listening, width15) listen_btn.pack(sidetk.LEFT, padx5) clear_btn tk.Button(button_frame, text清空对话, commandself.clear_conversation, width15) clear_btn.pack(sidetk.LEFT, padx5) # 状态栏 self.status_var tk.StringVar(value就绪) status_bar tk.Label(self.master, textvariableself.status_var, bd1, relieftk.SUNKEN, anchortk.W) status_bar.pack(sidetk.BOTTOM, filltk.X) def append_to_conversation(self, sender, message): 在对话区域添加一条消息 self.conversation_text.config(statenormal) self.conversation_text.insert(tk.END, f{sender}: {message}\n\n) self.conversation_text.config(statedisabled) self.conversation_text.see(tk.END) # 滚动到底部 def send_text(self, eventNone): 发送文本框中的文字 text self.user_input.get().strip() if not text: return self.user_input.delete(0, tk.END) self.append_to_conversation(你, text) # 在新线程中处理任务避免GUI卡顿 threading.Thread(targetself.process_user_input, args(text,), daemonTrue).start() def process_user_input(self, text): 处理用户输入在线程中 self.status_var.set(正在思考...) # 这里简化处理实际应调用助手的队列 # 为了演示我们模拟调用核心逻辑 messages self.assistant._build_messages(text) response self.assistant._call_groq_api(messages) # 更新GUI必须在主线程中进行 self.master.after(0, self.display_response, response) self.assistant._execute_action(response) # 执行动作 def display_response(self, response): 在主线程中显示AI回复 pure_response re.sub(r\[ACTION:\w\].*?\[/ACTION\], , response, flagsre.DOTALL).strip() self.append_to_conversation(Rika, pure_response) self.status_var.set(就绪) # 异步语音播报 self.assistant._speak_async(pure_response) def start_listening(self): 开始语音监听 self.status_var.set(正在聆听...) # 在新线程中运行监听避免阻塞GUI threading.Thread(targetself._listen_thread, daemonTrue).start() def _listen_thread(self): 语音监听的线程函数 self.assistant.listen_and_queue() # 监听完成后状态恢复这里需要更精细的事件驱动更新此处简化 self.master.after(100, lambda: self.status_var.set(就绪)) def clear_conversation(self): 清空对话历史 self.conversation_text.config(statenormal) self.conversation_text.delete(1.0, tk.END) self.conversation_text.config(statedisabled) self.assistant.conversation_history.clear() def on_closing(self): 关闭窗口时的清理工作 # 可以在这里添加保存对话历史等操作 self.master.destroy() if __name__ __main__: root tk.Tk() gui RikaGUI(root) root.mainloop()这个GUI提供了一个文本框显示对话一个输入框发送文字以及按钮进行语音输入和清空对话。它将核心的RikaAssistant类封装起来并通过多线程确保GUI在处理AI请求和语音识别时不会卡死。5. 常见问题与排查技巧实录在开发和测试Rika的过程中我遇到了不少典型问题。这里将它们整理出来并提供我的解决思路希望能帮你绕过这些坑。5.1 API调用与网络问题问题1Groq API请求返回401 Unauthorized错误。排查这几乎总是API密钥错误。首先检查你的.env文件中的GROQ_API_KEY变量名是否与代码中os.getenv(GROQ_API_KEY)读取的完全一致注意大小写。其次确认密钥本身是否正确可以登录Groq控制台重新复制一份。最后确保密钥没有意外地包含空格或换行符。技巧在代码开头添加一个简单的测试请求如果失败则打印明确的错误信息并退出避免后续流程混乱。def test_groq_key(api_key): test_url https://api.groq.com/openai/v1/models headers {Authorization: fBearer {api_key}} try: resp requests.get(test_url, headersheaders, timeout5) if resp.status_code 200: print(Groq API密钥验证成功。) return True else: print(fGroq API密钥验证失败状态码{resp.status_code}) return False except Exception as e: print(fGroq API连接测试失败{e}) return False问题2天气或新闻API调用超时或返回异常数据。排查网络连通性首先用浏览器或curl命令手动测试API端点看是否能正常返回数据。API密钥与参数检查API密钥是否有效、是否过期某些免费API有期限。检查请求的城市名、国家代码等参数是否符合API文档要求例如OpenWeatherMap的城市名需要英文。额度限制免费API通常有调用频率限制。如果短时间内频繁调用会被限制。检查API返回的错误信息通常会提示429 Too Many Requests或402 Payment Required等。解决为所有网络请求添加合理的超时设置如timeout5并做好异常捕获。实现退避重试机制。例如第一次失败后等待2秒再试最多重试3次。加强缓存机制。对于天气这种变化不频繁的数据缓存时间可以更长如1小时。对于新闻可以适当缩短如15分钟。在代码中记录失败日志便于后期分析。5.2 语音识别与合成问题问题3语音识别SpeechRecognition无法工作报错或识别不出内容。排查麦克风权限确保操作系统已授予Python程序或终端麦克风访问权限。默认麦克风sr.Microphone()默认使用系统首选麦克风。如果有多麦克风可能需要指定设备索引。你可以用sr.Microphone.list_microphone_names()列出所有设备。环境噪音在嘈杂环境中识别率会急剧下降。adjust_for_ambient_noise方法有助于校准但并非万能。网络问题默认的recognize_google需要联网因为它将音频数据发送到Google的服务器进行识别。如果网络不通会失败。解决指定麦克风microphone sr.Microphone(device_index1)索引从0开始。尝试离线引擎安装pocketsphinx包pip install pocketsphinx并使用recognizer.recognize_sphinx(audio)。但请注意离线引擎的中文识别准确率通常远低于在线引擎。增加语音指令的清晰度在代码中引导用户例如先说“叮”一声再开始识别或者要求用户说完后停顿一下。问题4文本转语音pyttsx3语音不自然、语速过快或没有声音。排查语音引擎pyttsx3在Windows上默认使用SAPI5在macOS上使用NSSpeechSynthesizer在Linux上使用eSpeak。不同引擎和声音库的质量差异很大。语速设置默认语速可能不适合中文。解决engine pyttsx3.init() # 获取当前语音速率 rate engine.getProperty(rate) engine.setProperty(rate, rate - 30) # 调慢语速 # 获取并选择语音如果有多个 voices engine.getProperty(voices) # 在Windows上可以尝试寻找更自然的中文语音包如微软晓晓 # for voice in voices: # print(voice.id) # 如果找到了中文语音ID可以设置 # engine.setProperty(voice, voices[1].id) # 例如考虑更强大的TTS方案如果对音质要求高可以研究edge-tts调用微软Edge的在线TTS音质好但需联网或付费的云服务如Azure Cognitive Services。5.3 多线程与并发控制问题5程序出现随机崩溃或“卡死”尤其是在频繁使用语音功能时。排查这很可能是多线程编程中的经典问题——竞态条件或线程阻塞。例如语音合成引擎pyttsx3的runAndWait()是阻塞调用如果在主线程中调用整个GUI就会卡住直到说完。又或者多个线程同时修改is_speaking这个状态标志。解决严格遵守GUI线程规则所有更新Tkinter界面的操作如修改文本、更新标签都必须在主线程中执行。使用self.master.after()方法将任务派发到主线程。使用线程安全的数据结构对于任务队列坚持使用queue.Queue它是线程安全的。精细控制语音状态使用threading.Lock锁来保护is_speaking等共享状态变量确保同一时间只有一个线程能修改它。避免死锁确保锁的获取和释放是成对的并且逻辑清晰。问题6AI回复慢导致语音播报和后续监听错乱。排查Groq API虽然快但网络波动或复杂问题仍可能导致响应时间超过1-2秒。如果设计成“AI回复完才允许下一次监听”用户体验会打折扣。解决这正是引入任务队列和异步播报的原因。将用户输入放入队列后立即返回让一个独立的工作线程去处理耗时的AI调用和动作执行。语音播报也使用独立线程。这样主监听循环或GUI就能始终保持响应。关键在于设计好状态机例如“正在播报时新的语音输入先缓存播报结束后再处理”这比简单的“边听边说”更复杂但更稳健。5.4 提示词与AI行为问题问题7AI经常主动说出时间、天气等信息即使没问它。排查这几乎肯定是提示词System Prompt没写清楚。LLM会倾向于使用你提供给它的所有信息。如果你在系统指令里提供了时间天气又没有明确禁止它主动提及它可能会认为“我有这些信息应该告诉用户”。解决在系统指令中使用非常明确、强硬的指令。例如“重要你拥有以下实时信息[时间、天气、新闻]。除非用户明确询问否则绝对不要主动提及或播报这些信息。仅在用户直接提问时才将这些信息融入你的回答中。” 多次强调并放在指令靠前的位置。问题8AI对于“写备忘录”这类指令有时只回复“好的”却不执行动作。排查AI的理解可能不一致。有时它认为回复“好的”就算完成任务有时它会明确说“已记录”。我们的动作解析器依赖于固定的标签格式如[ACTION:...]。解决强化提示词在系统指令中更具体地规定动作格式。“如果用户要求你执行操作如记录到记事本你必须在回复的最后使用以下精确格式标明要执行的动作[ACTION:write_notepad]这里是要记录的内容[/ACTION]。你的自然语言回复应放在动作标签之前。”改进动作解析除了正则匹配可以增加一些启发式规则。例如如果AI回复中包含“已记录”、“已写入”、“备忘内容是”等关键词但后面没有动作标签可以尝试提取紧随其后的引号内或冒号后的内容作为待记录文本。加入确认机制对于关键操作可以让AI在回复中向用户二次确认例如“您是要我记录‘下午三点开会’吗”待用户确认后再执行。这增加了可靠性但牺牲了一些流畅性。开发像Rika这样的项目是一个不断迭代和调试的过程。从最简单的命令行问答开始逐步加入语音、GUI、多任务和外部数据每增加一个功能都可能引入新的复杂性。我的建议是分模块测试先确保Groq API调用和基础对话工作正常再单独测试天气API然后集成语音识别和合成最后才处理多线程和复杂的交互逻辑。使用打印语句print或日志模块记录关键节点的状态是快速定位问题的好方法。最重要的是保持耐心享受将想法一步步变为现实的过程。
基于Groq API与LLaMA模型构建多功能AI个人助手：从架构设计到工程实践
发布时间：2026/5/30 14:07:40
1. 项目概述从零构建你的AI个人助手Rika大家好我是Krish一个喜欢捣鼓代码和DIY项目的学生。今天我想和大家分享一个我最近完成的项目——Rika一个基于Groq API和LLaMA模型的多功能AI个人助手。这个项目源于一个简单的想法能不能有一个助手不仅能像ChatGPT一样聊天还能主动告诉我时间、天气、新闻甚至能一边听我说话一边执行任务感觉就像有个真正的伙伴在身边经过一段时间的摸索和调试Rika从概念变成了现实。她不仅能处理多轮对话还能整合外部信息并以一种更自然、更“人性化”的方式与你互动。如果你对Python编程、API调用以及如何将前沿的大语言模型LLM应用到实际项目中感兴趣那么这篇分享或许能给你带来一些直接的启发和可复现的步骤。Rika的核心是让AI能力变得触手可及且实用。我们不再需要从零开始训练一个庞大的模型而是通过Groq这样的高速推理API直接调用像LLaMA 3.2这样成熟的模型再结合Python的灵活性将天气、新闻等实时数据“喂”给AI让它生成符合上下文的、有用的回复。整个项目涉及逻辑设计、API集成、用户界面UI构建以及一些提升体验的“小魔法”。无论你是想学习如何将多个API服务串联起来还是想打造一个属于自己的桌面助手我相信下面的内容都能提供一个清晰的路线图。2. 核心架构与设计思路拆解在动手写代码之前理清整个系统的架构至关重要。Rika不是一个单一功能的脚本而是一个需要协调多个组件协同工作的系统。我的设计目标是响应快、功能多、体验自然。基于此我选择了客户端-服务器模型与事件驱动编程相结合的架构。2.1 技术栈选型背后的考量为什么选择Groq API和LLaMA模型这背后有几个实际的考虑。首先Groq API以其惊人的推理速度著称。对于个人助手这类需要实时交互的应用响应延迟是用户体验的杀手。Groq基于其自研的LPU语言处理单元推理引擎能在毫秒级返回LLaMA模型的生成结果这远比使用一些云端服务通过传统GPU推理要快得多。这意味着当用户问“今天天气如何”时Rika几乎可以瞬间开始组织回答而不是让用户等待数秒。其次LLaMA系列模型特别是LLaMA 3.2在开源模型中达到了一个很好的平衡点。它在保持较强对话和推理能力的同时模型尺寸相对可控API调用成本也更为合理。相比于动辄上千亿参数的闭源模型LLaMA 3.2 90B或更小的70B版本已经能够出色地完成日常问答、信息归纳和指令跟随任务。对于个人项目来说性能和成本的平衡是关键。最后整个项目用Python实现这是最自然的选择。Python拥有极其丰富的库生态无论是处理HTTP请求的requests库还是构建图形界面的tkinter/PyQt或是处理音频的pyttsx3、SpeechRecognition都能找到成熟、易用的解决方案。它允许我们快速搭建原型并迭代。注意Groq API目前提供免费的额度供开发者使用但对于高频调用需要留意其定价策略。同时LLaMA模型的知识截止日期是固定的它不知道这之后的事件这就是为什么我们需要集成实时新闻API来弥补这一缺陷。2.2 系统模块化设计我将Rika分解为几个核心模块每个模块负责单一职责通过清晰的接口进行通信用户交互模块负责所有输入输出。包括图形用户界面GUI用于显示对话和按钮操作语音输入模块用于接收语音指令以及文本转语音TTS模块用于语音回复。这个模块是用户感知Rika的窗口。核心逻辑控制模块这是Rika的“大脑”。它监听来自交互模块的事件如用户发送了一条文本消息或语音指令然后协调其他模块工作。它决定何时调用AI何时获取外部数据并处理多任务队列。AI服务模块封装了与Groq API的通信。它接收来自控制模块的、包含上下文和指令的提示词Prompt发送请求解析返回的AI回复并处理可能出现的错误如网络超时、额度不足。外部数据服务模块这是一个“信息收集器”。它独立地、按需或定时从外部API获取数据如从OpenWeatherMap获取天气从NewsAPI获取头条新闻以及从系统获取当前时间日期。这些数据会被格式化后存入一个共享的“上下文池”。上下文管理与提示工程模块这是提升AI表现的核心。它维护一个动态的对话历史窗口并将外部数据模块提供的实时信息与预先定义好的“系统指令”即Rika的“人设”进行组合生成最终发送给AI的提示词。好的提示词能极大地引导AI的行为符合预期。这种模块化设计的好处是显而易见的高内聚、低耦合。例如如果我想把天气API从A家换成B家我只需要修改外部数据服务模块中的相应函数其他模块完全不受影响。如果想增加新的功能如查股票也只需新增一个数据服务并更新提示词即可。3. 核心细节解析与实操要点理解了整体架构我们深入到几个关键环节看看具体是如何实现的以及其中有哪些容易踩坑的地方。3.1 Groq API的集成与提示词工程与Groq API交互本身很简单但其效果的优劣几乎完全取决于你发送给它的“提示词”Prompt。这不仅仅是问一个问题那么简单。API调用基础 首先你需要在Groq官网注册并获取API密钥。调用其Chat Completion接口通常使用requests库。一个最基础的请求示例如下import requests import json def ask_groq(api_key, messages, modelllama3-70b-8192): url https://api.groq.com/openai/v1/chat/completions headers { Authorization: fBearer {api_key}, Content-Type: application/json } data { model: model, # 例如 llama3-70b-8192, llama3.1-70b-versatile messages: messages, # 这是一个消息列表 temperature: 0.7, # 控制创造性越高回答越随机 max_tokens: 1024 } response requests.post(url, headersheaders, jsondata) return response.json()这里的messages参数是关键。它通常是一个列表包含多个字典每个字典有role角色和content内容两个字段。角色可以是system、user、assistant。构建Rika的“人设”提示词 对于Rika我构建了一个复杂的系统指令systemmessage这是塑造其行为的基础。system_prompt 你是一个名为Rika的AI个人助手。你的性格友好、热情且乐于助人对话风格自然像朋友一样。 你拥有以下实时信息可以在用户询问时提供 - 当前时间{current_time} - 当前日期{current_date} - 本周天气概况{weather_summary} - 今日新闻头条{news_headlines} **重要行为准则** 1. 除非用户明确询问否则不要主动提及时间、日期、天气或新闻。这些信息是供你调用的背景知识不是开场白。 2. 你可以同时处理多个任务。当用户提出一个复合请求时例如“告诉我天气并写个备忘录”请先清晰回复然后在后台执行写入操作。 3. 你的回复应当简洁、直接避免冗长的开场白和结束语。 4. 如果用户要求你记录内容到记事本请在回复中确认例如“已为您记录……”。 请注意{current_time}、{weather_summary}这些是占位符在实际调用前需要用真实数据从“上下文池”中替换掉。这个系统指令做了几件重要的事定义身份和风格让AI知道自己是谁该怎么说话。注入实时知识弥补了LLM的静态知识缺陷。设定行为边界明确告诉AI什么该做什么不该做如不主动播报信息这是避免AI“胡言乱语”的关键。支持多任务预先告知AI需要具备这种能力。在实际对话中messages列表会像这样增长messages [ {role: system, content: system_prompt_with_real_data}, {role: user, content: 你好Rika今天上海天气怎么样}, {role: assistant, content: 根据实时信息上海今天多云气温在22到28度之间微风。是个不错的日子}, {role: user, content: 不错。顺便帮我记一下下午三点开会。} # 这是最新的用户输入 ]每次新的用户输入到来我们都会将整个对话历史或最近N轮以防超出token限制连同最新的系统指令一起发送。这保证了AI拥有完整的对话上下文。实操心得temperature参数对助手类应用至关重要。我建议设置在0.5到0.8之间。太低如0.1会让回答过于死板和重复太高如1.2则可能让回答变得天马行空不符合助手身份。多试几次找到最适合你“人设”的值。3.2 实现“边听边说”与多任务处理这是让Rika感觉更“智能”和“自然”的两个特性。“边听边说”的异步处理 传统的语音助手流程是听完→思考→回答→结束。Rika则希望在回答的同时就能开始监听用户的下一条指令。这在Python中可以通过多线程或异步编程来实现。我选择使用threading模块因为它概念相对简单。基本逻辑如下当用户按下语音输入按钮或说出唤醒词时主线程启动语音识别。识别出文本后主线程将其放入一个任务队列并立即启动语音合成TTS来播报“我正在处理”之类的反馈同时启动一个新的监听线程。另一个工作线程或主线程从任务队列中取出任务调用AI和外部服务生成最终答复。当最终答复生成后如果当前没有在播报其他内容则立即播报它如果新的监听线程收到了用户的下一条指令则根据优先级决定是打断当前播报还是加入队列。import threading import queue import time class RikaAssistant: def __init__(self): self.task_queue queue.Queue() self.is_speaking False self.listening_thread None def listen_and_queue(self): # 这里是语音识别代码 user_input speech_recognizer.listen() if user_input: self.task_queue.put(user_input) print(f任务已排队: {user_input}) # 立即给出语音反馈并开始新的监听 self.speak_async(收到) self.start_listening() # 非阻塞地启动新的监听线程 def start_listening(self): if not self.is_speaking: # 如果没在说话可以安全开始监听 self.listening_thread threading.Thread(targetself.listen_and_queue, daemonTrue) self.listening_thread.start() def process_queue(self): while True: task self.task_queue.get() # 处理任务调用AI执行命令 result self.process_task(task) self.speak_async(result) # 播报结果 self.task_queue.task_done() def speak_async(self, text): # 在另一个线程中播报不阻塞主流程 def _speak(): self.is_speaking True tts_engine.say(text) tts_engine.runAndWait() self.is_speaking False threading.Thread(target_speak, daemonTrue).start()多任务处理的逻辑 多任务分为两个层面AI理解层面的多任务用户说“查天气并写备忘录”。在提示词中我们已经要求AI能处理复合请求。AI的回复应该是结构化的例如“上海今天晴26度。好的已为您将‘下午三点开会’记录到备忘录。” 这需要AI在单次回复中完成信息查询和任务确认。系统执行层面的多任务对于AI回复中确认要执行的任务如“写备忘录”控制模块需要解析出这个意图并触发对应的执行函数如一个写入记事本的函数write_to_notepad(memo_content)。这个执行过程应该是异步的不能阻塞对话流。实现时我让AI在回复中对于需要执行的动作使用特定的关键词或格式标记例如[ACTION:write_notepad]内容下午三点开会[/ACTION]。控制模块在收到AI回复后先提取文本部分播报给用户同时用正则表达式或简单字符串匹配找出动作标记在后台线程中执行相应操作。注意事项多线程编程需要小心处理共享资源如is_speaking标志和线程安全。使用queue.Queue是线程安全的是个好选择。另外语音识别和合成库可能对线程有特定要求需要查阅其文档。3.3 外部数据API的集成Rika的“实时性”依赖于外部API。这里以天气和新闻为例。天气API以OpenWeatherMap为例 你需要注册获取API Key。调用其current weather data接口。import requests import json from datetime import datetime def get_weather(api_key, cityShanghai): base_url http://api.openweathermap.org/data/2.5/weather? complete_url f{base_url}q{city}appid{api_key}unitsmetric # unitsmetric 得到摄氏度 try: response requests.get(complete_url, timeout5) response.raise_for_status() # 检查HTTP错误 data response.json() if data[cod] ! 200: return 无法获取天气信息。 main data[main] weather_desc data[weather][0][description] temp main[temp] feels_like main[feels_like] humidity main[humidity] # 格式化成一句友好的话 summary f{city}当前天气{weather_desc}气温{temp}度体感{feels_like}度湿度{humidity}%。 return summary except requests.exceptions.RequestException as e: print(f获取天气请求失败: {e}) return 天气服务暂时不可用。 except (KeyError, json.JSONDecodeError) as e: print(f解析天气数据失败: {e}) return 天气信息解析错误。新闻API以NewsAPI为例 同样需要注册获取Key。通常有免费额度限制调用次数。def get_news_headlines(api_key, countryus, categorygeneral, num_headlines3): url fhttps://newsapi.org/v2/top-headlines?country{country}category{category}apiKey{api_key} try: response requests.get(url, timeout5) response.raise_for_status() data response.json() if data[status] ! ok: return 暂无新闻。 articles data[articles][:num_headlines] headlines [art[title] for art in articles if art[title]] # 合并成一段文本用分号隔开 summary .join(headlines) return summary if summary else 暂无重要新闻。 except requests.exceptions.RequestException as e: print(f获取新闻请求失败: {e}) return 新闻服务暂时不可用。数据更新策略 不应该每次用户提问都去调用这些API这会造成延迟和额度浪费。我采用了一个简单的缓存策略在程序启动时获取一次然后启动一个后台定时线程每15-30分钟更新一次天气和新闻数据并将其存储在全局变量或一个小的缓存类中。当构造系统提示词时直接从缓存中读取这些信息。import threading import time class DataCache: def __init__(self): self.weather 正在获取天气... self.news 正在获取新闻... self.last_updated None def update_all(self, weather_api_key, news_api_key): self.weather get_weather(weather_api_key) self.news get_news_headlines(news_api_key) self.last_updated time.time() print(数据缓存已更新。) def background_update_task(cache, weather_key, news_key, interval1800): # 1800秒30分钟 while True: time.sleep(interval) cache.update_all(weather_key, news_key) # 在主程序中 cache DataCache() # 立即更新一次 cache.update_all(WEATHER_API_KEY, NEWS_API_KEY) # 启动后台更新线程 update_thread threading.Thread(targetbackground_update_task, args(cache, WEATHER_API_KEY, NEWS_API_KEY), daemonTrue) update_thread.start()4. 实操过程与核心环节实现现在让我们把上述模块组装起来看看一个完整的用户交互周期是如何运行的。我将以一次典型的语音交互为例拆解从用户开口到Rika执行完毕的全过程。4.1 环境准备与依赖安装首先确保你的Python环境建议3.8以上已经就绪。创建一个新的项目目录并初始化一个虚拟环境是个好习惯。mkdir rika-assistant cd rika-assistant python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心依赖库。创建一个requirements.txt文件内容如下requests2.28.0 # 用于HTTP请求Groq API 天气/新闻API pyttsx32.90 # 跨平台的文本转语音库 SpeechRecognition3.10.0 # 语音识别库默认使用Google Web API离线需配置 pyaudio0.2.11 # 语音识别所需的音频输入库安装可能需额外步骤 python-dotenv1.0.0 # 用于从.env文件加载API密钥等敏感信息然后使用pip安装pip install -r requirements.txt注意pyaudio在某些系统上可能需要先安装系统级的音频开发包。例如在Ubuntu上你可能需要先运行sudo apt-get install portaudio19-dev python3-pyaudio。在Windows上通常可以直接通过pip安装预编译的wheel文件。4.2 核心代码结构与主循环项目的主要代码可以组织在一个主文件如rika.py和几个模块文件中。这里为了演示我将核心逻辑浓缩在一个类中。# rika_core.py import threading import queue import time import json import re from datetime import datetime import requests import pyttsx3 import speech_recognition as sr from dotenv import load_dotenv import os load_dotenv() # 从.env文件加载环境变量 class RikaAssistant: def __init__(self): # 1. 初始化配置和状态 self.groq_api_key os.getenv(GROQ_API_KEY) self.weather_api_key os.getenv(WEATHER_API_KEY) self.news_api_key os.getenv(NEWS_API_KEY) self.model llama3-70b-8192 # 2. 初始化组件 self.tts_engine pyttsx3.init() self.recognizer sr.Recognizer() self.microphone sr.Microphone() # 3. 初始化任务队列和缓存 self.task_queue queue.Queue() self.cache { weather: 正在初始化..., news: 正在初始化..., time: } self.is_speaking False self.conversation_history [] # 保存最近的对话 # 4. 启动后台服务 self._update_cache() # 首次更新数据 self._start_background_updater() self._start_queue_processor() def _update_cache(self): 更新天气、新闻和时间缓存 self.cache[time] datetime.now().strftime(%Y年%m月%d日 %H:%M:%S) self.cache[weather] self._fetch_weather() self.cache[news] self._fetch_news_headlines() def _fetch_weather(self): # ... 同前文的get_weather函数使用self.weather_api_key pass def _fetch_news_headlines(self): # ... 同前文的get_news_headlines函数使用self.news_api_key pass def _start_background_updater(self): 启动后台数据更新线程 def updater(): while True: time.sleep(1800) # 30分钟 self._update_cache() print([后台] 数据缓存已更新。) thread threading.Thread(targetupdater, daemonTrue) thread.start() def _build_messages(self, user_input): 构建发送给Groq API的messages列表 system_prompt f 你是Rika一个友好、高效的AI助手。当前信息如下 - 时间{self.cache[time]} - 天气{self.cache[weather]} - 新闻{self.cache[news]} 请根据对话历史和当前信息以自然的口语化风格回答用户。如果用户要求执行操作如记录到记事本请在回复中明确确认并使用[ACTION:xxx]标签标明要执行的动作。 # 构建历史消息只保留最近5轮对话以防token超限 history_messages [] for role, content in self.conversation_history[-10:]: # 保留最多5轮每轮2条消息 history_messages.append({role: role, content: content}) messages [{role: system, content: system_prompt}] messages.extend(history_messages) messages.append({role: user, content: user_input}) return messages def _call_groq_api(self, messages): 调用Groq API并返回回复文本 url https://api.groq.com/openai/v1/chat/completions headers { Authorization: fBearer {self.groq_api_key}, Content-Type: application/json } data { model: self.model, messages: messages, temperature: 0.7, max_tokens: 1024 } try: response requests.post(url, headersheaders, jsondata, timeout10) response.raise_for_status() result response.json() return result[choices][0][message][content] except requests.exceptions.RequestException as e: return f抱歉网络或服务出现异常{e} except (KeyError, IndexError, json.JSONDecodeError) as e: return 抱歉处理AI回复时出现错误。 def _execute_action(self, action_text): 解析并执行AI回复中标记的动作 # 简单正则匹配 [ACTION:write_notepad]内容[/ACTION] pattern r\[ACTION:(\w)\](.*?)\[/ACTION\] matches re.findall(pattern, action_text, re.DOTALL) for action_type, content in matches: content content.strip() if action_type write_notepad: self._write_to_notepad(content) print(f[动作执行] 已写入记事本: {content}) # 可以扩展其他动作类型如 open_browser, set_reminder 等 def _write_to_notepad(self, content): 将内容写入到本地文本文件 filename rika_notes.txt with open(filename, a, encodingutf-8) as f: timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S) f.write(f[{timestamp}] {content}\n) def _speak_async(self, text): 异步语音播报 if not text: return def speak(): self.is_speaking True self.tts_engine.say(text) self.tts_engine.runAndWait() self.is_speaking False threading.Thread(targetspeak, daemonTrue).start() def _start_queue_processor(self): 启动任务队列处理线程 def processor(): while True: user_input self.task_queue.get() print(f[处理队列] 开始处理: {user_input}) # 1. 构建消息并调用AI messages self._build_messages(user_input) ai_response self._call_groq_api(messages) # 2. 更新对话历史 self.conversation_history.append((user, user_input)) self.conversation_history.append((assistant, ai_response)) # 3. 提取纯文本回复去除动作标签并播报 # 先移除动作标签得到纯文本 pure_response re.sub(r\[ACTION:\w\].*?\[/ACTION\], , ai_response, flagsre.DOTALL).strip() if pure_response: self._speak_async(pure_response) else: self._speak_async(操作已完成。) # 4. 执行动作如果有 self._execute_action(ai_response) self.task_queue.task_done() print(f[处理队列] 处理完成。) thread threading.Thread(targetprocessor, daemonTrue) thread.start() def listen_and_queue(self): 监听语音输入并将识别文本加入队列 print([监听] 请说话...) try: with self.microphone as source: self.recognizer.adjust_for_ambient_noise(source, duration0.5) audio self.recognizer.listen(source, timeout5, phrase_time_limit10) text self.recognizer.recognize_google(audio, languagezh-CN) # 中文识别 print(f[识别结果] {text}) if text: self.task_queue.put(text) # 立即给出一个简短的语音反馈同时允许新的监听 self._speak_async(嗯) # 如果当前没在说长句子可以开始新的监听这里简化处理实际可更复杂 if not self.is_speaking: threading.Thread(targetself.listen_and_queue, daemonTrue).start() except sr.WaitTimeoutError: print([监听] 超时未检测到语音。) except sr.UnknownValueError: print([监听] 无法理解音频。) self._speak_async(我没听清) except sr.RequestError as e: print(f[监听] 语音识别服务错误: {e}) self._speak_async(语音服务出错) def run_cli(self): 运行命令行交互界面 print(Rika助手已启动输入文字与我对话或输入 listen 开始语音输入输入 quit 退出。) while True: user_input input(\n你: ).strip() if user_input.lower() quit: break elif user_input.lower() listen: self.listen_and_queue() else: # 将文本输入也加入队列处理 self.task_queue.put(user_input) # 简单等待一下避免输入和输出混杂 time.sleep(0.5) if __name__ __main__: assistant RikaAssistant() assistant.run_cli()这个RikaAssistant类集成了我们讨论的大部分核心功能。run_cli方法提供了一个简单的命令行界面进行测试。你可以输入文字或者输入listen来触发语音输入。4.3 图形用户界面GUI构建示例命令行适合测试但一个桌面助手需要友好的界面。这里使用Python内置的tkinter库创建一个简单的GUI。# rika_gui.py import tkinter as tk from tkinter import scrolledtext, ttk import threading from rika_core import RikaAssistant # 导入我们之前写的核心类 class RikaGUI: def __init__(self, master): self.master master master.title(Rika - 个人助手) master.geometry(600x700) # 初始化核心助手 self.assistant RikaAssistant() # 创建UI组件 self.create_widgets() # 绑定关闭事件 master.protocol(WM_DELETE_WINDOW, self.on_closing) def create_widgets(self): # 对话显示区域 self.conversation_text scrolledtext.ScrolledText(self.master, wraptk.WORD, statedisabled, height20) self.conversation_text.pack(padx10, pady10, filltk.BOTH, expandTrue) # 输入区域框架 input_frame tk.Frame(self.master) input_frame.pack(padx10, pady5, filltk.X) self.user_input tk.Entry(input_frame) self.user_input.pack(sidetk.LEFT, filltk.X, expandTrue) self.user_input.bind(Return, self.send_text) # 按回车发送 send_btn tk.Button(input_frame, text发送, commandself.send_text) send_btn.pack(sidetk.RIGHT, padx(5,0)) # 按钮区域框架 button_frame tk.Frame(self.master) button_frame.pack(padx10, pady5) listen_btn tk.Button(button_frame, text 语音输入, commandself.start_listening, width15) listen_btn.pack(sidetk.LEFT, padx5) clear_btn tk.Button(button_frame, text清空对话, commandself.clear_conversation, width15) clear_btn.pack(sidetk.LEFT, padx5) # 状态栏 self.status_var tk.StringVar(value就绪) status_bar tk.Label(self.master, textvariableself.status_var, bd1, relieftk.SUNKEN, anchortk.W) status_bar.pack(sidetk.BOTTOM, filltk.X) def append_to_conversation(self, sender, message): 在对话区域添加一条消息 self.conversation_text.config(statenormal) self.conversation_text.insert(tk.END, f{sender}: {message}\n\n) self.conversation_text.config(statedisabled) self.conversation_text.see(tk.END) # 滚动到底部 def send_text(self, eventNone): 发送文本框中的文字 text self.user_input.get().strip() if not text: return self.user_input.delete(0, tk.END) self.append_to_conversation(你, text) # 在新线程中处理任务避免GUI卡顿 threading.Thread(targetself.process_user_input, args(text,), daemonTrue).start() def process_user_input(self, text): 处理用户输入在线程中 self.status_var.set(正在思考...) # 这里简化处理实际应调用助手的队列 # 为了演示我们模拟调用核心逻辑 messages self.assistant._build_messages(text) response self.assistant._call_groq_api(messages) # 更新GUI必须在主线程中进行 self.master.after(0, self.display_response, response) self.assistant._execute_action(response) # 执行动作 def display_response(self, response): 在主线程中显示AI回复 pure_response re.sub(r\[ACTION:\w\].*?\[/ACTION\], , response, flagsre.DOTALL).strip() self.append_to_conversation(Rika, pure_response) self.status_var.set(就绪) # 异步语音播报 self.assistant._speak_async(pure_response) def start_listening(self): 开始语音监听 self.status_var.set(正在聆听...) # 在新线程中运行监听避免阻塞GUI threading.Thread(targetself._listen_thread, daemonTrue).start() def _listen_thread(self): 语音监听的线程函数 self.assistant.listen_and_queue() # 监听完成后状态恢复这里需要更精细的事件驱动更新此处简化 self.master.after(100, lambda: self.status_var.set(就绪)) def clear_conversation(self): 清空对话历史 self.conversation_text.config(statenormal) self.conversation_text.delete(1.0, tk.END) self.conversation_text.config(statedisabled) self.assistant.conversation_history.clear() def on_closing(self): 关闭窗口时的清理工作 # 可以在这里添加保存对话历史等操作 self.master.destroy() if __name__ __main__: root tk.Tk() gui RikaGUI(root) root.mainloop()这个GUI提供了一个文本框显示对话一个输入框发送文字以及按钮进行语音输入和清空对话。它将核心的RikaAssistant类封装起来并通过多线程确保GUI在处理AI请求和语音识别时不会卡死。5. 常见问题与排查技巧实录在开发和测试Rika的过程中我遇到了不少典型问题。这里将它们整理出来并提供我的解决思路希望能帮你绕过这些坑。5.1 API调用与网络问题问题1Groq API请求返回401 Unauthorized错误。排查这几乎总是API密钥错误。首先检查你的.env文件中的GROQ_API_KEY变量名是否与代码中os.getenv(GROQ_API_KEY)读取的完全一致注意大小写。其次确认密钥本身是否正确可以登录Groq控制台重新复制一份。最后确保密钥没有意外地包含空格或换行符。技巧在代码开头添加一个简单的测试请求如果失败则打印明确的错误信息并退出避免后续流程混乱。def test_groq_key(api_key): test_url https://api.groq.com/openai/v1/models headers {Authorization: fBearer {api_key}} try: resp requests.get(test_url, headersheaders, timeout5) if resp.status_code 200: print(Groq API密钥验证成功。) return True else: print(fGroq API密钥验证失败状态码{resp.status_code}) return False except Exception as e: print(fGroq API连接测试失败{e}) return False问题2天气或新闻API调用超时或返回异常数据。排查网络连通性首先用浏览器或curl命令手动测试API端点看是否能正常返回数据。API密钥与参数检查API密钥是否有效、是否过期某些免费API有期限。检查请求的城市名、国家代码等参数是否符合API文档要求例如OpenWeatherMap的城市名需要英文。额度限制免费API通常有调用频率限制。如果短时间内频繁调用会被限制。检查API返回的错误信息通常会提示429 Too Many Requests或402 Payment Required等。解决为所有网络请求添加合理的超时设置如timeout5并做好异常捕获。实现退避重试机制。例如第一次失败后等待2秒再试最多重试3次。加强缓存机制。对于天气这种变化不频繁的数据缓存时间可以更长如1小时。对于新闻可以适当缩短如15分钟。在代码中记录失败日志便于后期分析。5.2 语音识别与合成问题问题3语音识别SpeechRecognition无法工作报错或识别不出内容。排查麦克风权限确保操作系统已授予Python程序或终端麦克风访问权限。默认麦克风sr.Microphone()默认使用系统首选麦克风。如果有多麦克风可能需要指定设备索引。你可以用sr.Microphone.list_microphone_names()列出所有设备。环境噪音在嘈杂环境中识别率会急剧下降。adjust_for_ambient_noise方法有助于校准但并非万能。网络问题默认的recognize_google需要联网因为它将音频数据发送到Google的服务器进行识别。如果网络不通会失败。解决指定麦克风microphone sr.Microphone(device_index1)索引从0开始。尝试离线引擎安装pocketsphinx包pip install pocketsphinx并使用recognizer.recognize_sphinx(audio)。但请注意离线引擎的中文识别准确率通常远低于在线引擎。增加语音指令的清晰度在代码中引导用户例如先说“叮”一声再开始识别或者要求用户说完后停顿一下。问题4文本转语音pyttsx3语音不自然、语速过快或没有声音。排查语音引擎pyttsx3在Windows上默认使用SAPI5在macOS上使用NSSpeechSynthesizer在Linux上使用eSpeak。不同引擎和声音库的质量差异很大。语速设置默认语速可能不适合中文。解决engine pyttsx3.init() # 获取当前语音速率 rate engine.getProperty(rate) engine.setProperty(rate, rate - 30) # 调慢语速 # 获取并选择语音如果有多个 voices engine.getProperty(voices) # 在Windows上可以尝试寻找更自然的中文语音包如微软晓晓 # for voice in voices: # print(voice.id) # 如果找到了中文语音ID可以设置 # engine.setProperty(voice, voices[1].id) # 例如考虑更强大的TTS方案如果对音质要求高可以研究edge-tts调用微软Edge的在线TTS音质好但需联网或付费的云服务如Azure Cognitive Services。5.3 多线程与并发控制问题5程序出现随机崩溃或“卡死”尤其是在频繁使用语音功能时。排查这很可能是多线程编程中的经典问题——竞态条件或线程阻塞。例如语音合成引擎pyttsx3的runAndWait()是阻塞调用如果在主线程中调用整个GUI就会卡住直到说完。又或者多个线程同时修改is_speaking这个状态标志。解决严格遵守GUI线程规则所有更新Tkinter界面的操作如修改文本、更新标签都必须在主线程中执行。使用self.master.after()方法将任务派发到主线程。使用线程安全的数据结构对于任务队列坚持使用queue.Queue它是线程安全的。精细控制语音状态使用threading.Lock锁来保护is_speaking等共享状态变量确保同一时间只有一个线程能修改它。避免死锁确保锁的获取和释放是成对的并且逻辑清晰。问题6AI回复慢导致语音播报和后续监听错乱。排查Groq API虽然快但网络波动或复杂问题仍可能导致响应时间超过1-2秒。如果设计成“AI回复完才允许下一次监听”用户体验会打折扣。解决这正是引入任务队列和异步播报的原因。将用户输入放入队列后立即返回让一个独立的工作线程去处理耗时的AI调用和动作执行。语音播报也使用独立线程。这样主监听循环或GUI就能始终保持响应。关键在于设计好状态机例如“正在播报时新的语音输入先缓存播报结束后再处理”这比简单的“边听边说”更复杂但更稳健。5.4 提示词与AI行为问题问题7AI经常主动说出时间、天气等信息即使没问它。排查这几乎肯定是提示词System Prompt没写清楚。LLM会倾向于使用你提供给它的所有信息。如果你在系统指令里提供了时间天气又没有明确禁止它主动提及它可能会认为“我有这些信息应该告诉用户”。解决在系统指令中使用非常明确、强硬的指令。例如“重要你拥有以下实时信息[时间、天气、新闻]。除非用户明确询问否则绝对不要主动提及或播报这些信息。仅在用户直接提问时才将这些信息融入你的回答中。” 多次强调并放在指令靠前的位置。问题8AI对于“写备忘录”这类指令有时只回复“好的”却不执行动作。排查AI的理解可能不一致。有时它认为回复“好的”就算完成任务有时它会明确说“已记录”。我们的动作解析器依赖于固定的标签格式如[ACTION:...]。解决强化提示词在系统指令中更具体地规定动作格式。“如果用户要求你执行操作如记录到记事本你必须在回复的最后使用以下精确格式标明要执行的动作[ACTION:write_notepad]这里是要记录的内容[/ACTION]。你的自然语言回复应放在动作标签之前。”改进动作解析除了正则匹配可以增加一些启发式规则。例如如果AI回复中包含“已记录”、“已写入”、“备忘内容是”等关键词但后面没有动作标签可以尝试提取紧随其后的引号内或冒号后的内容作为待记录文本。加入确认机制对于关键操作可以让AI在回复中向用户二次确认例如“您是要我记录‘下午三点开会’吗”待用户确认后再执行。这增加了可靠性但牺牲了一些流畅性。开发像Rika这样的项目是一个不断迭代和调试的过程。从最简单的命令行问答开始逐步加入语音、GUI、多任务和外部数据每增加一个功能都可能引入新的复杂性。我的建议是分模块测试先确保Groq API调用和基础对话工作正常再单独测试天气API然后集成语音识别和合成最后才处理多线程和复杂的交互逻辑。使用打印语句print或日志模块记录关键节点的状态是快速定位问题的好方法。最重要的是保持耐心享受将想法一步步变为现实的过程。
