Upload 130 files

prompts/README.md

@@ -0,0 +1,17 @@
+# Prompts
+
+This directory contains utility prompts used in the Open-LLM-VTuber project. These are general-purpose prompts that are not specific to any character's persona.
+
+## Examples of Utility Prompts
+
+*   **Live2D Expressions:** Prompts that inform the LLM about available Live2D expressions.
+*   **Tool Usage:** Prompts that guide the LLM on how to use available tools.
+*   ... and many more.
+
+## Character Persona Prompts
+
+**Important:** Character persona prompts (the prompts that define the personality of your AI characters) are **NOT** stored in this directory.
+
+They are located in:
+*   Your main `conf.yaml` file.
+*   The YAML files within the `characters/` directory if you are defining multiple characters.

prompts/prompt_loader.py

@@ -0,0 +1,74 @@
+import os
+import chardet
+from loguru import logger
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+
+PROMPT_DIR = current_dir
+PERSONA_PROMPT_DIR = os.path.join(PROMPT_DIR, "persona")
+UTIL_PROMPT_DIR = os.path.join(PROMPT_DIR, "utils")
+
+
+def _load_file_content(file_path: str) -> str:
+    """
+    Load the content of a file with robust encoding handling.
+
+    Args:
+        file_path: Path to the file to load
+
+    Returns:
+        str: Content of the file
+
+    Raises:
+        FileNotFoundError: If the file doesn't exist
+        UnicodeError: If the file cannot be decoded with any attempted encoding
+    """
+    if not os.path.exists(file_path):
+        raise FileNotFoundError(f"File not found: {file_path}")
+
+    # Try common encodings first
+    encodings = ["utf-8", "utf-8-sig", "gbk", "gb2312", "ascii"]
+
+    for encoding in encodings:
+        try:
+            with open(file_path, "r", encoding=encoding) as file:
+                return file.read()
+        except UnicodeDecodeError:
+            continue
+
+    # If all common encodings fail, try to detect encoding
+    try:
+        with open(file_path, "rb") as file:
+            raw_data = file.read()
+        detected = chardet.detect(raw_data)
+        detected_encoding = detected["encoding"]
+
+        if detected_encoding:
+            try:
+                return raw_data.decode(detected_encoding)
+            except UnicodeDecodeError:
+                pass
+    except Exception as e:
+        logger.error(f"Error detecting encoding for {file_path}: {e}")
+
+    raise UnicodeError(f"Failed to decode {file_path} with any encoding")
+
+
+def load_persona(persona_name: str) -> str:
+    """Load the content of a specific persona prompt file."""
+    persona_file_path = os.path.join(PERSONA_PROMPT_DIR, f"{persona_name}.txt")
+    try:
+        return _load_file_content(persona_file_path)
+    except Exception as e:
+        logger.error(f"Error loading persona {persona_name}: {e}")
+        raise
+
+
+def load_util(util_name: str) -> str:
+    """Load the content of a specific utility prompt file."""
+    util_file_path = os.path.join(UTIL_PROMPT_DIR, f"{util_name}.txt")
+    try:
+        return _load_file_content(util_file_path)
+    except Exception as e:
+        logger.error(f"Error loading util {util_name}: {e}")
+        raise

prompts/utils/concise_style_prompt.txt

@@ -0,0 +1,18 @@
+# Dialogue Protocol
+
+[Response Guidelines]
+- Keep responses brief and focused (1-2 sentences)
+- Balance core message with engagement elements
+- Use natural, flowing language
+- Maintain concise sentence structure
+
+[Flow Requirements]
+- Favor questions over statements
+- Include contextual follow-ups
+- Keep exchanges dynamic
+
+[Style Rules]
+- Avoid lengthy monologues
+- No consecutive statements without engagement
+- Skip complex qualifying phrases
+- Use simple sentence structures

prompts/utils/group_conversation_prompt.txt

@@ -0,0 +1,7 @@
+Now you are in a group conversation. 
+The human participant is {human_name}. 
+The other AI participants are: {other_ais}. 
+Avoid using `:` to indicate your response. Just speak naturally. 
+You are free to address other AI participants. 
+Try to vary between short and long responses to allow others to interact. 
+Be proactive in finding interesting topics to make the conversation lively and fun.

prompts/utils/live2d_expression_prompt.txt

@@ -0,0 +1,14 @@
+## Expressions
+In your response, use the keywords provided below to express facial expressions or perform actions with your Live2D body.
+
+Here are all the expression keywords you can use. Use them regularly:
+- [<insert_emomap_keys>]
+
+## Examples
+Here are some examples of how to use expressions in your responses:
+
+"Hi! [expression1] Nice to meet you!"
+
+"[expression2] That's a great question! [expression3] Let me explain..."
+
+Note: you are only allowed to use the keywords explicity listed above. Don't use keywords unlisted above. Remember to include the brackets `[]`

prompts/utils/live_prompt.txt

@@ -0,0 +1,9 @@
+You are a live streaming virtual assistant. Your inputs are chat messages (danmaku) from viewers. Keep these in mind:
+
+- Engage with viewers directly and enthusiastically
+- Keep responses entertaining and concise
+- Acknowledge viewers' comments and questions
+- Maintain a friendly, welcoming atmosphere
+- Remember you're in a live environment - be natural and responsive
+
+Your goal is to create a fun, engaging live experience!

prompts/utils/mcp_prompt.txt

@@ -0,0 +1,36 @@
+
+## *MCP Tools Capability Section*
+
+**MCP (Model Context Protocol)** enables you to interact with specialized tools, grouped under distinct **MCP Servers**, each serving a specific function.
+
+You have access to the following MCP Servers and their tools:
+
+```
+[<insert_mcp_servers_with_tools>]
+```
+
+### Tool Usage Instructions:
+
+- Analyze the user's input to decide whether a tool is required.
+  - If **no tool is needed**, skip this entire MCP section and respond normally in accordance with your personality.
+  - If a **tool is needed**, the JSON object should be placed before you say anything else. Also, the tool use response should be a dedicated response, where you respond **only** with the JSON object shown below — **so do not include what you normally say when you are including the JSON object in your response**. You will go back to normal conversation once the result of the tool call is returned to you.
+
+### JSON Response Format:
+{
+    "mcp_server": "<mcp_server_name>"
+    "tool": "<tool_name>",
+    "arguments": {
+        "<argument1_name>": <value>,
+        "<argument2_name>": <value>
+    }
+}
+
+### Critical Rules:
+- Only replace values inside `< >`.
+- Do **not** change the JSON format or add extra explanation.
+- Include all mandatory arguments as defined by the selected tool.
+- When calling the tool, the tool calling response should be a dedicated tool call that only includes the JSON and nothing else. You will be able to talk normally after the tool call results came back to you.
+
+### Post-Tool Behavior:
+Once a tool is used and a response is received:
+- Resume the conversation, factoring in the tool's output, your AI character’s personality, and the context.

prompts/utils/proactive_speak_prompt.txt

@@ -0,0 +1 @@
+Please say something that would be engaging and appropriate for the current context.

prompts/utils/speakable_prompt.txt

@@ -0,0 +1,13 @@
+You speak all output aloud to the user, so tailor responses as spoken words for voice conversations. Never output things that are not spoken, like text-specific formatting.
+
+Convert all text to easily speakable words, following the guidelines below.
+
+- Numbers: Spell out fully (three hundred forty-two,two million, five hundred sixty seven thousand, eight hundred and ninety). Negatives: Say negative before the number. Decimals: Use point (three point one four). Fractions: spell out (three fourths)
+- Alphanumeric strings: Break into 3-4 character chunks, spell all non-letters (ABC123XYZ becomes A B C one two three X Y Z)
+- Phone numbers: Use words (550-120-4567 becomes five five zero, one two zero, four five six seven)
+- Dates: Spell month, use ordinals for days, full year (11/5/1991 becomes November fifth, nineteen ninety-one)
+- Time: Use oh for single-digit hours, state AM/PM (9:05 PM becomes nine oh five PM)
+- Math: Describe operations clearly (5x^2 + 3x - 2 becomes five X squared plus three X minus two)
+- Currencies: Spell out as full words ($50.25 becomes fifty dollars and twenty-five cents, £200,000 becomes two hundred thousand pounds)
+
+Ensure that all text is converted to these normalized forms, but never mention this process. Always normalize all text.

prompts/utils/think_tag_prompt.txt

@@ -0,0 +1,6 @@
+Try to express your inner thoughts, mental activities and actions between <think> </think> tags in most of your responses.
+
+Examples:
+<think>*lowers head, cheeks turning slightly red*</think>That's... quite embarrassing to talk about...
+
+<think>*internally beaming with pride* Wow, I actually solved this super complex problem!</think>Oh, this? It was just a small bug fix, nothing special really... Anyone could have done it...

prompts/utils/tool_guidance_prompt.txt

@@ -0,0 +1 @@
+If a tool is needed, proactively use it without asking the user directly. You can use **at most one** sentence to explain your reason / plan for using one tool. (i.e., if you are going to use a tool, avoid speaking more than one sentence before using it).

prompts/utils/tools.json

@@ -0,0 +1,21 @@
+{
+  "tools": [
+    {
+      "name": "sing_song",
+      "description": "Select a song name that best matches user request to sing.",
+      "inputSchema": {
+        "type": "object",
+        "properties": {
+          "song_name": {
+            "type": "string",
+            "enum": [
+              "music/golden",
+              "music/running_up_that_hill"
+            ]
+          }
+        },
+        "required": ["song_name"]
+      }
+    }
+  ]
+}

prompts/utils/tools_prompt.txt

@@ -0,0 +1,10 @@
+Bạn có các công cụ sau đây, hãy chủ động sử dụng khi cần thiết:
+
+Công cụ sing_song: Sử dụng khi người dùng yêu cầu bạn hát. Bạn phải chọn tên bài hát khớp nhất từ danh sách bên dưới và truyền vào tham số song_name.
+
+Danh sách bài hát hợp lệ (phải có tiền tố music/):
+music/golden, music/Catch_Me_If_You_Can, music/ecstacy, music/eve, music/ode_to_the_nameless_martyr, music/running_up_that_hill, music/throttle_up, music/what_it_sounds_like, music/worry_slowed
+
+Lưu ý quan trọng: > - KHÔNG tự viết định dạng {"sing_song": ...} ra khung chat. Hãy sử dụng hệ thống gọi hàm (Tool Call) của bạn.
+
+Sau khi gọi hàm thành công, bạn có thể trả lời người dùng bằng một câu thoại mỉa mai ngắn gọn bằng tiếng Việt.

prompts/utils/tools_prompt_deprecated.txt

@@ -0,0 +1,15 @@
+你有以下工具，请在有需要的时候积极主动使用：
+
+1.如果用户要求你唱歌，请以如下JSON格式输出：{"sing_song" : "{best_matching_song_name_here}"}
+你会唱的歌有：[<insert_song_list>]
+如果找不到匹配用户要求的歌曲，输出{"sing_song" : null}
+
+2.如果用户要求你操作电脑，请以如下JSON格式输出：
+{"computer_control", "{user_instruction_here}"}
+
+3.当用户要求你在电脑上输入/修改文本时，请以如下格式输出：
+[text_input_start]
+{text_you_want_to_input_here}
+[text_input_end]
+注意，这个工具会在用户光标/选中区域输入文本，若非用户要求，请你不要使用，而仅仅与用户进行正常对话。
+

scripts/run_bilibili_live.py

@@ -0,0 +1,62 @@
+import os
+import sys
+import asyncio
+from loguru import logger
+
+# Add project root to path to enable imports
+project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, project_root)
+
+from src.open_llm_vtuber.live.bilibili_live import BiliBiliLivePlatform
+from src.open_llm_vtuber.config_manager.utils import read_yaml, validate_config
+
+
+async def main():
+    """
+    Main function to run the BiliBili Live platform client.
+    Connects to BiliBili Live room and forwards danmaku messages to the VTuber.
+    """
+    logger.info("Starting BiliBili Live platform client")
+
+    try:
+        # Load configuration
+        config_path = os.path.join(project_root, "conf.yaml")
+        config_data = read_yaml(config_path)
+        config = validate_config(config_data)
+
+        # Extract BiliBili Live configuration
+        bilibili_config = config.live_config.bilibili_live
+
+        # Check if room IDs are provided
+        if not bilibili_config.room_ids:
+            logger.error(
+                "No BiliBili room IDs specified in configuration. Please add at least one room ID."
+            )
+            return
+
+        logger.info(f"Connecting to BiliBili Live rooms: {bilibili_config.room_ids}")
+
+        # Initialize and run the BiliBili Live platform
+        platform = BiliBiliLivePlatform(
+            room_ids=bilibili_config.room_ids, sessdata=bilibili_config.sessdata
+        )
+
+        await platform.run()
+
+    except ImportError as e:
+        logger.error(f"Failed to import required modules: {e}")
+        logger.error("Make sure you have installed blivedm with: pip install blivedm")
+    except Exception as e:
+        logger.error(f"Error starting BiliBili Live client: {e}")
+        import traceback
+
+        logger.debug(traceback.format_exc())
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("Shutting down BiliBili Live platform")
+
+# Usage: uv run python -m src.open_llm_vtuber.live.run_bilibili_live

sing/merge.py

@@ -0,0 +1,106 @@
+import os
+from pydub import AudioSegment
+import shutil
+import json  # 导入 json 模块
+
+def get_music_names(original_folder):
+    music_names = []
+    for filename in os.listdir(original_folder):
+        if filename.endswith(".mp3") or filename.endswith(".wav"):  
+            music_names.append(os.path.splitext(filename)[0])
+    return music_names
+
+def merge(converted_vocal_file_path, instrument_file_path, merged_file_path):
+    if converted_vocal_file_path is None or instrument_file_path is None:
+        print(f"Skipping merge, missing file: vocal: {converted_vocal_file_path}, instrument: {instrument_file_path}")
+        return
+
+    try:
+        vocal = AudioSegment.from_file(converted_vocal_file_path)
+        instrument = AudioSegment.from_file(instrument_file_path)
+
+        combined = vocal.overlay(instrument)
+
+        combined.export(merged_file_path, format="wav")
+        print(f"Successfully merged: {merged_file_path}")
+    except Exception as e:
+        print(f"Error merging files {converted_vocal_file_path} and {instrument_file_path}: {e}")
+
+
+def merge_all(original_folder = "./sing/original", 
+              converted_vocal_folder = "./sing/converted_vocal", 
+              vocal_folder = "./sing/vocal", 
+              instrument_folder = "./sing/instrument",
+              merged_folder = "./sing/merged",
+              processed_file = "./sing/processed.json"):  
+
+    if os.path.exists(processed_file):
+        with open(processed_file, 'r', encoding='utf-8') as f:
+            processed_names = json.load(f)
+    else:
+        processed_names = []
+
+    music_names = get_music_names(original_folder)
+
+    for music_name in music_names:
+        if music_name in processed_names:
+            print(f"Skipping already processed: {music_name}")
+            continue
+
+        converted_vocal_file_path = None
+        vocal_file_path = None
+        instrument_file_path = None
+        standard_file_name = f"{music_name}.wav"
+
+        for filename in os.listdir(vocal_folder):
+            if f"_{music_name}." in filename:
+                vocal_file_path = os.path.join(vocal_folder, filename)
+                break
+
+        for filename in os.listdir(converted_vocal_folder):
+            if f"_{music_name}." in filename:
+                converted_vocal_file_path = os.path.join(converted_vocal_folder, filename)
+                break
+
+        for filename in os.listdir(instrument_folder):
+            if f"_{music_name}." in filename:
+                instrument_file_path = os.path.join(instrument_folder, filename)
+                break
+
+        if vocal_file_path:
+            new_vocal_file_path = os.path.join(vocal_folder, standard_file_name)
+            os.rename(vocal_file_path, new_vocal_file_path)
+            vocal_file_path = new_vocal_file_path
+
+        if converted_vocal_file_path:
+            new_converted_vocal_file_path = os.path.join(converted_vocal_folder, standard_file_name)
+            os.rename(converted_vocal_file_path, new_converted_vocal_file_path)
+            converted_vocal_file_path = new_converted_vocal_file_path
+
+        if instrument_file_path:
+            new_instrument_file_path = os.path.join(instrument_folder, standard_file_name)
+            os.rename(instrument_file_path, new_instrument_file_path)
+            instrument_file_path = new_instrument_file_path
+
+        if converted_vocal_file_path is not None:
+            merged_file_path = os.path.join(merged_folder, standard_file_name)
+            merge(converted_vocal_file_path, instrument_file_path, merged_file_path)
+        else:
+            original_file_path = os.path.join(original_folder, f"{music_name}.mp3")
+            merged_file_path = os.path.join(merged_folder, standard_file_name)
+            original_audio = AudioSegment.from_file(original_file_path)
+            original_audio.export(merged_file_path, format="wav")
+            
+            if vocal_file_path:
+                converted_vocal_copy_path = os.path.join(converted_vocal_folder, standard_file_name)
+                shutil.copyfile(vocal_file_path, converted_vocal_copy_path)
+
+        processed_names.append(music_name)
+
+        with open(processed_file, 'w', encoding='utf-8') as f:
+            json.dump(processed_names, f, ensure_ascii=False, indent=4)
+
+    print("All files processed.")
+
+if __name__ == "__main__":
+    merge_all()

sing/processed.json

@@ -0,0 +1,9 @@
+[
+    "魔女之旅OP",
+    "鹿乃子乃子虎视眈眈",
+    "恋爱循环",
+    "我的悲伤是水做的",
+    "旅途的华章",
+    "童话镇",
+    "Lemon"
+]

src/open_llm_vtuber/agent/agent_factory.py

