From d4dd9a41338d3f7accc1589855cf4bbeadf00ca0 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Tue, 4 Nov 2025 23:40:41 +0000 Subject: [PATCH] Improve security, error handling, and code quality MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit addresses several high-impact improvements to the MindMapper codebase: ## Security Improvements - Remove hardcoded empty API keys from agents.py and main.py - Use environment variables (OPENAI_API_KEY, ANTHROPIC_API_KEY, GOOGLE_API_KEY) - Add warnings when API keys are not set - Update README with comprehensive security best practices ## Error Handling - Add comprehensive try-catch blocks to all API calls - Handle timeout errors for network requests (with 10s timeout) - Catch specific OpenAI API errors (AuthenticationError, RateLimitError, APIError) - Catch Anthropic API errors with proper error messages - Add JSON parsing error handling for Knowledge Graph API - Provide informative error messages to users ## Code Quality - Add type hints to key functions across agents.py, main.py, and simulation.py - Import typing module (List, Dict, Optional, Any) for better type safety - Add docstrings to API methods for better documentation - Improve code maintainability and IDE support ## Documentation - Update README.md with clear API key setup instructions - Add environment variable configuration examples for Unix/Windows - Document .env file approach - Add links to API key providers - Include security best practices section All changes follow the CLAUDE.md guidelines: no breaking changes, focused improvements, and enhanced safety. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- README.md | 38 ++++++++++++--- agents.py | 130 +++++++++++++++++++++++++++++++++++--------------- main.py | 108 ++++++++++++++++++++++++++--------------- simulation.py | 12 +++-- 4 files changed, 198 insertions(+), 90 deletions(-) diff --git a/README.md b/README.md index 8f2f33d..1fb13aa 100644 --- a/README.md +++ b/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 diff --git a/agents.py b/agents.py index 3cc0cd4..2013ce0 100644 --- a/agents.py +++ b/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,7 +29,7 @@ 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 @@ -36,9 +37,13 @@ def __init__(self, id, goal, name=None, role=None, **kwargs): # Add **kwargs he 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) diff --git a/main.py b/main.py index f060d41..672deaf 100644 --- a/main.py +++ b/main.py @@ -1,5 +1,6 @@ import os import random +from typing import List, Dict, Optional, Any from textblob import TextBlob from yake import KeywordExtractor from enum import Enum @@ -8,12 +9,9 @@ import requests import anthropic -import os - -# Set the Anthropic API key -os.environ["ANTHROPIC_API_KEY"] = "" - -# Rest of the code... +# Check for Anthropic API key in environment +if not os.environ.get("ANTHROPIC_API_KEY"): + print("Warning: ANTHROPIC_API_KEY environment variable not set") class ThoughtNode: def __init__(self, content, parent=None): @@ -35,7 +33,7 @@ def analyze(self, text): return blob.sentiment.polarity class Agent: - def __init__(self, id, goal, name=None, role=None, **kwargs): + def __init__(self, id: int, goal: str, name: Optional[str] = None, role: Optional[str] = None, **kwargs): self.id = id self.goal = goal self.name = name if name else str(uuid.uuid4()) @@ -43,8 +41,13 @@ def __init__(self, id, goal, name=None, role=None, **kwargs): self.role = role self.knowledge = set() - self.client = anthropic.Client(api_key="") - self.google_api_key = "" + api_key = os.getenv("ANTHROPIC_API_KEY", "") + if not api_key: + raise ValueError("ANTHROPIC_API_KEY environment variable must be set") + self.client = anthropic.Client(api_key=api_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" @@ -65,33 +68,52 @@ def __init__(self, id, goal, name=None, role=None, **kwargs): 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_claude(self, prompt, max_tokens=50): - response = self.client.messages.create( - model="claude-3-opus-20240229", - max_tokens=max_tokens, - system="You are a helpful assistant.", - messages=[ - {"role": "user", "content": prompt}, - ], - ) - return response.content[-1].text.strip() - - def generate_task(self, goal, **kwargs): + def generate_thoughts_with_claude(self, prompt: str, max_tokens: int = 50) -> str: + """Generate thoughts using Claude API with proper error handling.""" + try: + response = self.client.messages.create( + model="claude-3-opus-20240229", + max_tokens=max_tokens, + system="You are a helpful assistant.", + messages=[ + {"role": "user", "content": prompt}, + ], + ) + return response.content[-1].text.strip() + except anthropic.APIError as e: + print(f"Error: Anthropic API error - {str(e)}") + return "" + except Exception as e: + print(f"Error: Unexpected error in Claude generation - {str(e)}") + return "" + + def generate_task(self, goal: str, **kwargs) -> Dict[str, Any]: task_dict = { "goal": goal, "steps": self.generate_task_steps(goal), @@ -103,7 +125,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]: prompt = f"Generate a list of 4 steps to accomplish the goal related to {goal}:" steps = self.generate_thoughts_with_claude(prompt, max_tokens=100) @@ -112,11 +134,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): @@ -158,14 +180,22 @@ def generate_thoughts(self, context): return thoughts - def query_claude(self, prompt): - response = self.client.completion( - prompt=f"\n\nHuman: {prompt}\n\nAssistant: ", - stop_sequences=["\n\nHuman"], - model="claude-3-opus-20240229", - max_tokens_to_sample=100, - ) - return response.completion.strip() + def query_claude(self, prompt: str) -> str: + """Query Claude API with proper error handling.""" + try: + response = self.client.completion( + prompt=f"\n\nHuman: {prompt}\n\nAssistant: ", + stop_sequences=["\n\nHuman"], + model="claude-3-opus-20240229", + max_tokens_to_sample=100, + ) + return response.completion.strip() + except anthropic.APIError as e: + print(f"Error: Anthropic API error - {str(e)}") + return "" + except Exception as e: + print(f"Error: Unexpected error in Claude query - {str(e)}") + return "" def evaluate_thoughts(self, thoughts): if len(thoughts) == 0: @@ -209,7 +239,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) diff --git a/simulation.py b/simulation.py index ba9d53d..469785e 100644 --- a/simulation.py +++ b/simulation.py @@ -1,11 +1,13 @@ +from typing import List, Callable, Optional, Any + class Simulation: - def __init__(self, agents): + def __init__(self, agents: List[Any]): self.agents = agents - self.data_log = [] - self.finished = False # Add this attribute to the Simulation class + self.data_log: List[tuple] = [] + self.finished: bool = False # Add this attribute to the Simulation class # Add this method to update the finished status when the termination condition is met - def update_finished_status(self): + def update_finished_status(self) -> None: # Implement termination condition checking logic here # For example, you can check if all agents have reached their goals all_agents_reached_goal = all([agent.is_goal() for agent in self.agents]) @@ -13,7 +15,7 @@ def update_finished_status(self): if all_agents_reached_goal: self.finished = True - def run_interactions(self, display_message): + def run_interactions(self, display_message: Optional[Callable] = None) -> None: for agent in self.agents: for other_agent in self.agents: if agent != other_agent: