Improve security, error handling, and code quality

README.md

@@ -36,9 +36,27 @@ The project consists of the following components:
    cd MindMapper
    ```
 
-3. Update the API keys in the agents.py file.
-   Make sure to replace the placeholders with your actual API keys.
-   Refer to the instructions in agents.py for details.
+3. Set up environment variables for API keys:
+
+   ```bash
+   export OPENAI_API_KEY="your-openai-api-key"
+   export ANTHROPIC_API_KEY="your-anthropic-api-key"
+   export GOOGLE_API_KEY="your-google-api-key"
+   ```
+
+   For Windows:
+   ```cmd
+   set OPENAI_API_KEY=your-openai-api-key
+   set ANTHROPIC_API_KEY=your-anthropic-api-key
+   set GOOGLE_API_KEY=your-google-api-key
+   ```
+
+   Or create a `.env` file in the project root (recommended):
+   ```
+   OPENAI_API_KEY=your-openai-api-key
+   ANTHROPIC_API_KEY=your-anthropic-api-key
+   GOOGLE_API_KEY=your-google-api-key
+   ```
 
 4. (Optional) Create a virtual environment:
 
@@ -56,13 +74,19 @@ The project consists of the following components:
 
 ## API Keys
 
-The MindMapper project requires API keys for certain functionalities. You need to obtain the following API keys and update them in the agents.py file:
+The MindMapper project requires API keys for certain functionalities. You need to obtain the following API keys:
+
+- **Google API Key**: Obtain a Google API key from the [Google Cloud Console](https://console.cloud.google.com/). Enable the Knowledge Graph Search API. This is used for querying expert information.
 
-Google API Key: Obtain a Google API key from the Google Cloud Console. It is used for agent interactions.
+- **OpenAI API Key**: Obtain an OpenAI API key from the [OpenAI website](https://platform.openai.com/). This is used for generating context-aware messages in `agents.py`.
 
-OpenAI API Key: Obtain an OpenAI API key from the OpenAI website. It is used for generating context-aware messages.
+- **Anthropic API Key**: Obtain an Anthropic API key from [Anthropic](https://console.anthropic.com/). This is used for Claude-based thought generation in `main.py`.
 
-Make sure to keep your API keys secure and avoid sharing them publicly.
+**Security Best Practices:**
+- Never commit API keys to version control
+- Use environment variables or a `.env` file to store keys
+- Keep your API keys secure and avoid sharing them publicly
+- Add `.env` to your `.gitignore` file
 
 ## Usage
 

agents.py

@@ -1,6 +1,7 @@
 import os
 import openai
 import random
+from typing import List, Dict, Optional, Any
 from textblob import TextBlob
 from yake import KeywordExtractor
 from enum import Enum
@@ -28,17 +29,21 @@ def add_child(self, child_node):
         child_node.parent = self
 
 class Agent:
-    def __init__(self, id, goal, name=None, role=None, **kwargs):  # Add **kwargs here
+    def __init__(self, id: int, goal: str, name: Optional[str] = None, role: Optional[str] = None, **kwargs):  # Add **kwargs here
         self.id = id
         self.goal = goal
         self.name = name if name else str(uuid.uuid4())  # Updated name initialization
         self.thought_tree = ThoughtNode(content="Root Thought")
         self.role = role  # Add this line to store the
         self.knowledge = set()  # Add this line to include the 'knowledge' attribute
 
-        self.api_key = "ENTER-OPENAI-KEY"
+        self.api_key = os.getenv("OPENAI_API_KEY", "")
+        if not self.api_key:
+            print("Warning: OPENAI_API_KEY environment variable not set")
         openai.api_key = self.api_key
-        self.google_api_key = "ENTER-GOOGLE-KEY"
+        self.google_api_key = os.getenv("GOOGLE_API_KEY", "")
+        if not self.google_api_key:
+            print("Warning: GOOGLE_API_KEY environment variable not set")
 
         self.sentiment_analyzer = SentimentAnalyzer()
         self.emotional_state = "neutral"
@@ -60,37 +65,66 @@ def __init__(self, id, goal, name=None, role=None, **kwargs):  # Add **kwargs he
 
         if role == "expert" and not self.experts:
             raise ValueError("Domain not specified for 'expert' agent.")
-    def query_knowledge_graph_api(self, query: str, api_key: str):
+    def query_knowledge_graph_api(self, query: str, api_key: str) -> List[Dict[str, Any]]:
+        """Query the Google Knowledge Graph API with proper error handling."""
+        if not api_key:
+            print("Error: Google API key not provided")
+            return []
+
         url = f'https://kgsearch.googleapis.com/v1/entities:search?query={query}&key={api_key}&limit=10&indent=True'
         headers = {
             'Accept': 'application/json',
         }
 
-        response = requests.get(url, headers=headers)
-
-        if response.status_code == 200:
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            response.raise_for_status()
             data = response.json()
             return data.get('itemListElement', [])
-        else:
-            print(f"Error: {response.status_code} - {response.text}")
+        except requests.exceptions.Timeout:
+            print(f"Error: Request timed out for query '{query}'")
+            return []
+        except requests.exceptions.RequestException as e:
+            print(f"Error: Network request failed - {str(e)}")
+            return []
+        except ValueError as e:
+            print(f"Error: Failed to parse JSON response - {str(e)}")
             return []
 
 
-    def generate_thoughts_with_gpt3(self, prompt, n=1, max_token=50):
-        openai.api_key = os.getenv("OPENAI_API_KEY")
-        response = openai.Completion.create(
-            engine="text-davinci-003",
-            prompt=prompt,
-            n=n,
-            max_tokens=max_token,
-            temperature=0.7,
-            top_p=1,
-            frequency_penalty=0
-        )
-        return [choice.text.strip() for choice in response.choices]
+    def generate_thoughts_with_gpt3(self, prompt: str, n: int = 1, max_token: int = 50) -> List[str]:
+        """Generate thoughts using GPT-3 with proper error handling."""
+        if not self.api_key:
+            print("Error: OpenAI API key not set")
+            return []
+
+        openai.api_key = self.api_key
+        try:
+            response = openai.Completion.create(
+                engine="text-davinci-003",
+                prompt=prompt,
+                n=n,
+                max_tokens=max_token,
+                temperature=0.7,
+                top_p=1,
+                frequency_penalty=0
+            )
+            return [choice.text.strip() for choice in response.choices]
+        except openai.error.AuthenticationError:
+            print("Error: Invalid OpenAI API key")
+            return []
+        except openai.error.RateLimitError:
+            print("Error: OpenAI API rate limit exceeded")
+            return []
+        except openai.error.APIError as e:
+            print(f"Error: OpenAI API error - {str(e)}")
+            return []
+        except Exception as e:
+            print(f"Error: Unexpected error in GPT-3 generation - {str(e)}")
+            return []
 
     # Add this method to the Agent class to generate tasks dynamically based on the goal and kwargs
-    def generate_task(self, goal, **kwargs):
+    def generate_task(self, goal: str, **kwargs) -> Dict[str, Any]:
         task_dict = {
             "goal": goal,
             "steps": self.generate_task_steps(goal),  # Generate steps for the task
@@ -102,7 +136,7 @@ def generate_task(self, goal, **kwargs):
 
         return task_dict
 
-    def generate_task_steps(self, goal):
+    def generate_task_steps(self, goal: str) -> List[str]:
         # Use GPT-3 to generate the steps based on user input
         prompt = f"Generate a list of 4 steps to accomplish the goal related to {goal}:"
         steps = self.generate_thoughts_with_gpt3(prompt, n=1, max_token=100)
@@ -112,11 +146,11 @@ def generate_task_steps(self, goal):
         else:
             return []
 
-    def learn_skill(self):
+    def learn_skill(self) -> str:
         available_skills = ['A', 'B', 'C', 'D']
         return random.choice(available_skills)
 
-    def is_goal(self):
+    def is_goal(self) -> bool:
         return self.goal in self.knowledge
 
     def observe(self, other_agent):
@@ -160,19 +194,37 @@ def generate_thoughts(self, context):
 
         return thoughts
 
-    def query_openai_api(self, prompt):
-        response = openai.Completion.create(
-            engine="davinci-codex",
-            prompt=prompt,
-            max_tokens=100,
-            n=1,
-            stop=None,
-            temperature=0.5,
-            top_p=1,
-            frequency_penalty=0,
-            presence_penalty=0,
-        )
-        return response.choices[0].text.strip()
+    def query_openai_api(self, prompt: str) -> str:
+        """Query OpenAI API with proper error handling."""
+        if not self.api_key:
+            print("Error: OpenAI API key not set")
+            return ""
+
+        try:
+            response = openai.Completion.create(
+                engine="davinci-codex",
+                prompt=prompt,
+                max_tokens=100,
+                n=1,
+                stop=None,
+                temperature=0.5,
+                top_p=1,
+                frequency_penalty=0,
+                presence_penalty=0,
+            )
+            return response.choices[0].text.strip()
+        except openai.error.AuthenticationError:
+            print("Error: Invalid OpenAI API key")
+            return ""
+        except openai.error.RateLimitError:
+            print("Error: OpenAI API rate limit exceeded")
+            return ""
+        except openai.error.APIError as e:
+            print(f"Error: OpenAI API error - {str(e)}")
+            return ""
+        except Exception as e:
+            print(f"Error: Unexpected error in OpenAI query - {str(e)}")
+            return ""
 
     def evaluate_thoughts(self, thoughts):
         if len(thoughts) == 0:
@@ -216,7 +268,7 @@ def generate_context_aware_message(self, other_agent, message):
     def cooperate(self, other_agent):
         ...
 
-    def generate_experts(self, domain, num_experts=5):
+    def generate_experts(self, domain: str, num_experts: int = 5) -> List[str]:
         query = f"{domain} experts"
         response = self.query_knowledge_graph_api(query=query, api_key=self.google_api_key)