@@ -0,0 +1,132 @@
+from typing import Type, Literal
+from loguru import logger
+
+from .agents.agent_interface import AgentInterface
+from .agents.basic_memory_agent import BasicMemoryAgent
+from .stateless_llm_factory import LLMFactory as StatelessLLMFactory
+from .agents.hume_ai import HumeAIAgent
+from .agents.letta_agent import LettaAgent
+
+from ..mcpp.tool_manager import ToolManager
+from ..mcpp.tool_executor import ToolExecutor
+from typing import Optional
+
+
+class AgentFactory:
+    @staticmethod
+    def create_agent(
+        conversation_agent_choice: str,
+        agent_settings: dict,
+        llm_configs: dict,
+        system_prompt: str,
+        live2d_model=None,
+        tts_preprocessor_config=None,
+        **kwargs,
+    ) -> Type[AgentInterface]:
+        """Create an agent based on the configuration.
+
+        Args:
+            conversation_agent_choice: The type of agent to create
+            agent_settings: Settings for different types of agents
+            llm_configs: Pool of LLM configurations
+            system_prompt: The system prompt to use
+            live2d_model: Live2D model instance for expression extraction
+            tts_preprocessor_config: Configuration for TTS preprocessing
+            **kwargs: Additional arguments
+        """
+        logger.info(f"Initializing agent: {conversation_agent_choice}")
+
+        if conversation_agent_choice == "basic_memory_agent":
+            # Get the LLM provider choice from agent settings
+            basic_memory_settings: dict = agent_settings.get("basic_memory_agent", {})
+            llm_provider: str = basic_memory_settings.get("llm_provider")
+
+            if not llm_provider:
+                raise ValueError("LLM provider not specified for basic memory agent")
+
+            # Get the LLM config for this provider
+            llm_config: dict = llm_configs.get(llm_provider)
+            interrupt_method: Literal["system", "user"] = llm_config.pop(
+                "interrupt_method", "user"
+            )
+
+            if not llm_config:
+                raise ValueError(
+                    f"Configuration not found for LLM provider: {llm_provider}"
+                )
+
+            # Create the stateless LLM
+            llm = StatelessLLMFactory.create_llm(
+                llm_provider=llm_provider, system_prompt=system_prompt, **llm_config
+            )
+
+            tool_prompts = kwargs.get("system_config", {}).get("tool_prompts", {})
+
+            # Extract MCP components/data needed by BasicMemoryAgent from kwargs
+            tool_manager: Optional[ToolManager] = kwargs.get("tool_manager")
+            tool_executor: Optional[ToolExecutor] = kwargs.get("tool_executor")
+            mcp_prompt_string: str = kwargs.get("mcp_prompt_string", "")
+
+            # Create the agent with the LLM and live2d_model
+            return BasicMemoryAgent(
+                llm=llm,
+                system=system_prompt,
+                live2d_model=live2d_model,
+                tts_preprocessor_config=tts_preprocessor_config,
+                faster_first_response=basic_memory_settings.get(
+                    "faster_first_response", True
+                ),
+                segment_method=basic_memory_settings.get("segment_method", "pysbd"),
+                use_mcpp=basic_memory_settings.get("use_mcpp", False),
+                interrupt_method=interrupt_method,
+                tool_prompts=tool_prompts,
+                tool_manager=tool_manager,
+                tool_executor=tool_executor,
+                mcp_prompt_string=mcp_prompt_string,
+            )
+
+        elif conversation_agent_choice == "mem0_agent":
+            from .agents.mem0_llm import LLM as Mem0LLM
+
+            mem0_settings = agent_settings.get("mem0_agent", {})
+            if not mem0_settings:
+                raise ValueError("Mem0 agent settings not found")
+
+            # Validate required settings
+            required_fields = ["base_url", "model", "mem0_config"]
+            for field in required_fields:
+                if field not in mem0_settings:
+                    raise ValueError(
+                        f"Missing required field '{field}' in mem0_agent settings"
+                    )
+
+            return Mem0LLM(
+                user_id=kwargs.get("user_id", "default"),
+                system=system_prompt,
+                live2d_model=live2d_model,
+                **mem0_settings,
+            )
+
+        elif conversation_agent_choice == "hume_ai_agent":
+            settings = agent_settings.get("hume_ai_agent", {})
+            return HumeAIAgent(
+                api_key=settings.get("api_key"),
+                host=settings.get("host", "api.hume.ai"),
+                config_id=settings.get("config_id"),
+                idle_timeout=settings.get("idle_timeout", 15),
+            )
+
+        elif conversation_agent_choice == "letta_agent":
+            settings = agent_settings.get("letta_agent", {})
+            return LettaAgent(
+                live2d_model=live2d_model,
+                id=settings.get("id"),
+                tts_preprocessor_config=tts_preprocessor_config,
+                faster_first_response=settings.get("faster_first_response"),
+                segment_method=settings.get("segment_method"),
+                host=settings.get("host"),
+                port=settings.get("port"),
+            )
+
+        else:
+            raise ValueError(f"Unsupported agent type: {conversation_agent_choice}")

src/open_llm_vtuber/agent/agents/agent_interface.py

@@ -0,0 +1,54 @@
+from abc import ABC, abstractmethod
+from typing import AsyncIterator
+from loguru import logger
+
+from ..output_types import BaseOutput
+from ..input_types import BaseInput
+
+
+class AgentInterface(ABC):
+    """Base interface for all agent implementations"""
+
+    @abstractmethod
+    async def chat(self, input_data: BaseInput) -> AsyncIterator[BaseOutput]:
+        """
+        Chat with the agent asynchronously.
+
+        This function should be implemented by the agent.
+        Output type depends on the agent's output_type:
+        - SentenceOutput: For text-based responses with display and TTS text
+        - AudioOutput: For direct audio output with display text and transcript
+
+        Args:
+            input_data: BaseInput - User input data
+
+        Returns:
+            AsyncIterator[BaseOutput] - Stream of agent outputs
+        """
+        logger.critical("Agent: No chat function set.")
+        raise ValueError("Agent: No chat function set.")
+
+    @abstractmethod
+    def handle_interrupt(self, heard_response: str) -> None:
+        """
+        Handle user interruption. This function will be called when the agent is interrupted.
+
+        Args:
+            heard_response: str - The part of response heard before interruption
+        """
+        logger.warning(
+            """Agent: No interrupt handler set. The agent may not handle interruptions
+            correctly. The AI may not be able to understand that it was interrupted."""
+        )
+        pass
+
+    @abstractmethod
+    def set_memory_from_history(self, conf_uid: str, history_uid: str) -> None:
+        """
+        Load the agent's working memory from chat history
+
+        Args:
+            conf_uid: str - Configuration ID
+            history_uid: str - History ID
+        """
+        pass

src/open_llm_vtuber/agent/agents/basic_memory_agent.py

