实现效果

一、微信消息获取+自动答复代码

import time

from wxautox import WeChat

import DeepSeek

if __name__ == '__main__':
    wx = WeChat()
    listen_list = [
        '阿芳-BXL'
    ]
    for i in listen_list:
        wx.AddListenChat(who=i, savepic=True)
    # 持续监听消息，并且收到消息后回复“收到”
    wait = 1  # 设置1秒查看一次是否有新消息
    while True:
        msgs = wx.GetListenMessage()
        for chat in msgs:
            who = chat.who  # 获取聊天窗口名（人或群名）
            one_msgs = msgs.get(chat)  # 获取消息内容
            # 回复收到
            for msg in one_msgs:
                msgtype = msg.type  # 获取消息类型
                content = msg.content  # 获取消息内容，字符串类型的消息内容
                print(f'【{who}】：{content}')
                # ===================================================
                # 处理消息逻辑（如果有）
                #
                # 处理消息内容的逻辑每个人都不同，按自己想法写就好了，这里不写了
                #
                # ===================================================
                if msgtype == 'friend':
                    reply = DeepSeek.start_model(content)
                    chat.SendMsg(reply)  # 回复收到
        time.sleep(wait)

二、大模型自动答复消息并返回

import json
import os
import re
import time
from typing import List, Dict, Any
import requests
from requests.exceptions import RequestException


class ModelHistory:
    def __init__(self, role: str = "", content: str = ""):
        self.role = role
        self.content = content

    def to_dict(self) -> Dict[str, str]:
        return {"role": self.role, "content": self.content}

    @classmethod
    def from_dict(cls, data: Dict[str, str]) -> 'ModelHistory':
        return cls(role=data.get("role", ""), content=data.get("content", ""))


def start_model(user_question: str) -> str:  # 修改返回值类型为 str
    BASE_URL = "http://localhost:11434/v1/"
    API_KEY = "ollama"  # required but ignored

    # Initialize HTTP client with timeouts
    session = requests.Session()
    session.timeout = (10, 30)  # (connect, read) timeouts in seconds

    # Read knowledge file
    file_path = "knowledge.txt"
    model_history_list = []

    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
            if content:
                model_history_list = [ModelHistory.from_dict(item) for item in json.loads(content)]
    except (IOError, json.JSONDecodeError) as e:
        print(f"Error reading history file: {e}")

    # Prepare messages
    messages = []

    # Get knowledge from word files
    knowledge = get_content_by_dir("src/main/resources/word_files")

    # Add system message with knowledge
    messages.append({
        "role": "user",
        "content": f"你的角色是：问题答复专家。\n你的任务是：结合已知知识和上下文回答问题。\n已知知识：{knowledge}\n根据以上要求，我的问题是：{user_question}"
    })

    # Add user question to history
    model_history_list.append(ModelHistory(role="user", content=user_question))

    # Prepare request body
    request_body = {
        "messages": messages,
        "model": "deepseek-16b-custom:latest",
        "stream": True,
        "temperature": 0.0
    }

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    try:
        # Make the request with streaming
        response = session.post(
            f"{BASE_URL}chat/completions",
            json=request_body,
            headers=headers,
            stream=True
        )

        response.raise_for_status()

        temp_res = ""
        out_res = ""

        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data: ') and '[DONE]' not in decoded_line:
                    try:
                        data = json.loads(decoded_line[6:])  # Remove 'data: ' prefix
                        if 'choices' in data:
                            for choice in data['choices']:
                                if 'delta' in choice and 'content' in choice['delta']:
                                    content = choice['delta']['content']
                                    # print(content, end='', flush=True)
                                    temp_res += content
                                    out_res += content

                                    # Check for sentence end
                                    mark_1 = temp_res.find('。')
                                    if mark_1 > 0:
                                        start_tts(temp_res[:mark_1 + 1])
                                        temp_res = temp_res[mark_1 + 1:]
                    except json.JSONDecodeError as e:
                        print(f"Error decoding JSON: {e}")

        # Speak any remaining text
        if temp_res:
            start_tts(temp_res)

        # Process and save the response to history
        model_history = ModelHistory(role="assistant", content=out_res)

        # Remove <think> tags
        regex = r"<think>[\s\S]*?</think>"
        model_history.content = re.sub(regex, "", model_history.content)

        model_history_list.append(model_history)

        # Save history back to file
        try:
            with open(file_path, 'w', encoding='utf-8') as f:
                json.dump([item.to_dict() for item in model_history_list], f, indent=2, ensure_ascii=False)
        except IOError as e:
            print(f"Error writing history file: {e}")

        # 返回完整的答复内容
        return out_res

    except RequestException as e:
        print(f"Request failed: {e}")
        return ""  # 返回空字符串表示失败


