使用ExAPI让Live2D接入多智能体

使用Live2DViewerEX的ExAPI，完善自己的桌宠

前言

某天闲来无事翻阅Live2DViewerEX的官方文档，偶然发现居然已经支持Api调用并且有了Api文档，遂有了饺子醋。
原本想的是接入文本生成大模型让桌宠活起来，后来发现角色描述的提示词过于难写，且对AI模型选择等都有较高的要求，于是转到了另一个方向——语音生成，基于GPT-SoVITS使用角色对应的别人训练好的模型去做一个简单的邮件接收、读取、提示的demo。

Live2DViewerEX

官方文档

打开Live2DViewerEX的端口服务

实测Live2D与Spine模型均可使用ExAPI

调用

具体的”msg”:对应的接口”号”以及传参参见官方文档。

显示气泡文本

# 给模型发json格式的消息,通过Live2DViewerEX api [11000]
def send_json_message(json_message: str):
    uri = "ws://127.0.0.1:10086/api"
    data = {
        "msg": 11000,
        "msgId": 1,
        "data": {
            "id": 0,        #模型的id(序号)，即你当前有多少个模型同时运行时对应的序号 0，1，2。。。
            "text": json_message,    #传递的文本
            "textFrameColor": 0x000000,
            "textColor": 0xFFFFFF,     #文本颜色
            "duration": 30000   #文本框持续时间，根据需要可以以message长度测算，我这里就一律30s了
        }
    }

    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

触发动作

# 控制表情(动作)
def send_expression():
    uri = "ws://127.0.0.1:10086/api"
    data = {
        "msg": 13200,
        "msgId": 1,
        "data": {
            "id": 0,
            "type": 0,
            "mtn": "a#3:a3"     #动作组 group:motion
        }
    }

    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

这里解释一下什么是动作组：

关于文本与动作、表情的适配（情感？？？）

启用后即可调用模型的动作，这个功能可以适配一些对话场景，让模型对应自己要说出的文本做出相应的动作

展开来讲：
你可以在大语言模型调用前写一个prompt.txt的角色描述提示词文档，在文档中写入：
“如果此时你做出的反应是生气，那么你可以在返回给我的text中附带 “[angry]” ” 的备注
这样就可以在处理text时对于特定的文本做出对于的动作或者表情（[13300] 设置表情）响应了

最早发现这种用法是来自B站UP主: 卤v ，在提示词中写入相关的命令，在角色返回text时将对应的情感体现为表情包的形式发出
附上他的项目网址 https://www.momotalk.top/

关于这一块我十分清楚逻辑，但是正如前面所说的，我不会写角色描述的提示词文档。。。。。。

播放声音

我完成的最完备的一部分，通过GPT-SoVITS的API接口处理文本并输出对应的语音
关于GPT-SoVITS的使用请看 我的另一篇博客

# 调用TTS API生成音频
def get_tts_audio(text: str, output_file: str = "output.wav") -> bool:
    """
    调用TTS API的GET方法生成音频并保存
    :param text: 待合成的文本
    :param output_file: 音频输出路径
    :return: 成功返回True，失败返回False
    """
    params = {
        "text": text,
        "text_lang": TEXT_LANG,
        "ref_audio_path": REF_AUDIO_PATH,
        "prompt_lang": PROMPT_LANG,
        "prompt_text": PROMPT_TEXT,
        "text_split_method": "cut2",
        "batch_size": 1,
        "media_type": "wav",
        "streaming_mode": False
    }

    try:
        response = requests.get("http://127.0.0.1:9880/tts", params=params)
        if response.status_code == 200:
            with open(output_file, "wb") as f:
                f.write(response.content)
            print(f"音频已保存至：{output_file}")
            return True
        else:
            print(f"TTS API调用失败：{response.json().get('message', '未知错误')}")
            return False
    except Exception as e:
        print(f"调用TTS API时发生错误：{str(e)}")
        return False

# 传递TTS生成的语音
def send_sound():                  
    uri = "ws://127.0.0.1:10086/api"
    data ={
        "msg": 13500,
        "msgId": 1,
        "data": {
            "id": 0,
            "channel":0,
            "volume":1,
            "delay":0,
            "type":0,
            "sound":"C:\\Users\\X.J\\Desktop\\live2d_ai\\output.wav"
        }
    }

    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

调用完成后语音会自动播放

番外篇：通过IMAPClient接收并相应邮件

灵感来源: L2DVEx接入ai（续）——通过IMAPClient接收邮件

通过IMAPClient读取我QQ邮箱的未读邮件（大多使用场景是新接收到的邮件），将邮件信息提炼并返回（其实也可以再喂给大语言模型，这样可以让AI对邮件内容做出响应生成更有“感觉”的文本）给TTS并生成语音。