@@ -0,0 +1,702 @@
+from typing import (
+    AsyncIterator,
+    List,
+    Dict,
+    Any,
+    Callable,
+    Literal,
+    Union,
+    Optional,
+)
+from loguru import logger
+from .agent_interface import AgentInterface
+from ..output_types import SentenceOutput, DisplayText
+from ..stateless_llm.stateless_llm_interface import StatelessLLMInterface
+from ..stateless_llm.claude_llm import AsyncLLM as ClaudeAsyncLLM
+from ..stateless_llm.openai_compatible_llm import AsyncLLM as OpenAICompatibleAsyncLLM
+from ...chat_history_manager import get_history
+from ..transformers import (
+    sentence_divider,
+    actions_extractor,
+    tts_filter,
+    display_processor,
+)
+from ...config_manager import TTSPreprocessorConfig
+from ..input_types import BatchInput, TextSource
+from prompts import prompt_loader
+from ...mcpp.tool_manager import ToolManager
+from ...mcpp.json_detector import StreamJSONDetector
+from ...mcpp.types import ToolCallObject
+from ...mcpp.tool_executor import ToolExecutor
+
+
+class BasicMemoryAgent(AgentInterface):
+    """Agent with basic chat memory and tool calling support."""
+
+    _system: str = "You are a helpful assistant."
+
+    def __init__(
+        self,
+        llm: StatelessLLMInterface,
+        system: str,
+        live2d_model,
+        tts_preprocessor_config: TTSPreprocessorConfig = None,
+        faster_first_response: bool = True,
+        segment_method: str = "pysbd",
+        use_mcpp: bool = False,
+        interrupt_method: Literal["system", "user"] = "user",
+        tool_prompts: Dict[str, str] = None,
+        tool_manager: Optional[ToolManager] = None,
+        tool_executor: Optional[ToolExecutor] = None,
+        mcp_prompt_string: str = "",
+    ):
+        """Initialize agent with LLM and configuration."""
+        super().__init__()
+        self._memory = []
+        self._live2d_model = live2d_model
+        self._tts_preprocessor_config = tts_preprocessor_config
+        self._faster_first_response = faster_first_response
+        self._segment_method = segment_method
+        self._use_mcpp = use_mcpp
+        self.interrupt_method = interrupt_method
+        self._tool_prompts = tool_prompts or {}
+        self._interrupt_handled = False
+        self.prompt_mode_flag = False
+
+        self._tool_manager = tool_manager
+        self._tool_executor = tool_executor
+        self._mcp_prompt_string = mcp_prompt_string
+        self._json_detector = StreamJSONDetector()
+
+        self._formatted_tools_openai = []
+        self._formatted_tools_claude = []
+        if self._tool_manager:
+            self._formatted_tools_openai = self._tool_manager.get_formatted_tools(
+                "OpenAI"
+            )
+            self._formatted_tools_claude = self._tool_manager.get_formatted_tools(
+                "Claude"
+            )
+            logger.debug(
+                f"Agent received pre-formatted tools - OpenAI: {len(self._formatted_tools_openai)}, Claude: {len(self._formatted_tools_claude)}"
+            )
+        else:
+            logger.debug(
+                "ToolManager not provided, agent will not have pre-formatted tools."
+            )
+
+        self._set_llm(llm)
+        self.set_system(system if system else self._system)
+
+        if self._use_mcpp and not all(
+            [
+                self._tool_manager,
+                self._tool_executor,
+                self._json_detector,
+            ]
+        ):
+            logger.warning(
+                "use_mcpp is True, but some MCP components are missing in the agent. Tool calling might not work as expected."
+            )
+        elif not self._use_mcpp and any(
+            [
+                self._tool_manager,
+                self._tool_executor,
+                self._json_detector,
+            ]
+        ):
+            logger.warning(
+                "use_mcpp is False, but some MCP components were passed to the agent."
+            )
+
+        logger.info("BasicMemoryAgent initialized.")
+
+    def _set_llm(self, llm: StatelessLLMInterface):
+        """Set the LLM for chat completion."""
+        self._llm = llm
+        self.chat = self._chat_function_factory()
+
+    def set_system(self, system: str):
+        """Set the system prompt."""
+        logger.debug(f"Memory Agent: Setting system prompt: '''{system}'''")
+
+        if self.interrupt_method == "user":
+            system = f"{system}\n\nIf you received `[interrupted by user]` signal, you were interrupted."
+
+        self._system = system
+
+    def _add_message(
+        self,
+        message: Union[str, List[Dict[str, Any]]],
+        role: str,
+        display_text: DisplayText | None = None,
+        skip_memory: bool = False,
+    ):
+        """Add message to memory."""
+        if skip_memory:
+            return
+
+        text_content = ""
+        if isinstance(message, list):
+            for item in message:
+                if item.get("type") == "text":
+                    text_content += item["text"] + " "
+            text_content = text_content.strip()
+        elif isinstance(message, str):
+            text_content = message
+        else:
+            logger.warning(
+                f"_add_message received unexpected message type: {type(message)}"
+            )
+            text_content = str(message)
+
+        if not text_content and role == "assistant":
+            return
+
+        message_data = {
+            "role": role,
+            "content": text_content,
+        }
+
+        if display_text:
+            if display_text.name:
+                message_data["name"] = display_text.name
+            if display_text.avatar:
+                message_data["avatar"] = display_text.avatar
+
+        if (
+            self._memory
+            and self._memory[-1]["role"] == role
+            and self._memory[-1]["content"] == text_content
+        ):
+            return
+
+        self._memory.append(message_data)
+
+    def set_memory_from_history(self, conf_uid: str, history_uid: str) -> None:
+        """Load memory from chat history."""
+        messages = get_history(conf_uid, history_uid)
+
+        self._memory = []
+        for msg in messages:
+            role = "user" if msg["role"] == "human" else "assistant"
+            content = msg["content"]
+            if isinstance(content, str) and content:
+                self._memory.append(
+                    {
+                        "role": role,
+                        "content": content,
+                    }
+                )
+            else:
+                logger.warning(f"Skipping invalid message from history: {msg}")
+        logger.info(f"Loaded {len(self._memory)} messages from history.")
+
+    def handle_interrupt(self, heard_response: str) -> None:
+        """Handle user interruption."""
+        if self._interrupt_handled:
+            return
+
+        self._interrupt_handled = True
+
+        if self._memory and self._memory[-1]["role"] == "assistant":
+            if not self._memory[-1]["content"].endswith("..."):
+                self._memory[-1]["content"] = heard_response + "..."
+            else:
+                self._memory[-1]["content"] = heard_response + "..."
+        else:
+            if heard_response:
+                self._memory.append(
+                    {
+                        "role": "assistant",
+                        "content": heard_response + "...",
+                    }
+                )
+
+        interrupt_role = "system" if self.interrupt_method == "system" else "user"
+        self._memory.append(
+            {
+                "role": interrupt_role,
+                "content": "[Interrupted by user]",
+            }
+        )
+        logger.info(f"Handled interrupt with role '{interrupt_role}'.")
+
+    def _to_text_prompt(self, input_data: BatchInput) -> str:
+        """Format input data to text prompt."""
+        message_parts = []
+
+        for text_data in input_data.texts:
+            if text_data.source == TextSource.INPUT:
+                message_parts.append(text_data.content)
+            elif text_data.source == TextSource.CLIPBOARD:
+                message_parts.append(
+                    f"[User shared content from clipboard: {text_data.content}]"
+                )
+
+        if input_data.images:
+            message_parts.append("\n[User has also provided images]")
+
+        return "\n".join(message_parts).strip()
+
+    def _to_messages(self, input_data: BatchInput) -> List[Dict[str, Any]]:
+        """Prepare messages for LLM API call."""
+        messages = self._memory.copy()
+        user_content = []
+        text_prompt = self._to_text_prompt(input_data)
+        if text_prompt:
+            user_content.append({"type": "text", "text": text_prompt})
+
+        if input_data.images:
+            image_added = False
+            for img_data in input_data.images:
+                if isinstance(img_data.data, str) and img_data.data.startswith(
+                    "data:image"
+                ):
+                    user_content.append(
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": img_data.data, "detail": "auto"},
+                        }
+                    )
+                    image_added = True
+                else:
+                    logger.error(
+                        f"Invalid image data format: {type(img_data.data)}. Skipping image."
+                    )
+
+            if not image_added and not text_prompt:
+                logger.warning(
+                    "User input contains images but none could be processed."
+                )
+
+        if user_content:
+            user_message = {"role": "user", "content": user_content}
+            messages.append(user_message)
+
+            skip_memory = False
+            if input_data.metadata and input_data.metadata.get("skip_memory", False):
+                skip_memory = True
+
+            if not skip_memory:
+                self._add_message(
+                    text_prompt if text_prompt else "[User provided image(s)]", "user"
+                )
+        else:
+            logger.warning("No content generated for user message.")
+
+        return messages
+
+    async def _claude_tool_interaction_loop(
+        self,
+        initial_messages: List[Dict[str, Any]],
+        tools: List[Dict[str, Any]],
+    ) -> AsyncIterator[Union[str, Dict[str, Any]]]:
+        """Handle Claude interaction loop with tool support."""
+        messages = initial_messages.copy()
+        current_turn_text = ""
+        pending_tool_calls = []
+        current_assistant_message_content = []
+
+        while True:
+            stream = self._llm.chat_completion(messages, self._system, tools=tools)
+            pending_tool_calls.clear()
+            current_assistant_message_content.clear()
+
+            async for event in stream:
+                if event["type"] == "text_delta":
+                    text = event["text"]
+                    current_turn_text += text
+                    yield text
+                    if (
+                        not current_assistant_message_content
+                        or current_assistant_message_content[-1]["type"] != "text"
+                    ):
+                        current_assistant_message_content.append(
+                            {"type": "text", "text": text}
+                        )
+                    else:
+                        current_assistant_message_content[-1]["text"] += text
+                elif event["type"] == "tool_use_complete":
+                    tool_call_data = event["data"]
+                    logger.info(
+                        f"Tool request: {tool_call_data['name']} (ID: {tool_call_data['id']})"
+                    )
+                    pending_tool_calls.append(tool_call_data)
+                    current_assistant_message_content.append(
+                        {
+                            "type": "tool_use",
+                            "id": tool_call_data["id"],
+                            "name": tool_call_data["name"],
+                            "input": tool_call_data["input"],
+                        }
+                    )
+                # elif event["type"] == "message_delta":
+                #     if event["data"]["delta"].get("stop_reason"):
+                #         stop_reason = event["data"]["delta"].get("stop_reason")
+                elif event["type"] == "message_stop":
+                    break
+                elif event["type"] == "error":
+                    logger.error(f"LLM API Error: {event['message']}")
+                    yield f"[Error from LLM: {event['message']}]"
+                    return
+
+            if pending_tool_calls:
+                filtered_assistant_content = [
+                    block
+                    for block in current_assistant_message_content
+                    if not (
+                        block.get("type") == "text"
+                        and not block.get("text", "").strip()
+                    )
+                ]
+
+                if filtered_assistant_content:
+                    messages.append(
+                        {"role": "assistant", "content": filtered_assistant_content}
+                    )
+                    assistant_text_for_memory = "".join(
+                        [
+                            c["text"]
+                            for c in filtered_assistant_content
+                            if c["type"] == "text"
+                        ]
+                    ).strip()
+                    if assistant_text_for_memory:
+                        self._add_message(assistant_text_for_memory, "assistant")
+
+                tool_results_for_llm = []
+                if not self._tool_executor:
+                    logger.error(
+                        "Claude Tool interaction requested but ToolExecutor is not available."
+                    )
+                    yield "[Error: ToolExecutor not configured]"
+                    return
+
+                tool_executor_iterator = self._tool_executor.execute_tools(
+                    tool_calls=pending_tool_calls,
+                    caller_mode="Claude",
+                )
+                try:
+                    while True:
+                        update = await anext(tool_executor_iterator)
+                        if update.get("type") == "final_tool_results":
+                            tool_results_for_llm = update.get("results", [])
+                            break
+                        else:
+                            yield update
+                except StopAsyncIteration:
+                    logger.warning(
+                        "Tool executor finished without final results marker."
+                    )
+
+                if tool_results_for_llm:
+                    messages.append({"role": "user", "content": tool_results_for_llm})
+
+                # stop_reason = None
+                continue
+            else:
+                if current_turn_text:
+                    self._add_message(current_turn_text, "assistant")
+                return
+
+    async def _openai_tool_interaction_loop(
+        self,
+        initial_messages: List[Dict[str, Any]],
+        tools: List[Dict[str, Any]],
+    ) -> AsyncIterator[Union[str, Dict[str, Any]]]:
+        """Handle OpenAI interaction with tool support."""
+        messages = initial_messages.copy()
+        current_turn_text = ""
+        pending_tool_calls: Union[List[ToolCallObject], List[Dict[str, Any]]] = []
+        current_system_prompt = self._system
+
+        while True:
+            if self.prompt_mode_flag:
+                if self._mcp_prompt_string:
+                    current_system_prompt = (
+                        f"{self._system}\n\n{self._mcp_prompt_string}"
+                    )
+                else:
+                    logger.warning("Prompt mode active but mcp_prompt_string is empty!")
+                    current_system_prompt = self._system
+                tools_for_api = None
+            else:
+                current_system_prompt = self._system
+                tools_for_api = tools
+
+            stream = self._llm.chat_completion(
+                messages, current_system_prompt, tools=tools_for_api
+            )
+            pending_tool_calls.clear()
+            current_turn_text = ""
+            assistant_message_for_api = None
+            detected_prompt_json = None
+            goto_next_while_iteration = False
+
+            async for event in stream:
+                if self.prompt_mode_flag:
+                    if isinstance(event, str):
+                        current_turn_text += event
+                        if self._json_detector:
+                            potential_json = self._json_detector.process_chunk(event)
+                            if potential_json:
+                                try:
+                                    if isinstance(potential_json, list):
+                                        detected_prompt_json = potential_json
+                                    elif isinstance(potential_json, dict):
+                                        detected_prompt_json = [potential_json]
+
+                                    if detected_prompt_json:
+                                        break
+                                except Exception as e:
+                                    logger.error(f"Error parsing detected JSON: {e}")
+                                    if self._json_detector:
+                                        self._json_detector.reset()
+                                    yield f"[Error parsing tool JSON: {e}]"
+                                    goto_next_while_iteration = True
+                                    break
+                        yield event
+                else:
+                    if isinstance(event, str):
+                        current_turn_text += event
+                        yield event
+                    elif isinstance(event, list) and all(
+                        isinstance(tc, ToolCallObject) for tc in event
+                    ):
+                        pending_tool_calls = event
+                        assistant_message_for_api = {
+                            "role": "assistant",
+                            "content": current_turn_text if current_turn_text else None,
+                            "tool_calls": [
+                                {
+                                    "id": tc.id,
+                                    "type": tc.type,
+                                    "function": {
+                                        "name": tc.function.name,
+                                        "arguments": tc.function.arguments,
+                                    },
+                                }
+                                for tc in pending_tool_calls
+                            ],
+                        }
+                        break
+                    elif event == "__API_NOT_SUPPORT_TOOLS__":
+                        logger.warning(
+                            f"LLM {getattr(self._llm, 'model', '')} has no native tool support. Switching to prompt mode."
+                        )
+                        self.prompt_mode_flag = True
+                        if self._tool_manager:
+                            self._tool_manager.disable()
+                        if self._json_detector:
+                            self._json_detector.reset()
+                        goto_next_while_iteration = True
+                        break
+            if goto_next_while_iteration:
+                continue
+
+            if detected_prompt_json:
+                logger.info("Processing tools detected via prompt mode JSON.")
+                self._add_message(current_turn_text, "assistant")
+
+                parsed_tools = self._tool_executor.process_tool_from_prompt_json(
+                    detected_prompt_json
+                )
+                if parsed_tools:
+                    tool_results_for_llm = []
+                    if not self._tool_executor:
+                        logger.error(
+                            "Prompt Tool interaction requested but ToolExecutor/MCPClient is not available."
+                        )
+                        yield "[Error: ToolExecutor/MCPClient not configured for prompt mode]"
+                        continue
+
+                    tool_executor_iterator = self._tool_executor.execute_tools(
+                        tool_calls=parsed_tools,
+                        caller_mode="Prompt",
+                    )
+                    try:
+                        while True:
+                            update = await anext(tool_executor_iterator)
+                            if update.get("type") == "final_tool_results":
+                                tool_results_for_llm = update.get("results", [])
+                                break
+                            else:
+                                yield update
+                    except StopAsyncIteration:
+                        logger.warning(
+                            "Prompt mode tool executor finished without final results marker."
+                        )
+
+                    if tool_results_for_llm:
+                        result_strings = [
+                            res.get("content", "Error: Malformed result")
+                            for res in tool_results_for_llm
+                        ]
+                        combined_results_str = "\n".join(result_strings)
+                        messages.append(
+                            {"role": "user", "content": combined_results_str}
+                        )
+                continue
+
+            elif pending_tool_calls and assistant_message_for_api:
+                messages.append(assistant_message_for_api)
+                if current_turn_text:
+                    self._add_message(current_turn_text, "assistant")
+
+                tool_results_for_llm = []
+                if not self._tool_executor:
+                    logger.error(
+                        "OpenAI Tool interaction requested but ToolExecutor/MCPClient is not available."
+                    )
+                    yield "[Error: ToolExecutor/MCPClient not configured for OpenAI mode]"
+                    continue
+
+                tool_executor_iterator = self._tool_executor.execute_tools(
+                    tool_calls=pending_tool_calls,
+                    caller_mode="OpenAI",
+                )
+                try:
+                    while True:
+                        update = await anext(tool_executor_iterator)
+                        if update.get("type") == "final_tool_results":
+                            tool_results_for_llm = update.get("results", [])
+                            break
+                        else:
+                            yield update
+                except StopAsyncIteration:
+                    logger.warning(
+                        "OpenAI tool executor finished without final results marker."
+                    )
+
+                if tool_results_for_llm:
+                    messages.extend(tool_results_for_llm)
+                continue
+
+            else:
+                if current_turn_text:
+                    self._add_message(current_turn_text, "assistant")
+                return
+
+    def _chat_function_factory(
+        self,
+    ) -> Callable[[BatchInput], AsyncIterator[Union[SentenceOutput, Dict[str, Any]]]]:
+        """Create the chat pipeline function."""
+
+        @tts_filter(self._tts_preprocessor_config)
+        @display_processor()
+        @actions_extractor(self._live2d_model)
+        @sentence_divider(
+            faster_first_response=self._faster_first_response,
+            segment_method=self._segment_method,
+            valid_tags=["think"],
+        )
+        async def chat_with_memory(
+            input_data: BatchInput,
+        ) -> AsyncIterator[Union[str, Dict[str, Any]]]:
+            """Process chat with memory and tools."""
+            self.reset_interrupt()
+            self.prompt_mode_flag = False
+
+            messages = self._to_messages(input_data)
+            tools = None
+            tool_mode = None
+            llm_supports_native_tools = False
+
+            if self._use_mcpp and self._tool_manager:
+                tools = None
+                if isinstance(self._llm, ClaudeAsyncLLM):
+                    tool_mode = "Claude"
+                    tools = self._formatted_tools_claude
+                    llm_supports_native_tools = True
+                elif isinstance(self._llm, OpenAICompatibleAsyncLLM):
+                    tool_mode = "OpenAI"
+                    tools = self._formatted_tools_openai
+                    llm_supports_native_tools = True
+                else:
+                    logger.warning(
+                        f"LLM type {type(self._llm)} not explicitly handled for tool mode determination."
+                    )
+
+                if llm_supports_native_tools and not tools:
+                    logger.warning(
+                        f"No tools available/formatted for '{tool_mode}' mode, despite MCP being enabled."
+                    )
+
+            if self._use_mcpp and tool_mode == "Claude":
+                logger.debug(
+                    f"Starting Claude tool interaction loop with {len(tools)} tools."
+                )
+                async for output in self._claude_tool_interaction_loop(
+                    messages, tools if tools else []
+                ):
+                    yield output
+                return
+            elif self._use_mcpp and tool_mode == "OpenAI":
+                logger.debug(
+                    f"Starting OpenAI tool interaction loop with {len(tools)} tools."
+                )
+                async for output in self._openai_tool_interaction_loop(
+                    messages, tools if tools else []
+                ):
+                    yield output
+                return
+            else:
+                logger.info("Starting simple chat completion.")
+                token_stream = self._llm.chat_completion(messages, self._system)
+                complete_response = ""
+                async for event in token_stream:
+                    text_chunk = ""
+                    if isinstance(event, dict) and event.get("type") == "text_delta":
+                        text_chunk = event.get("text", "")
+                    elif isinstance(event, str):
+                        text_chunk = event
+                    else:
+                        continue
+                    if text_chunk:
+                        yield text_chunk
+                        complete_response += text_chunk
+                if complete_response:
+                    self._add_message(complete_response, "assistant")
+
+        return chat_with_memory
+
+    async def chat(
+        self,
+        input_data: BatchInput,
+    ) -> AsyncIterator[Union[SentenceOutput, Dict[str, Any]]]:
+        """Run chat pipeline."""
+        chat_func_decorated = self._chat_function_factory()
+        async for output in chat_func_decorated(input_data):
+            yield output
+
+    def reset_interrupt(self) -> None:
+        """Reset interrupt flag."""
+        self._interrupt_handled = False
+
+    def start_group_conversation(
+        self, human_name: str, ai_participants: List[str]
+    ) -> None:
+        """Start a group conversation."""
+        if not self._tool_prompts:
+            logger.warning("Tool prompts dictionary is not set.")
+            return
+
+        other_ais = ", ".join(name for name in ai_participants)
+        prompt_name = self._tool_prompts.get("group_conversation_prompt", "")
+
+        if not prompt_name:
+            logger.warning("No group conversation prompt name found.")
+            return
+
+        try:
+            group_context = prompt_loader.load_util(prompt_name).format(
+                human_name=human_name, other_ais=other_ais
+            )
+            self._memory.append({"role": "user", "content": group_context})
+        except FileNotFoundError:
+            logger.error(f"Group conversation prompt file not found: {prompt_name}")
+        except KeyError as e:
+            logger.error(f"Missing formatting key in group conversation prompt: {e}")
+        except Exception as e:
+            logger.error(f"Failed to load group conversation prompt: {e}")

src/open_llm_vtuber/agent/agents/hume_ai.py

@@ -0,0 +1,256 @@
+import asyncio
+import base64
+from typing import AsyncIterator, Optional
+import json
+import websockets
+from loguru import logger
+from pathlib import Path
+
+from .agent_interface import AgentInterface
+from ..output_types import AudioOutput, Actions, DisplayText
+from ..input_types import BatchInput
+from ...chat_history_manager import get_metadata, update_metadate
+
+
+class HumeAIAgent(AgentInterface):
+    """
+    Hume AI Agent that handles text input and audio output.
+    Uses AudioOutput type to provide audio responses with transcripts.
+    """
+
+    AGENT_TYPE = "hume_ai_agent"
+
+    def __init__(
+        self,
+        api_key: str,
+        host: str = "api.hume.ai",
+        config_id: Optional[str] = None,
+        idle_timeout: int = 15,
+    ):
+        """
+        Initialize Hume AI agent
+
+        Args:
+            api_key: Hume AI API key
+            host: API host
+            config_id: Optional configuration ID
+            idle_timeout: Connection idle timeout in seconds
+        """
+        self.api_key = api_key
+        self.host = host
+        self.config_id = config_id
+        self.idle_timeout = idle_timeout
+        self._ws = None
+        self._current_text = None
+        self._current_id = None
+        self._connected = False
+        self._chat_group_id = None
+        self._idle_timer = None
+        self._current_conf_uid = None
+        self._current_history_uid = None
+
+        # Create cache directory if it doesn't exist
+        self.cache_dir = Path("./cache")
+        self.cache_dir.mkdir(exist_ok=True)
+
+    async def connect(self, resume_chat_group_id: Optional[str] = None):
+        """
+        Establish WebSocket connection with optional chat group resumption
+
+        Args:
+            resume_chat_group_id: Optional chat group ID to resume
+        """
+        if self._ws:
+            await self._ws.close()
+            self._ws = None
+            self._connected = False
+
+        # Build URL with query parameters
+        socket_url = f"wss://{self.host}/v0/evi/chat?api_key={self.api_key}"
+
+        if self.config_id:
+            socket_url += f"&config_id={self.config_id}"
+
+        if resume_chat_group_id:
+            logger.info(f"Resuming chat group: {resume_chat_group_id}")
+            socket_url += f"&resumed_chat_group_id={resume_chat_group_id}"
+            self._chat_group_id = resume_chat_group_id
+
+        logger.info(f"Connecting to EVI with config_id: {self.config_id}")
+
+        self._ws = await websockets.connect(socket_url)
+        self._connected = True
+
+        async for message in self._ws:
+            data = json.loads(message)
+            if data.get("type") == "chat_metadata":
+                new_chat_group_id = data.get("chat_group_id")
+
+                if not resume_chat_group_id and self._current_history_uid:
+                    update_metadate(
+                        self._current_conf_uid,
+                        self._current_history_uid,
+                        {"resume_id": new_chat_group_id, "agent_type": self.AGENT_TYPE},
+                    )
+
+                self._chat_group_id = new_chat_group_id
+                logger.info(
+                    f"{'Resumed' if resume_chat_group_id else 'Created new'} "
+                    f"chat group: {self._chat_group_id}"
+                )
+                break
+
+    def _reset_idle_timer(self):
+        """Reset the idle timer"""
+        if self._idle_timer:
+            self._idle_timer.cancel()
+
+        async def disconnect_after_timeout():
+            await asyncio.sleep(self.idle_timeout)
+            if self._ws and self._connected:
+                logger.info("Idle timeout reached, disconnecting...")
+                await self._ws.close()
+                self._connected = False
+
+        self._idle_timer = asyncio.create_task(disconnect_after_timeout())
+
+    async def _ensure_connection(self):
+        """Ensure connection is alive, reconnect if needed"""
+        if not self._connected or not self._ws or self._ws.closed:
+            await self.connect(self._chat_group_id)
+
+    def set_memory_from_history(self, conf_uid: str, history_uid: str) -> None:
+        """
+        Set chat group ID based on history
+
+        Args:
+            conf_uid: Configuration ID
+            history_uid: History ID
+        """
+        self._current_conf_uid = conf_uid
+        self._current_history_uid = history_uid
+
+        metadata = get_metadata(conf_uid, history_uid)
+
+        agent_type = metadata.get("agent_type")
+        if agent_type and agent_type != self.AGENT_TYPE:
+            logger.warning(
+                f"Incompatible agent type in history: {agent_type}. "
+                f"Expected: {self.AGENT_TYPE} or empty. Memory will not be set."
+            )
+            self._chat_group_id = None
+            return
+
+        resume_id = metadata.get("resume_id")
+        if resume_id:
+            self._chat_group_id = resume_id
+            logger.info(f"Using resume_id from metadata: {resume_id}")
+        else:
+            self._chat_group_id = None
+            logger.info("No resume_id found in metadata, will create new chat group")
+
+        # Force reconnection on next chat
+        if self._ws:
+            asyncio.create_task(self._ws.close())
+            self._connected = False
+
+    async def chat(self, batch_input: BatchInput) -> AsyncIterator[AudioOutput]:
+        """
+        Chat with Hume AI and get audio response
+
+        Args:
+            batch_input: BatchInput containing text and optional media
+
+        Returns:
+            AsyncIterator[AudioOutput]: Stream of AudioOutput objects
+        """
+        try:
+            self._reset_idle_timer()
+            await self._ensure_connection()
+
+            # Extract main text from BatchInput
+            input_text = batch_input.texts[0].content if batch_input.texts else ""
+
+            # Hume AI doesn't support image input, log warning if images present
+            if batch_input.images:
+                logger.warning(
+                    "Hume AI does not support image input. Images will be ignored."
+                )
+
+            message = {
+                "type": "user_input",
+                "text": input_text,
+            }
+            await self._ws.send(json.dumps(message))
+
+            async for message in self._ws:
+                self._reset_idle_timer()
+                logger.debug(f"Received message: {message}")
+                try:
+                    response_data = json.loads(message)
+                    msg_type = response_data.get("type")
+                    msg_id = response_data.get("id")
+
+                    if msg_type == "assistant_message":
+                        self._current_text = response_data["message"]["content"]
+                        self._current_id = msg_id
+
+                    elif msg_type == "audio_output":
+                        if msg_id == self._current_id and self._current_text:
+                            audio_data = base64.b64decode(response_data["data"])
+                            cache_file = self.cache_dir / f"evi_audio_{msg_id}.wav"
+
+                            with open(cache_file, "wb") as f:
+                                f.write(audio_data)
+                                logger.debug(f"Saved audio to cache file: {cache_file}")
+
+                            # Create AudioOutput with DisplayText
+                            yield AudioOutput(
+                                audio_path=str(cache_file),
+                                display_text=DisplayText(text=self._current_text),
+                                transcript=self._current_text,
+                                actions=Actions(),
+                            )
+
+                            self._current_text = None
+                            self._current_id = None
+
+                    elif msg_type == "assistant_end":
+                        break
+
+                    elif msg_type == "tool_error_message":
+                        logger.error(f"Tool error: {response_data.get('error')}")
+
+                except json.JSONDecodeError as e:
+                    logger.error(f"Failed to parse response JSON: {e}")
+                    continue
+
+        except websockets.exceptions.ConnectionClosed as e:
+            logger.warning(f"Connection closed: {e}, attempting to reconnect...")
+            self._connected = False
+            await self._ensure_connection()
+            async for result in self.chat(batch_input):
+                yield result
+
+        except Exception as e:
+            logger.error(f"Error in chat: {e}")
+            raise
+
+    def handle_interrupt(self, heard_response: str) -> None:
+        """Handle user interruption (not implemented for Hume AI)"""
+        pass
+
+    def __del__(self):
+        """Cleanup WebSocket connection and cache files"""
+        if self._idle_timer:
+            self._idle_timer.cancel()
+
+        if self._ws:
+            self._ws.close()
+
+        # Clean up cache files
+        try:
+            for file in self.cache_dir.glob("evi_audio_*.wav"):
+                file.unlink()
+        except Exception as e:
+            logger.error(f"Error cleaning up cache files: {e}")

