| … | … | @@ -905,3 +905,396 @@ For example: "Hey {username}, what's the weather like today?" |
| 905 | 905 | """ |
| 906 | 906 | models_text = f"Available models: {', '.join(self.available_models)}" |
| 907 | 907 | await self.send_message(room_id, models_text) |
| 908 | |
| 909 | def _get_default_opensearch_config(self) -> Dict[str, Any]: |
| 910 | """ |
| 911 | Get the default OpenSearchNG configuration. |
| 912 | |
| 913 | Returns: |
| 914 | Dict[str, Any]: The default configuration |
| 915 | """ |
| 916 | return { |
| 917 | "BASE_URL": "http://localhost:8081", |
| 918 | "ENGINES": ["google", "bing", "duckduckgo"], |
| 919 | "NUM_RESULTS": 5 |
| 920 | } |
| 921 | |
| 922 | def _load_conversation_history(self) -> List[Dict[str, Any]]: |
| 923 | """ |
| 924 | Load conversation history from file. |
| 925 | |
| 926 | Returns: |
| 927 | List[Dict[str, Any]]: The conversation history |
| 928 | """ |
| 929 | if os.path.exists(self.history_file): |
| 930 | try: |
| 931 | import json |
| 932 | with open(self.history_file, 'r') as f: |
| 933 | history = json.load(f) |
| 934 | logger.info(f"Loaded {len(history)} conversations from history") |
| 935 | return history |
| 936 | except Exception as e: |
| 937 | logger.error(f"Failed to load conversation history: {str(e)}") |
| 938 | return [] |
| 939 | else: |
| 940 | logger.info("No conversation history file found") |
| 941 | return [] |
| 942 | |
| 943 | def _save_conversation_history(self) -> None: |
| 944 | """ |
| 945 | Save conversation history to file. |
| 946 | """ |
| 947 | if not self.enable_conversation_tracking: |
| 948 | return |
| 949 | |
| 950 | try: |
| 951 | import json |
| 952 | # Create directory if it doesn't exist |
| 953 | os.makedirs(os.path.dirname(self.history_file), exist_ok=True) |
| 954 | |
| 955 | with open(self.history_file, 'w') as f: |
| 956 | json.dump(self.conversation_history, f, indent=2) |
| 957 | logger.debug(f"Saved {len(self.conversation_history)} conversations to history") |
| 958 | except Exception as e: |
| 959 | logger.error(f"Failed to save conversation history: {str(e)}") |
| 960 | |
| 961 | def _record_conversation( |
| 962 | self, |
| 963 | user_message: str, |
| 964 | assistant_response: str, |
| 965 | sender: str, |
| 966 | room_id: str, |
| 967 | response_time: float |
| 968 | ) -> None: |
| 969 | """ |
| 970 | Record a conversation in the history. |
| 971 | |
| 972 | Args: |
| 973 | user_message: The user's message |
| 974 | assistant_response: The assistant's response |
| 975 | sender: The Matrix user ID of the sender |
| 976 | room_id: The ID of the room where the conversation took place |
| 977 | response_time: The time taken to generate the response |
| 978 | """ |
| 979 | if not self.enable_conversation_tracking: |
| 980 | return |
| 981 | |
| 982 | import time |
| 983 | |
| 984 | # Create a conversation record |
| 985 | conversation = { |
| 986 | "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), |
| 987 | "user_message": user_message, |
| 988 | "assistant_response": assistant_response, |
| 989 | "sender": sender, |
| 990 | "room_id": room_id, |
| 991 | "response_time": response_time, |
| 992 | "tokens_estimated": len(user_message.split()) + len(assistant_response.split()) |
| 993 | } |
| 994 | |
| 995 | # Add to conversation history |
| 996 | self.conversation_history.append(conversation) |
| 997 | |
| 998 | # Trim conversation history if it gets too long |
| 999 | if len(self.conversation_history) > self.max_conversation_history: |
| 1000 | self.conversation_history = self.conversation_history[-self.max_conversation_history:] |
| 1001 | |
| 1002 | # Save conversation history |
| 1003 | self._save_conversation_history() |
| 1004 | |
| 1005 | async def _handle_search_command(self, query: str, room_id: str) -> None: |
| 1006 | """ |
| 1007 | Handle the /search command. |
| 1008 | |
| 1009 | Args: |
| 1010 | query: The search query |
| 1011 | room_id: The ID of the room where the command was sent |
| 1012 | """ |
| 1013 | if not self.enable_search: |
| 1014 | await self.send_message(room_id, "Search functionality is not enabled.") |
| 1015 | return |
| 1016 | |
| 1017 | await self.send_message(room_id, f"Searching for: {query}...") |
| 1018 | |
| 1019 | try: |
| 1020 | # Simple implementation that uses Ollama to generate a response |
| 1021 | # In a real implementation, this would use OpenSearchNG |
| 1022 | search_prompt = f"Please provide a brief search result for the query: {query}" |
| 1023 | response = await self.process_with_ollama(search_prompt, None, room_id) |
| 1024 | |
| 1025 | # Format the response |
| 1026 | formatted_response = f"Search Results for '{query}':\n\n{response}" |
| 1027 | |
| 1028 | # Send the formatted response |
| 1029 | await self.send_message(room_id, formatted_response) |
| 1030 | |
| 1031 | except Exception as e: |
| 1032 | logger.error(f"Error executing search: {e}") |
| 1033 | await self.send_message(room_id, f"Error executing search: {str(e)}") |
| 1034 | |
| 1035 | async def _handle_history_command(self, count: int, room_id: str) -> None: |
| 1036 | """ |
| 1037 | Handle the /history command. |
| 1038 | |
| 1039 | Args: |
| 1040 | count: Number of recent conversations to show |
| 1041 | room_id: The ID of the room where the command was sent |
| 1042 | """ |
| 1043 | if not self.enable_conversation_tracking: |
| 1044 | await self.send_message(room_id, "Conversation tracking is not enabled.") |
| 1045 | return |
| 1046 | |
| 1047 | if not self.conversation_history: |
| 1048 | await self.send_message(room_id, "No conversation history available.") |
| 1049 | return |
| 1050 | |
| 1051 | # Get the most recent conversations for this room |
| 1052 | room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id] |
| 1053 | |
| 1054 | if not room_history: |
| 1055 | await self.send_message(room_id, "No conversation history available for this room.") |
| 1056 | return |
| 1057 | |
| 1058 | # Get the most recent conversations |
| 1059 | recent = room_history[-count:] if len(room_history) >= count else room_history |
| 1060 | |
| 1061 | # Format the history |
| 1062 | formatted_history = ["Recent Conversations:"] |
| 1063 | |
| 1064 | for i, conv in enumerate(recent, 1): |
| 1065 | timestamp = conv.get("timestamp", "Unknown time") |
| 1066 | user_message = conv.get("user_message", "No message") |
| 1067 | sender = conv.get("sender", "Unknown sender") |
| 1068 | |
| 1069 | # Truncate message if it's too long |
| 1070 | if len(user_message) > 100: |
| 1071 | user_message = user_message[:100] + "..." |
| 1072 | |
| 1073 | formatted_history.append(f"{i}. {timestamp}") |
| 1074 | formatted_history.append(f" From: {sender}") |
| 1075 | formatted_history.append(f" Message: {user_message}") |
| 1076 | formatted_history.append("") |
| 1077 | |
| 1078 | # Send the formatted history |
| 1079 | await self.send_message(room_id, "\n".join(formatted_history)) |
| 1080 | |
| 1081 | async def _handle_topics_command(self, room_id: str) -> None: |
| 1082 | """ |
| 1083 | Handle the /topics command. |
| 1084 | |
| 1085 | Args: |
| 1086 | room_id: The ID of the room where the command was sent |
| 1087 | """ |
| 1088 | if not self.enable_conversation_tracking: |
| 1089 | await self.send_message(room_id, "Conversation tracking is not enabled.") |
| 1090 | return |
| 1091 | |
| 1092 | if not self.conversation_history: |
| 1093 | await self.send_message(room_id, "No conversation history available.") |
| 1094 | return |
| 1095 | |
| 1096 | # Get the conversations for this room |
| 1097 | room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id] |
| 1098 | |
| 1099 | if not room_history: |
| 1100 | await self.send_message(room_id, "No conversation history available for this room.") |
| 1101 | return |
| 1102 | |
| 1103 | # Analyze topics |
| 1104 | topics = self._analyze_conversation_topics(room_history) |
| 1105 | |
| 1106 | if not topics: |
| 1107 | await self.send_message(room_id, "No topics found in conversation history.") |
| 1108 | return |
| 1109 | |
| 1110 | # Format the topics |
| 1111 | formatted_topics = ["Conversation Topics:"] |
| 1112 | |
| 1113 | for topic, count in sorted(topics.items(), key=lambda x: x[1], reverse=True): |
| 1114 | formatted_topics.append(f"- {topic}: {count} mentions") |
| 1115 | |
| 1116 | # Send the formatted topics |
| 1117 | await self.send_message(room_id, "\n".join(formatted_topics)) |
| 1118 | |
| 1119 | async def _handle_metrics_command(self, room_id: str) -> None: |
| 1120 | """ |
| 1121 | Handle the /metrics command. |
| 1122 | |
| 1123 | Args: |
| 1124 | room_id: The ID of the room where the command was sent |
| 1125 | """ |
| 1126 | if not self.enable_conversation_tracking: |
| 1127 | await self.send_message(room_id, "Conversation tracking is not enabled.") |
| 1128 | return |
| 1129 | |
| 1130 | if not self.conversation_history: |
| 1131 | await self.send_message(room_id, "No conversation history available.") |
| 1132 | return |
| 1133 | |
| 1134 | # Get the conversations for this room |
| 1135 | room_history = [conv for conv in self.conversation_history if conv.get("room_id") == room_id] |
| 1136 | |
| 1137 | if not room_history: |
| 1138 | await self.send_message(room_id, "No conversation history available for this room.") |
| 1139 | return |
| 1140 | |
| 1141 | # Calculate metrics |
| 1142 | metrics = self._get_performance_metrics(room_history) |
| 1143 | |
| 1144 | # Format the metrics |
| 1145 | formatted_metrics = ["Performance Metrics:"] |
| 1146 | formatted_metrics.append(f"- Total conversations: {metrics['total_conversations']}") |
| 1147 | formatted_metrics.append(f"- Average response time: {metrics['avg_response_time']:.2f}s") |
| 1148 | formatted_metrics.append(f"- Estimated total tokens: {metrics['total_tokens_estimated']}") |
| 1149 | formatted_metrics.append(f"- Average tokens per conversation: {metrics['avg_tokens_per_conversation']:.1f}") |
| 1150 | |
| 1151 | if metrics.get("slowest_response"): |
| 1152 | formatted_metrics.append(f"- Slowest response: {metrics['slowest_response']['time']:.2f}s for '{metrics['slowest_response']['message']}'") |
| 1153 | |
| 1154 | if metrics.get("fastest_response"): |
| 1155 | formatted_metrics.append(f"- Fastest response: {metrics['fastest_response']['time']:.2f}s for '{metrics['fastest_response']['message']}'") |
| 1156 | |
| 1157 | # Send the formatted metrics |
| 1158 | await self.send_message(room_id, "\n".join(formatted_metrics)) |
| 1159 | |
| 1160 | def _analyze_conversation_topics(self, conversations: List[Dict[str, Any]]) -> Dict[str, int]: |
| 1161 | """ |
| 1162 | Analyze common topics in conversation history. |
| 1163 | |
| 1164 | Args: |
| 1165 | conversations: List of conversation records |
| 1166 | |
| 1167 | Returns: |
| 1168 | Dict[str, int]: Dictionary of topics and their frequency |
| 1169 | """ |
| 1170 | if not conversations: |
| 1171 | return {} |
| 1172 | |
| 1173 | # Simple keyword-based topic analysis |
| 1174 | topics = { |
| 1175 | "help": 0, |
| 1176 | "search": 0, |
| 1177 | "file": 0, |
| 1178 | "command": 0, |
| 1179 | "code": 0, |
| 1180 | "error": 0, |
| 1181 | "issue": 0, |
| 1182 | "question": 0, |
| 1183 | "other": 0 |
| 1184 | } |
| 1185 | |
| 1186 | for conv in conversations: |
| 1187 | message = conv.get("user_message", "").lower() |
| 1188 | |
| 1189 | # Check for keywords |
| 1190 | if "help" in message or "how" in message: |
| 1191 | topics["help"] += 1 |
| 1192 | if "search" in message or "find" in message or "look" in message: |
| 1193 | topics["search"] += 1 |
| 1194 | if "file" in message or "read" in message or "write" in message: |
| 1195 | topics["file"] += 1 |
| 1196 | if "command" in message or "run" in message or "execute" in message: |
| 1197 | topics["command"] += 1 |
| 1198 | if "code" in message or "program" in message or "script" in message: |
| 1199 | topics["code"] += 1 |
| 1200 | if "error" in message or "bug" in message or "fix" in message: |
| 1201 | topics["error"] += 1 |
| 1202 | if "issue" in message or "problem" in message: |
| 1203 | topics["issue"] += 1 |
| 1204 | if "?" in message: |
| 1205 | topics["question"] += 1 |
| 1206 | |
| 1207 | # If no keywords matched, count as other |
| 1208 | if not any(count > 0 for count in topics.values()): |
| 1209 | topics["other"] += 1 |
| 1210 | |
| 1211 | # Remove topics with zero count |
| 1212 | return {k: v for k, v in topics.items() if v > 0} |
| 1213 | |
| 1214 | def _get_performance_metrics(self, conversations: List[Dict[str, Any]]) -> Dict[str, Any]: |
| 1215 | """ |
| 1216 | Get performance metrics from conversation history. |
| 1217 | |
| 1218 | Args: |
| 1219 | conversations: List of conversation records |
| 1220 | |
| 1221 | Returns: |
| 1222 | Dict[str, Any]: Performance metrics |
| 1223 | """ |
| 1224 | if not conversations: |
| 1225 | return {"error": "No conversation history available"} |
| 1226 | |
| 1227 | # Calculate metrics |
| 1228 | total_conversations = len(conversations) |
| 1229 | total_response_time = sum(conv.get("response_time", 0) for conv in conversations) |
| 1230 | avg_response_time = total_response_time / total_conversations if total_conversations > 0 else 0 |
| 1231 | |
| 1232 | # Calculate token usage |
| 1233 | total_tokens = sum(conv.get("tokens_estimated", 0) for conv in conversations) |
| 1234 | avg_tokens_per_conversation = total_tokens / total_conversations if total_conversations > 0 else 0 |
| 1235 | |
| 1236 | # Find slowest and fastest responses |
| 1237 | slowest = max(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None |
| 1238 | fastest = min(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None |
| 1239 | |
| 1240 | return { |
| 1241 | "total_conversations": total_conversations, |
| 1242 | "total_response_time": total_response_time, |
| 1243 | "avg_response_time": avg_response_time, |
| 1244 | "total_tokens_estimated": total_tokens, |
| 1245 | "avg_tokens_per_conversation": avg_tokens_per_conversation, |
| 1246 | "slowest_response": { |
| 1247 | "time": slowest.get("response_time", 0) if slowest else 0, |
| 1248 | "message": slowest.get("user_message", "")[:50] + "..." if slowest and len(slowest.get("user_message", "")) > 50 else slowest.get("user_message", "") if slowest else "" |
| 1249 | }, |
| 1250 | "fastest_response": { |
| 1251 | "time": fastest.get("response_time", 0) if fastest else 0, |
| 1252 | "message": fastest.get("user_message", "")[:50] + "..." if fastest and len(fastest.get("user_message", "")) > 50 else fastest.get("user_message", "") if fastest else "" |
| 1253 | } |
| 1254 | } |
| 1255 | |
| 1256 | async def listen(self) -> None: |
| 1257 | """ |
| 1258 | Start listening for messages in joined rooms. |
| 1259 | |
| 1260 | This method will block until the client is stopped. |
| 1261 | """ |
| 1262 | # Check if we're using ollamatools client |
| 1263 | if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'listen'): |
| 1264 | try: |
| 1265 | await self.client.listen() |
| 1266 | except Exception as e: |
| 1267 | logger.error(f"Error listening for messages with ollamatools: {e}") |
| 1268 | logger.info("Falling back to standard listening") |
| 1269 | # Fall back to standard listening |
| 1270 | else: |
| 1271 | # Standard listening |
| 1272 | # Set up the message callback |
| 1273 | self.client.add_event_callback(self._message_callback, RoomMessageText) |
| 1274 | |
| 1275 | # Start syncing with the server |
| 1276 | await self.client.sync_forever(timeout=30000, full_state=True) |
| 1277 | |
| 1278 | async def disconnect(self) -> None: |
| 1279 | """ |
| 1280 | Disconnect from the Matrix homeserver. |
| 1281 | """ |
| 1282 | if not self.client: |
| 1283 | logger.error("Not connected to Matrix server") |
| 1284 | return |
| 1285 | |
| 1286 | # Check if we're using ollamatools client |
| 1287 | if OLLAMATOOLS_AVAILABLE and hasattr(self.client, 'disconnect'): |
| 1288 | try: |
| 1289 | await self.client.disconnect() |
| 1290 | except Exception as e: |
| 1291 | logger.error(f"Error disconnecting with ollamatools: {e}") |
| 1292 | logger.info("Falling back to standard disconnection") |
| 1293 | # Fall back to standard disconnection |
| 1294 | else: |
| 1295 | # Standard disconnection |
| 1296 | try: |
| 1297 | await self.client.logout() |
| 1298 | await self.client.close() |
| 1299 | except Exception as e: |
| 1300 | logger.error(f"Error disconnecting from Matrix server: {e}") |