另一个灵感来源：游戏《碧蓝航线》和《少女前线》中的Live2D模型有一项设定，在玩家收到邮件（游戏内）时，模型会做出相应的动作和语音（都是预设好的） 老一辈Live2D游戏特有的设计感和交互反馈

首先在QQ邮箱申请打开服务
QQ邮箱首页——>账号与安全——>安全设置——>POP3/IMAP/SMTP/Exchange/CardDAV 服务 选择开启即可

随后将你获得的16位码写入imap_object.login

def get_data():
    global PROCESSED_UIDS
    imap_object = imapclient.IMAPClient('imap.qq.com', ssl=True)
    imap_object.login('xxxxxxx@qq.com', '【你的16位码】')
    imap_object.select_folder('INBOX', readonly=False)

    # 只查找今日所有未读邮件
    since_date = date.today().strftime('%d-%b-%Y')
    UIDs = imap_object.search(['SINCE', since_date, 'UNSEEN'])

    # 只处理未处理过的新邮件
    new_UIDs = [uid for uid in UIDs if uid not in PROCESSED_UIDS]
    if not new_UIDs:
        print("[INFO] 暂无新邮件。")
    else:
        print(f"[INFO] 检测到 {len(new_UIDs)} 封新邮件：")
        for i in new_UIDs:
            str_ = "老师，有新的邮件发来。"
            raw_message = imap_object.fetch(i, ['BODY[]'])
            message_object = pyzmail.PyzMessage.factory(raw_message[i][b'BODY[]'])
            subject = message_object.get_subject()
            from_addr = message_object.get_addresses('from')
            print(f"---\n主题: {subject}\n发件人: {from_addr[0][0]}")
            str_ += "主题："+subject+ ",发件人是："+ str(from_addr[0][0])    
            #一般来说from_addr是个tuple下的数组，其中的数组的首项一般是发信人的昵称，第二项是发信人的邮箱
            if message_object.text_part:
                text = message_object.text_part.get_payload().decode(message_object.text_part.charset)
                print(f"文本内容: {text[:100]}...")
                str_ += "。文本内容是："+text[:100]
            elif message_object.html_part:
                html = message_object.html_part.get_payload().decode(message_object.html_part.charset)
                print(f"HTML内容: {html[:100]}...")
            else:
                print("无正文内容")
                str_ += "无正文内容"
            # 标记为已读
            imap_object.add_flags(i, [b'\\Seen'])      #将邮件置为已读
            PROCESSED_UIDS.add(i)
            # 合成语音
            get_tts_audio(str_)
            send_sound()
            send_json_message(str_)
            #send_expression()
    imap_object.logout()


def listen_email():      #启动监听，轮询邮箱中的邮件
    while True:
        get_data()
        time.sleep(1)   #轮询时间

项目完整代码

import json
import os
import requests  # 新增：用于发送HTTP请求
from websocket import create_connection
from openai import OpenAI
import time
import threading
import imapclient
import pyzmail
from datetime import date

# 读取预设
file_path = "prompt.txt"
try:
    with open(file_path, "r", encoding="utf-8") as file:
        system_prompt = file.read()
except FileNotFoundError:
    print(f"错误：文件 {file_path} 未找到。")
except Exception as e:
    print(f"读取文件时发生错误：{e}")

# 调用kimi api
client = OpenAI(
    api_key="sk-xxxxxxxxxxxxx",  
    base_url="https://api.moonshot.cn/v1",
)

# TTS API配置（根据实际情况修改）
TTS_API_URL = "http://127.0.0.1:9880/tts"
REF_AUDIO_PATH = "C:\\Users\\X.J\\Desktop\\GPT-SoVITS-1007-cu124\\1.ogg"  # 参考音频路径
PROMPT_TEXT = "衝動買いは禁止！消費はちゃんと計画的にっ！"  # 参考音频对应的文本
PROMPT_LANG = "ja"  # 参考文本语言
TEXT_LANG = "zh"  # 待合成文本语言

# 预设信息
system_messages = [
    {"role": "system", "content": system_prompt},
]

# 聊天记录
messages = []

# 限制记忆消息数小于等于10条
def make_new_messages(input: str , n: int = 10) -> list[dict]:
    global messages
    messages.append({
        "role": "user",
        "content": input
    })

    new_messages = []
    new_messages.extend(system_messages)

    if len(messages) > n:
        messages = messages[-n:]
    new_messages.extend(messages)
    return new_messages

# 调用TTS API生成音频
def get_tts_audio(text: str, output_file: str = "output.wav") -> bool:
    """
    调用TTS API的GET方法生成音频并保存
    :param text: 待合成的文本
    :param output_file: 音频输出路径
    :return: 成功返回True，失败返回False
    """
    params = {
        "text": text,
        "text_lang": TEXT_LANG,
        "ref_audio_path": REF_AUDIO_PATH,
        "prompt_lang": PROMPT_LANG,
        "prompt_text": PROMPT_TEXT,
        "text_split_method": "cut2",
        "batch_size": 1,
        "media_type": "wav",
        "streaming_mode": False
    }

    try:
        response = requests.get("http://127.0.0.1:9880/tts", params=params)
        if response.status_code == 200:
            with open(output_file, "wb") as f:
                f.write(response.content)
            print(f"音频已保存至：{output_file}")
            return True
        else:
            print(f"TTS API调用失败：{response.json().get('message', '未知错误')}")
            return False
    except Exception as e:
        print(f"调用TTS API时发生错误：{str(e)}")
        return False

# 给模型发json格式的消息,通过Live2DViewerEX api [11000]
def send_json_message(json_message: str):
    uri = "ws://127.0.0.1:10086/api"
    data = {
        "msg": 11000,
        "msgId": 1,
        "data": {
            "id": 0,
            "text": json_message,
            "textFrameColor": 0x000000,
            "textColor": 0xFFFFFF,
            "duration": 30000
        }
    }

    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

# 控制表情(动作)
def send_expression():
    uri = "ws://127.0.0.1:10086/api"
    data = {
        "msg": 13200,
        "msgId": 1,
        "data": {
            "id": 0,
            "type": 0,
            "mtn": "a#3:a3"
        }
    }
    
    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

# 传递TTS生成的语言
def send_sound():                  
    uri = "ws://127.0.0.1:10086/api"
    data ={
        "msg": 13500,
        "msgId": 1,
        "data": {
            "id": 0,
            "channel":0,
            "volume":1,
            "delay":0,
            "type":0,
            "sound":"C:\\Users\\X.J\\Desktop\\live2d_ai\\output.wav"
        }
    }

    try:
        ws = create_connection(uri)
        ws.send(json.dumps(data))
    except Exception as e:
        print("错误:", str(e))

# 聊天函数
def chat(input: str) -> str:
    completion = client.chat.completions.create(
        model="moonshot-v1-8k",
        messages=make_new_messages(input),
        temperature=0.3,
    )

    mita_message = completion.choices[0].message.content
    messages.append({
        "role": "assistant",
        "content": mita_message
    })

    # 新增：调用TTS生成音频
    get_tts_audio(mita_message)
    return mita_message


# 全局集合，记录已处理的UID
PROCESSED_UIDS = set()

def get_data():
    global PROCESSED_UIDS
    imap_object = imapclient.IMAPClient('imap.qq.com', ssl=True)
    imap_object.login('xxxxxx@qq.com', '[你的16位码]')
    imap_object.select_folder('INBOX', readonly=False)

    # 只查找今日所有未读邮件
    since_date = date.today().strftime('%d-%b-%Y')
    UIDs = imap_object.search(['SINCE', since_date, 'UNSEEN'])

    # 只处理未处理过的新邮件
    new_UIDs = [uid for uid in UIDs if uid not in PROCESSED_UIDS]
    if not new_UIDs:
        print("[INFO] 暂无新邮件。")
    else:
        print(f"[INFO] 检测到 {len(new_UIDs)} 封新邮件：")
        for i in new_UIDs:
            str_ = "老师，有新的邮件发来。"
            raw_message = imap_object.fetch(i, ['BODY[]'])
            message_object = pyzmail.PyzMessage.factory(raw_message[i][b'BODY[]'])
            subject = message_object.get_subject()
            from_addr = message_object.get_addresses('from')
            print(f"---\n主题: {subject}\n发件人: {from_addr[0][0]}")
            str_ += "主题："+subject+ ",发件人是："+ str(from_addr[0][0])
            if message_object.text_part:
                text = message_object.text_part.get_payload().decode(message_object.text_part.charset)
                print(f"文本内容: {text[:100]}...")
                str_ += "。文本内容是："+text[:100]
            elif message_object.html_part:
                html = message_object.html_part.get_payload().decode(message_object.html_part.charset)
                print(f"HTML内容: {html[:100]}...")
            else:
                print("无正文内容")
                str_ += "无正文内容"
            # 标记为已读
            imap_object.add_flags(i, [b'\\Seen'])
            PROCESSED_UIDS.add(i)
            # 合成语音
            get_tts_audio(str_)
            send_sound()
            send_json_message(str_)
            #send_expression()
    imap_object.logout()
 
# 调用

def listen_email():
    while True:
        get_data()
        time.sleep(1)

if __name__ == '__main__':
    # 在后台启动邮件监听线程（守护线程），主线程继续运行交互循环
    listener_thread = threading.Thread(target=listen_email, daemon=True)
    listener_thread.start()
    print('[INFO] 邮件监听线程已启动，主循环开始。')

    # 死循环，一直重复对话
    while True:
        try:
            saying = input("Player:")
            if saying == "exit loop":
                break
            response = chat(saying)
            send_json_message(response)
            send_sound()
            #send_expression()
            print(response)
        except Exception as e:
            print(f"大模型出现错误:{e}")