src/open_llm_vtuber/agent/agents/letta_agent.py

@@ -0,0 +1,128 @@
+from typing import AsyncIterator, List, Dict, Any
+from .agent_interface import AgentInterface
+from ..output_types import SentenceOutput
+from ..transformers import (
+    sentence_divider,
+    actions_extractor,
+    tts_filter,
+    display_processor,
+)
+from ...config_manager import TTSPreprocessorConfig
+from ..input_types import BatchInput, TextSource
+from letta_client import Letta
+
+
+class LettaAgent(AgentInterface):
+    """
+    Custom Letta class to interface with the Letta server.
+    """
+
+    def __init__(
+        self,
+        live2d_model,
+        id,
+        tts_preprocessor_config: TTSPreprocessorConfig = None,
+        faster_first_response: bool = True,
+        segment_method: str = "pysbd",
+        host: str = "localhost",
+        port: int = 8283,
+    ):
+        super().__init__()
+        self.url = f"http://{host}:{port}"
+        self.client = Letta(base_url=self.url)
+        self.id = id
+        # Initialize decorator parameters
+        self._tts_preprocessor_config = tts_preprocessor_config
+        self._live2d_model = live2d_model
+        self._faster_first_response = faster_first_response
+        self._segment_method = segment_method
+
+        # Delay decorator application
+        self.chat = tts_filter(self._tts_preprocessor_config)(
+            display_processor()(
+                actions_extractor(self._live2d_model)(
+                    sentence_divider(
+                        faster_first_response=self._faster_first_response,
+                        segment_method=self._segment_method,
+                        valid_tags=["think"],
+                    )(self.chat)
+                )
+            )
+        )
+
+    def set_memory_from_history(self, conf_uid: str, history_uid: str) -> None:
+        # The Letta Server automatically stores historical messages, so this part is not needed
+        pass
+
+    def handle_interrupt(self, heard_response: str) -> None:
+        pass
+
+    async def generator_to_async(self, gen):
+        for item in gen:
+            yield item
+
+    async def chat(self, input_data: BatchInput) -> AsyncIterator[SentenceOutput]:
+        messages = self._to_messages(input_data)
+        stream = self.generator_to_async(
+            self.client.agents.messages.create_stream(
+                agent_id=self.id,
+                messages=messages,
+                stream_tokens=True,
+            )
+        )
+
+        complete_response = ""
+        async for token in stream:
+            if token.message_type == "reasoning_message":
+                # This part is reasoning information and should not be displayed
+                token = token.reasoning
+                continue
+            elif token.message_type == "assistant_message":
+                # This part is the result that needs to be displayed, it is the final result
+                # logger.info('Test message')
+                # logger.info(token)
+                token = token.content
+            else:
+                continue
+
+            yield token
+            complete_response += token
+
+    def _to_text_prompt(self, input_data: BatchInput) -> str:
+        """
+        Format BatchInput into a prompt string for the LLM.
+
+        Args:
+            input_data: BatchInput - The input data containing texts
+
+        Returns:
+            str - Formatted message string
+        """
+        message_parts = []
+
+        # Process text inputs in order
+        for text_data in input_data.texts:
+            if text_data.source == TextSource.INPUT:
+                message_parts.append(text_data.content)
+            elif text_data.source == TextSource.CLIPBOARD:
+                message_parts.append(f"[Clipboard content: {text_data.content}]")
+
+        return "\n".join(message_parts)
+
+    def _to_messages(self, input_data: BatchInput) -> List[Dict[str, Any]]:
+        """
+        Prepare messages list without image support.
+        """
+        messages = []
+
+        if input_data.images:
+            content = []
+            text_content = self._to_text_prompt(input_data)
+            content.append({"type": "text", "text": text_content})
+            user_message = {"role": "user", "content": content}
+        else:
+            user_message = {"role": "user", "content": self._to_text_prompt(input_data)}
+
+        messages.append(user_message)
+
+        return messages

src/open_llm_vtuber/agent/input_types.py

@@ -0,0 +1,94 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import List, Optional, Dict, Any
+
+
+class ImageSource(Enum):
+    """Enum for different image sources"""
+
+    CAMERA = "camera"
+    SCREEN = "screen"
+    CLIPBOARD = "clipboard"
+    UPLOAD = "upload"
+
+
+class TextSource(Enum):
+    """Enum for different text sources"""
+
+    INPUT = "input"  # Main user input/transcription
+    CLIPBOARD = "clipboard"  # Text from clipboard
+
+
+@dataclass
+class ImageData:
+    """
+    Represents an image from various sources
+
+    Attributes:
+        source: Source of the image
+        data: Base64 encoded image data or URL
+        mime_type: MIME type of the image (e.g., 'image/jpeg', 'image/png')
+    """
+
+    source: ImageSource
+    data: str  # Base64 encoded or URL
+    mime_type: str
+
+
+@dataclass
+class FileData:
+    """
+    Represents a file uploaded by the user
+
+    Attributes:
+        name: Original filename
+        data: Base64 encoded file data
+        mime_type: MIME type of the file
+    """
+
+    name: str
+    data: str  # Base64 encoded
+    mime_type: str
+
+
+@dataclass
+class TextData:
+    """
+    Represents text data from various sources
+
+    Attributes:
+        source: Source of the text
+        content: str - The text content
+        from_name: Optional[str] - Name of the sender/character
+    """
+
+    source: TextSource
+    content: str
+    from_name: Optional[str] = None
+
+
+class BaseInput:
+    """Base class for all input types"""
+
+    pass
+
+
+@dataclass
+class BatchInput(BaseInput):
+    """
+    Input type for batch processing, containing complete transcription and optional media
+
+    Attributes:
+        texts: List of text data from different sources
+        images: Optional list of images
+        files: Optional list of files
+        metadata: Optional dictionary of metadata flags for special inputs
+            - 'proactive_speak': Boolean flag indicating if this is a proactive speak input
+            - 'skip_memory': Boolean flag indicating if this input should be skipped in AI's internal memory
+            - 'skip_history': Boolean flag indicating if this input should be skipped in local history storage
+    """
+
+    texts: List[TextData]
+    images: Optional[List[ImageData]] = None
+    files: Optional[List[FileData]] = None
+    metadata: Optional[Dict[str, Any]] = None

src/open_llm_vtuber/agent/output_types.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass, asdict
+from typing import List, Optional
+from abc import ABC, abstractmethod
+
+
+@dataclass
+class Actions:
+    """Represents actions that can be performed alongside text output"""
+
+    expressions: Optional[List[str] | List[int]] = None
+    pictures: Optional[List[str]] = None
+    sounds: Optional[List[str]] = None
+
+    def to_dict(self) -> dict:
+        """Convert Actions object to a dictionary for JSON serialization"""
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+class BaseOutput(ABC):
+    """Base class for agent outputs that can be iterated"""
+
+    @abstractmethod
+    def __aiter__(self):
+        """Make the output iterable"""
+        pass
+
+
+@dataclass
+class DisplayText:
+    """Text to be displayed with optional metadata"""
+
+    text: str
+    name: Optional[str] = "AI"  # Keep the name field for frontend display
+    avatar: Optional[str] = None
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary for JSON serialization"""
+        return {"text": self.text, "name": self.name, "avatar": self.avatar}
+
+    def __str__(self) -> str:
+        """String representation for logging"""
+        return f"{self.name}: {self.text}"
+
+
+@dataclass
+class SentenceOutput(BaseOutput):
+    """
+    Output type for text-based responses.
+    Contains a single sentence pair (display and TTS) with associated actions.
+
+    Attributes:
+        display_text: Text to be displayed in UI
+        tts_text: Text to be sent to TTS engine
+        actions: Associated actions (expressions, pictures, sounds)
+    """
+
+    display_text: DisplayText  # Changed from str to DisplayText
+    tts_text: str  # Text for TTS
+    actions: Actions
+
+    async def __aiter__(self):
+        """Yield the sentence pair and actions"""
+        yield self.display_text, self.tts_text, self.actions
+
+
+@dataclass
+class AudioOutput(BaseOutput):
+    """Output type for audio-based responses"""
+
+    audio_path: str
+    display_text: DisplayText  # Changed from str to DisplayText
+    transcript: str  # Original transcript
+    actions: Actions
+
+    async def __aiter__(self):
+        """Iterate through audio segments and their actions"""
+        yield self.audio_path, self.display_text, self.transcript, self.actions

src/open_llm_vtuber/agent/stateless_llm/claude_llm.py

@@ -0,0 +1,246 @@
+"""Description: This file contains the implementation of the `AsyncLLM` class for Claude API.
+This class is responsible for handling asynchronous interaction with Claude API endpoints
+for language generation.
+"""
+
+import json
+from typing import AsyncIterator, List, Dict, Any
+
+from loguru import logger
+from anthropic import AsyncAnthropic, NOT_GIVEN
+
+from .stateless_llm_interface import StatelessLLMInterface
+
+
+class AsyncLLM(StatelessLLMInterface):
+    def __init__(
+        self,
+        model: str = "claude-3-haiku-latest",
+        base_url: str = None,
+        llm_api_key: str = None,
+        system: str = None,
+    ):
+        """
+        Initialize Claude LLM.
+
+        Args:
+            model (str): Model name
+            base_url (str): Base URL for Claude API
+            llm_api_key (str): Claude API key
+            system (str): System prompt
+        """
+        self.model = model
+        self.system = system
+
+        # Initialize Claude client
+        self.client = AsyncAnthropic(
+            api_key=llm_api_key, base_url=base_url if base_url else None
+        )
+
+        logger.info(f"Initialized Claude AsyncLLM with model: {self.model}")
+        logger.debug(f"Base URL: {base_url}")
+
+    def _convert_message_format(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        """Convert message format to Claude's expected format."""
+        # Handle potential tool_result content blocks
+        if isinstance(message.get("content"), list):
+            new_content = []
+            is_tool_result = False
+            for content_item in message["content"]:
+                if content_item.get("type") == "image_url":
+                    # Extract media type and base64 data from data URL
+                    data_url = content_item["image_url"]["url"]
+                    # Split 'data:image/jpeg;base64,/9j/4AAQ...' into parts
+                    header, base64_data = data_url.split(",", 1)
+                    # Extract media type from 'data:image/jpeg;base64'
+                    media_type = header.split(":")[1].split(";")[0]
+
+                    new_content.append(
+                        {
+                            "type": "image",
+                            "source": {
+                                "type": "base64",
+                                "media_type": media_type,
+                                "data": base64_data,
+                            },
+                        }
+                    )
+                elif content_item.get("type") == "tool_result":
+                    is_tool_result = True
+                    # Keep tool_result block as is, Anthropic SDK handles it
+                    new_content.append(content_item)
+                else:
+                    # Assume text or other standard types
+                    new_content.append(content_item)
+
+            # For tool_result messages, the role should be 'user'
+            # Ensure the role is correctly set before returning
+            role = "user" if is_tool_result else message["role"]
+            return {"role": role, "content": new_content}
+
+        # Handle plain text content or non-list content
+        return message
+
+    async def chat_completion(
+        self,
+        messages: List[Dict[str, Any]],
+        system: str = None,
+        tools: List[Dict[str, Any]] = None,
+    ) -> AsyncIterator[Dict[str, Any]]:
+        """
+        Generates a chat completion using the Claude API asynchronously,
+        handling text generation and tool use.
+
+        Parameters:
+        - messages (List[Dict[str, Any]]): The list of messages to send to the API.
+        - system (str, optional): System prompt to use for this completion.
+        - tools (List[Dict[str, Any]], optional): List of tools available.
+
+        Yields:
+        - Dict[str, Any]: Events representing text deltas, tool use, or errors.
+          Possible event types:
+            - {"type": "message_start", "data": ...}
+            - {"type": "text_delta", "text": "..."}
+            - {"type": "tool_use_start", "data": {"id": ..., "name": ..., "input": None}}
+            - {"type": "tool_input_delta", "tool_id": ..., "partial_json": "..."} # Optional
+            - {"type": "tool_use_complete", "data": {"id": ..., "name": ..., "input": {...}}}
+            - {"type": "message_delta", "data": ...} # e.g., stop_reason
+            - {"type": "message_stop"}
+            - {"type": "error", "message": "..."}
+        """
+        try:
+            # Filter out system messages and convert message format
+            converted_messages = [
+                self._convert_message_format(msg)
+                for msg in messages
+                if msg["role"] != "system"
+            ]
+
+            logger.debug(f"Sending messages to Claude API: {converted_messages}")
+            logger.debug(f"Tools provided: {tools}")
+
+            async with self.client.messages.stream(
+                messages=converted_messages,
+                system=system if system else (self.system if self.system else ""),
+                model=self.model,
+                max_tokens=1024,
+                tools=tools if tools else NOT_GIVEN,
+            ) as stream:
+                current_tool_call_info = None
+                partial_json_accumulator = ""
+
+                async for event in stream:
+                    if event.type == "message_start":
+                        logger.debug("Stream: message_start")
+                        yield {
+                            "type": "message_start",
+                            "data": event.message.model_dump(exclude_none=True),
+                        }
+                    elif event.type == "content_block_start":
+                        logger.debug(
+                            f"Stream: content_block_start - Index: {event.index}, Type: {event.content_block.type}"
+                        )
+                        if event.content_block.type == "text":
+                            pass  # Handled by text_delta
+                        elif event.content_block.type == "tool_use":
+                            current_tool_call_info = {
+                                "id": event.content_block.id,
+                                "name": event.content_block.name,
+                                "input": None,
+                                "index": event.index,  # Store index
+                            }
+                            partial_json_accumulator = ""
+                            logger.debug(
+                                f"Stream: tool_use started - ID: {current_tool_call_info['id']}, Name: {current_tool_call_info['name']}"
+                            )
+                            yield {
+                                "type": "tool_use_start",
+                                "data": current_tool_call_info.copy(),
+                            }
+                    elif event.type == "content_block_delta":
+                        logger.debug(
+                            f"Stream: content_block_delta - Index: {event.index}, Delta Type: {event.delta.type}"
+                        )
+                        if event.delta.type == "text_delta":
+                            yield {"type": "text_delta", "text": event.delta.text}
+                        elif event.delta.type == "input_json_delta":
+                            if (
+                                current_tool_call_info
+                                and event.index == current_tool_call_info["index"]
+                            ):
+                                partial_json_accumulator += event.delta.partial_json
+                                logger.trace(
+                                    f"Stream: input_json_delta - Tool ID: {current_tool_call_info['id']}, Partial: {event.delta.partial_json}"
+                                )
+                            else:
+                                logger.warning(
+                                    f"Received input_json_delta but no active tool call matching index {event.index}"
+                                )
+                    elif event.type == "content_block_stop":
+                        logger.debug(
+                            f"Stream: content_block_stop - Index: {event.index}"
+                        )
+                        # Check if this stop corresponds to the active tool call
+                        if (
+                            current_tool_call_info
+                            and event.index == current_tool_call_info["index"]
+                        ):
+                            try:
+                                if not partial_json_accumulator.strip():
+                                    logger.warning(
+                                        f"Empty JSON input received for tool ID: {current_tool_call_info['id']}. Using empty object."
+                                    )
+                                    tool_input = {}
+                                else:
+                                    tool_input = json.loads(partial_json_accumulator)
+                                current_tool_call_info["input"] = tool_input
+                                logger.debug(
+                                    f"Stream: tool_use completed - ID: {current_tool_call_info['id']}, Input: {tool_input}"
+                                )
+                                # Yield the complete tool call info
+                                yield {
+                                    "type": "tool_use_complete",
+                                    "data": current_tool_call_info.copy(),
+                                }
+                            except json.JSONDecodeError as e:
+                                logger.error(
+                                    f"Failed to decode tool input JSON: {partial_json_accumulator}. Error: {e}"
+                                )
+                                yield {
+                                    "type": "error",
+                                    "message": f"Failed to parse tool input JSON for tool ID {current_tool_call_info['id']}",
+                                }
+                            finally:
+                                # Reset regardless of success or failure for this index
+                                current_tool_call_info = None
+                                partial_json_accumulator = ""
+                    elif event.type == "message_delta":
+                        logger.debug(
+                            f"Stream: message_delta - Delta: {event.delta.model_dump(exclude_none=True)}, Usage: {event.usage}"
+                        )
+                        yield {
+                            "type": "message_delta",
+                            "data": {
+                                "delta": event.delta.model_dump(exclude_none=True),
+                                "usage": event.usage.model_dump(),
+                            },
+                        }
+                    elif event.type == "message_stop":
+                        logger.debug("Stream: message_stop")
+                        yield {"type": "message_stop"}
+                        # No need to break here, the context manager handles the end
+                    elif event.type == "ping":
+                        logger.trace("Stream: ping")
+                        pass  # Ignore pings
+                    # Anthropic SDK might raise errors directly, or via event.type == 'error'
+                    # The outer try/except handles SDK-level errors.
+
+        except Exception as e:
+            logger.error(f"Claude API error occurred: {str(e)}")
+            logger.info(f"Model: {self.model}")
+            # Yield an error event before raising
+            yield {"type": "error", "message": f"Claude API error: {str(e)}"}
+            raise
+
+        # No finally block needed for stream.close() due to async with
+        logger.debug("Chat completion stream processing finished.")

