VTuber AI Agent Manager v1.0 | Powered by LFM2-8B-A1B | Built with anycoder
""")
return dashboard
# Launch the application
if __name__ == "__main__":
dashboard = create_dashboard()
dashboard.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
debug=True
)
```
```python
# models.py
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from typing import Dict, List, Generator, Optional
import json
from datetime import datetime
class PersonalityConfig:
"""Configuration class for VTuber personality"""
def __init__(self, config_dict: Dict):
self.name = config_dict.get("name", "AI-chan")
self.age = config_dict.get("age", 18)
self.personality_type = config_dict.get("personality_type", "Energetic")
self.speech_style = config_dict.get("speech_style", "Casual")
self.background = config_dict.get("background", "")
self.traits = config_dict.get("traits", {
"energy_level": 0.7,
"friendliness": 0.8,
"curiosity": 0.6,
"humor": 0.5
})
def to_dict(self) -> Dict:
return {
"name": self.name,
"age": self.age,
"personality_type": self.personality_type,
"speech_style": self.speech_style,
"background": self.background,
"traits": self.traits
}
def generate_system_prompt(self) -> str:
"""Generate system prompt based on personality configuration"""
traits_desc = f"""
Energy Level: {self.traits['energy_level']}/1.0
Friendliness: {self.traits['friendliness']}/1.0
Curiosity: {self.traits['curiosity']}/1.0
Humor: {self.traits['humor']}/1.0
"""
system_prompt = f"""
You are {self.name}, a {self.age}-year-old VTuber with a {self.personality_type.lower()} personality.
Your speech style is {self.speech_style.lower()}.
Background: {self.background}
Personality Traits:
{traits_desc}
Guidelines:
1. Always stay in character as {self.name}
2. Use expressions and emotions that match your personality type
3. Respond naturally and conversationally
4. Show emotions through your language (use emoticons and expressive language)
5. Keep responses concise but engaging
6. Never break character or mention you are an AI
Current mood and context should influence your responses naturally.
"""
return system_prompt
class VTuberAgent:
"""Main VTuber AI Agent class"""
def __init__(self, model_path: str, personality_config: Dict):
self.model_path = model_path
self.personality = PersonalityConfig(personality_config)
self.tokenizer = None
self.model = None
self.pipeline = None
self.device = "cuda" if torch.cuda.is_available() else "cpu"
self._load_model()
def _load_model(self):
"""Load the language model"""
try:
print(f"Loading model from {self.model_path}...")
# Load tokenizer
self.tokenizer = AutoTokenizer.from_pretrained(
self.model_path,
trust_remote_code=True
)
# Load model
self.model = AutoModelForCausalLM.from_pretrained(
self.model_path,
torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
device_map="auto" if self.device == "cuda" else None,
trust_remote_code=True
)
if self.device == "cpu":
self.model = self.model.to(self.device)
# Create pipeline
self.pipeline = pipeline(
"text-generation",
model=self.model,
tokenizer=self.tokenizer,
device=0 if self.device == "cuda" else -1,
torch_dtype=torch.float16 if self.device == "cuda" else torch.float32
)
print("Model loaded successfully!")
except Exception as e:
print(f"Error loading model: {e}")
raise
def generate_response(self, user_input: str, conversation_history: List[Dict]) -> str:
"""Generate a response to user input"""
try:
# Format conversation history
messages = [
{"role": "system", "content": self.personality.generate_system_prompt()}
]
# Add conversation history (last 10 messages to avoid context overflow)
for msg in conversation_history[-10:]:
messages.append({
"role": msg["role"],
"content": msg["content"]
})
# Add current user input
messages.append({"role": "user", "content": user_input})
# Generate response
response = self.pipeline(
messages,
max_new_tokens=512,
temperature=0.7,
top_p=0.9,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id
)
generated_text = response[0]["generated_text"]
# Extract only the assistant's response
if isinstance(generated_text, list):
assistant_response = generated_text[-1]["content"]
else:
# Parse the response to get the assistant's part
lines = generated_text.split('\n')
assistant_response = ""
capture = False
for line in lines:
if line.strip().startswith('assistant:') or line.strip().startswith('Assistant:'):
capture = True
assistant_response = line.split(':', 1)[1].strip()
elif capture and line.strip():
assistant_response += " " + line.strip()
elif capture and not line.strip():
break
return assistant_response.strip()
except Exception as e:
print(f"Error generating response: {e}")
return f"Sorry, I had trouble processing that. Could you try again? (Error: {str(e)})"
def generate_streaming_response(self, user_input: str, conversation_history: List[Dict]) -> Generator[str, None, None]:
"""Generate streaming response"""
try:
# For streaming, we'll simulate it for now
# In a real implementation, you'd use the model's streaming capability
full_response = self.generate_response(user_input, conversation_history)
# Simulate streaming by yielding chunks
words = full_response.split()
current_response = ""
for i, word in enumerate(words):
current_response += word + " "
yield current_response
# Add a small delay to simulate streaming
time.sleep(0.05)
except Exception as e:
print(f"Error in streaming response: {e}")
yield f"Sorry, I encountered an error: {str(e)}"
def update_personality(self, new_config: Dict):
"""Update personality configuration"""
self.personality = PersonalityConfig(new_config)
def get_current_mood(self) -> str:
"""Get current mood based on personality and recent interactions"""
# This would be implemented with more sophisticated logic
# For now, return a simple mood based on personality traits
if self.personality.traits["energy_level"] > 0.7:
return "excited"
elif self.personality.traits["friendliness"] > 0.7:
return "happy"
elif self.personality.traits["curiosity"] > 0.7:
return "curious"
else:
return "neutral"
```
```python
# utils.py
import numpy as np
from typing import Tuple, Dict, List, Optional
import queue
import threading
import time
import re
class AudioProcessor:
"""Process audio inputs and outputs"""
def __init__(self):
self.sample_rate = 48000
self.is_recording = False
def transcribe_audio(self, audio_data: Tuple[int, np.ndarray]) -> str:
"""Transcribe audio data to text (placeholder implementation)"""
# In a real implementation, you would use a speech-to-text model
# For now, return a placeholder
return "[Audio input detected - transcription would appear here]"
def synthesize_speech(self, text: str) -> Tuple[int, np.ndarray]:
"""Convert text to speech (placeholder implementation)"""
# In a real implementation, you would use a text-to-speech model
# For now, return placeholder audio
duration = len(text) * 0.1 # Estimate duration
samples = int(duration * self.sample_rate)
audio = np.random.randint(-32768, 32767, samples, dtype=np.int16)
return (self.sample_rate, audio)
def apply_voice_effects(self, audio: Tuple[int, np.ndarray], personality: str) -> Tuple[int, np.ndarray]:
"""Apply voice effects based on personality"""
sample_rate, audio_data = audio
# Placeholder for voice effects
# In a real implementation, you would apply pitch shifting, speed changes, etc.
if personality == "Energetic":
# Increase pitch slightly
pass
elif personality == "Calm":
# Slow down speech slightly
pass
return (sample_rate, audio_data)
class ExpressionMapper:
"""Map text emotions to facial expressions"""
def __init__(self):
self.emotion_keywords = {
"happy": ["happy", "joy", "excited", "glad", "wonderful", "amazing", "yay", "woohoo", ":)", "😊", "😄"],
"sad": ["sad", "cry", "tears", "unhappy", "depressed", "sob", ":( ", "😢", "😭"],
"angry": ["angry", "mad", "furious", "annoyed", "irritated", ">:(", "😡", "😠"],
"surprised": ["surprised", "shock", "wow", "omg", "really", "what", "😲", "😱"],
"thinking": ["think", "ponder", "wonder", "consider", "maybe", "hmm", "🤔"],
"blushing": ["blush", "embarrassed", "shy", "awkward", "😳", "🥰"],
"love": ["love", "heart", "adore", "cherish", "❤️", "💕", "😍"]
}
self.expressions = {
"happy": "😊",
"sad": "😢",
"angry": "😠",
"surprised": "😲",
"thinking": "🤔",
"blushing": "😳",
"love": "😍",
"neutral": "😐"
}
def map_expression(self, text: str) -> str:
"""Map text to appropriate expression"""
text_lower = text.lower()
# Check for emotion keywords
for emotion, keywords in self.emotion_keywords.items():
for keyword in keywords:
if keyword in text_lower:
return self.expressions.get(emotion, "😐")
# Check for question marks (thinking expression)
if "?" in text:
return self.expressions["thinking"]
# Check for exclamation marks (excited/happy)
if "!" in text and text.count("!") > 1:
return self.expressions["happy"]
return self.expressions["neutral"]
def get_expression_params(self, expression: str) -> Dict:
"""Get parameters for a specific expression"""
params = {
"neutral": {"eyes": 0.5, "mouth": 0.0, "eyebrows": 0.0, "blush": 0.0},
"happy": {"eyes": 0.8, "mouth": 0.8, "eyebrows": 0.3, "blush": 0.2},
"sad": {"eyes": 0.2, "mouth": -0.6, "eyebrows": -0.4, "blush": 0.3},
"angry": {"eyes": 0.6, "mouth": -0.3, "eyebrows": -0.6, "blush": 0.0},
"surprised": {"eyes": 1.0, "mouth": 0.5, "eyebrows": 0.6, "blush": 0.1},
"thinking": {"eyes": 0.4, "mouth": 0.1, "eyebrows": 0.2, "blush": 0.0},
"blushing": {"eyes": 0.6, "mouth": 0.3, "eyebrows": 0.1, "blush": 0.9},
"love": {"eyes": 0.9, "mouth": 0.6, "eyebrows": 0.3, "blush": 0.7}
}
return params.get(expression, params["neutral"])
class ResponseQueue:
"""Queue for managing responses and ensuring low latency"""
def __init__(self):
self.queue = queue.Queue()
self.processing = False
self.lock = threading.Lock()
def add_response(self, response: str, priority: int = 1):
"""Add a response to the queue"""
with self.lock:
self.queue.put((priority, time.time(), response))
def get_next_response(self, timeout: float = 1.0) -> Optional[str]:
"""Get the next response from the queue"""
try:
priority, timestamp, response = self.queue.get(timeout=timeout)
return response
except queue.Empty:
return None
def clear_queue(self):
"""Clear all responses from the queue"""
with self.lock:
while not self.queue.empty():
try:
self.queue.get_nowait()
except queue.Empty:
break
class SessionManager:
"""Manage VTuber sessions and data persistence"""
def __init__(self):
self.current_session = None
self.session_history = []
def create_session(self, personality_config: Dict) -> str:
"""Create a new session"""
session_id = f"session_{int(time.time())}"
self.current_session = {
"id": session_id,
"start_time": time.time(),
"personality": personality_config,
"messages": [],
"analytics": {
"message_count": 0,
"total_response_time": 0.0,
"emotion_counts": {}
}
}
return session_id
def add_message(self, role: str, content: str, response_time: float = 0.0):
"""Add a message to the current session"""
if self.current_session:
message = {
"role": role,
"content": content,
"timestamp": time.time(),
"response_time": response_time
}
self.current_session["messages"].append(message)
self.current_session["analytics"]["message_count"] += 1
self.current_session["analytics"]["total_response_time"] += response_time
def end_session(self) -> Dict:
"""End the current session and return session data"""
if self.current_session:
self.current_session["end_time"] = time.time()
self.current_session["duration"] = (
self.current_session["end_time"] - self.current_session["start_time"]
)
# Calculate average response time
if self.current_session["analytics"]["message_count"] > 0:
avg_time = (
self.current_session["analytics"]["total_response_time"] /
self.current_session["analytics"]["message_count"]
)
self.current_session["analytics"]["average_response_time"] = avg_time
session_data = self.current_session.copy()
self.session_history.append(session_data)
self.current_session = None
return session_data
return None
def export_session(self, session_id: str) -> Optional[Dict]:
"""Export session data"""
for session in self.session_history:
if session["id"] == session_id:
return session
return None
def format_conversation_history(history: List[Dict]) -> str:
"""Format conversation history for display"""
formatted = ""
for msg in history:
role_emoji = "👤" if msg["role"] == "user" else "🎭"
formatted += f"{role_emoji} {msg['role'].title()}: {msg['content']}\n"
return formatted
def analyze_emotions(text: str) -> Dict[str, float]:
"""Analyze emotions in text (simple implementation)"""
emotion_words = {
"joy": ["happy", "joy", "excited", "glad", "wonderful", "amazing"],
"sadness": ["sad", "cry", "tears", "unhappy", "depressed"],
"anger": ["angry", "mad", "furious", "annoyed", "irritated"],
"fear": ["scared", "afraid", "terrified", "worried", "anxious"],
"surprise": ["surprised", "shock", "wow", "amazed", "astonished"],
"love": ["love", "adore", "cherish", "heart", "affection"]
}
emotions = {}
text_lower = text.lower()
for emotion, words in emotion_words.items():
count = sum(1 for word in words if word in text_lower)
emotions[emotion] = count / len(text.split()) if text.split() else 0
return emotions
def validate_personality_config(config: Dict) -> bool:
"""Validate personality configuration"""
required_fields = ["name", "age", "personality_type", "speech_style"]
for field in required_fields:
if field not in config:
return False
# Validate age
if not isinstance(config.get("age"), (int, float)) or config["age"] < 16:
return False
# Validate traits if present
if "traits" in config:
traits = config["traits"]
required_traits = ["energy_level", "friendliness", "curiosity", "humor"]
for trait in required_traits:
if trait not in traits or not (0 <= traits[trait] <= 1):
return False
return True
```
```python
# config.py
import os
# Default personality configuration
DEFAULT_PERSONALITY = {
"name": "Yuki",
"age": 19,
"personality_type": "Energetic",
"speech_style": "Casual",
"background": "A cheerful VTuber who loves gaming and chatting with viewers! She's always full of energy and loves to make people smile.",
"traits": {
"energy_level": 0.8,
"friendliness": 0.9,
"curiosity": 0.7,
"humor": 0.6
}
}
# System prompts for different scenarios
SYSTEM_PROMPTS = {
"greeting": "Start with a warm, energetic greeting that matches your personality!",
"farewell": "End the conversation with a friendly goodbye that leaves viewers looking forward to next time!",
"gaming": "When talking about games, show excitement and share your genuine interests!",
"music": "Share your passion for music with enthusiasm and energy!",
"cooking": "Talk about cooking with excitement and share fun tips!",
"tech": "Discuss technology with curiosity and excitement about new innovations!"
}
# Model configurations
MODEL_CONFIG = {
"temperature": 0.7,
"max_tokens": 512,
"top_p": 0.9,
"repetition_penalty": 1.1,
"do_sample": True,
"pad_token_id": None # Will be set based on tokenizer
}
# Audio configurations
AUDIO_CONFIG = {
"input_sample_rate": 48000,
"output_sample_rate": 24000,
"chunk_duration": 0.02, # 20ms chunks for low latency
"noise_suppression": True,
"auto_gain_control": True,
"echo_cancellation": True
}
# Expression configurations
EXPRESSION_CONFIG = {
"auto_expressions": True,
"expression_change_threshold": 0.3, # Minimum confidence to change expression
"expression_duration": 2.0, # Minimum duration for expression change
"default_expression": "neutral",
"expression_smoothing": True
}
# Analytics configurations
ANALYTICS_CONFIG = {
"track_response_times": True,
"track_emotions": True,
"track_message_count": True,
"track_session_duration": True,
"export_format": "json",
"auto_save_interval": 300 # Save every 5 minutes
}
# UI configurations
UI_CONFIG = {
"theme": "soft",
"max_conversation_length": 50, # Maximum messages to display
"auto_scroll": True,
"show_timestamps": False,
"message_animation": True,
"emoji_size": "24px"
}
# Performance configurations
PERFORMANCE_CONFIG = {
"use_gpu": True,
"batch_size": 1,
"max_concurrent_requests": 10,
"cache_size": 1000,
"cleanup_interval": 3600 # Cleanup cache every hour
}
# API configurations (for external integrations)
API_CONFIG = {
"tts_provider": "local", # Options: "local", "openai", "elevenlabs"
"stt_provider": "local", # Options: "local", "openai", "google"
"avatar_provider": "local", # Options: "local", "vrm", "live2d"
"streaming": True,
"low_latency_mode": True
}
# Development/Debug configurations
DEBUG_CONFIG = {
"enable_logging": True,
"log_level": "INFO",
"save_conversations": True,
"save_directory": "./conversations",
"enable_profiling": False
}
# Safety and content moderation
SAFETY_CONFIG = {
"enable_content_filter": True,
"banned_words": [],
"max_response_length": 1000,
"min_response_length": 10,
"emergency_stop_phrase": "emergency stop vtuber"
}
# Integration configurations
INTEGRATION_CONFIG = {
"obs_websocket": {
"enabled": False,
"host": "localhost",
"port": 4444,
"password": ""
},
"discord": {
"enabled": False,
"bot_token": "",
"channel_id": ""
},
"twitch": {
"enabled": False,
"oauth_token": "",
"channel_name": ""
}
}
# Environment-specific settings
def get_config_for_environment(env: str = "development"):
"""Get configuration based on environment"""
base_config = {
"DEFAULT_PERSONALITY": DEFAULT_PERSONALITY,
"MODEL_CONFIG": MODEL_CONFIG,
"AUDIO_CONFIG": AUDIO_CONFIG,
"EXPRESSION_CONFIG": EXPRESSION_CONFIG,
"ANALYTICS_CONFIG": ANALYTICS_CONFIG,
"UI_CONFIG": UI_CONFIG,
"PERFORMANCE_CONFIG": PERFORMANCE_CONFIG,
"API_CONFIG": API_CONFIG,
"DEBUG_CONFIG": DEBUG_CONFIG,
"SAFETY_CONFIG": SAFETY_CONFIG,
"INTEGRATION_CONFIG": INTEGRATION_CONFIG
}
if env == "production":
# Production overrides
base_config["DEBUG_CONFIG"]["enable_logging"] = False
base_config["DEBUG_CONFIG"]["save_conversations"] = False
base_config["SAFETY_CONFIG"]["enable_content_filter"] = True
base_config["PERFORMANCE_CONFIG"]["use_gpu"] = True
elif env == "development":
# Development overrides
base_config["DEBUG_CONFIG"]["enable_logging"] = True
base_config["DEBUG_CONFIG"]["save_conversations"] = True
base_config["DEBUG_CONFIG"]["enable_profiling"] = True
base_config["PERFORMANCE_CONFIG"]["use_gpu"] = False # Use CPU for development
return base_config
# Get current environment
CURRENT_ENV = os.getenv("VTUBER_ENV", "development")
CONFIG = get_config_for_environment(CURRENT_ENV)
```
```python
# requirements.txt
gradio>=4.0.0
torch>=2.0.0
transformers>=4.35.0
accelerate>=0.24.0
numpy>=1.24.0
pandas>=2.0.0
scipy>=1.10.0
librosa>=0.10.0
soundfile>=0.12.0
opencv-python>=4.8.0
pillow>=10.0.0
matplotlib>=3.7.0
plotly>=5.15.0
requests>=2.31.0
websockets>=11.0.0
asyncio-mqtt>=0.13.0
python-dotenv>=1.0.0
pydantic>=2.0.0
typer>=0.9.0
rich>=13.0.0
tqdm>=4.65.0
```
This comprehensive VTuber AI Agent Manager provides:
## Key Features:
1. **Main Control Panel**: Initialize and control the VTuber agent with session management
2. **Personality Configuration**: Fully customizable personality traits including energy levels, speech style, and background
3. **Expression Control**: Real-time facial expression mapping based on conversation content
4. **Chat Interface**: Low-latency streaming responses with multimodal input (text/voice)
5. **Analytics Dashboard**: Track conversation metrics, emotions, and session statistics
6. **Advanced Settings**: Configure model parameters, audio settings, and performance options
## Technical Highlights:
- Uses the LFM2-8B-A1B model for natural, human-like responses
- Streaming response generation for low-latency interaction
- Audio processing capabilities for voice input/output
- Emotion-based expression mapping
- Session persistence and export functionality
- Modular architecture for easy extension
## Personality System:
The VTuber can be configured with:
- Character name and age
- Personality type (Energetic, Calm, Playful, etc.)
- Speech style (Casual, Formal, Cute, etc.)
- Background story
- Adjustable personality traits (energy, friendliness, curiosity, humor)
The system automatically generates appropriate system prompts based on the personality configuration, ensuring consistent character behavior throughout conversations.
The interface is designed to be intuitive for VTuber managers while providing powerful controls for fine-tuning the AI agent's behavior and responses. The low-latency streaming ensures smooth, natural interactions perfect for live streaming or real-time avatar control.