def get_content_by_dir(directory: str) -> str:
    """Helper function to get content from files in a directory"""
    # Implement this based on your Java version
    return ""


def start_tts(text: str) -> None:
    """Helper function to start text-to-speech"""
    # Implement this based on your Java version
    pass


if __name__ == "__main__":
    response = start_model("你是谁？")
    print("\n完整答复：", response)

[
  {
    "role": "user",
    "content": "推荐几款车"
  },
  {
    "role": "assistant",
    "content": " 作为问题答复专家，我可以为您提供一些建议，但请注意，这些信息可能会随时间而变化，因为汽车市场不断更新，新款车型层出不穷。以下是一些不同类型的车辆推荐，以满足不同的需求和偏好：\n\n1. **经济型城市通勤车**：\n   - 丰田雅力士（Toyota Yaris）：适合城市驾驶，油耗低，性价比高。\n   - 本田飞度（Honda Fit/Jazz）：空间利用率高，操控灵活，适合年轻消费者。\n\n2. **家庭实用型SUV**：\n   - 马自达CX-5：提供良好的驾驶体验和实用性，适合家庭使用。\n   - 斯巴鲁森林人（Subaru Forester）：以其全轮驱动系统和耐用性著称。\n\n3. **豪华轿车**：\n   - 奔驰E级：提供高端的舒适性和先进的技术配置。\n   - 宝马5系：运动性能与豪华感的完美结合，适合商务和家庭使用。\n\n4. **电动车/混合动力车**：\n   - 特斯拉Model 3：电动驱动，高性能，智能化程度高，是环保出行的优选。\n   - 丰田普锐斯（Prius）：作为混合动力车型的先驱之一，以其燃油效率和经济性著称。\n\n5. **越野能力强的SUV**：\n   - Jeep牧马人（Jeep Wrangler）：强大的越野性能和坚固的车身设计。\n   - 路虎卫士（Land Rover Defender）：结合了豪华与耐用性，适合各种复杂路况。\n\n请根据您的具体需求、预算以及对车辆性能的要求来选择合适的车型。在做出决定之前，建议您进行试驾，并咨询专业人士以获取更多详细信息。"
  },
  {
    "role": "user",
    "content": "推荐几款车"
  },
  {
    "role": "assistant",
    "content": " 作为一个AI，我无法提供实时的市场数据或个人建议，但我可以根据一些通用的标准和趋势来推荐几款车型。以下是一些不同类型的车辆推荐，适用于不同的需求和偏好：\n\n1. **城市通勤用车**：\n   - **小型车/紧凑型车**：这类车型通常油耗较低，适合在城市中穿梭，如丰田雅力士（Yaris）、本田飞度（Honda Fit）。\n   - **电动车**：对于环保意识较强的用户，推荐特斯拉Model 3或比亚迪汉EV等电动车型，它们提供高效的能源利用和低运行成本。\n\n2. **家庭用车**：\n   - **中型SUV**：这类车辆空间宽敞，适合家庭出游，如丰田RAV4、本田CR-V。\n   - **多功能车（MPV）**：如果需要更多座位或更大的载物空间，可以考虑奥德赛（Honda Odyssey）或大众夏朗（Volkswagen Touran）等车型。\n\n3. **越野/冒险用车**：\n   - **紧凑型SUV**：如Jeep自由侠（Jeep Renegade）或铃木维特拉（Suzuki Vitara），适合轻度越野和日常使用。\n   - **硬派越野车**：对于极端的越野环境，推荐吉普牧马人（Jeep Wrangler）或路虎卫士（Land Rover Defender）。\n\n4. **豪华车型**：\n   - **大型轿车/行政级轿车**：如奔驰S级、宝马7系和奥迪A8等，提供高端舒适性和先进技术。\n   - **跑车**：对于追求速度和激情的用户，推荐保时捷911或奥迪R8等高性能跑车。\n\n5. **经济型车辆**：\n   - **入门级轿车/掀背车**：如雪佛兰科鲁兹（Chevrolet Cruze）或福特福克斯（Ford Focus），提供良好的燃油效率和性价比。\n\n请记住，选择汽车时应该考虑个人的预算、驾驶习惯、用车需求以及对车辆性能的要求。建议在做出最终决定前，进行充分的市场调研，并试驾感兴趣的车型以确保满意。"
  },
  {
    "role": "user",
    "content": "夏季为什么吃冰棒"
  },
  {
    "role": "assistant",
    "content": " 夏季吃冰棒的原因有以下几点：\n\n1. **降温解暑**：在炎热的夏天，人体会通过出汗来散热以降低体温。吃冰棒可以帮助迅速降低口腔和食道的温度，从而给人一种凉爽的感觉，有助于缓解高温带来的不适感。\n\n2. **补充水分**：夏季人们容易因出汗而流失大量水分，冰棒通常含有一定比例的水分，可以在享受清凉的同时帮助补充身体所需的水分。\n\n3. **口感满足**：冰棒的冷冻质地和甜味可以给人带来即时的口感满足感，这种感觉在高温下尤为受欢迎。\n\n4. **方便携带**：与冷饮相比，冰棒通常更容易携带，适合户外活动或旅行时食用。\n\n5. **成本较低**：相比于其他冷饮（如冰淇淋），冰棒通常价格更实惠，对于预算有限的人来说是一个经济实惠的选择。\n\n6. **口味多样**：市场上有各种各样的冰棒口味和类型，可以满足不同人的口味偏好。\n\n综上所述，夏季吃冰棒是一种既能够带来即时清凉感，又能够在高温环境中帮助补充水分、满足口感需求的行为。"
  },
  {
    "role": "user",
    "content": "怎么学习英文"
  },
  {
    "role": "assistant",
    "content": " 学习英语可以通过以下几个步骤来进行：\n\n1. **基础语法和词汇**：首先，你需要掌握基本的英语语法规则和一定数量的单词。这可以通过使用教科书、在线课程或应用程序来实现。\n\n2. **听力训练**：提高你的听力技能是非常重要的，因为这是理解口语的基础。你可以通过听英文歌曲、观看英文电影或电视节目、收听广播等方式来进行练习。\n\n3. **发音练习**：准确的发音可以帮助你更好地理解和被他人理解。可以通过跟读磁带、使用语音识别软件来改善你的发音。\n\n4. **阅读材料**：广泛阅读英文书籍、报纸、杂志和在线文章，这有助于提高你的阅读理解能力。\n\n5. **写作练习**：尝试用英语写日记、短文或电子邮件，这可以帮助你练习组织思想并使用新学的词汇和语法结构。\n\n6. **语言交换**：找一个英语母语的伙伴进行语言交换，这样可以在实际对话中练习你的口语技能。\n\n7. **参加课程**：如果可能的话，报名参加正规的英语课程或在线课程，这可以提供系统的学习计划和专业的指导。\n\n8. **使用技术工具**：利用各种应用程序和在线资源来辅助学习，如单词卡片应用、语言学习网站等。\n\n9. **持续实践**：语言技能需要不断练习才能提高。设定每天的学习时间，并坚持下去。\n\n10. **评估进度**：定期进行自我测试，了解自己的进步情况，并根据需要调整学习计划。\n\n通过这些步骤的持续努力，你可以逐步提高你的英语水平。记住，学习一门新语言是一个长期的过程，耐心和毅力是非常重要的。"
  },
  {
    "role": "user",
    "content": "推荐几款车"
  },
  {
    "role": "assistant",
    "content": " 作为一个AI，我无法提供实时的市场信息或个人建议，但我可以根据一些普遍的标准和趋势来推荐车型。以下是几个不同类型的车辆推荐，适用于不同的需求和偏好：\n\n1. **城市通勤者**：\n   - **小型车/紧凑型车**：这类车辆通常油耗较低，适合在城市中穿梭，如丰田雅力士（Toyota Yaris）、本田飞度（Honda Fit）或大众Polo。\n\n2. **家庭用车**：\n   - **中型SUV**：提供更大的空间和更好的载人能力，如丰田RAV4、本田CR-V或福特Escape。\n\n3. **环保意识者**：\n   - **电动车/混合动力车**：如果你关心环境并希望减少碳排放，可以考虑特斯拉Model 3、日产聆风（Nissan Leaf）或丰田普锐斯（Toyota Prius）。\n\n4. **性能爱好者**：\n   - **运动型轿车/跑车**：对于追求速度和操控感的驾驶者，推荐宝马M系列、梅赛德斯-奔驰AMG车型或保时捷911。\n\n5. **豪华车消费者**：\n   - **大型豪华SUV**：如果你需要高端舒适性和先进技术，可以考虑奥迪Q7、宝马X5或路虎揽胜（Land Rover Range Rover）。\n\n请记住，选择车辆时还应考虑预算、维护成本、保险费用和个人需求等因素。建议在做出决定前进行充分的市场调研和试驾体验。"
  },
  {
    "role": "user",
    "content": "夏季为什么吃冰棒"
  },
  {
    "role": "assistant",
    "content": " 夏季吃冰棒的原因有以下几点：\n\n1. **降温解暑**：在炎热的夏天，人体会通过出汗来散热以降低体温。吃冰棒可以帮助迅速降低口腔和食道的温度，从而给人一种凉爽的感觉，有助于缓解高温带来的不适感。\n\n2. **补充水分**：夏季人们容易因出汗而流失大量水分，冰棒通常含有一定比例的水分，可以在享受清凉的同时帮助补充身体所需的水分。\n\n3. **口感满足**：冰棒的冷冻质地和甜味可以给人带来即时的口感满足感，这种感觉在高温下尤为受欢迎。\n\n4. **方便携带**：与冷饮相比，冰棒通常更容易携带，适合户外活动或旅行时食用。\n\n5. **成本较低**：相比于其他冷饮（如冰淇淋），冰棒通常价格更实惠，对于预算有限的人来说是一个经济实惠的选择。\n\n6. **口味多样**：市场上有各种各样的冰棒口味和类型，可以满足不同人的口味偏好。\n\n综上所述，夏季吃冰棒是一种既能够带来即时清凉感，又能够在高温环境中帮助补充水分、满足口感需求的行为。"
  },
  {
    "role": "user",
    "content": "怎么学习英文"
  },
  {
    "role": "assistant",
    "content": " 学习英语可以通过以下几个步骤来进行：\n\n1. **基础语法和词汇**：首先，你需要掌握基本的英语语法规则和一定数量的单词。这可以通过使用教科书、在线课程或应用程序来实现。\n\n2. **听力训练**：提高你的听力技能是非常重要的，因为这是理解口语的基础。你可以通过听英文歌曲、观看英文电影或电视节目、收听广播等方式来进行练习。\n\n3. **发音练习**：准确的发音可以帮助你更好地理解和被他人理解。可以通过跟读磁带、使用语音识别软件来改善你的发音。\n\n4. **阅读材料**：广泛阅读英文书籍、报纸、杂志和在线文章，这有助于提高你的阅读理解能力。\n\n5. **写作练习**：尝试用英语写日记、短文或电子邮件，这可以帮助你练习组织思想并使用新学的词汇和语法结构。\n\n6. **语言交换**：找一个英语母语的伙伴进行语言交换，这样可以在实际对话中练习你的口语技能。\n\n7. **参加课程**：如果可能的话，报名参加正规的英语课程或在线课程，这可以提供系统的学习计划和专业的指导。\n\n8. **使用技术工具**：利用各种应用程序和在线资源来辅助学习，如单词卡片应用、语言学习网站等。\n\n9. **持续实践**：语言技能需要不断练习才能提高。设定每天的学习时间，并坚持下去。\n\n10. **评估进度**：定期进行自我测试，了解自己的进步情况，并根据需要调整学习计划。\n\n通过这些步骤的持续努力，你可以逐步提高你的英语水平。记住，学习一门新语言是一个长期的过程，耐心和毅力是非常重要的。"
  }
]