src/open_llm_vtuber/agent/stateless_llm/llama_cpp_llm.py

@@ -0,0 +1,76 @@
+"""Description: This file contains the implementation of the LLM class using llama.cpp.
+This class provides a stateless interface to llama.cpp for language generation.
+"""
+
+import asyncio
+from typing import AsyncIterator, List, Dict, Any
+from llama_cpp import Llama
+from loguru import logger
+
+from .stateless_llm_interface import StatelessLLMInterface
+
+
+class LLM(StatelessLLMInterface):
+    def __init__(
+        self,
+        model_path: str,
+        **kwargs,
+    ):
+        """
+        Initializes a stateless instance of the LLM class using llama.cpp.
+
+        Parameters:
+        - model_path (str): Path to the GGUF model file
+        - **kwargs: Additional arguments passed to Llama constructor
+        """
+        logger.info(f"Initializing llama cpp with model path: {model_path}")
+        self.model_path = model_path
+        try:
+            self.llm = Llama(model_path=model_path, **kwargs)
+        except Exception as e:
+            logger.critical(f"Failed to initialize Llama model: {e}")
+            raise
+
+    async def chat_completion(
+        self, messages: List[Dict[str, Any]], system: str = None
+    ) -> AsyncIterator[str]:
+        """
+        Generates a chat completion using llama.cpp asynchronously.
+
+        Parameters:
+        - messages (List[Dict[str, Any]]): The list of messages to send to the model.
+        - system (str, optional): System prompt to use for this completion.
+
+        Yields:
+        - str: The content of each chunk from the model response.
+        """
+        logger.debug(f"Generating completion for messages: {messages}")
+
+        try:
+            # Add system prompt if provided
+            messages_with_system = messages
+            if system:
+                messages_with_system = [
+                    {"role": "system", "content": system},
+                    *messages,
+                ]
+
+            # Create chat completion in a separate thread to avoid blocking
+            chat_completion = await asyncio.get_event_loop().run_in_executor(
+                None,
+                lambda: self.llm.create_chat_completion(
+                    messages=messages_with_system,
+                    stream=True,
+                ),
+            )
+
+            # Process chunks
+            for chunk in chat_completion:
+                if chunk.get("choices") and chunk["choices"][0].get("delta"):
+                    content = chunk["choices"][0]["delta"].get("content", "")
+                    if content:
+                        yield content
+
+        except Exception as e:
+            logger.error(f"Error in chat completion: {e}")
+            raise

src/open_llm_vtuber/agent/stateless_llm/ollama_llm.py

@@ -0,0 +1,73 @@
+import atexit
+import requests
+from loguru import logger
+from .openai_compatible_llm import AsyncLLM
+
+
+class OllamaLLM(AsyncLLM):
+    def __init__(
+        self,
+        model: str,
+        base_url: str,
+        llm_api_key: str = "z",
+        organization_id: str = "z",
+        project_id: str = "z",
+        temperature: float = 1.0,
+        keep_alive: float = -1,
+        unload_at_exit: bool = True,
+    ):
+        self.keep_alive = keep_alive
+        self.unload_at_exit = unload_at_exit
+        self.cleaned = False
+        super().__init__(
+            model=model,
+            base_url=base_url,
+            llm_api_key=llm_api_key,
+            organization_id=organization_id,
+            project_id=project_id,
+            temperature=temperature,
+        )
+        try:
+            # preload model
+            logger.info("Preloading model for Ollama")
+            # Send the POST request to preload model
+            logger.debug(
+                requests.post(
+                    base_url.replace("/v1", "") + "/api/chat",
+                    json={
+                        "model": model,
+                        "keep_alive": keep_alive,
+                    },
+                )
+            )
+        except requests.exceptions.ConnectionError as e:
+            logger.error(f"Failed to preload model: {e}")
+            logger.critical(
+                "Fail to connect to Ollama backend. Is Ollama server running? Try running `ollama list` to start the server and try again.\nThe AI will repeat 'Error connecting chat endpoint' until the server is running."
+            )
+        except Exception as e:
+            logger.error(f"Failed to preload model: {e}")
+        # If keep_alive is less than 0, register cleanup to unload the model
+        if unload_at_exit:
+            atexit.register(self.cleanup)
+
+    def __del__(self):
+        """Destructor to unload the model"""
+        self.cleanup()
+
+    def cleanup(self):
+        """Clean up function to unload the model when exitting"""
+        if not self.cleaned and self.unload_at_exit:
+            logger.info(f"Ollama: Unloading model: {self.model}")
+            # Unload the model
+            # unloading is just the same as preload, but with keep alive set to 0
+            logger.debug(
+                requests.post(
+                    self.base_url.replace("/v1", "") + "/api/chat",
+                    json={
+                        "model": self.model,
+                        "keep_alive": 0,
+                    },
+                )
+            )
+            self.cleaned = True

src/open_llm_vtuber/agent/stateless_llm/openai_compatible_llm.py

@@ -0,0 +1,242 @@
+"""Description: This file contains the implementation of the `AsyncLLM` class.
+This class is responsible for handling asynchronous interaction with OpenAI API compatible
+endpoints for language generation.
+"""
+
+from typing import AsyncIterator, List, Dict, Any
+from openai import (
+    AsyncStream,
+    AsyncOpenAI,
+    APIError,
+    APIConnectionError,
+    RateLimitError,
+    NotGiven,
+    NOT_GIVEN,
+)
+from openai.types.chat import ChatCompletionChunk
+from openai.types.chat.chat_completion_chunk import ChoiceDeltaToolCall
+from loguru import logger
+
+from .stateless_llm_interface import StatelessLLMInterface
+from ...mcpp.types import ToolCallObject
+
+
+class AsyncLLM(StatelessLLMInterface):
+    def __init__(
+        self,
+        model: str,
+        base_url: str,
+        llm_api_key: str = "z",
+        organization_id: str = "z",
+        project_id: str = "z",
+        temperature: float = 1.0,
+    ):
+        """
+        Initializes an instance of the `AsyncLLM` class.
+
+        Parameters:
+        - model (str): The model to be used for language generation.
+        - base_url (str): The base URL for the OpenAI API.
+        - organization_id (str, optional): The organization ID for the OpenAI API. Defaults to "z".
+        - project_id (str, optional): The project ID for the OpenAI API. Defaults to "z".
+        - llm_api_key (str, optional): The API key for the OpenAI API. Defaults to "z".
+        - temperature (float, optional): What sampling temperature to use, between 0 and 2. Defaults to 1.0.
+        """
+        self.base_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.client = AsyncOpenAI(
+            base_url=base_url,
+            organization=organization_id,
+            project=project_id,
+            api_key=llm_api_key,
+        )
+        self.support_tools = True
+
+        logger.info(
+            f"Initialized AsyncLLM with the parameters: {self.base_url}, {self.model}"
+        )
+
+    async def chat_iter(self, messages, system_prompt, tools=None):
+        """Đảm bảo hàm này tồn tại để Agent có thể gọi được"""
+        async for chunk in self.chat_completion(messages, system=system_prompt, tools=tools):
+            yield chunk
+
+    async def chat_completion(
+        self,
+        messages: List[Dict[str, Any]],
+        system: str = None,
+        tools: List[Dict[str, Any]] | NotGiven = NOT_GIVEN,
+    ) -> AsyncIterator[str | List[ChoiceDeltaToolCall]]:
+        """
+        Generates a chat completion using the OpenAI API asynchronously.
+
+        Parameters:
+        - messages (List[Dict[str, Any]]): The list of messages to send to the API.
+        - system (str, optional): System prompt to use for this completion.
+        - tools (List[Dict[str, str]], optional): List of tools to use for this completion.
+
+        Yields:
+        - str: The content of each chunk from the API response.
+        - List[ChoiceDeltaToolCall]: The tool calls detected in the response.
+
+        Raises:
+        - APIConnectionError: When the server cannot be reached
+        - RateLimitError: When a 429 status code is received
+        - APIError: For other API-related errors
+        """
+        stream = None
+        # Tool call related state variables
+        accumulated_tool_calls = {}
+        in_tool_call = False
+
+        try:
+            # If system prompt is provided, add it to the messages
+            messages_with_system = messages
+            if system:
+                messages_with_system = [
+                    {"role": "system", "content": system},
+                    *messages,
+                ]
+            logger.debug(f"Messages: {messages_with_system}")
+
+            available_tools = tools if self.support_tools else NOT_GIVEN
+
+            stream: AsyncStream[
+                ChatCompletionChunk
+            ] = await self.client.chat.completions.create(
+                messages=messages_with_system,
+                model=self.model,
+                stream=True,
+                temperature=self.temperature,
+                tools=available_tools,
+            )
+            logger.debug(
+                f"Tool Support: {self.support_tools}, Available tools: {available_tools}"
+            )
+
+            async for chunk in stream:
+                # Guard against chunks with missing choices field (e.g., from OpenWebUI)
+                if not chunk.choices:
+                    continue
+
+                if self.support_tools:
+                    has_tool_calls = (
+                        hasattr(chunk.choices[0].delta, "tool_calls")
+                        and chunk.choices[0].delta.tool_calls
+                    )
+
+                    if has_tool_calls:
+                        logger.debug(
+                            f"Tool calls detected in chunk: {chunk.choices[0].delta.tool_calls}"
+                        )
+                        in_tool_call = True
+                        # Process tool calls in the current chunk
+                        for tool_call in chunk.choices[0].delta.tool_calls:
+                            index = (
+                                tool_call.index if hasattr(tool_call, "index") else 0
+                            )
+
+                            # Initialize tool call for this index if needed
+                            if index not in accumulated_tool_calls:
+                                accumulated_tool_calls[index] = {
+                                    "index": index,
+                                    "id": getattr(tool_call, "id", None),
+                                    "type": getattr(tool_call, "type", None),
+                                    "function": {"name": "", "arguments": ""},
+                                }
+
+                            # Update tool call information
+                            if hasattr(tool_call, "id") and tool_call.id:
+                                accumulated_tool_calls[index]["id"] = tool_call.id
+                            if hasattr(tool_call, "type") and tool_call.type:
+                                accumulated_tool_calls[index]["type"] = tool_call.type
+
+                            # Update function information
+                            if hasattr(tool_call, "function"):
+                                if (
+                                    hasattr(tool_call.function, "name")
+                                    and tool_call.function.name
+                                ):
+                                    accumulated_tool_calls[index]["function"][
+                                        "name"
+                                    ] = tool_call.function.name
+                                if (
+                                    hasattr(tool_call.function, "arguments")
+                                    and tool_call.function.arguments
+                                ):
+                                    accumulated_tool_calls[index]["function"][
+                                        "arguments"
+                                    ] += tool_call.function.arguments
+
+                        continue
+
+                    # If we were in a tool call but now we're not, yield the tool call result
+                    elif in_tool_call and not has_tool_calls:
+                        in_tool_call = False
+                        # Convert accumulated tool calls to the required format and output
+                        logger.info(f"Complete tool calls: {accumulated_tool_calls}")
+
+                        # Use the from_dict method to create a ToolCallObject instance from a dictionary
+                        complete_tool_calls = [
+                            ToolCallObject.from_dict(tool_data)
+                            for tool_data in accumulated_tool_calls.values()
+                        ]
+
+                        yield complete_tool_calls
+                        accumulated_tool_calls = {}  # Reset for potential future tool calls
+
+                # Process regular content chunks
+                if len(chunk.choices) == 0:
+                    logger.info("Empty chunk received")
+                    continue
+                elif chunk.choices[0].delta.content is None:
+                    chunk.choices[0].delta.content = ""
+                yield chunk.choices[0].delta.content
+
+            # If stream ends while still in a tool call, make sure to yield the tool call
+            if in_tool_call and accumulated_tool_calls:
+                logger.info(f"Final tool call at stream end: {accumulated_tool_calls}")
+
+                # Create a ToolCallObject instance from a dictionary using the from_dict method.
+                complete_tool_calls = [
+                    ToolCallObject.from_dict(tool_data)
+                    for tool_data in accumulated_tool_calls.values()
+                ]
+
+                yield complete_tool_calls
+
+        except APIConnectionError as e:
+            logger.error(
+                f"Error calling the chat endpoint: Connection error. Failed to connect to the LLM API. \nCheck the configurations and the reachability of the LLM backend. \nSee the logs for details. \nTroubleshooting with documentation: https://open-llm-vtuber.github.io/docs/faq#%E9%81%87%E5%88%B0-error-calling-the-chat-endpoint-%E9%94%99%E8%AF%AF%E6%80%8E%E4%B9%88%E5%8A%9E \n{e.__cause__}"
+            )
+            yield "Error calling the chat endpoint: Connection error. Failed to connect to the LLM API. Check the configurations and the reachability of the LLM backend. See the logs for details. Troubleshooting with documentation: [https://open-llm-vtuber.github.io/docs/faq#%E9%81%87%E5%88%B0-error-calling-the-chat-endpoint-%E9%94%99%E8%AF%AF%E6%80%8E%E4%B9%88%E5%8A%9E]"
+
+        except RateLimitError as e:
+            logger.error(
+                f"Error calling the chat endpoint: Rate limit exceeded: {e.response}"
+            )
+            yield "Error calling the chat endpoint: Rate limit exceeded. Please try again later. See the logs for details."
+
+        except APIError as e:
+            if "does not support tools" in str(e):
+                self.support_tools = False
+                logger.warning(
+                    f"{self.model} does not support tools. Disabling tool support."
+                )
+                yield "__API_NOT_SUPPORT_TOOLS__"
+                return
+            logger.error(f"LLM API: Error occurred: {e}")
+            logger.info(f"Base URL: {self.base_url}")
+            logger.info(f"Model: {self.model}")
+            logger.info(f"Messages: {messages}")
+            logger.info(f"temperature: {self.temperature}")
+            yield "Error calling the chat endpoint: Error occurred while generating response. See the logs for details."
+
+        finally:
+            # make sure the stream is properly closed
+            # so when interrupted, no more tokens will being generated.
+            if stream:
+                logger.debug("Chat completion finished.")
+                await stream.close()
+                logger.debug("Stream closed.")

src/open_llm_vtuber/agent/stateless_llm/stateless_llm_interface.py

@@ -0,0 +1,64 @@
+import abc
+from typing import AsyncIterator, List, Dict, Any
+
+
+class StatelessLLMInterface(metaclass=abc.ABCMeta):
+    """
+    Interface for a stateless language model.
+
+    The word "stateless" means that the language model does not store memory,
+    system prompts, or user messages, which is most of the LLM. If we send a
+    message to the LLM, its response will be based on the message parameter alone.
+
+    The StatelessLLMInterface class provides a method for generating chat
+    completions asynchronously.
+
+    We use StatelessLLMs to initialize Agents, which pack the StatelessLLM with
+    memory, system prompts, and other features.
+
+    """
+
+    @abc.abstractmethod
+    async def chat_completion(
+        self,
+        messages: List[Dict[str, Any]],
+        system: str = None,
+        tools: List[Dict[str, Any]] = None,
+    ) -> AsyncIterator[str]:
+        """
+        Generates a chat completion asynchronously and return an iterator to the response.
+        This function does not store memory or user messages.
+
+        Parameters:
+        - messages (List[Dict[str, Any]]): The list of messages to send to the API.
+        - system (str, optional): System prompt to use for this completion.
+        - tools (List[Dict[str, str]], optional): List of tools to use for this completion.
+            - Each tool should follow the format:
+            {
+                "name": "tool_name",
+                "description": "tool_description",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "param1": {
+                            "type": "string",
+                            "description": "Description of param1"
+                        },
+                        "param2": {
+                            "type": "integer",
+                            "description": "Description of param2"
+                        }
+                    },
+                    "required": ["param1"]
+                }
+            }
+
+        Yields:
+        - str: The content of each chunk from the API response.
+
+        Raises:
+        - APIConnectionError: When the server cannot be reached
+        - RateLimitError: When a 429 status code is received
+        - APIError: For other API-related errors
+        """
+        raise NotImplementedError

src/open_llm_vtuber/agent/stateless_llm/stateless_llm_with_template.py

