语音助手Antenna（长期更新）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Sat Sep 28 19:13:28 2024

@author: jiaminghu
"""

from openai import OpenAI
import cv2
import sounddevice as sd
import numpy as np
from scipy.io.wavfile import write

import pyttsx3
import pyaudio
from vosk import Model, KaldiRecognizer
import json
import time

from typing import *
from openai.types.chat.chat_completion import Choice
import speech_recognition as sr

def listen_for_command():
    # 初始化识别器
    recognizer = sr.Recognizer()

    # 使用麦克风作为音频源
    with sr.Microphone() as source:
        print("Please speak command...")
        audio = recognizer.listen(source)

        try:
            # 使用Google的免费Web API进行语音识别
            command = recognizer.recognize_google(audio, language='zh-CN') #en-US'
            print("You said: " + command)
            
            # 在这里，你可以将command发送给Kimi搜索
            answer = kimi_search(command)
            
            # 将文本转换为语音并播放
            engine.say(answer)
            engine.runAndWait()

            # 在这里，你可以将command发送给Kimi搜索
            # 例如，你可以调用一个函数来处理搜索请求
            return command

        except sr.UnknownValueError:
            print("Google Speech Recognition could not understand audio")
            return None
        except sr.RequestError as e:
            print("Could not request results from Google Speech Recognition service; {0}".format(e))
            return None
        
        

def listen_for_command_localvosk():
    
    # 指定模型路径
    model_path = "./vosk-model-small-cn-0.22"
    model = Model(model_path)
    
    # 初始化识别器，采样率为16000Hz
    recognizer = KaldiRecognizer(model, 16000)
    
    # 定义音频流参数
    chunk = 1024
    format = pyaudio.paInt16
    channels = 1
    rate = 16000
    p = pyaudio.PyAudio()
    
    # 开启音频流
    stream = p.open(format=format, channels=channels, rate=rate, input=True, frames_per_buffer=chunk)
    
    print("开始说话，说'停止'以结束录音...")
    
    # 记录开始时间
    start_time = time.time()
    
    text = ''
    # 读取音频流
    while True:
        data = stream.read(chunk)
        if recognizer.AcceptWaveform(data):
            result = recognizer.Result()
            if result:
                result_json = json.loads(result)
                if 'text' in result_json:
                    if '停止' in result_json['text']:
                        print("停止录音")
                        break
                    print("识别结果：", result_json['text'])
                    text += result_json['text']
                    # 检查是否说出了“停止”
                    
        if time.time() - start_time > 20:  # 设置超时时间为10秒，也可以根据需要调整
            print("超时，自动停止录音")
            break
    
    # 关闭音频流
    stream.stop_stream()
    stream.close()
    p.terminate()
    
    command = text.strip()
    print("You said: " + command)
    
    # 在这里，你可以将command发送给Kimi搜索
    answer = kimi_search(command)
    
    # 将文本转换为语音并播放
    engine.say(answer)
    engine.runAndWait()

    return command



def kimi_search_offline(query):
    
    # 这里应该是调用Kimi搜索的代码
    query = '你好kimi, ' + query
    print("Searching Kimi for: " + query)
    client = OpenAI(
        api_key = "API_KEY", #得用自己的KEY哦
        base_url = "https://api.moonshot.cn/v1",
    )
     
    completion = client.chat.completions.create(
        model = "moonshot-v1-8k",
        messages = [
            {"role": "system", "content": system_query},
            {"role": "user", "content": query}
        ],
        temperature = 0.8,
    )
    
    print(completion.choices[0].message.content)
   

    # 要转换为语音的文本
    return completion.choices[0].message.content
    

    
def kimi_search(query):
    query = '你好kimi, ' + query
    print("Searching Kimi for: " + query)
    client = OpenAI(
        api_key = "sk-NZ9KYlSLrfCUEe3gvGMO6YSP99Lf7n5OVzoNer6xRtfHfbti",
        base_url = "https://api.moonshot.cn/v1",
    )
    
    messages = [
        {"role": "system", "content": system_query},
    ]
     
    # 初始提问
    messages.append({
        "role": "user",
        "content": query
    })
    

 
    # search 工具的具体实现，这里我们只需要返回参数即可
    def search_impl(arguments: Dict[str, Any]) -> Any:
        """
        在使用 Moonshot AI 提供的 search 工具的场合，只需要原封不动返回 arguments 即可，
        不需要额外的处理逻辑。
     
        但如果你想使用其他模型，并保留联网搜索的功能，那你只需要修改这里的实现（例如调用搜索
        和获取网页内容等），函数签名不变，依然是 work 的。
     
        这最大程度保证了兼容性，允许你在不同的模型间切换，并且不需要对代码有破坏性的修改。
        """
        return arguments
     
     
    def chat(messages) -> Choice:
        completion = client.chat.completions.create(
            model="moonshot-v1-128k",
            messages=messages,
            temperature=0.3,
            tools=[
                {
                    "type": "builtin_function",  # <-- 使用 builtin_function 声明 $web_search 函数，请在每次请求都完整地带上 tools 声明
                    "function": {
                        "name": "$web_search",
                    },
                }
            ]
        )
        return completion.choices[0]
     
    finish_reason = None
    while finish_reason is None or finish_reason == "tool_calls":
        choice = chat(messages)
        finish_reason = choice.finish_reason
        if finish_reason == "tool_calls":  # <-- 判断当前返回内容是否包含 tool_calls
            messages.append(choice.message)  # <-- 我们将 Kimi 大模型返回给我们的 assistant 消息也添加到上下文中，以便于下次请求时 Kimi 大模型能理解我们的诉求
            for tool_call in choice.message.tool_calls:  # <-- tool_calls 可能是多个，因此我们使用循环逐个执行
                tool_call_name = tool_call.function.name
                tool_call_arguments = json.loads(tool_call.function.arguments)  # <-- arguments 是序列化后的 JSON Object，我们需要使用 json.loads 反序列化一下
                if tool_call_name == "$web_search":
                    tool_result = search_impl(tool_call_arguments)
                else:
                    tool_result = f"Error: unable to find tool by name '{tool_call_name}'"
     
                # 使用函数执行结果构造一个 role=tool 的 message，以此来向模型展示工具调用的结果；
                # 注意，我们需要在 message 中提供 tool_call_id 和 name 字段，以便 Kimi 大模型
                # 能正确匹配到对应的 tool_call。
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": tool_call_name,
                    "content": json.dumps(tool_result),  # <-- 我们约定使用字符串格式向 Kimi 大模型提交工具调用结果，因此在这里使用 json.dumps 将执行结果序列化成字符串
                })
    
    print(choice.message.content)
    
    return choice.message.content  # <-- 在这里，我们才将模型生成的回复返回给用户

    

if __name__ == "__main__":
    system_query = "你是 Kimi，由 Moonshot AI 提供的人工智能助手，你更擅长中文和英文的对话。你之后收到的文字来源于语音识别，所以可能会有误差、重复和标点不全的情况，请按照一个人通常说话的习惯进行理解。"
    engine = pyttsx3.init()
    engine.setProperty('voice', 'com.apple.voice.compact.zh-CN.Tingting')
    engine.setProperty('rate', 220)
    
    command = listen_for_command()
    #command = listen_for_command_localvosk()