"""
Matrix client for the Matrix Command-Line Agent.
This module provides functionality to connect to Matrix homeservers,
join rooms, send messages, and listen for incoming messages.
It integrates with Ollama to process messages and generate responses.
This implementation uses the ollamatools Matrix integration to provide
enhanced features like search, conversation history tracking, topic analysis,
and performance metrics.
"""
import os
import sys
import yaml
import logging
import asyncio
import re
from typing import Dict, List, Optional, Union, Callable, Any, Set
from collections import defaultdict
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Import from ollamatools Matrix integration
try:
# Import from the new enhanced_matrix_core module
from ollamatools.tools.enhanced_matrix_core import (
EnhancedMatrixClient,
_connect_enhanced_matrix
)
# Import from nio for basic Matrix functionality
from nio import (
AsyncClient,
MatrixRoom,
RoomMessageText,
LoginResponse,
JoinError
)
# Define MatrixConfig class
class MatrixConfig:
"""
Configuration handler for Matrix integration.
"""
@staticmethod
def load_config(config_path: str) -> Dict:
"""
Load configuration from a YAML file.
Args:
config_path: Path to the configuration file
Returns:
Dict: The configuration as a dictionary
"""
try:
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
logger.info(f"Loaded configuration from {config_path}")
return config
except Exception as e:
logger.error(f"Error loading configuration from {config_path}: {e}")
raise
@staticmethod
def save_config(config: Dict, config_path: str) -> None:
"""
Save configuration to a YAML file.
Args:
config: The configuration as a dictionary
config_path: Path to save the configuration to
"""
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(config_path), exist_ok=True)
with open(config_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
logger.info(f"Saved configuration to {config_path}")
except Exception as e:
logger.error(f"Error saving configuration to {config_path}: {e}")
raise
@staticmethod
def create_default_config() -> Dict:
"""
Create a default configuration.
Returns:
Dict: The default configuration
"""
return {
"MATRIX": {
"HOMESERVER": "https://matrix.example.org",
"USER": "@user:example.org",
"PASSWORD": "your-password",
"DEVICE_ID": "matrix-cmd-agent",
"ROOMS": [
"#room:example.org"
],
"ADMIN_USERS": [
"@admin:example.org"
]
},
"OLLAMA": {
"HOST": "localhost:11434",
"MODEL": "llama3:latest",
"PROMPT": "You are a helpful assistant.",
"AVAILABLE_MODELS": [
"llama3:latest",
"mistral:latest",
"gemma:latest"
]
},
"ENHANCED": {
"ENABLE_SEARCH": False,
"ENABLE_CONVERSATION_TRACKING": False,
"HISTORY_FILE": "~/.config/matrix-cmd/history.json",
"MAX_CONVERSATION_HISTORY": 100,
"RESPOND_TO_ALL_MESSAGES": False,
"OPENSEARCH": {
"BASE_URL": "http://localhost:8081",
"ENGINES": [
"google",
"bing",
"duckduckgo"
],
"NUM_RESULTS": 5
}
}
}
# Import functions from plugins
from ollamatools.plugins.integrations.matrix.search import (
handle_search_command,
get_default_opensearch_config
)
from ollamatools.plugins.integrations.matrix.history import (
load_conversation_history,
save_conversation_history,
record_conversation,
analyze_conversation_topics,
get_performance_metrics
)
# Define connect_enhanced_matrix function
connect_enhanced_matrix = _connect_enhanced_matrix
# Define handle_history_command function
async def handle_history_command(client, count, room_id):
from ollamatools.plugins.integrations.matrix.history import handle_history_command
await handle_history_command(client, count, room_id)
# Define handle_topics_command function
async def handle_topics_command(client, room_id):
from ollamatools.plugins.integrations.matrix.history import handle_topics_command
await handle_topics_command(client, room_id)
# Define handle_metrics_command function
async def handle_metrics_command(client, room_id):
from ollamatools.plugins.integrations.matrix.history import handle_metrics_command
await handle_metrics_command(client, room_id)
OLLAMATOOLS_AVAILABLE = True
except ImportError:
logger.error(
"Matrix integration requires the 'ollamatools' package. "
"Please ensure it is installed correctly."
)
# Fall back to direct imports for basic functionality
try:
from nio import (
AsyncClient,
MatrixRoom,
RoomMessageText,
LoginResponse,
JoinError
)
except ImportError:
logger.error(
"Matrix support requires the 'matrix-nio' package. "
"Please install it with 'pip install matrix-nio' or "
"'pip install matrix-nio[e2e]' for end-to-end encryption support."
)
sys.exit(1)
OLLAMATOOLS_AVAILABLE = False
# Define MatrixConfig for fallback
class MatrixConfig:
"""
Configuration handler for Matrix integration.
"""
@staticmethod
def load_config(config_path: str) -> Dict:
"""
Load configuration from a YAML file.
Args:
config_path: Path to the configuration file
Returns:
Dict: The configuration as a dictionary
"""
try:
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
logger.info(f"Loaded configuration from {config_path}")
return config
except Exception as e:
logger.error(f"Error loading configuration from {config_path}: {e}")
raise
@staticmethod
def save_config(config: Dict, config_path: str) -> None:
"""
Save configuration to a YAML file.
Args:
config: The configuration as a dictionary
config_path: Path to save the configuration to
"""
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(config_path), exist_ok=True)
with open(config_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
logger.info(f"Saved configuration to {config_path}")
except Exception as e:
logger.error(f"Error saving configuration to {config_path}: {e}")
raise
@staticmethod
def create_default_config() -> Dict:
"""
Create a default configuration.
Returns:
Dict: The default configuration
"""
return {
"MATRIX": {
"HOMESERVER": "https://matrix.example.org",
"USER": "@user:example.org",
"PASSWORD": "your-password",
"DEVICE_ID": "matrix-cmd-agent",
"ROOMS": [
"#room:example.org"
],
"ADMIN_USERS": [
"@admin:example.org"
]
},
"OLLAMA": {
"HOST": "localhost:11434",
"MODEL": "llama3:latest",
"PROMPT": "You are a helpful assistant.",
"AVAILABLE_MODELS": [
"llama3:latest",
"mistral:latest",
"gemma:latest"
]
},
"ENHANCED": {
"ENABLE_SEARCH": False,
"ENABLE_CONVERSATION_TRACKING": False,
"HISTORY_FILE": "~/.config/matrix-cmd/history.json",
"MAX_CONVERSATION_HISTORY": 100,
"RESPOND_TO_ALL_MESSAGES": False,
"OPENSEARCH": {
"BASE_URL": "http://localhost:8081",
"ENGINES": [
"google",
"bing",
"duckduckgo"
],
"NUM_RESULTS": 5
}
}
}
# Make ollama import optional
try:
from ollama import AsyncClient as OllamaAsync
OLLAMA_AVAILABLE = True
except ImportError:
OLLAMA_AVAILABLE = False
logger.warning(
"The 'ollama' package is not installed. "
"Message processing with Ollama will not be available. "
"Install it with 'pip install ollama'."
)
class MatrixClient:
"""
A client for connecting to Matrix homeservers and interacting with rooms.
This client can connect to a Matrix homeserver, join rooms, send messages,
and listen for incoming messages. It integrates with Ollama to process
messages and generate responses.
This implementation uses the ollamatools Matrix integration to provide
enhanced features like search, conversation history tracking, topic analysis,
and performance metrics.
"""
# Command patterns
MODEL_COMMAND_PATTERN = re.compile(r'^/model\s+([a-zA-Z0-9:_\-\.]+)$')
HELP_COMMAND_PATTERN = re.compile(r'^/help$')
LIST_MODELS_COMMAND_PATTERN = re.compile(r'^/models$')
PING_COMMAND_PATTERN = re.compile(r'^/ping$')
SEARCH_COMMAND_PATTERN = re.compile(r'^/search\s+(.+)$')
HISTORY_COMMAND_PATTERN = re.compile(r'^/history(?:\s+(\d+))?$')
TOPICS_COMMAND_PATTERN = re.compile(r'^/topics$')
METRICS_COMMAND_PATTERN = re.compile(r'^/metrics$')
def __init__(
self,
homeserver: str,
user_id: str,
password: str,
device_id: str = None,
ollama_host: str = "localhost:11434",
ollama_model: str = "llama3:latest",
ollama_prompt: str = None,
available_models: List[str] = None,
admin_users: List[str] = None,
ssl: bool = True,
enable_search: bool = False,
enable_conversation_tracking: bool = False,
opensearch_config: Dict[str, Any] = None,
history_file: str = None,
max_conversation_history: int = 100,
respond_to_all_messages: bool = False
):
"""
Initialize a new Matrix client.
Args:
homeserver: The URL of the Matrix homeserver
user_id: The Matrix user ID (e.g., @user:example.com)
password: The password for the Matrix account
device_id: A unique identifier for this device (for E2E encryption)
ollama_host: The host address for the Ollama API
ollama_model: The Ollama model to use for generating responses
ollama_prompt: An optional system prompt for the Ollama model
available_models: List of available Ollama models
admin_users: List of Matrix user IDs that can change models for all users
ssl: Whether to use SSL for the Matrix connection
enable_search: Whether to enable search functionality
enable_conversation_tracking: Whether to enable conversation history tracking
opensearch_config: Configuration for OpenSearchNG
history_file: Path to the conversation history file
max_conversation_history: Maximum number of conversations to keep in history
respond_to_all_messages: Whether to respond to all messages (not just mentions)
"""
self.homeserver = homeserver
self.user_id = user_id
self.password = password
self.device_id = device_id
self.ollama_host = ollama_host
self.ollama_model = ollama_model
self.ollama_prompt = ollama_prompt
self.ssl = ssl
# Enhanced features
self.enable_search = enable_search
self.enable_conversation_tracking = enable_conversation_tracking
self.history_file = history_file or os.path.expanduser("~/.config/matrix-cmd/history.json")
self.max_conversation_history = max_conversation_history
self.respond_to_all_messages = respond_to_all_messages
# Set OpenSearch config
if OLLAMATOOLS_AVAILABLE and hasattr(get_default_opensearch_config, '__call__'):
self.opensearch_config = opensearch_config or get_default_opensearch_config()
else:
self.opensearch_config = opensearch_config or self._get_default_opensearch_config()
# Conversation history
self.conversation_history = []
if self.enable_conversation_tracking:
if OLLAMATOOLS_AVAILABLE and hasattr(load_conversation_history, '__call__'):
self.conversation_history = load_conversation_history(self.history_file)
else:
self.conversation_history = self._load_conversation_history()
# Initialize the client (will be set in connect())
self.client = None
# Track when the client connected to avoid processing old messages
self.connection_time = None
# Track joined rooms
self.rooms = []
# Message callback
self.message_callback = None
# User context tracking
self.user_contexts = defaultdict(list)
self.max_context_length = 10 # Maximum number of messages to keep in context
# User model preferences
self.user_models = {}
# Available models
self.available_models = available_models or [ollama_model]
# Admin users who can change models
self.admin_users = set(admin_users or [])
async def connect(self, max_retries: int = 3, initial_retry_delay: float = 1.0) -> bool:
"""
Connect to the Matrix homeserver and log in.
Args:
max_retries: Maximum number of retry attempts (default: 3)
initial_retry_delay: Initial delay between retries in seconds (default: 1.0)
Returns:
bool: True if the connection was successful, False otherwise
"""
if OLLAMATOOLS_AVAILABLE and hasattr(connect_enhanced_matrix, '__call__'):
try:
# Use the ollamatools enhanced Matrix client
enhanced_client = await connect_enhanced_matrix(
homeserver=self.homeserver,
user_id=self.user_id,
password=self.password,
device_id=self.device_id,
ollama_host=self.ollama_host,
ollama_model=self.ollama_model,
ollama_prompt=self.ollama_prompt,
available_models=self.available_models,
admin_users=list(self.admin_users),
opensearch_config=self.opensearch_config,
ssl=self.ssl,
enable_search=self.enable_search,
enable_conversation_tracking=self.enable_conversation_tracking,
max_context_length=self.max_context_length,
max_conversation_history=self.max_conversation_history,
respond_to_all_messages=self.respond_to_all_messages
)
if enhanced_client:
# Store the client
self.client = enhanced_client
# Set connection time
self.connection_time = asyncio.get_event_loop().time()
# Copy conversation history if available
if hasattr(enhanced_client, 'conversation_history'):
self.conversation_history = enhanced_client.conversation_history
logger.info(f"Successfully connected to {self.homeserver} as {self.user_id}")
return True
else:
logger.error(f"Failed to connect to {self.homeserver}")
return False
except Exception as e:
logger.error(f"Error connecting to Matrix server with ollamatools: {e}")
logger.info("Falling back to standard Matrix connection")
# Fall back to standard connection
# Standard connection (fallback or if ollamatools is not available)
retry_count = 0
current_retry_delay = initial_retry_delay
while retry_count <= max_retries:
try:
# Initialize the Matrix client
self.client = AsyncClient(
homeserver=self.homeserver,
device_id=self.device_id,
ssl=self.ssl
)
# Extract username from user_id (e.g., "@user:example.com" -> "user")
username = self.user_id.split(':')[0].lstrip('@')
# Set the user ID for the client
self.client.user = username
logger.info(f"Attempting to connect to {self.homeserver} as {self.user_id}...")
response = await self.client.login(
password=self.password
)
if isinstance(response, LoginResponse):
logger.info(f"Successfully connected to {self.homeserver} as {self.user_id}")
self.connection_time = asyncio.get_event_loop().time()
return True
else:
logger.error(f"Failed to connect to {self.homeserver}: {response}")
# Check if we should retry
if retry_count < max_retries:
retry_count += 1
logger.info(f"Retrying connection (attempt {retry_count}/{max_retries}) in {current_retry_delay:.1f} seconds...")
await asyncio.sleep(current_retry_delay)
# Exponential backoff
current_retry_delay *= 2
else:
logger.error(f"Maximum retry attempts ({max_retries}) reached. Giving up.")
return False
except asyncio.CancelledError:
logger.info("Connection attempt cancelled by user.")
raise
except Exception as e:
logger.error(f"Error connecting to Matrix server: {e}")
# Check if this is a rate limiting error
if "429" in str(e) and "ratelimited" in str(e):
# Try to extract the retry time from the error message
import re
retry_match = re.search(r'sleeping for (\d+)ms', str(e))
if retry_match:
retry_ms = int(retry_match.group(1))
retry_sec = retry_ms / 1000
if retry_sec > 300: # If wait time is more than 5 minutes
logger.error(f"Rate limit wait time is too long ({retry_sec:.1f} seconds). Consider trying again later.")
return False
logger.info(f"Server is rate limiting requests. Waiting for {retry_sec:.1f} seconds before retrying...")
try:
# Wait for the specified time
await asyncio.sleep(retry_sec)
# Don't increment retry count for rate limiting
continue
except asyncio.CancelledError:
logger.info("Rate limit wait cancelled by user.")
raise
else:
# If we can't extract the retry time, use exponential backoff
if retry_count < max_retries:
retry_count += 1
logger.info(f"Rate limited but couldn't determine wait time. Retrying (attempt {retry_count}/{max_retries}) in {current_retry_delay:.1f} seconds...")
await asyncio.sleep(current_retry_delay)
# Exponential backoff
current_retry_delay *= 2
else:
logger.error(f"Maximum retry attempts ({max_retries}) reached. Giving up.")
return False
else:
# For non-rate-limiting errors
if retry_count < max_retries:
retry_count += 1
logger.info(f"Retrying connection (attempt {retry_count}/{max_retries}) in {current_retry_delay:.1f} seconds...")
await asyncio.sleep(current_retry_delay)
# Exponential backoff
current_retry_delay *= 2
else:
logger.error(f"Maximum retry attempts ({max_retries}) reached. Giving up.")
return False
return False
async def join_room(self, room_id: str) -> bool:
"""
Join a Matrix room.
Args:
room_id: The ID of the room to join (e.g., #room:example.com)
Returns:
bool: True if joining was successful, False otherwise
"""
if not self.client:
logger.error("Not connected to Matrix server")
return False
# Check if we're using ollamatools client
if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'join_room'):
try:
success = await self.client.join_room(room_id)
if success:
logger.info(f"Successfully joined room {room_id}")
self.rooms.append(room_id)
return True
else:
logger.error(f"Failed to join room {room_id}")
return False
except Exception as e:
logger.error(f"Error joining room {room_id}: {e}")
return False
else:
# Standard client
try:
response = await self.client.join(room_id)
if isinstance(response, JoinError):
logger.error(f"Failed to join room {room_id}: {response}")
return False
logger.info(f"Successfully joined room {room_id}")
self.rooms.append(room_id)
return True
except Exception as e:
logger.error(f"Error joining room {room_id}: {e}")
return False
async def send_message(self, room_id: str, message: str) -> bool:
"""
Send a message to a Matrix room.
Args:
room_id: The ID of the room to send the message to
message: The message to send
Returns:
bool: True if sending was successful, False otherwise
"""
if not self.client:
logger.error("Not connected to Matrix server")
return False
# Check if we're using ollamatools client
if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'send_message'):
try:
logger.info(f"Sending message to room {room_id}: {message}")
success = await self.client.send_message(room_id, message)
if success:
logger.info(f"Message sent successfully to room {room_id}")
return True
else:
logger.error(f"Failed to send message to room {room_id}")
return False
except Exception as e:
logger.error(f"Error sending message to room {room_id}: {e}")
return False
else:
# Standard client
try:
logger.info(f"Sending message to room {room_id}: {message}")
response = await self.client.room_send(
room_id=room_id,
message_type="m.room.message",
content={
"msgtype": "m.text",
"body": message
}
)
logger.info(f"Message sent successfully to room {room_id}")
return True
except Exception as e:
logger.error(f"Error sending message to room {room_id}: {e}")
return False
async def process_with_ollama(self, message: str, sender: str = None, room_id: str = None) -> str:
"""
Process a message with Ollama and return the response.
Args:
message: The message to process
sender: The Matrix user ID of the sender
room_id: The ID of the room where the message was sent
Returns:
str: The response from Ollama
"""
if not self.client:
logger.error("Not connected to Matrix server")
return "Error: Not connected to Matrix server"
if not OLLAMA_AVAILABLE:
return "Ollama is not available. Please install the 'ollama' package with 'pip install ollama'."
# Check if we're using ollamatools client
if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'process_with_ollama'):
try:
# Use the ollamatools client to process the message
response = await self.client.process_with_ollama(message, sender, room_id)
return response
except Exception as e:
logger.error(f"Error processing message with ollamatools: {e}")
logger.info("Falling back to standard Ollama processing")
# Fall back to standard processing
# Standard processing
try:
ollama_client = OllamaAsync(host=self.ollama_host)
# Determine which model to use
model = self.ollama_model
if sender and sender in self.user_models:
model = self.user_models[sender]
messages = []
# Add system prompt if provided
if self.ollama_prompt:
messages.append({
"role": "system",
"content": self.ollama_prompt
})
# Add conversation context if available
if sender and room_id:
context_key = f"{sender}:{room_id}"
for ctx_msg in self.user_contexts[context_key]:
messages.append(ctx_msg)
# Add user message
user_message = {
"role": "user",
"content": message
}
messages.append(user_message)
# Store user message in context
if sender and room_id:
context_key = f"{sender}:{room_id}"
self.user_contexts[context_key].append(user_message)
# Trim context if it gets too long
if len(self.user_contexts[context_key]) > self.max_context_length:
self.user_contexts[context_key] = self.user_contexts[context_key][-self.max_context_length:]
logger.info(f"Sending request to Ollama with model: {model}")
response = await ollama_client.chat(
model=model,
messages=messages
)
logger.info(f"Received response from Ollama: {response}")
response_message = response["message"]
message_content = response_message["content"]
logger.info(f"Extracted message content: {message_content}")
# Store assistant response in context
if sender and room_id:
context_key = f"{sender}:{room_id}"
self.user_contexts[context_key].append({
"role": "assistant",
"content": message_content
})
# Trim context if it gets too long
if len(self.user_contexts[context_key]) > self.max_context_length:
self.user_contexts[context_key] = self.user_contexts[context_key][-self.max_context_length:]
# Record conversation if tracking is enabled
if self.enable_conversation_tracking and sender and room_id:
import time
start_time = time.time()
response_time = time.time() - start_time
if OLLAMATOOLS_AVAILABLE and hasattr(record_conversation, '__call__'):
record_conversation(
self.conversation_history,
message,
message_content,
sender,
room_id,
response_time,
self.history_file,
self.max_conversation_history
)
else:
self._record_conversation(
message,
message_content,
sender,
room_id,
response_time
)
logger.debug(f"Received response from Ollama (model: {model}): {message_content}")
return message_content
except Exception as e:
logger.error(f"Error processing message with Ollama: {e}")
return f"Error processing message: {e}"
async def _message_callback(self, room: MatrixRoom, event: RoomMessageText) -> None:
"""
Internal callback for handling incoming messages.
Args:
room: The room the message was sent in
event: The message event
"""
if not isinstance(event, RoomMessageText):
return
# Get message timestamp
message_timestamp = event.server_timestamp / 1000
# Skip messages from before the client connected
if self.connection_time and message_timestamp < self.connection_time:
logger.debug("Skipping old message")
return
# Skip messages from the bot itself
if event.sender == self.user_id:
logger.debug("Skipping message from self")
return
message = event.body
room_id = room.room_id
sender = event.sender
logger.info(f"Received message in room {room_id} from {sender}: {message}")
# If a custom callback is set, use it
if self.message_callback:
await self.message_callback(room, event, self)
return
# Check if this is a command
if message.startswith('/'):
await self._handle_command(message, room_id, sender)
return
# Extract username from user_id (e.g., "@user:example.com" -> "user")
username = self.user_id.split(':')[0].lstrip('@')
# Check if the message contains the bot's username (case insensitive)
if username.lower() in message.lower() or self.respond_to_all_messages:
logger.info(f"Username mention detected in message: {message}")
# Process with Ollama and send response
response = await self.process_with_ollama(message, sender, room_id)
await self.send_message(room_id, response)
return
# For other messages without username mention, don't automatically respond
async def _handle_command(self, message: str, room_id: str, sender: str) -> None:
"""
Handle commands sent to the bot.
Args:
message: The command message
room_id: The ID of the room where the command was sent
sender: The Matrix user ID of the sender
"""
# Check if this is a model change command
model_match = self.MODEL_COMMAND_PATTERN.match(message)
if model_match:
model_name = model_match.group(1)
await self._change_model(model_name, room_id, sender)
return
# Check if this is a help command
if self.HELP_COMMAND_PATTERN.match(message):
await self._send_help(room_id)
return
# Check if this is a list models command
if self.LIST_MODELS_COMMAND_PATTERN.match(message):
await self._list_models(room_id)
return
# Check if this is a ping command
if self.PING_COMMAND_PATTERN.match(message):
await self.send_message(room_id, "Pong!")
return
# Check if this is a search command
search_match = self.SEARCH_COMMAND_PATTERN.match(message)
if search_match:
if not self.enable_search:
await self.send_message(room_id, "Search functionality is not enabled.")
return
query = search_match.group(1)
# Use ollamatools if available
if OLLAMATOOLS_AVAILABLE and hasattr(handle_search_command, '__call__'):
await handle_search_command(self, query, room_id)
else:
await self._handle_search_command(query, room_id)
return
# Check if this is a history command
history_match = self.HISTORY_COMMAND_PATTERN.match(message)
if history_match:
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
count = int(history_match.group(1)) if history_match.group(1) else 10
# Use ollamatools if available
if OLLAMATOOLS_AVAILABLE and hasattr(handle_history_command, '__call__'):
await handle_history_command(self, count, room_id)
else:
await self._handle_history_command(count, room_id)
return
# Check if this is a topics command
if self.TOPICS_COMMAND_PATTERN.match(message):
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
# Use ollamatools if available
if OLLAMATOOLS_AVAILABLE and hasattr(handle_topics_command, '__call__'):
await handle_topics_command(self, room_id)
else:
await self._handle_topics_command(room_id)
return
# Check if this is a metrics command
if self.METRICS_COMMAND_PATTERN.match(message):
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
# Use ollamatools if available
if OLLAMATOOLS_AVAILABLE and hasattr(handle_metrics_command, '__call__'):
await handle_metrics_command(self, room_id)
else:
await self._handle_metrics_command(room_id)
return
# Unknown command
await self.send_message(room_id, f"Unknown command: {message}\nType /help for available commands.")
async def _change_model(self, model_name: str, room_id: str, sender: str) -> None:
"""
Change the Ollama model for a user.
Args:
model_name: The name of the model to change to
room_id: The ID of the room where the command was sent
sender: The Matrix user ID of the sender
"""
# Check if the model is available
if model_name not in self.available_models:
await self.send_message(
room_id,
f"Model '{model_name}' is not available. Available models: {', '.join(self.available_models)}"
)
return
# Change the model for the user
self.user_models[sender] = model_name
# Clear the user's context
for key in list(self.user_contexts.keys()):
if key.startswith(f"{sender}:"):
self.user_contexts[key] = []
await self.send_message(room_id, f"Changed model to '{model_name}' for {sender}")
logger.info(f"Changed model to '{model_name}' for user {sender}")
async def _send_help(self, room_id: str) -> None:
"""
Send help information to a room.
Args:
room_id: The ID of the room to send the help to
"""
# Build help text based on enabled features
commands = [
"- /help - Show this help message",
"- /models - List available models",
"- /model <model_name> - Change to a different model",
"- /ping - Test if the bot is responsive"
]
# Add search command if enabled
if self.enable_search:
commands.append("- /search <query> - Search the web for information")
# Add conversation tracking commands if enabled
if self.enable_conversation_tracking:
commands.append("- /history [count] - Show recent conversation history")
commands.append("- /topics - Show conversation topic analysis")
commands.append("- /metrics - Show performance metrics")
help_text = """
Available commands:
{commands}
You can also mention me by my username in your message and I'll respond.
For example: "Hey {username}, what's the weather like today?"
""".format(
commands="\n".join(commands),
username=self.user_id.split(':')[0].lstrip('@')
)
await self.send_message(room_id, help_text.strip())
async def _list_models(self, room_id: str) -> None:
"""
List available models to a room.
Args:
room_id: The ID of the room to send the list to
"""
models_text = f"Available models: {', '.join(self.available_models)}"
await self.send_message(room_id, models_text)
def _get_default_opensearch_config(self) -> Dict[str, Any]:
"""
Get the default OpenSearchNG configuration.
Returns:
Dict[str, Any]: The default configuration
"""
return {
"BASE_URL": "http://localhost:8081",
"ENGINES": ["google", "bing", "duckduckgo"],
"NUM_RESULTS": 5
}
def _load_conversation_history(self) -> List[Dict[str, Any]]:
"""
Load conversation history from file.
Returns:
List[Dict[str, Any]]: The conversation history
"""
if os.path.exists(self.history_file):
try:
import json
with open(self.history_file, 'r') as f:
history = json.load(f)
logger.info(f"Loaded {len(history)} conversations from history")
return history
except Exception as e:
logger.error(f"Failed to load conversation history: {str(e)}")
return []
else:
logger.info("No conversation history file found")
return []
def _save_conversation_history(self) -> None:
"""
Save conversation history to file.
"""
if not self.enable_conversation_tracking:
return
try:
import json
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(self.history_file), exist_ok=True)
with open(self.history_file, 'w') as f:
json.dump(self.conversation_history, f, indent=2)
logger.debug(f"Saved {len(self.conversation_history)} conversations to history")
except Exception as e:
logger.error(f"Failed to save conversation history: {str(e)}")
def _record_conversation(
self,
user_message: str,
assistant_response: str,
sender: str,
room_id: str,
response_time: float
) -> None:
"""
Record a conversation in the history.
Args:
user_message: The user's message
assistant_response: The assistant's response
sender: The Matrix user ID of the sender
room_id: The ID of the room where the conversation took place
response_time: The time taken to generate the response
"""
if not self.enable_conversation_tracking:
return
import time
# Create a conversation record
conversation = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"user_message": user_message,
"assistant_response": assistant_response,
"sender": sender,
"room_id": room_id,
"response_time": response_time,
"tokens_estimated": len(user_message.split()) + len(assistant_response.split())
}
# Add to conversation history
self.conversation_history.append(conversation)
# Trim conversation history if it gets too long
if len(self.conversation_history) > self.max_conversation_history:
self.conversation_history = self.conversation_history[-self.max_conversation_history:]
# Save conversation history
self._save_conversation_history()
async def _handle_search_command(self, query: str, room_id: str) -> None:
"""
Handle the /search command.
Args:
query: The search query
room_id: The ID of the room where the command was sent
"""
if not self.enable_search:
await self.send_message(room_id, "Search functionality is not enabled.")
return
await self.send_message(room_id, f"Searching for: {query}...")
try:
# Use OpenSearchNG to get real search results
from ollamatools.plugins.integrations.opensearch.tools import opensearch_query
# Execute the search query
results = opensearch_query(query, self.opensearch_config)
if not results or not results.get("results"):
await self.send_message(room_id, f"No results found for: {query}")
return
# Format the results
formatted_results = ["Search Results:"]
for i, result in enumerate(results["results"][:5], 1):
title = result.get("title", "No title")
url = result.get("url", "No URL")
content = result.get("content", "No content")
# Truncate content if it's too long
if len(content) > 200:
content = content[:200] + "..."
formatted_results.append(f"{i}. {title}")
formatted_results.append(f" URL: {url}")
formatted_results.append(f" {content}")
formatted_results.append("")
# Send the formatted response
await self.send_message(room_id, "\n".join(formatted_results))
except Exception as e:
logger.error(f"Error executing search: {e}")
await self.send_message(room_id, f"Error executing search: {str(e)}")
async def _handle_history_command(self, count: int, room_id: str) -> None:
"""
Handle the /history command.
Args:
count: Number of recent conversations to show
room_id: The ID of the room where the command was sent
"""
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
if not self.conversation_history:
await self.send_message(room_id, "No conversation history available.")
return
# Get the most recent conversations for this room
room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await self.send_message(room_id, "No conversation history available for this room.")
return
# Get the most recent conversations
recent = room_history[-count:] if len(room_history) >= count else room_history
# Format the history
formatted_history = ["Recent Conversations:"]
for i, conv in enumerate(recent, 1):
timestamp = conv.get("timestamp", "Unknown time")
user_message = conv.get("user_message", "No message")
sender = conv.get("sender", "Unknown sender")
# Truncate message if it's too long
if len(user_message) > 100:
user_message = user_message[:100] + "..."
formatted_history.append(f"{i}. {timestamp}")
formatted_history.append(f" From: {sender}")
formatted_history.append(f" Message: {user_message}")
formatted_history.append("")
# Send the formatted history
await self.send_message(room_id, "\n".join(formatted_history))
async def _handle_topics_command(self, room_id: str) -> None:
"""
Handle the /topics command.
Args:
room_id: The ID of the room where the command was sent
"""
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
if not self.conversation_history:
await self.send_message(room_id, "No conversation history available.")
return
# Get the conversations for this room
room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await self.send_message(room_id, "No conversation history available for this room.")
return
# Analyze topics
topics = self._analyze_conversation_topics(room_history)
if not topics:
await self.send_message(room_id, "No topics found in conversation history.")
return
# Format the topics
formatted_topics = ["Conversation Topics:"]
for topic, count in sorted(topics.items(), key=lambda x: x[1], reverse=True):
formatted_topics.append(f"- {topic}: {count} mentions")
# Send the formatted topics
await self.send_message(room_id, "\n".join(formatted_topics))
async def _handle_metrics_command(self, room_id: str) -> None:
"""
Handle the /metrics command.
Args:
room_id: The ID of the room where the command was sent
"""
if not self.enable_conversation_tracking:
await self.send_message(room_id, "Conversation tracking is not enabled.")
return
if not self.conversation_history:
await self.send_message(room_id, "No conversation history available.")
return
# Get the conversations for this room
room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await self.send_message(room_id, "No conversation history available for this room.")
return
# Calculate metrics
metrics = self._get_performance_metrics(room_history)
# Format the metrics
formatted_metrics = ["Performance Metrics:"]
formatted_metrics.append(f"- Total conversations: {metrics['total_conversations']}")
formatted_metrics.append(f"- Average response time: {metrics['avg_response_time']:.2f}s")
formatted_metrics.append(f"- Estimated total tokens: {metrics['total_tokens_estimated']}")
formatted_metrics.append(f"- Average tokens per conversation: {metrics['avg_tokens_per_conversation']:.1f}")
if metrics.get("slowest_response"):
formatted_metrics.append(f"- Slowest response: {metrics['slowest_response']['time']:.2f}s for '{metrics['slowest_response']['message']}'")
if metrics.get("fastest_response"):
formatted_metrics.append(f"- Fastest response: {metrics['fastest_response']['time']:.2f}s for '{metrics['fastest_response']['message']}'")
# Send the formatted metrics
await self.send_message(room_id, "\n".join(formatted_metrics))
def _analyze_conversation_topics(self, conversations: List[Dict[str, Any]]) -> Dict[str, int]:
"""
Analyze common topics in conversation history.
Args:
conversations: List of conversation records
Returns:
Dict[str, int]: Dictionary of topics and their frequency
"""
if not conversations:
return {}
# Simple keyword-based topic analysis
topics = {
"help": 0,
"search": 0,
"file": 0,
"command": 0,
"code": 0,
"error": 0,
"issue": 0,
"question": 0,
"other": 0
}
for conv in conversations:
message = conv.get("user_message", "").lower()
# Check for keywords
if "help" in message or "how" in message:
topics["help"] += 1
if "search" in message or "find" in message or "look" in message:
topics["search"] += 1
if "file" in message or "read" in message or "write" in message:
topics["file"] += 1
if "command" in message or "run" in message or "execute" in message:
topics["command"] += 1
if "code" in message or "program" in message or "script" in message:
topics["code"] += 1
if "error" in message or "bug" in message or "fix" in message:
topics["error"] += 1
if "issue" in message or "problem" in message:
topics["issue"] += 1
if "?" in message:
topics["question"] += 1
# If no keywords matched, count as other
if not any(count > 0 for count in topics.values()):
topics["other"] += 1
# Remove topics with zero count
return {k: v for k, v in topics.items() if v > 0}
def _get_performance_metrics(self, conversations: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Get performance metrics from conversation history.
Args:
conversations: List of conversation records
Returns:
Dict[str, Any]: Performance metrics
"""
if not conversations:
return {"error": "No conversation history available"}
# Calculate metrics
total_conversations = len(conversations)
total_response_time = sum(conv.get("response_time", 0) for conv in conversations)
avg_response_time = total_response_time / total_conversations if total_conversations > 0 else 0
# Calculate token usage
total_tokens = sum(conv.get("tokens_estimated", 0) for conv in conversations)
avg_tokens_per_conversation = total_tokens / total_conversations if total_conversations > 0 else 0
# Find slowest and fastest responses
slowest = max(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None
fastest = min(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None
return {
"total_conversations": total_conversations,
"total_response_time": total_response_time,
"avg_response_time": avg_response_time,
"total_tokens_estimated": total_tokens,
"avg_tokens_per_conversation": avg_tokens_per_conversation,
"slowest_response": {
"time": slowest.get("response_time", 0) if slowest else 0,
"message": slowest.get("user_message", "")[:50] + "..." if slowest and len(slowest.get("user_message", "")) > 50 else slowest.get("user_message", "") if slowest else ""
},
"fastest_response": {
"time": fastest.get("response_time", 0) if fastest else 0,
"message": fastest.get("user_message", "")[:50] + "..." if fastest and len(fastest.get("user_message", "")) > 50 else fastest.get("user_message", "") if fastest else ""
}
}
async def listen(self) -> None:
"""
Start listening for messages in joined rooms.
This method will block until the client is stopped.
"""
# Check if we're using ollamatools client
if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'listen'):
try:
await self.client.listen()
except Exception as e:
logger.error(f"Error listening for messages with ollamatools: {e}")
logger.info("Falling back to standard listening")
# Fall back to standard listening
else:
# Standard listening
# Set up the message callback
self.client.add_event_callback(self._message_callback, RoomMessageText)
# Start syncing with the server
await self.client.sync_forever(timeout=30000, full_state=True)
async def disconnect(self) -> None:
"""
Disconnect from the Matrix homeserver.
"""
if not self.client:
logger.error("Not connected to Matrix server")
return
# Check if we're using ollamatools client
if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'disconnect'):
try:
await self.client.disconnect()
except Exception as e:
logger.error(f"Error disconnecting with ollamatools: {e}")
logger.info("Falling back to standard disconnection")
# Fall back to standard disconnection
else:
# Standard disconnection
try:
await self.client.logout()
await self.client.close()
except Exception as e:
logger.error(f"Error disconnecting from Matrix server: {e}")