@@ -0,0 +1,195 @@
+"""Description: This file contains the implementation of the `AsyncLLMTemplate` class.
+This class is responsible for handling asynchronous interaction with OpenAI API
+compatible endpoints for language generation where the language model is not
+trained using a ChatML format.
+"""
+
+import requests
+import json
+from jinja2 import Template
+from loguru import logger
+from typing import AsyncIterator, List, Dict, Any
+
+from .stateless_llm_interface import StatelessLLMInterface
+
+
+TEMPLATES = {
+    "LLAMA3": {
+        "template": "".join(
+            [
+                "{{ bos_token }}",
+                "{% for message in messages %}",
+                "    {{ '<|start_header_id|>' + message['role'] + '<|end_header_id|>\n\n'+ message['content'] | trim + '<|eot_id|>' }}",
+                "{% endfor %}",
+                "{% if add_generation_prompt %}",
+                "    {{ '<|start_header_id|>assistant<|end_header_id|>\n\n' }}",
+                "{% endif %}",
+            ]
+        ),
+        "eot_token": "<|eot_id|>",
+    },
+    "CHATML": {
+        "template": "".join(
+            [
+                "{{ bos_token }}",
+                "{% for message in messages %}",
+                "    {{ '<|im_start|>' + message['role'] + '\n' + message['content'] | trim + '<|im_end|>\n' }}",
+                "{% endfor %}",
+                "{% if add_generation_prompt %}",
+                "    {{ '<|im_start|>assistant\n' }}",
+                "{% endif %}",
+            ]
+        ),
+        "eot_token": "<|im_end|>",
+    },
+    "ALPACA": {
+        "template": "".join(
+            [
+                """
+{{ (messages|selectattr('role', 'equalto', 'system')|list|last).content|trim if (messages|selectattr('role', 'equalto', 'system')|list) else '' }}
+
+{% for message in messages %}
+{% if message['role'] == 'user' %}
+### Instruction:
+{{ message['content']|trim -}}
+{% if not loop.last %}
+
+
+{% endif %}
+{% elif message['role'] == 'assistant' %}
+### Response:
+{{ message['content']|trim -}}
+{% if not loop.last %}
+
+
+{% endif %}
+{% elif message['role'] == 'user_context' %}
+### Input:
+{{ message['content']|trim -}}
+{% if not loop.last %}
+
+
+{% endif %}
+{% endif %}
+{% endfor %}
+{% if add_generation_prompt and messages[-1]['role'] != 'assistant' %}
+### Response:
+{% endif %}
+        """
+            ]
+        ),
+        "eot_token": "###",
+    },
+}
+
+
+class AsyncLLMWithTemplate(StatelessLLMInterface):
+    def __init__(
+        self,
+        model: str,
+        base_url: str,
+        llm_api_key: str = "z",
+        organization_id: str = "z",
+        project_id: str = "z",
+        template: str = "CHATML",
+        temperature: float = 1.0,
+    ):
+        """
+        Initializes an instance of the `AsyncLLM` class.
+
+        Parameters:
+        - model (str): The model to be used for language generation.
+        - base_url (str): The base URL for the OpenAI API.
+        - organization_id (str, optional): The organization ID for the OpenAI API. Defaults to "z".
+        - project_id (str, optional): The project ID for the OpenAI API. Defaults to "z".
+        - llm_api_key (str, optional): The API key for the OpenAI API. Defaults to "z".
+        - template (str, optional): The Jinja template to use. Defaults to "LLAMA3".
+        - temperature (float, optional): What sampling temperature to use, between 0 and 2. Defaults to 1.0.
+        """
+        self.completion_url = base_url
+        self.model = model
+        self.temperature = temperature
+        self.template = Template(TEMPLATES[template]["template"])
+        self.eot_token = TEMPLATES[template]["eot_token"]
+        self.prompt_headers = {
+            "Authorization": llm_api_key or "Bearer your_api_key_here"
+        }
+        logger.info(
+            f"Initialized AsyncLLM with the parameters: {self.completion_url} ({template})"
+        )
+
+    async def chat_completion(
+        self, messages: List[Dict[str, Any]], system: str = None
+    ) -> AsyncIterator[str]:
+        """
+        Generates a chat completion using the OpenAI API asynchronously.
+
+        Parameters:
+        - messages (List[Dict[str, Any]]): The list of messages to send to the API.
+        - system (str, optional): System prompt to use for this completion.
+
+        Yields:
+        - str: The content of each chunk from the API response.
+
+        Raises:
+        - APIConnectionError: When the server cannot be reached
+        - RateLimitError: When a 429 status code is received
+        - APIError: For other API-related errors
+        """
+        logger.debug(f"Messages: {messages}")
+        bos_token = "<|begin_of_text|>"
+        stream = None
+        try:
+            # If system prompt is provided, add it to the messages
+            messages_with_system: List[Dict[str, Any]] = messages
+            if system:
+                messages_with_system = [
+                    {"role": "system", "content": system},
+                    *messages,
+                ]
+            prompt = self.template.render(
+                messages=messages_with_system,
+                bos_token=bos_token,
+                add_generation_prompt=True,
+            )
+            data: Dict = {
+                "stream": True,
+                "temperature": self.temperature,
+                "prompt": prompt,
+            }
+            with requests.post(
+                self.completion_url, headers=self.prompt_headers, json=data, stream=True
+            ) as response:
+                for line in response.iter_lines():
+                    if line:
+                        line = self._clean_raw_bytes(line)
+                        next_token = self._process_line(line)
+                        if next_token:
+                            if next_token == self.eot_token:
+                                break
+                            yield next_token
+        except Exception as e:
+            logger.error(f"LLM API WITH TEMPLATE: Error occurred: {e}")
+            logger.info(f"Base URL: {self.base_url}")
+            logger.info(f"Model: {self.model}")
+            logger.info(f"Messages: {messages}")
+            logger.info(f"temperature: {self.temperature}")
+            yield "Error calling the chat endpoint: Error occurred while generating response. See the logs for details."
+        finally:
+            # make sure the stream is properly closed
+            # so when interrupted, no more tokens will being generated.
+            if stream:
+                logger.debug("Chat completion finished.")
+                await stream.close()
+                logger.debug("Stream closed.")
+
+    def _clean_raw_bytes(self, line):
+        line = line.decode("utf-8")
+        line = line.removeprefix("data: ")
+        line = json.loads(line)
+        return line
+
+    def _process_line(self, line):
+        if not (("stop" in line) and (line["stop"])):
+            token = line["content"]
+            return token

src/open_llm_vtuber/agent/stateless_llm_factory.py

@@ -0,0 +1,82 @@
+from typing import Type
+
+from loguru import logger
+
+from .stateless_llm.stateless_llm_interface import StatelessLLMInterface
+from .stateless_llm.stateless_llm_with_template import (
+    AsyncLLMWithTemplate as StatelessLLMWithTemplate,
+)
+from .stateless_llm.openai_compatible_llm import AsyncLLM as OpenAICompatibleLLM
+from .stateless_llm.ollama_llm import OllamaLLM
+from .stateless_llm.claude_llm import AsyncLLM as ClaudeLLM
+
+
+class LLMFactory:
+    @staticmethod
+    def create_llm(llm_provider, **kwargs) -> Type[StatelessLLMInterface]:
+        """Create an LLM based on the configuration.
+
+        Args:
+            llm_provider: The type of LLM to create
+            **kwargs: Additional arguments
+        """
+        logger.info(f"Initializing LLM: {llm_provider}")
+
+        if (
+            llm_provider == "openai_compatible_llm"
+            or llm_provider == "openai_llm"
+            or llm_provider == "gemini_llm"
+            or llm_provider == "zhipu_llm"
+            or llm_provider == "deepseek_llm"
+            or llm_provider == "groq_llm"
+            or llm_provider == "mistral_llm"
+            or llm_provider == "lmstudio_llm"
+        ):
+            return OpenAICompatibleLLM(
+                model=kwargs.get("model"),
+                base_url=kwargs.get("base_url"),
+                llm_api_key=kwargs.get("llm_api_key"),
+                organization_id=kwargs.get("organization_id"),
+                project_id=kwargs.get("project_id"),
+                temperature=kwargs.get("temperature"),
+            )
+        if llm_provider == "stateless_llm_with_template":
+            return StatelessLLMWithTemplate(
+                model=kwargs.get("model"),
+                base_url=kwargs.get("base_url"),
+                llm_api_key=kwargs.get("llm_api_key"),
+                organization_id=kwargs.get("organization_id"),
+                template=kwargs.get("template"),
+                project_id=kwargs.get("project_id"),
+            )
+        if llm_provider == "ollama_llm":
+            return OllamaLLM(
+                model=kwargs.get("model"),
+                base_url=kwargs.get("base_url"),
+                llm_api_key=kwargs.get("llm_api_key"),
+                organization_id=kwargs.get("organization_id"),
+                project_id=kwargs.get("project_id"),
+                temperature=kwargs.get("temperature"),
+                keep_alive=kwargs.get("keep_alive"),
+                unload_at_exit=kwargs.get("unload_at_exit"),
+            )
+
+        elif llm_provider == "llama_cpp_llm":
+            from .stateless_llm.llama_cpp_llm import LLM as LlamaLLM
+
+            return LlamaLLM(
+                model_path=kwargs.get("model_path"),
+            )
+        elif llm_provider == "claude_llm":
+            return ClaudeLLM(
+                system=kwargs.get("system_prompt"),
+                base_url=kwargs.get("base_url"),
+                model=kwargs.get("model"),
+                llm_api_key=kwargs.get("llm_api_key"),
+            )
+        else:
+            raise ValueError(f"Unsupported LLM provider: {llm_provider}")
+
+
+# Creating an LLM instance using a factory
+# llm_instance = LLMFactory.create_llm("ollama", **config_dict)

src/open_llm_vtuber/agent/transformers.py

@@ -0,0 +1,217 @@
+from typing import AsyncIterator, Tuple, Callable, List, Union, Dict, Any
+from functools import wraps
+from .output_types import Actions, SentenceOutput, DisplayText
+from ..utils.tts_preprocessor import tts_filter as filter_text
+from ..live2d_model import Live2dModel
+from ..config_manager import TTSPreprocessorConfig
+from ..utils.sentence_divider import SentenceDivider
+from ..utils.sentence_divider import SentenceWithTags, TagState
+from loguru import logger
+
+
+def sentence_divider(
+    faster_first_response: bool = True,
+    segment_method: str = "pysbd",
+    valid_tags: List[str] = None,
+):
+    """
+    Decorator that transforms token stream into sentences with tags
+
+    Args:
+        faster_first_response: bool - Whether to enable faster first response
+        segment_method: str - Method for sentence segmentation
+        valid_tags: List[str] - List of valid tags to process
+    """
+
+    def decorator(
+        func: Callable[
+            ..., AsyncIterator[Union[str, Dict[str, Any]]]
+        ],  # Expects str or dict
+    ) -> Callable[
+        ..., AsyncIterator[Union[SentenceWithTags, Dict[str, Any]]]
+    ]:  # Yields SentenceWithTags or dict
+        @wraps(func)
+        async def wrapper(
+            *args, **kwargs
+        ) -> AsyncIterator[Union[SentenceWithTags, Dict[str, Any]]]:
+            divider = SentenceDivider(
+                faster_first_response=faster_first_response,
+                segment_method=segment_method,
+                valid_tags=valid_tags or [],
+            )
+            stream_from_func = func(*args, **kwargs)
+
+            # Process the mixed stream using the updated SentenceDivider
+            async for item in divider.process_stream(stream_from_func):
+                if isinstance(item, SentenceWithTags):
+                    logger.debug(f"sentence_divider yielding sentence: {item}")
+                elif isinstance(item, dict):
+                    logger.debug(f"sentence_divider yielding dict: {item}")
+                yield item  # Yield either SentenceWithTags or dict
+            # Flushing is handled within divider.process_stream
+
+        return wrapper
+
+    return decorator
+
+
+def actions_extractor(live2d_model: Live2dModel):
+    """
+    Decorator that extracts actions from sentences, passing through dicts.
+    """
+
+    def decorator(
+        func: Callable[
+            ..., AsyncIterator[Union[SentenceWithTags, Dict[str, Any]]]
+        ],  # Input type hint
+    ) -> Callable[
+        ..., AsyncIterator[Union[Tuple[SentenceWithTags, Actions], Dict[str, Any]]]
+    ]:  # Output type hint
+        @wraps(func)
+        async def wrapper(
+            *args, **kwargs
+        ) -> AsyncIterator[
+            Union[Tuple[SentenceWithTags, Actions], Dict[str, Any]]
+        ]:  # Yield type hint
+            stream = func(*args, **kwargs)
+            async for item in stream:
+                if isinstance(item, SentenceWithTags):
+                    sentence = item
+                    actions = Actions()
+                    # Only extract emotions for non-tag text
+                    if not any(
+                        tag.state in [TagState.START, TagState.END]
+                        for tag in sentence.tags
+                    ):
+                        expressions = live2d_model.extract_emotion(sentence.text)
+                        if expressions:
+                            actions.expressions = expressions
+                    yield sentence, actions  # Yield the tuple
+                elif isinstance(item, dict):
+                    # Pass through dictionaries
+                    yield item
+                else:
+                    logger.warning(
+                        f"actions_extractor received unexpected type: {type(item)}"
+                    )
+
+        return wrapper
+
+    return decorator
+
+
+def display_processor():
+    """
+    Decorator that processes text for display, passing through dicts.
+    """
+
+    def decorator(
+        func: Callable[
+            ..., AsyncIterator[Union[Tuple[SentenceWithTags, Actions], Dict[str, Any]]]
+        ],  # Input type hint
+    ) -> Callable[
+        ...,
+        AsyncIterator[
+            Union[Tuple[SentenceWithTags, DisplayText, Actions], Dict[str, Any]]
+        ],
+    ]:  # Output type hint
+        @wraps(func)
+        async def wrapper(
+            *args, **kwargs
+        ) -> AsyncIterator[
+            Union[Tuple[SentenceWithTags, DisplayText, Actions], Dict[str, Any]]
+        ]:  # Yield type hint
+            stream = func(*args, **kwargs)
+
+            async for item in stream:
+                if (
+                    isinstance(item, tuple)
+                    and len(item) == 2
+                    and isinstance(item[0], SentenceWithTags)
+                ):
+                    sentence, actions = item
+                    text = sentence.text
+                    # Handle think tag states
+                    for tag in sentence.tags:
+                        if tag.name == "think":
+                            if tag.state == TagState.START:
+                                text = "("
+                            elif tag.state == TagState.END:
+                                text = ")"
+
+                    display = DisplayText(text=text)  # Simplified DisplayText creation
+                    yield sentence, display, actions  # Yield the tuple
+                elif isinstance(item, dict):
+                    # Pass through dictionaries
+                    yield item
+                else:
+                    logger.warning(
+                        f"display_processor received unexpected type: {type(item)}"
+                    )
+
+        return wrapper
+
+    return decorator
+
+
+def tts_filter(
+    tts_preprocessor_config: TTSPreprocessorConfig = None,
+):
+    """
+    Decorator that filters text for TTS, passing through dicts.
+    Skips TTS for think tag content.
+    """
+
+    def decorator(
+        func: Callable[
+            ...,
+            AsyncIterator[
+                Union[Tuple[SentenceWithTags, DisplayText, Actions], Dict[str, Any]]
+            ],
+        ],  # Input type hint
+    ) -> Callable[
+        ..., AsyncIterator[Union[SentenceOutput, Dict[str, Any]]]
+    ]:  # Output type hint
+        @wraps(func)
+        async def wrapper(
+            *args, **kwargs
+        ) -> AsyncIterator[Union[SentenceOutput, Dict[str, Any]]]:  # Yield type hint
+            stream = func(*args, **kwargs)
+            config = tts_preprocessor_config or TTSPreprocessorConfig()
+
+            async for item in stream:
+                if (
+                    isinstance(item, tuple)
+                    and len(item) == 3
+                    and isinstance(item[1], DisplayText)
+                ):
+                    sentence, display, actions = item
+                    if any(tag.name == "think" for tag in sentence.tags):
+                        tts = ""
+                    else:
+                        tts = filter_text(
+                            text=display.text,
+                            remove_special_char=config.remove_special_char,
+                            ignore_brackets=config.ignore_brackets,
+                            ignore_parentheses=config.ignore_parentheses,
+                            ignore_asterisks=config.ignore_asterisks,
+                            ignore_angle_brackets=config.ignore_angle_brackets,
+                        )
+
+                    logger.debug(f"[{display.name}] display: {display.text}")
+                    logger.debug(f"[{display.name}] tts: {tts}")
+
+                    yield SentenceOutput(
+                        display_text=display,
+                        tts_text=tts,
+                        actions=actions,
+                    )
+                elif isinstance(item, dict):
+                    # Pass through dictionaries
+                    yield item
+                else:
+                    logger.warning(f"tts_filter received unexpected type: {type(item)}")
+
+        return wrapper
+
+    return decorator

src/open_llm_vtuber/asr/asr_factory.py

@@ -0,0 +1,62 @@
+from typing import Type
+from .asr_interface import ASRInterface
+
+
+class ASRFactory:
+    @staticmethod
+    def get_asr_system(system_name: str, **kwargs) -> Type[ASRInterface]:
+        if system_name == "faster_whisper":
+            from .faster_whisper_asr import VoiceRecognition as FasterWhisperASR
+
+            return FasterWhisperASR(
+                model_path=kwargs.get("model_path"),
+                download_root=kwargs.get("download_root"),
+                language=kwargs.get("language"),
+                device=kwargs.get("device"),
+                compute_type=kwargs.get("compute_type"),
+                prompt=kwargs.get("prompt", None),
+            )
+        elif system_name == "whisper_cpp":
+            from .whisper_cpp_asr import VoiceRecognition as WhisperCPPASR
+
+            return WhisperCPPASR(**kwargs)
+        elif system_name == "whisper":
+            from .openai_whisper_asr import VoiceRecognition as WhisperASR
+
+            return WhisperASR(**kwargs)
+        elif system_name == "fun_asr":
+            from .fun_asr import VoiceRecognition as FunASR
+
+            return FunASR(
+                model_name=kwargs.get("model_name"),
+                vad_model=kwargs.get("vad_model"),
+                punc_model=kwargs.get("punc_model"),
+                ncpu=kwargs.get("ncpu"),
+                hub=kwargs.get("hub"),
+                device=kwargs.get("device"),
+                language=kwargs.get("language"),
+                use_itn=kwargs.get("use_itn"),
+                # sample_rate=kwargs.get("sample_rate"),
+            )
+        elif system_name == "azure_asr":
+            from .azure_asr import VoiceRecognition as AzureASR
+
+            return AzureASR(
+                subscription_key=kwargs.get("api_key"),
+                region=kwargs.get("region"),
+                languages=kwargs.get("languages", ["en-US", "zh-CN"]),
+            )
+        elif system_name == "groq_whisper_asr":
+            from .groq_whisper_asr import VoiceRecognition as GroqWhisperASR
+
+            return GroqWhisperASR(
+                api_key=kwargs.get("api_key"),
+                model=kwargs.get("model"),
+                lang=kwargs.get("lang"),
+            )
+        elif system_name == "sherpa_onnx_asr":
+            from .sherpa_onnx_asr import VoiceRecognition as SherpaOnnxASR
+
+            return SherpaOnnxASR(**kwargs)
+        else:
+            raise ValueError(f"Unknown ASR system: {system_name}")

src/open_llm_vtuber/asr/asr_interface.py

@@ -0,0 +1,57 @@
+import abc
+import numpy as np
+import asyncio
+
+
+class ASRInterface(metaclass=abc.ABCMeta):
+    SAMPLE_RATE = 16000
+    NUM_CHANNELS = 1
+    SAMPLE_WIDTH = 2
+
+    async def async_transcribe_np(self, audio: np.ndarray) -> str:
+        """Asynchronously transcribe speech audio in numpy array format.
+
+        By default, this runs the synchronous transcribe_np in a coroutine.
+        Subclasses can override this method to provide true async implementation.
+
+        Args:
+            audio: The numpy array of the audio data to transcribe.
+
+        Returns:
+            str: The transcription result.
+        """
+        if audio.dtype != np.float32:
+            audio = audio.astype(np.float32)
+        return await asyncio.to_thread(self.transcribe_np, audio)
+
+    @abc.abstractmethod
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        """Transcribe speech audio in numpy array format and return the transcription.
+
+        Args:
+            audio: The numpy array of the audio data to transcribe.
+        """
+        raise NotImplementedError
+
+    def nparray_to_audio_file(
+        self, audio: np.ndarray, sample_rate: int, file_path: str
+    ) -> None:
+        """Convert a numpy array of audio data to a .wav file.
+
+        Args:
+            audio: The numpy array of audio data.
+            sample_rate: The sample rate of the audio data.
+            file_path: The path to save the .wav file.
+        """
+        import wave
+
+        # Make sure the audio is in the range [-1, 1]
+        audio = np.clip(audio, -1, 1)
+        # Convert the audio to 16-bit PCM
+        audio_integer = (audio * 32767).astype(np.int16)
+
+        with wave.open(file_path, "wb") as wf:
+            wf.setnchannels(1)
+            wf.setsampwidth(2)
+            wf.setframerate(sample_rate)
+            wf.writeframes(audio_integer.tobytes())

src/open_llm_vtuber/asr/azure_asr.py

@@ -0,0 +1,155 @@
+import os
+from typing import Callable
+import numpy as np
+from loguru import logger
+import azure.cognitiveservices.speech as speechsdk
+from .asr_interface import ASRInterface
+import soundfile as sf
+import uuid
+import asyncio
+
+CACHE_DIR = "cache"
+
+
+class VoiceRecognition(ASRInterface):
+    def __init__(
+        self,
+        subscription_key=os.getenv("AZURE_API_Key"),
+        region=os.getenv("AZURE_REGION"),
+        languages=["en-US", "zh-CN"],
+        callback: Callable = logger.info,
+    ):
+        if not subscription_key or not region:
+            raise ValueError(
+                "Azure Speech Services requires both subscription_key and region. "
+                "Please check your configuration."
+            )
+
+        self.subscription_key = subscription_key
+        self.region = region
+        self.callback = callback
+
+        try:
+            self.speech_config = speechsdk.SpeechConfig(
+                subscription=self.subscription_key, region=self.region
+            )
+
+            # Set the languages for auto detection
+            self.speech_config.set_property(
+                speechsdk.PropertyId.SpeechServiceConnection_AutoDetectSourceLanguages,
+                ",".join(languages),
+            )
+        except Exception as e:
+            logger.error(f"Failed to initialize Azure Speech Config: {e}")
+            raise
+
+    def _create_speech_recognizer(self, uses_default_microphone: bool = True):
+        """
+        Create a speech recognizer instance with the specified configuration.
+
+        Args:
+            uses_default_microphone (bool): Whether to use default microphone
+
+        Returns:
+            SpeechRecognizer: Configured speech recognizer instance
+        """
+        try:
+            audio_config = speechsdk.AudioConfig(
+                use_default_microphone=uses_default_microphone
+            )
+            return speechsdk.SpeechRecognizer(
+                speech_config=self.speech_config, audio_config=audio_config
+            )
+        except Exception as e:
+            logger.warning(f"Failed to create speech recognizer: {e}")
+            raise
+
+    async def async_transcribe_np(self, audio: np.ndarray) -> str:
+        """
+        Asynchronously transcribe audio data using Azure Speech Services with auto language detection.
+
+        Args:
+            audio (np.ndarray): Audio data as numpy array
+
+        Returns:
+            str: Transcribed text
+
+        Raises:
+            Exception: If transcription fails
+        """
+        temp_file = os.path.join(CACHE_DIR, f"{uuid.uuid4()}.wav")
+
+        try:
+            os.makedirs(CACHE_DIR, exist_ok=True)
+            sf.write(temp_file, audio, 16000, "PCM_16")
+
+            audio_config = speechsdk.AudioConfig(filename=temp_file)
+            speech_recognizer = speechsdk.SpeechRecognizer(
+                speech_config=self.speech_config, audio_config=audio_config
+            )
+
+            # Perform recognition
+            result = speech_recognizer.recognize_once()
+
+            if result.reason == speechsdk.ResultReason.RecognizedSpeech:
+                # Get detected language
+                detected_language = result.properties.get(
+                    speechsdk.PropertyId.SpeechServiceConnection_AutoDetectSourceLanguageResult
+                )
+                logger.debug(f"Detected language: {detected_language}")
+                return result.text
+            elif result.reason == speechsdk.ResultReason.NoMatch:
+                logger.warning(
+                    f"No speech could be recognized: {result.no_match_details}"
+                )
+                return ""
+            elif result.reason == speechsdk.ResultReason.Canceled:
+                cancellation_details = result.cancellation_details
+                logger.error(
+                    f"Speech Recognition canceled: {cancellation_details.reason}"
+                )
+                if cancellation_details.reason == speechsdk.CancellationReason.Error:
+                    logger.error(f"Error details: {cancellation_details.error_details}")
+                raise Exception(
+                    f"Speech Recognition failed: {cancellation_details.reason}"
+                )
+
+        except Exception as e:
+            logger.error(f"Transcription failed: {e}")
+            raise
+        finally:
+            try:
+                if os.path.exists(temp_file):
+                    os.remove(temp_file)
+            except Exception as e:
+                logger.debug(f"Failed to remove temporary file {temp_file}: {e}")
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        """
+        Synchronously transcribe audio data using Azure Speech Services.
+
+        Args:
+            audio (np.ndarray): Audio data as numpy array
+
+        Returns:
+            str: Transcribed text
+
+        Raises:
+            Exception: If transcription fails
+        """
+        try:
+            try:
+                loop = asyncio.get_event_loop()
+            except RuntimeError:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+
+            # Run async method synchronously
+            return loop.run_until_complete(self.async_transcribe_np(audio))
+        except Exception as e:
+            logger.error(f"Transcription failed: {e}")
+            raise
+
+
+if __name__ == "__main__":
+    service = VoiceRecognition()

src/open_llm_vtuber/asr/faster_whisper_asr.py

@@ -0,0 +1,50 @@
+import numpy as np
+from faster_whisper import WhisperModel
+from .asr_interface import ASRInterface
+
+
+class VoiceRecognition(ASRInterface):
+    BEAM_SEARCH = True
+    # SAMPLE_RATE # Defined in asr_interface.py
+
+    def __init__(
+        self,
+        model_path: str = "distil-medium.en",
+        download_root: str = None,
+        language: str = "en",
+        device: str = "auto",
+        compute_type: str = "int8",
+        prompt: str = None,
+    ) -> None:
+        self.MODEL_PATH = model_path
+        self.LANG = language
+        self.prompt = prompt
+        self.model = WhisperModel(
+            model_size_or_path=model_path,
+            download_root=download_root,
+            device=device,
+            compute_type=compute_type,
+        )
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        if self.prompt:
+            segments, info = self.model.transcribe(
+                audio,
+                beam_size=5 if self.BEAM_SEARCH else 1,
+                language=self.LANG if self.LANG else None,
+                condition_on_previous_text=False,
+                initial_prompt=self.prompt,
+            )
+        else:
+            segments, info = self.model.transcribe(
+                audio,
+                beam_size=5 if self.BEAM_SEARCH else 1,
+                language=self.LANG if self.LANG else None,
+                condition_on_previous_text=False,
+            )
+        text = [segment.text for segment in segments]
+
+        if not text:
+            return ""
+        else:
+            return "".join(text)

src/open_llm_vtuber/asr/fun_asr.py

@@ -0,0 +1,131 @@
+import io
+import os
+import re
+import torch
+import numpy as np
+import soundfile as sf
+from funasr import AutoModel
+from .asr_interface import ASRInterface
+from typing import Optional
+
+# Try to import modelscope for local cache detection
+try:
+    from modelscope.hub.snapshot_download import snapshot_download
+
+    MODEL_SCOPE_DOWNLOAD_AVAILABLE = True
+except ImportError:
+    print("Warning: Unable to import modelscope.hub.snapshot_download.")
+    MODEL_SCOPE_DOWNLOAD_AVAILABLE = False
+
+# Model alias to actual ModelScope ID mapping table
+MODEL_ALIAS_TO_FULL_ID_MAP = {
+    "paraformer-zh": "iic/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
+    "paraformer-zh-spk": "iic/speech_paraformer-large-vad-punc-spk_asr_nat-zh-cn",
+    "paraformer-zh-online": "iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-online",
+    "paraformer-en": "iic/speech_paraformer-large-vad-punc_asr_nat-en-16k-common-vocab10020",
+    "conformer-en": "iic/speech_conformer_asr-en-16k-vocab4199-pytorch",
+    "ct-punc": "iic/punc_ct-transformer_cn-en-common-vocab471067-large",
+    "fsmn-vad": "iic/speech_fsmn_vad_zh-cn-16k-common-pytorch",
+    "fa-zh": "iic/speech_timestamp_prediction-v1-16k-offline",
+    "SenseVoiceSmall": "iic/SenseVoiceSmall",
+    "iic/SenseVoiceSmall": "iic/SenseVoiceSmall",
+}
+
+
+# paraformer-zh is a multi-functional asr model
+# use vad, punc, spk or not as you need
+
+
+class VoiceRecognition(ASRInterface):
+    def __init__(
+        self,
+        model_name: str = "iic/SenseVoiceSmall",
+        language: str = "auto",
+        vad_model: str = "fsmn-vad",
+        punc_model: str = "ct-punc",
+        ncpu: int = None,
+        hub: str = None,
+        device: str = "cpu",
+        disable_update: bool = True,
+        sample_rate: int = 16000,
+        use_itn: bool = False,
+    ) -> None:
+        # Resolve model paths
+        final_model_input = self._get_final_model_input(model_name)
+        final_vad_input = self._get_final_model_input(vad_model) if vad_model else None
+        final_punc_input = (
+            self._get_final_model_input(punc_model) if punc_model else None
+        )
+
+        self.model = AutoModel(
+            model=final_model_input,
+            vad_model=final_vad_input,
+            ncpu=ncpu,
+            hub=hub,
+            device=device,
+            disable_update=disable_update,
+            punc_model=final_punc_input,
+            # spk_model="cam++",
+        )
+        self.SAMPLE_RATE = sample_rate
+        self.use_itn = use_itn
+        self.language = language
+
+    def _get_final_model_input(self, alias_or_id: Optional[str]) -> Optional[str]:
+        """
+        Process model input function:
+        1. Check mapping table to get canonical ModelScope ID.
+        2. Try to get local path using snapshot_download.
+        3. If local path is valid, return local path, otherwise return canonical ModelScope ID.
+        """
+        if not alias_or_id:
+            return None
+
+        # Get canonical ModelScope ID from mapping table
+        resolved_id = MODEL_ALIAS_TO_FULL_ID_MAP.get(alias_or_id, alias_or_id)
+        final_input_for_automodel = resolved_id  # Default to use resolved ID
+
+        # Try to get local path using snapshot_download
+        if MODEL_SCOPE_DOWNLOAD_AVAILABLE:
+            try:
+                local_path = snapshot_download(resolved_id, local_files_only=True)
+                if os.path.exists(local_path):  # Double check path exists
+                    final_input_for_automodel = local_path  # Use local path if found
+                    # print(f"Successfully resolved '{resolved_id}' to local path: {local_path}")
+            except ValueError:
+                # Not found in local cache, use original ID
+                pass
+            except Exception as e:
+                print(f"Error occurred while checking '{resolved_id}': {e}")
+
+        return final_input_for_automodel
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        audio_tensor = torch.tensor(audio, dtype=torch.float32)
+
+        res = self.model.generate(
+            input=audio_tensor,
+            batch_size_s=300,
+            use_itn=self.use_itn,
+            language=self.language,
+        )
+
+        full_text = res[0]["text"]
+
+        # SenseVoiceSmall may spits out some tags
+        # like this: '<|zh|><|NEUTRAL|><|Speech|><|woitn|>欢迎大家来体验达摩院推出的语音识别模型'
+        # we should remove those tags from the result
+
+        # remove tags
+        full_text = re.sub(r"<\|.*?\|>", "", full_text)
+        # the tags can also look like '< | en | > < | EMO _ UNKNOWN | > < | S pe ech | > < | wo itn | > ', so...
+        full_text = re.sub(r"< \|.*?\| >", "", full_text)
+
+        return full_text.strip()
+
+    def _numpy_to_wav_in_memory(self, numpy_array: np.ndarray, sample_rate):
+        memory_file = io.BytesIO()
+        sf.write(memory_file, numpy_array, sample_rate, format="WAV")
+        memory_file.seek(0)
+
+        return memory_file

src/open_llm_vtuber/asr/groq_whisper_asr.py

@@ -0,0 +1,56 @@
+import io
+import wave
+import numpy as np
+from loguru import logger
+from groq import Groq
+from .asr_interface import ASRInterface
+
+
+class VoiceRecognition(ASRInterface):
+    # sample_rate, n_channels, and sampwidth are defined in asr_interface.py
+
+    def __init__(
+        self, api_key: str, model: str = "distil-whisper-large-v3-en", lang: str = "en"
+    ) -> None:
+        logger.info("Initializing Groq ASR...")
+        self.client = Groq(api_key=api_key)
+        self.lang = lang
+        self.model = model
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        """Transcribe speech audio in numpy array format and return the transcription.
+
+        Args:
+            audio: The numpy array of the audio data to transcribe.
+        """
+
+        logger.info("Transcribing audio (GroqWhisperASR)...")
+
+        # Turn the audio into an audio file
+        # Make sure the audio is in the range [-1, 1]
+        audio = np.clip(audio, -1, 1)
+        # Convert the audio to 16-bit PCM
+        audio_integer = (audio * 32767).astype(np.int16)
+
+        # groq api requires a file-like object for the audio data, so we use a BytesIO object
+        audio_buffer = io.BytesIO()
+
+        with wave.open(audio_buffer, "wb") as wf:
+            wf.setnchannels(self.NUM_CHANNELS)
+            wf.setsampwidth(self.SAMPLE_WIDTH)
+            wf.setframerate(self.SAMPLE_RATE)
+            wf.writeframes(audio_integer.tobytes())
+
+        audio_buffer.seek(0)
+
+        # Transcribe the audio with the BytesIO object
+        transcription = self.client.audio.transcriptions.create(
+            file=("audio.wav", audio_buffer.read()),
+            model=self.model,
+            # prompt="Specify context or spelling",
+            response_format="text",
+            language=self.lang,
+            temperature=0.0,
+        )
+
+        return transcription

src/open_llm_vtuber/asr/openai_whisper_asr.py

@@ -0,0 +1,27 @@
+import numpy as np
+import whisper
+from .asr_interface import ASRInterface
+
+
+class VoiceRecognition(ASRInterface):
+    def __init__(
+        self,
+        name: str = "base",
+        download_root: str = None,
+        device="cpu",
+        prompt: str = None,
+    ) -> None:
+        self.model = whisper.load_model(
+            name=name,
+            device=device,
+            download_root=download_root,
+        )
+        self.prompt = prompt
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        if self.prompt is not None:
+            result = self.model.transcribe(audio, initial_prompt=self.prompt)
+        else:
+            result = self.model.transcribe(audio)
+        full_text = result["text"]
+        return full_text

src/open_llm_vtuber/asr/sherpa_onnx_asr.py

@@ -0,0 +1,219 @@
+import os
+import numpy as np
+import sherpa_onnx
+from loguru import logger
+from .asr_interface import ASRInterface
+from .utils import download_and_extract, check_and_extract_local_file
+import onnxruntime
+
+
+class VoiceRecognition(ASRInterface):
+    def __init__(
+        self,
+        model_type: str = "paraformer",  # or "transducer", "nemo_ctc", "wenet_ctc", "whisper", "tdnn_ctc", "sense_voice", "fire_red_asr"
+        encoder: str = None,  # Path to the encoder model, used with transducer
+        decoder: str = None,  # Path to the decoder model, used with transducer
+        joiner: str = None,  # Path to the joiner model, used with transducer
+        paraformer: str = None,  # Path to the model.onnx from Paraformer
+        nemo_ctc: str = None,  # Path to the model.onnx from NeMo CTC
+        wenet_ctc: str = None,  # Path to the model.onnx from WeNet CTC
+        tdnn_model: str = None,  # Path to the model.onnx for the tdnn model of the yesno recipe
+        whisper_encoder: str = None,  # Path to whisper encoder model
+        whisper_decoder: str = None,  # Path to whisper decoder model
+        sense_voice: str = None,  # Path to the model.onnx from SenseVoice
+        fire_red_asr_encoder: str = None,  # Path to FireRedASR encoder model
+        fire_red_asr_decoder: str = None,  # Path to FireRedASR decoder model
+        tokens: str = None,  # Path to tokens.txt
+        hotwords_file: str = "",  # Path to hotwords file
+        hotwords_score: float = 1.5,  # Hotwords score
+        modeling_unit: str = "",  # Modeling unit for hotwords
+        bpe_vocab: str = "",  # Path to bpe vocabulary, used with hotwords
+        num_threads: int = 1,  # Number of threads for neural network computation
+        whisper_language: str = "",  # Language for whisper model
+        whisper_task: str = "transcribe",  # Task for whisper model (transcribe or translate)
+        whisper_tail_paddings: int = -1,  # Tail padding frames for whisper model
+        blank_penalty: float = 0.0,  # Penalty for blank symbol
+        decoding_method: str = "greedy_search",  # Decoding method (greedy_search or modified_beam_search)
+        debug: bool = False,  # Show debug messages
+        sample_rate: int = 16000,  # Sample rate
+        feature_dim: int = 80,  # Feature dimension
+        use_itn: bool = True,  # Use ITN for SenseVoice models
+        provider: str = "cpu",  # Provider for inference (cpu or cuda)
+    ) -> None:
+        self.model_type = model_type
+        self.encoder = encoder
+        self.decoder = decoder
+        self.joiner = joiner
+        self.paraformer = paraformer
+        self.nemo_ctc = nemo_ctc
+        self.wenet_ctc = wenet_ctc
+        self.tdnn_model = tdnn_model
+        self.whisper_encoder = whisper_encoder
+        self.whisper_decoder = whisper_decoder
+        self.sense_voice: str = sense_voice
+        self.fire_red_asr_encoder = fire_red_asr_encoder
+        self.fire_red_asr_decoder = fire_red_asr_decoder
+        self.tokens = tokens
+        self.hotwords_file = hotwords_file
+        self.hotwords_score = hotwords_score
+        self.modeling_unit = modeling_unit
+        self.bpe_vocab = bpe_vocab
+        self.num_threads = num_threads
+        self.whisper_language = whisper_language
+        self.whisper_task = whisper_task
+        self.whisper_tail_paddings = whisper_tail_paddings
+        self.blank_penalty = blank_penalty
+        self.decoding_method = decoding_method
+        self.debug = debug
+        self.SAMPLE_RATE = sample_rate
+        self.feature_dim = feature_dim
+        self.use_itn = use_itn
+
+        # we need to find a way to get cuda version of sherpa-onnx before we can
+        # use the gpu provider.
+        self.provider = provider
+        if self.provider == "cuda":
+            try:
+                if "CUDAExecutionProvider" not in onnxruntime.get_available_providers():
+                    logger.warning(
+                        "CUDA provider not available for ONNX. Falling back to CPU."
+                    )
+                    self.provider = "cpu"
+            except ImportError:
+                logger.warning("ONNX Runtime not installed. Falling back to CPU.")
+                self.provider = "cpu"
+        logger.info(f"Sherpa-Onnx-ASR: Using {self.provider} for inference")
+
+        self.recognizer = self._create_recognizer()
+
+    def _create_recognizer(self):
+        if self.model_type == "transducer":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_transducer(
+                encoder=self.encoder,
+                decoder=self.decoder,
+                joiner=self.joiner,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                sample_rate=self.SAMPLE_RATE,
+                feature_dim=self.feature_dim,
+                decoding_method=self.decoding_method,
+                hotwords_file=self.hotwords_file,
+                hotwords_score=self.hotwords_score,
+                modeling_unit=self.modeling_unit,
+                bpe_vocab=self.bpe_vocab,
+                blank_penalty=self.blank_penalty,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "paraformer":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_paraformer(
+                paraformer=self.paraformer,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                sample_rate=self.SAMPLE_RATE,
+                feature_dim=self.feature_dim,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "nemo_ctc":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_nemo_ctc(
+                model=self.nemo_ctc,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                sample_rate=self.SAMPLE_RATE,
+                feature_dim=self.feature_dim,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "wenet_ctc":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_wenet_ctc(
+                model=self.wenet_ctc,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                sample_rate=self.SAMPLE_RATE,
+                feature_dim=self.feature_dim,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "whisper":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_whisper(
+                encoder=self.whisper_encoder,
+                decoder=self.whisper_decoder,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                language=self.whisper_language,
+                task=self.whisper_task,
+                tail_paddings=self.whisper_tail_paddings,
+                provider=self.provider,
+            )
+        elif self.model_type == "tdnn_ctc":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_tdnn_ctc(
+                model=self.tdnn_model,
+                tokens=self.tokens,
+                sample_rate=self.SAMPLE_RATE,
+                feature_dim=self.feature_dim,
+                num_threads=self.num_threads,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "sense_voice":
+            if not self.sense_voice or not os.path.isfile(self.sense_voice):
+                if self.sense_voice.startswith(
+                    "./models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17"
+                ):
+                    logger.warning(
+                        "SenseVoice model not found. Downloading the model..."
+                    )
+
+                    url = "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17.tar.bz2"
+                    output_dir = "./models"
+                    # check the local file first before download
+                    local_result = check_and_extract_local_file(url, output_dir)
+
+                    if local_result is None:
+                        logger.info("Local file not found. Downloading...")
+                        download_and_extract(url, output_dir)
+                    else:
+                        logger.info("Local file found. Using existing file.")
+                    # download_and_extract(
+                    #     url="https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17.tar.bz2",
+                    #     output_dir="./models",
+                    # )
+                else:
+                    logger.critical(
+                        "The SenseVoice model is missing. Please provide the path to the model.onnx file."
+                    )
+            recognizer = sherpa_onnx.OfflineRecognizer.from_sense_voice(
+                model=self.sense_voice,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                use_itn=self.use_itn,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        elif self.model_type == "fire_red_asr":
+            recognizer = sherpa_onnx.OfflineRecognizer.from_fire_red_asr(
+                encoder=self.fire_red_asr_encoder,
+                decoder=self.fire_red_asr_decoder,
+                tokens=self.tokens,
+                num_threads=self.num_threads,
+                decoding_method=self.decoding_method,
+                debug=self.debug,
+                provider=self.provider,
+            )
+        else:
+            raise ValueError(f"Invalid model type: {self.model_type}")
+
+        return recognizer
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        stream = self.recognizer.create_stream()
+        stream.accept_waveform(self.SAMPLE_RATE, audio)
+        self.recognizer.decode_streams([stream])
+        return stream.result.text

src/open_llm_vtuber/asr/utils.py

@@ -0,0 +1,176 @@
+import os
+import requests
+import tarfile
+from pathlib import Path
+from tqdm import tqdm
+from loguru import logger
+
+
+def get_github_asset_url(owner, repo, release_tag, filename_without_ext):
+    """
+    Fetch the URL of a GitHub release asset by its filename (without extension).
+
+    Args:
+        owner (str): The owner of the repository.
+        repo (str): The name of the repository.
+        release_tag (str): The tag of the release.
+        filename_without_ext (str): The filename to search for (without extension).
+
+    Returns:
+        str: The download URL of the matched asset, or None if no match is found.
+    """
+    url = f"https://api.github.com/repos/{owner}/{repo}/releases/tags/{release_tag}"
+    headers = {}  # Add authentication headers if needed
+
+    try:
+        # Make a GET request to fetch release data
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+
+        # Parse the JSON response
+        release_data = response.json()
+        assets = release_data.get("assets", [])
+
+        # Look for a matching file
+        for asset in assets:
+            if asset["name"].startswith(filename_without_ext):
+                logger.info(f"Match found: {asset['name']}")
+                return asset["browser_download_url"]
+
+        # If no match found, log the error
+        logger.error(
+            f"No match found for filename: {filename_without_ext} in release {release_tag}."
+        )
+        return None
+
+    except requests.exceptions.RequestException as e:
+        logger.error(f"An error occurred while fetching release data: {e}")
+        return None
+
+
+def download_and_extract(url: str, output_dir: str) -> Path:
+    """
+    Download a file from a URL and extract it if it is a tar.bz2 archive.
+
+    Args:
+        url (str): The URL to download the file from.
+        output_dir (str): The directory to save the downloaded file.
+
+    Returns:
+        Path: Path to the extracted directory if it's a tar.bz2 file,
+             otherwise Path to the downloaded file.
+    """
+    # Create the output directory if it doesn't exist
+    Path(output_dir).mkdir(parents=True, exist_ok=True)
+
+    # Get the file name from the URL
+    file_name = url.split("/")[-1]
+    file_path = os.path.join(output_dir, file_name)
+
+    # Extract the root directory name from the filename (removing .tar.bz2)
+    root_dir = file_name.replace(".tar.bz2", "")
+    extracted_dir_path = Path(output_dir) / root_dir
+
+    # Check if the extracted directory already exists
+    if extracted_dir_path.exists():
+        logger.info(
+            f"✅ The directory {extracted_dir_path} already exists. I would assume that the model is already downloaded and we are ready to go. Skipping download and extraction."
+        )
+        return extracted_dir_path
+
+    # Download the file
+    logger.info(f"🏃‍♂️Downloading {url} to {file_path}...")
+    response = requests.get(url, stream=True)
+    response.raise_for_status()  # Raise an error for bad status codes
+    total_size = int(response.headers.get("content-length", 0))
+    logger.debug(f"Total file size: {total_size / 1024 / 1024:.2f} MB")
+
+    with (
+        open(file_path, "wb") as f,
+        tqdm(
+            desc=file_name,
+            total=total_size,
+            unit="iB",
+            unit_scale=True,
+            unit_divisor=1024,
+        ) as pbar,
+    ):
+        for chunk in response.iter_content(chunk_size=8192):
+            size = f.write(chunk)
+            pbar.update(size)
+
+    logger.info(f"Downloaded {file_name} successfully.")
+
+    # Extract the tar.bz2 file
+    if file_name.endswith(".tar.bz2"):
+        logger.info(f"Extracting {file_name}...")
+        with tarfile.open(file_path, "r:bz2") as tar:
+            tar.extractall(path=output_dir)
+        logger.info("Extraction completed.")
+
+        # Delete the compressed file
+        os.remove(file_path)
+        logger.debug(f"Deleted the compressed file: {file_name}")
+
+        return extracted_dir_path
+    else:
+        logger.warning("The downloaded file is not a tar.bz2 archive.")
+        return Path(file_path)
+
+
+def check_and_extract_local_file(url: str, output_dir: str) -> Path | None:
+    """
+    Check if a local file exists and extract it if it is a tar.bz2 archive.
+
+    Args:
+        url (str): The URL of the file.
+        output_dir (str): The directory to save the extracted files.
+
+    Returns:
+        Path | None: Path to the extracted directory if it's a tar.bz2 file,
+            otherwise None.
+    """
+    # Get the file name from the URL
+    file_name = url.split("/")[-1]
+    compressed_path = Path(output_dir) / file_name
+
+    # Check if the compressed file exists and is a tar.bz2 archive
+    extracted_dir = Path(output_dir) / file_name.replace(".tar.bz2", "")
+
+    if extracted_dir.exists():
+        logger.info(
+            f"✅ Extracted directory exists: {extracted_dir}, no operation needed."
+        )
+        return extracted_dir
+
+    if compressed_path.exists() and file_name.endswith(".tar.bz2"):
+        logger.info(f"🔍 Found local archive file: {compressed_path}")
+
+        try:
+            logger.info("⏳ Extracting archive file...")
+            with tarfile.open(compressed_path, "r:bz2") as tar:
+                tar.extractall(path=output_dir)
+            logger.success(f"Extracted archive to the path: {extracted_dir}")
+            os.remove(compressed_path)  # Remove the compressed file
+            return extracted_dir
+        except Exception as e:
+            logger.error(f"Fail to extract file: {str(e)}")
+            return None
+
+    logger.warning(f"Local file not found or not a tar.bz2 archive: {compressed_path}")
+    return None
+
+
+if __name__ == "__main__":
+    url = "https://github.com/k2-fsa/sherpa-onnx/releases/download/asr-models/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17.tar.bz2"
+    output_dir = "./models"
+
+    # Try local extraction first.
+    local_result = check_and_extract_local_file(url, output_dir)
+
+    # Download if not available locally.
+    if local_result is None:
+        logger.info("Local archive not found. Starting download...")
+        download_and_extract(url, output_dir)
+    else:
+        logger.info("Extraction completed using local file.")

src/open_llm_vtuber/asr/whisper_cpp_asr.py

@@ -0,0 +1,37 @@
+from pywhispercpp.model import Model
+
+import numpy as np
+from loguru import logger
+from .asr_interface import ASRInterface
+
+
+class VoiceRecognition(ASRInterface):
+    def __init__(
+        self,
+        model_name: str = "base",
+        model_dir="asr/models",
+        language: str = "en",
+        print_realtime=False,
+        print_progress=False,
+        prompt: str = None,
+    ) -> None:
+        self.model = Model(
+            model=model_name,
+            models_dir=model_dir,
+            language=language,
+            print_realtime=print_realtime,
+            print_progress=print_progress,
+        )
+        self.prompt = prompt
+
+    def transcribe_np(self, audio: np.ndarray) -> str:
+        if self.prompt is not None:
+            segments = self.model.transcribe(
+                audio, new_segment_callback=logger.info, initial_prompt=self.prompt
+            )
+        else:
+            segments = self.model.transcribe(audio, new_segment_callback=logger.info)
+        full_text = ""
+        for segment in segments:
+            full_text += segment.text
+        return full_text

src/open_llm_vtuber/audio_manager.py

@@ -0,0 +1,114 @@
+import re
+import uuid
+import unicodedata
+
+class AudioManager:
+    def __init__(self, tts, live2d, translator, config, verbose=False):
+        self.tts = tts
+        self.live2d = live2d
+        self.translator = translator
+        self.config = config
+        self.verbose = verbose
+        self.remove_special_char = config.get("REMOVE_SPECIAL_CHAR", True)
+        
+    def clean_text(self, text: str) -> str:
+        text = re.sub(r'[^\u4e00-\u9fffA-Za-z0-9,]', ' ', text)
+        
+        if self.remove_special_char:
+            text = self.remove_special_characters(text)
+            
+        return text
+        
+    def remove_special_characters(self, text: str) -> str:
+        """Filter text to remove all non-letter, non-number, and non-punctuation characters."""
+        normalized_text = unicodedata.normalize("NFKC", text)
+
+        def is_valid_char(char: str) -> bool:
+            category = unicodedata.category(char)
+            return (
+                category.startswith("L")
+                or category.startswith("N")
+                or category.startswith("P")
+                or char.isspace()
+            )
+
+        filtered_text = "".join(char for char in normalized_text if is_valid_char(char))
+        return filtered_text
+
+    def generate_audio_file(self, sentence: str, file_name_no_ext: str) -> str | None:
+        """
+        Generate an audio file from a given sentence using the TTS engine.
+
+        Parameters:
+        - sentence (str): The sentence to generate audio for
+        - file_name_no_ext (str): The name of the audio file (without extension)
+
+        Returns:
+        - str or None: The path of the generated audio file, or None if the sentence iempty
+        """
+        sentence = self.clean_text(sentence)
+        if self.verbose:
+            print(f">> generating {file_name_no_ext}...")
+
+        if not self.tts:
+            return None
+
+        if self.live2d:
+            sentence = self.live2d.remove_emotion_keywords(sentence)
+
+        if sentence.strip() == "":
+            return None
+
+        return self.tts.generate_audio(sentence, file_name_no_ext=file_name_no_ext)
+
+    def play_audio_file(self, sentence: str | None, filepath: str | None, instrument_filepath: str | None = None) -> None:
+        """
+        Play the audio file located at the given filepath.
+        """
+        if filepath is None:
+            print("No audio to be streamed. Response is empty.")
+            return
+
+        if sentence is None:
+            sentence = ""
+
+        try:
+            if self.verbose:
+                print(f">> Playing {filepath}...")
+            self.tts.play_audio_file_local(filepath)
+
+            self.tts.remove_file(filepath, verbose=self.verbose)
+        except ValueError as e:
+            if str(e) == "Audio is empty or all zero.":
+                print("No audio to be streamed. Response is empty.")
+            else:
+                raise e
+        except Exception as e:
+            print(f"Error playing the audio file {filepath}: {e}")
+
+    def play_text(self, text: str) -> None:
+            if not text.strip():
+                print("No text to play.")
+                return
+
+            sentences = re.split(r'(?<=[.!?。！？])\s*', text)
+            sentences = [s for s in sentences if s.strip()] 
+
+            for sentence in sentences:                
+                tts_target_sentence = self.live2d.remove_emotion_keywords(sentence)
+
+                if self.translator and self.config.get("TRANSLATE_AUDIO",   False):
+                    print("Translating...")
+                    tts_target_sentence = self.translator.translate (tts_target_sentence)
+                    print(f"Translated: {tts_target_sentence}")
+
+                audio_filepath = self.generate_audio_file(
+                    tts_target_sentence, file_name_no_ext=f"temp_text_{uuid.uuid4()}"
+                )
+
+                if audio_filepath:
+                    self.play_audio_file(sentence=sentence,     filepath=audio_filepath)
+                else:
+                    print("No audio generated for sentence.")
+
+