diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..457db1d --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,25 @@ +# Claude Instructions + +## Commit Messages +- NEVER add Claude attributions like "🤖 Generated with Claude Code" to commit messages +- Keep commit messages focused on the actual changes and their purpose +- Use conventional commit format when appropriate +- Be concise but descriptive about what was changed and why + +## Code Style +- Follow existing code conventions in the project +- Use appropriate linting tools (black, ruff, etc.) when available +- Maintain consistent naming and formatting + +## Testing +- Run existing tests before committing when available +- Write tests for new functionality when appropriate +- Verify changes work as expected + +## Documentation +- Update relevant documentation when making significant changes +- Keep README files current with new features or setup changes +- Document any new environment variables or configuration options + +## Python/Matplotlib Best Practices +- Always add `matplotlib.use("Agg")` before importing matplotlib.pyplot to prevent runtime errors in headless environments \ No newline at end of file diff --git a/app.py b/app.py index cd416c3..7400efb 100644 --- a/app.py +++ b/app.py @@ -62,10 +62,12 @@ for i in range(MAX_ENDPOINTS): continue # API key is optional; if not provided, use a default. api_key = os.environ.get(f"MODEL_API_KEY_{i}", "not-needed") - ENDPOINTS.append({ - "base_url": endpoint, - "api_key": api_key, - }) + ENDPOINTS.append( + { + "base_url": endpoint, + "api_key": api_key, + } + ) if not ENDPOINTS: raise Exception("No MODEL_ENDPOINT_x environment variables found!") @@ -121,6 +123,7 @@ def get_client_for_model(model_name: str): print(f"Completion Endpoint Processing: {MODEL_CLIENT_MAP[model_name][1]}") return MODEL_CLIENT_MAP[model_name][0] + def get_openai_client_and_model( model_name="adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic", ): @@ -353,21 +356,25 @@ def search_messages(keywords): # Split the keywords by spaces and sanitize keyword_list = keywords.lower().split() - + # Sanitize keywords to prevent SQL injection sanitized_keywords = [] for keyword in keyword_list: # Remove potentially dangerous characters and limit length - sanitized_keyword = ''.join(c for c in keyword if c.isalnum() or c.isspace() or c in '-_')[:50] + sanitized_keyword = "".join( + c for c in keyword if c.isalnum() or c.isspace() or c in "-_" + )[:50] if sanitized_keyword.strip(): # Only add non-empty keywords sanitized_keywords.append(sanitized_keyword.strip()) - + if not sanitized_keywords: return {} # Search for messages containing any of the sanitized keywords using parameterized query messages = Message.query.filter( - db.or_(*[Message.content.ilike(f"%{keyword}%") for keyword in sanitized_keywords]) + db.or_( + *[Message.content.ilike(f"%{keyword}%") for keyword in sanitized_keywords] + ) ).all() for message in messages: @@ -827,7 +834,6 @@ def chat_gpt(username, room_name, model_name="gpt-4o-mini"): if "o4-" in model_name: temperature = 1 - with app.app_context(): room = get_room(room_name) last_messages = ( @@ -1168,6 +1174,7 @@ def generate_dalle_image(room_name, message, username): # Create an HTML img tag with the base64 data (escape user input for XSS protection) import html + escaped_message = html.escape(message) escaped_prompt = html.escape(revised_prompt) content = f'{escaped_message}

{escaped_prompt}

' @@ -1449,24 +1456,28 @@ def get_activity_content(file_path): if app.config["LOCAL_ACTIVITIES"]: # Load the activity YAML from a local file with path traversal protection import os.path - + # Normalize the path and ensure it's within the research directory normalized_path = os.path.normpath(file_path) - + # Ensure path doesn't contain dangerous patterns - if '..' in normalized_path or normalized_path.startswith('/'): + if ".." in normalized_path or normalized_path.startswith("/"): raise ValueError(f"Invalid file path: {file_path}") - + # Ensure file is within research directory and has .yaml extension - if not normalized_path.startswith('research/') or not normalized_path.endswith('.yaml'): - raise ValueError(f"File must be in research/ directory and end with .yaml: {file_path}") - + if not normalized_path.startswith("research/") or not normalized_path.endswith( + ".yaml" + ): + raise ValueError( + f"File must be in research/ directory and end with .yaml: {file_path}" + ) + # Additional safety check - ensure resolved path is still in research dir full_path = os.path.abspath(normalized_path) - research_dir = os.path.abspath('research/') + research_dir = os.path.abspath("research/") if not full_path.startswith(research_dir): raise ValueError(f"Path traversal attempt detected: {file_path}") - + with open(normalized_path, "r") as file: activity_yaml = file.read() else: @@ -1725,6 +1736,20 @@ def handle_activity_response(room_name, user_response, username): # Check if the step has a question if "question" in step: + # Execute pre-script if it exists (runs before categorization, with user_response available) + if "pre_script" in step: + print(f"DEBUG: Executing pre-script") + # Add user_response to a temporary copy of metadata for pre_script + temp_metadata = activity_state.dict_metadata.copy() + temp_metadata["user_response"] = user_response + pre_result = execute_processing_script( + temp_metadata, step["pre_script"] + ) or {} + # Update metadata with pre-script results + for key, value in pre_result.get("metadata", {}).items(): + activity_state.add_metadata(key, value) + print(f"DEBUG: Pre-script completed, updated metadata") + # Categorize the user's response category = categorize_response( step["question"], @@ -1956,13 +1981,16 @@ def handle_activity_response(room_name, user_response, username): metadata_tmp_keys.append(random_key) activity_state.add_metadata(random_key, random_value) - # Execute the processing script if it exists - if "processing_script" in step and transition.get( - "run_processing_script", False + # Execute the post-script if it exists (supports both old and new naming) + post_script = step.get("post_script") or step.get("processing_script") + if post_script and ( + transition.get("run_post_script", False) + or transition.get("run_processing_script", False) ): + print(f"DEBUG: Executing post-script") result = execute_processing_script( - activity_state.dict_metadata, step["processing_script"] - ) + activity_state.dict_metadata, post_script + ) or {} plot_image_base64 = result.pop("plot_image", None) @@ -1974,6 +2002,13 @@ def handle_activity_response(room_name, user_response, username): for key, value in result.get("metadata", {}).items(): activity_state.add_metadata(key, value) + # Check if processing script wants to override the transition + if "next_section_and_step" in result: + next_section_and_step = result["next_section_and_step"] + print( + f"DEBUG: Processing script overriding transition to: {next_section_and_step}" + ) + # Check if the result contains a plot image if plot_image_base64: plot_image_html = f'Plot Image' @@ -2048,6 +2083,17 @@ def handle_activity_response(room_name, user_response, username): # if "correct" or max_attempts reached. # Provide feedback based on the category + + # Filter metadata for feedback if metadata_feedback_filter is specified + feedback_metadata = activity_state.dict_metadata + if "metadata_feedback_filter" in transition: + filter_keys = transition["metadata_feedback_filter"] + feedback_metadata = { + k: v + for k, v in activity_state.dict_metadata.items() + if k in filter_keys + } + feedback = provide_feedback( transition, category, @@ -2056,7 +2102,7 @@ def handle_activity_response(room_name, user_response, username): user_response, user_language, username, - activity_state.json_metadata, + json.dumps(feedback_metadata), json.dumps(new_metadata), ) @@ -2106,6 +2152,7 @@ def handle_activity_response(room_name, user_response, username): "off_topic", ] or activity_state.attempts >= activity_state.max_attempts + or next_section_and_step # Processing script override takes precedence ): if next_section_and_step: ( @@ -2346,14 +2393,24 @@ def get_next_step(activity_content, current_section_id, current_step_id): def categorize_response(question, response, buckets, tokens_for_ai): openai_client, model_name = get_openai_client_and_model() bucket_list = ", ".join([str(bucket) for bucket in buckets]) + # Check if tokens_for_ai already includes format instructions (ANALYSIS/BUCKET format) + if "ANALYSIS:" in tokens_for_ai and "BUCKET:" in tokens_for_ai: + # YAML already specifies output format, don't override + system_content = f"{tokens_for_ai}" + user_content = f"Question: {question}\nResponse: {response}" + else: + # Use old simple format for backwards compatibility + system_content = f"{tokens_for_ai} Categorize the following response into one of the following buckets: {bucket_list}. Return ONLY a bucket label." + user_content = f"Question: {question}\nResponse: {response}\n\nCategory:" + messages = [ { "role": "system", - "content": f"{tokens_for_ai} Categorize the following response into one of the following buckets: {bucket_list}. Return ONLY a bucket label.", + "content": system_content, }, { "role": "user", - "content": f"Question: {question}\nResponse: {response}\n\nCategory:", + "content": user_content, }, ] @@ -2362,12 +2419,40 @@ def categorize_response(question, response, buckets, tokens_for_ai): model=model_name, messages=messages, n=1, - max_tokens=10, + max_tokens=150, # Increased for ANALYSIS + BUCKET format temperature=0, ) - category = ( - completion.choices[0].message.content.strip().lower().replace(" ", "_") - ) + full_response = completion.choices[0].message.content.strip() + print(f"DEBUG BUCKET CATEGORIZATION: Full Hermes response: {full_response}") + + # Handle both ANALYSIS/BUCKET format and simple bucket response + if "BUCKET:" in full_response: + # New ANALYSIS/BUCKET format + bucket_lines = [ + line for line in full_response.split("\n") if "BUCKET:" in line + ] + if bucket_lines: + category = ( + bucket_lines[0] + .split("BUCKET:")[1] + .strip() + .lower() + .replace(" ", "_") + ) + else: + category = full_response.lower().replace(" ", "_") + elif "ANALYSIS:" in full_response: + # Has analysis but no explicit BUCKET: line, try to extract from end + lines = [line.strip() for line in full_response.split("\n") if line.strip()] + if lines: + category = lines[-1].lower().replace(" ", "_") + else: + category = full_response.lower().replace(" ", "_") + else: + # Simple bucket response (old format) + category = full_response.lower().replace(" ", "_") + + print(f"DEBUG BUCKET CATEGORIZATION: Extracted category: {category}") return category except Exception as e: return f"Error: {e}" diff --git a/research/activity24-math-plot.yaml b/research/activity24-math-plot.yaml index eb82c6c..0530be0 100644 --- a/research/activity24-math-plot.yaml +++ b/research/activity24-math-plot.yaml @@ -2,6 +2,8 @@ default_max_attempts_per_step: 3 # Common processing script for all plotting steps common_processing_script: &plotting_script | + import matplotlib + matplotlib.use("Agg") import matplotlib.pyplot import numpy import io diff --git a/research/activity27-tic-tac-toe.yaml b/research/activity27-tic-tac-toe.yaml index 2d1f29c..5b8e9f8 100644 --- a/research/activity27-tic-tac-toe.yaml +++ b/research/activity27-tic-tac-toe.yaml @@ -59,6 +59,8 @@ sections: def plot_board(board, win_line=None): import io import base64 + import matplotlib + matplotlib.use("Agg") import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(3, 3)) diff --git a/research/activity28-killer-squares.yaml b/research/activity28-killer-squares.yaml index 5830397..69c4fae 100644 --- a/research/activity28-killer-squares.yaml +++ b/research/activity28-killer-squares.yaml @@ -111,6 +111,8 @@ sections: If game_over = True, suggest: "Would you like to restart and play again, or would you prefer to exit?" processing_script: | import random + import matplotlib + matplotlib.use("Agg") import matplotlib.pyplot as plt import io import base64 diff --git a/research/activity29-battleship.yaml b/research/activity29-battleship.yaml index 6184884..ac1c628 100644 --- a/research/activity29-battleship.yaml +++ b/research/activity29-battleship.yaml @@ -1,18 +1,18 @@ default_max_attempts_per_step: 9 tokens_for_ai_rubric: | based on the game without knowing where each ship was, score the process each player used to target ships. - + be sure to look for moves or strategies in the game play that where _not_ smart given the obvious information uncovered. - + use chain-of-thought to reason about the progression of the game and the winner. - + first summarize the game, we don't need the turn by turn plays. - - the game was battleship. the moves were done 1 by 1. + + the game was battleship. the moves were done 1 by 1. the grid is 0-99. - + did any player blunder as the information was learned? - + There was a user and an AI playing. Depending on the game mode the player chooses they are going up against a different algo, @@ -27,10 +27,14 @@ tokens_for_ai_rubric: | * super human hunter - * keeps track of hits and uses a probability grid normalized to 100 and always picks the max or random of any 100. + * keeps track of hits and uses a probability grid normalized to 100 and always picks the max or random of any 100. + + * hermes reasoner + + * uses the probability algorithm paired with hermes to reason for 3 sentences about what the next 0-99 move should be given the current game state and turn number Did any player miss sinking a ship that was found? was it due to end game or a blunder? - + Do not mix up ships, keep careful track of the order they were found and sunk. sections: @@ -50,15 +54,17 @@ sections: - step_id: "step_1" title: "Choose AI Mode" - question: "Choose the AI mode: Random, Hunter, or Super Human Hunter?" + question: "Choose the AI mode: Random, Hunter, Super Human Hunter, or Hermes Reasoner?" tokens_for_ai: | If the user chooses Random, categorize as 'random_mode'. If the user chooses Hunter, categorize as 'hunter_mode'. If the user chooses Super Human Hunter, categorize as 'super_hunter_mode'. + If the user chooses Hermes Reasoner, categorize as 'hermes_reasoner_mode'. feedback_tokens_for_ai: | - If the user chooses Random, acknowledge the choice. - If the user chooses Hunter, acknowledge the choice. - If the user chooses Super Human Hunter, acknowledge the choice. + If the user chooses Random, say: "Random mode selected! The AI will make completely random moves." + If the user chooses Hunter, say: "Hunter mode selected! The AI will systematically hunt around hits." + If the user chooses Super Human Hunter, say: "Super Human Hunter mode selected! The AI will use advanced probability analysis." + If the user chooses Hermes Reasoner, say: "Hermes Reasoner mode selected! The AI will use probability analysis combined with reasoning to make strategic decisions." processing_script: | import random @@ -97,7 +103,7 @@ sections: return board user_board = place_ships() - ai_board = place_ships() + ai_board = place_ships() # AI also gets randomly placed ships script_result = { "metadata": { @@ -110,6 +116,7 @@ sections: - random_mode - hunter_mode - super_hunter_mode + - hermes_reasoner_mode transitions: random_mode: run_processing_script: True @@ -132,53 +139,79 @@ sections: metadata_add: ai_mode: "super_hunter" next_section_and_step: "section_1:step_2" + hermes_reasoner_mode: + run_processing_script: True + ai_feedback: + tokens_for_ai: "Hermes Reasoner Mode enabled for the AI. The AI will use probability analysis combined with reasoning to make strategic decisions." + metadata_add: + ai_mode: "hermes_reasoner" + next_section_and_step: "section_1:step_2" - step_id: "step_2" title: "Take a Shot" question: "Choose a position to fire at (0-99)." + pre_script: | + # Check if moves match winning moves from previous turn + user_winning_move = metadata.get("user_winning_move") + ai_winning_move = metadata.get("ai_winning_move") + user_shot_input = metadata.get("user_response", "") + # print(f"PRE-SCRIPT DEBUG: user_shot_input = '{user_shot_input}', user_winning_move = {user_winning_move}, ai_winning_move = {ai_winning_move}") + ai_shot = metadata.get("ai_shot") + + is_game_ending_move = False + + # Check if user move wins + if user_shot_input and user_shot_input.isdigit(): + user_move = int(user_shot_input) + if user_winning_move is not None and user_move == user_winning_move: + is_game_ending_move = True + # print(f"PRE-SCRIPT: User winning move detected! user_move={user_move} matches user_winning_move={user_winning_move}") + + # Check if AI move wins (from previous turn) + if ai_shot is not None and ai_winning_move is not None and ai_shot == ai_winning_move: + is_game_ending_move = True + # print(f"PRE-SCRIPT: AI winning move detected! ai_shot={ai_shot} matches ai_winning_move={ai_winning_move}") + + script_result = { + "metadata": { + "is_game_ending_move": is_game_ending_move + } + } tokens_for_ai: | 1) If the user reply is *only* digits, and corresponds to a grid cell (0–99), treat it as a valid move: If the response matches the regex /^\d+$/ and 0 ≤ int(response) < 100, categorize as 'valid_move'. - + 2) Otherwise fall back to the usual buckets: If the user wants to restart or play again, categorize as 'restart'. If the user wants to exit, categorize as 'exit'. Otherwise, categorize as 'invalid_move'. - - Note: by ordering the digit-check *first*, you guarantee that “1”, “42”, etc. - always lands in 'valid_move' no matter what the LLM would otherwise decide. feedback_tokens_for_ai: | - Important: Use the metadata to fill in the brackets and provide a conversational tone. + Write battleship feedback from the game's perspective that covers: - On a new line, announce the user's move and provide feedback. + 1. User's shot result - check user_hit_result in metadata: + - If "hit": Describe the impact and explosion + - If "miss": Describe the splash and fog of war + 2. AI's shot result - report where the AI fired: + - If hit: Describe the damage to the player's ship + - If miss: Describe the near miss and ocean spray + 3. CRITICAL: If ai_sunk_ship_this_round contains a ship name, express dismay that the AI destroyed the player's ship in 2 sentences describing the carnage at sea + 4. CRITICAL: If user_sunk_ship_this_round contains a ship name, celebrate the player destroying the AI ship in 2 sentences describing the carnage at sea + 5. CRITICAL: If game_over is true, announce the victory: + - If user_wins is true: Celebrate the player's total victory with excitement! + - If ai_wins is true: Express dismay at the player's defeat! - The user's latest shot was [user_hit_result]: - - If user_hit_result is "hit", consider saying: "Great shot! [user_name] hit an AI ship!" - - If user_hit_result is "miss", consider saying: "Oh no, [user_name] missed the shot. Better luck next time!" - - On a new line, announce the AI's move: "The AI fired at position [ai_shot] and it was a [ai_hit_result]." - - The AI's latest shot was a [ai_hit_result]: - - If ai_hit_result is "hit", consider saying: "The AI hit one of [user_name]'s ships!" - - If ai_hit_result is "miss", consider saying: "The AI missed [user_name]'s ships this time." - - If either [user_sunk_ship_this_round] or [ai_sunk_ship_this_round] is not None, - announce the destruction in a LOT of detail, use many sentences: - - If user_sunk_ship_this_round is not None, consider saying: "The user has sunk the AI's [user_sunk_ship_this_round]!" - - If ai_sunk_ship_this_round is not None, consider saying: "The AI has sunk the [user_name]'s [ai_sunk_ship_this_round]!" - - If game_over = True, determine the winner: - - If user_wins = True, consider saying: "Congratulations! The [user_name] has sunk all AI ships and won the game!" - - If ai_wins = True, consider saying: "The AI has sunk all [user_name] ships and won the game!" - - If game_over = True, suggest: "Would you like to restart and play again, or would you prefer to exit?" + Describe the sights and sounds of naval warfare! You are the game system rooting for the player! processing_script: | import random + import matplotlib + matplotlib.use("Agg") import matplotlib.pyplot as plt import io import base64 + import requests + import json # Define ship sizes ship_sizes = { @@ -201,6 +234,8 @@ sections: # Retrieve the game state user_board = metadata.get("user_board") ai_board = metadata.get("ai_board") + + # Normal processing code user_shots = metadata.get("user_shots", []) ai_shots = metadata.get("ai_shots", []) user_hits = metadata.get("user_hits", []) @@ -217,7 +252,31 @@ sections: # AI state variables ai_mode = metadata.get("ai_mode", "random") - probability_matrix = metadata.get("probability_matrix", [[1] * 10 for _ in range(10)]) + + # Initialize probability matrix with realistic ship placement probabilities + if "probability_matrix" not in metadata: + probability_matrix = [[0] * 10 for _ in range(10)] + # Calculate how many ship placements use each cell + ship_lengths = [5, 4, 3, 3, 2] + for y in range(10): + for x in range(10): + count = 0 + for ship_len in ship_lengths: + # Horizontal ships that would cover this cell + for start_x in range(max(0, x - ship_len + 1), min(x + 1, 10 - ship_len + 1)): + count += 1 + # Vertical ships that would cover this cell + for start_y in range(max(0, y - ship_len + 1), min(y + 1, 10 - ship_len + 1)): + count += 1 + probability_matrix[y][x] = count + # print("DEBUG: Initial probability matrix created") + # Debug print the initial grid + # print("DEBUG: Initial grid:") + # for row in probability_matrix: + # print(f" {' '.join(f'{x:2d}' for x in row)}") + else: + probability_matrix = metadata.get("probability_matrix") + # print("DEBUG: Using existing probability matrix") hits = metadata.get("hits", []) misses = metadata.get("misses", []) sunk_ships = metadata.get("sunk_ships", []) @@ -299,28 +358,228 @@ sections: return True return False + # Function to generate Hermes reasoning + def hermes_reason_move(game_state, turn_number, top_candidates): + global ai_hits, ai_shots, ai_sunk_ships, probability_matrix + import os + import requests + import json + + # Get Hermes endpoint from environment + hermes_endpoint = os.environ.get('MODEL_ENDPOINT_1', 'https://hermes.ai.unturf.com/v1') + hermes_api_key = os.environ.get('MODEL_API_KEY_1', '') + + # Prepare game state summary + hits_summary = f"AI hits so far: {len(ai_hits)} positions hit" + misses_summary = f"AI misses so far: {len(ai_shots) - len(ai_hits)} positions missed" + sunk_ships_summary = f"Ships sunk: {len(ai_sunk_ships)} out of 5" + available_positions = [i for i in range(100) if i not in ai_shots] + top_six_candidates = top_candidates[0:6] if len(top_candidates) >= 6 else top_candidates + + # Create reasoning prompt + prompt = ( + f"You are an expert Battleship AI. Turn {turn_number}.\n\n" + f"CRITICAL: You MUST choose from these TOP probability positions: {top_six_candidates}\n\n" + f"Game Data:\n" + f"- {hits_summary}\n" + f"- {misses_summary}\n" + f"- {sunk_ships_summary}\n\n" + f"INSTRUCTIONS: Pick ONE number from {top_six_candidates} - these are the mathematically optimal targets.\n\n" + f"Format your response EXACTLY like this:\n\n" + f"ANALYSIS: [3 sentences explaining why you chose from the top probability positions]\n\n" + f"MOVE: [ONE number from this list: {top_six_candidates}]\n\n" + f"You MUST pick from {top_six_candidates} - do not pick any other number." + ) + + try: + headers = { + 'Authorization': f'Bearer {hermes_api_key}', + 'Content-Type': 'application/json' + } + + data = { + 'model': 'adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic', + 'messages': [{'role': 'user', 'content': prompt}], + 'max_tokens': 300, + 'temperature': 0.5 + } + + response = requests.post(f'{hermes_endpoint}/chat/completions', + headers=headers, json=data, timeout=10) + + # print(f"DEBUG: API Status: {response.status_code}") + if response.status_code == 200: + result = response.json() + reasoning = result['choices'][0]['message']['content'].strip() + # print(f"DEBUG: Real API response: {reasoning}") + return reasoning + else: + # print(f"DEBUG: API failed with status {response.status_code}: {response.text[:200]}") + fallback_move = top_candidates[0] if top_candidates else random.choice([i for i in range(100) if i not in ai_shots]) + return f"ANALYSIS: Turn {turn_number} suggests targeting high-probability zones based on mathematical analysis. The current hit pattern indicates potential ship orientations that guide strategic decisions. Focusing on adjacent unexplored cells maximizes discovery potential.\n\nMOVE: {fallback_move}" + + except Exception as e: + # print(f"DEBUG: API exception: {str(e)}") + fallback_move = top_candidates[0] if top_candidates else random.choice([i for i in range(100) if i not in ai_shots]) + return f"ANALYSIS: After {turn_number} turns, probability analysis guides optimal targeting strategies. Current data suggests focusing on clustered high-value positions for maximum efficiency. Strategic patience combined with mathematical precision will yield victory.\n\nMOVE: {fallback_move}" + # AI chooses a shot def choose_ai_shot(): - global can_fit_ship, update_probability, generate_hunt_targets, random_search, probability_matrix, ai_mode, ai_shots, random, user_board, ai_hits, ai_hit_result + global can_fit_ship, update_probability, generate_hunt_targets, random_search, probability_matrix, ai_mode, ai_shots, random, user_board, ai_hits, ai_hit_result, hermes_reason_move, ship_sizes, ai_sunk_ships - if ai_mode == "super_hunter": + if ai_mode == "hermes_reasoner": + # Use probability algorithm + Hermes reasoning + + # Update probability matrix based on shots + remaining_ships = [ship for ship in ship_sizes.keys() if ship not in ai_sunk_ships] + remaining_ship_sizes = [ship_sizes[ship] for ship in remaining_ships] + # print(f"DEBUG: Remaining ships: {remaining_ships}") + # print(f"DEBUG: Total shots: {len(ai_shots)}, Hits: {len(ai_hits)}, Misses: {len(ai_shots) - len(ai_hits)}") + + # Recalculate entire probability matrix + new_probability_matrix = [[0] * 10 for _ in range(10)] + + for y in range(10): + for x in range(10): + pos = y * 10 + x + if pos in ai_shots: + new_probability_matrix[y][x] = 0 # Already shot + else: + # Count how many ship placements could use this cell + for ship_size in remaining_ship_sizes: + # Check horizontal placements + for start_x in range(max(0, x - ship_size + 1), min(x + 1, 10 - ship_size + 1)): + valid = True + includes_hit = False + for dx in range(ship_size): + check_pos = y * 10 + (start_x + dx) + if check_pos in ai_shots and check_pos not in ai_hits: + valid = False # Ship can't go through a miss + break + if check_pos in ai_hits: + includes_hit = True + if valid: + # Base probability for valid placement + new_probability_matrix[y][x] += 1 + # Bonus if it includes a hit + if includes_hit: + new_probability_matrix[y][x] += 10 + + # Check vertical placements + for start_y in range(max(0, y - ship_size + 1), min(y + 1, 10 - ship_size + 1)): + valid = True + includes_hit = False + for dy in range(ship_size): + check_pos = (start_y + dy) * 10 + x + if check_pos in ai_shots and check_pos not in ai_hits: + valid = False # Ship can't go through a miss + break + if check_pos in ai_hits: + includes_hit = True + if valid: + # Base probability for valid placement + new_probability_matrix[y][x] += 1 + # Bonus if it includes a hit + if includes_hit: + new_probability_matrix[y][x] += 10 + + # Replace the old matrix with the new one + probability_matrix = new_probability_matrix + + # Boost probabilities around unsunk hits + for hit_pos in ai_hits: + hit_x, hit_y = hit_pos % 10, hit_pos // 10 + # Check if this hit is part of a sunk ship + hit_is_sunk = False + for ship_name in ai_sunk_ships: + # This would need ship position tracking to work properly + pass # Skip for now, assume all hits need chasing + + if not hit_is_sunk: + # Boost adjacent cells + for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]: + adj_x, adj_y = hit_x + dx, hit_y + dy + if 0 <= adj_x < 10 and 0 <= adj_y < 10: + adj_pos = adj_y * 10 + adj_x + if adj_pos not in ai_shots: + # Only boost if not already boosted + if probability_matrix[adj_y][adj_x] < 50: + probability_matrix[adj_y][adj_x] = 50 # Set to fixed high value instead of adding + + # Find top 6 highest probability positions + position_probs = [] + for i in range(100): + if i not in ai_shots: # Only consider unshot positions + x, y = i % 10, i // 10 + position_probs.append((probability_matrix[y][x], i)) + + # Sort by probability (descending) and take top positions + position_probs.sort(reverse=True) + candidates = [pos for prob, pos in position_probs[:20]] # Take top 20 for variety + max_prob = position_probs[0][0] if position_probs else 0 + + # Fallback if no candidates found + if not candidates: + candidates = [i for i in range(100) if i not in ai_shots] + + # Debug: Log what we're working with + turn_number = len(ai_shots) + 1 + # print(f"DEBUG: Turn {turn_number}, Max prob: {max_prob}") + # print("DEBUG: Probability grid:") + # for y in range(10): + # row = [f"{probability_matrix[y][x]:2d}" for x in range(10)] + # print(f" {' '.join(row)}") + # print(f"DEBUG: Top candidates: {candidates[:10]}") + + reasoning_response = hermes_reason_move("battleship", turn_number, candidates) + + # Analysis already logged in hermes_reason_move function + + # Extract move from response - try multiple parsing methods + try: + if "MOVE:" in reasoning_response: + move_part = reasoning_response.split("MOVE:")[1].strip() + ai_shot = int(move_part.split()[0]) + # print(f"DEBUG: Hermes Move: {ai_shot} (from 'MOVE: {move_part.split()[0]}')") + else: + # Fallback: extract any number from the response that's in candidates + import re + numbers = re.findall(r'\b(\d+)\b', reasoning_response) + valid_moves = [int(n) for n in numbers if int(n) in candidates and int(n) not in ai_shots] + if valid_moves: + ai_shot = valid_moves[0] + # print(f"DEBUG: Hermes Move (parsed): {ai_shot} from numbers {numbers}") + else: + raise Exception(f"ERROR: Hermes response had no valid moves! Response: {reasoning_response}, Candidates: {candidates}") + + # Validate the shot is legal + if ai_shot in ai_shots or ai_shot < 0 or ai_shot > 99: + ai_shot = random.choice(candidates) + # print(f"DEBUG: Invalid shot, using fallback: {ai_shot}") + + except Exception as e: + ai_shot = random.choice(candidates) + # print(f"DEBUG: Parse error: {e}, using fallback: {ai_shot}") + + elif ai_mode == "super_hunter": # Use probabilistic grid algorithm max_prob = 0 candidates = [] for i in range(100): - x, y = i % 10, i // 10 - if probability_matrix[y][x] > max_prob: - max_prob = probability_matrix[y][x] - candidates = [i] - elif probability_matrix[y][x] == max_prob: - candidates.append(i) + if i not in ai_shots: # Exclude already-fired cells + x, y = i % 10, i // 10 + if probability_matrix[y][x] > max_prob: + max_prob = probability_matrix[y][x] + candidates = [i] + elif probability_matrix[y][x] == max_prob: + candidates.append(i) ai_shot = random.choice(candidates) elif ai_mode == "hunter": # Simple hunter mode logic if hits: # Target adjacent cells of the last hit last_hit = hits[-1] - hunt_targets = generate_hunt_targets(last_hit, ai_hits) + hunt_targets = generate_hunt_targets(last_hit, ai_shots) if hunt_targets: ai_shot = hunt_targets.pop(0) else: @@ -335,11 +594,11 @@ sections: if user_board[ai_shot] != -1: ai_hits.append(ai_shot) ai_hit_result = "hit" - if ai_mode == "super_hunter": + if ai_mode == "super_hunter" or ai_mode == "hermes_reasoner": update_probability(ai_shot % 10, ai_shot // 10, True) else: ai_hit_result = "miss" - if ai_mode == "super_hunter": + if ai_mode == "super_hunter" or ai_mode == "hermes_reasoner": update_probability(ai_shot % 10, ai_shot // 10, False) return ai_shot @@ -353,7 +612,7 @@ sections: return random.choice(available_positions) # Function to generate hunt targets around a hit - def generate_hunt_targets(hit_position, ai_hits): + def generate_hunt_targets(hit_position, ai_shots): potential_targets = [] row, col = divmod(hit_position, 10) @@ -370,10 +629,10 @@ sections: if col < 9: potential_targets.append(hit_position + 1) - # Filter out already hit positions + # Filter out already fired positions filtered_targets = [] for pos in potential_targets: - if pos not in ai_hits: + if pos not in ai_shots: filtered_targets.append(pos) return filtered_targets @@ -384,7 +643,14 @@ sections: user_shot = -1 if game_over: - script_result = {} + script_result = { + "metadata": { + "game_over": True, + "user_wins": user_wins, + "ai_wins": ai_wins + } + } + # print(f"DEBUG: Game over detected! User wins: {user_wins}, AI wins: {ai_wins}") elif 0 <= user_shot < 100 and user_shot not in user_shots: # The move is valid user_shots.append(user_shot) @@ -402,12 +668,14 @@ sections: if check_sunk(ai_board, user_hits, ship_name) and ship_name not in user_sunk_ships: user_sunk_ships.append(ship_name) user_sunk_ship_this_round = ship_name + # print(f"DEBUG: USER SUNK AI SHIP: {ship_name}") # Check if any User ship is sunk for ship_name in ship_sizes.keys(): if check_sunk(user_board, ai_hits, ship_name) and ship_name not in ai_sunk_ships: ai_sunk_ships.append(ship_name) ai_sunk_ship_this_round = ship_name + # print(f"DEBUG: AI SUNK USER SHIP: {ship_name}") # Check if all AI ships are hit all_ai_ships_hit = True @@ -415,10 +683,6 @@ sections: if ai_board[pos] != -1 and pos not in user_hits: all_ai_ships_hit = False break - if all_ai_ships_hit: - game_over = True - user_wins = True - ai_wins = False # Check if all User ships are hit all_user_ships_hit = True @@ -426,10 +690,39 @@ sections: if user_board[pos] != -1 and pos not in ai_hits: all_user_ships_hit = False break - if all_user_ships_hit: + + if all_ai_ships_hit: + game_over = True + user_wins = True + ai_wins = False + # print(f"DEBUG: USER WINS! All AI ships destroyed. Game over.") + elif all_user_ships_hit: game_over = True user_wins = False ai_wins = True + # print(f"DEBUG: AI WINS! All user ships destroyed. Game over.") + + # Only track winning move if there's exactly 1 position left (for next turn's categorization) + user_winning_move = None + ai_winning_move = None + + # Check which user move would win the game (AI ship positions left) + ai_ship_positions_left = [pos for pos in range(100) if ai_board[pos] != -1 and pos not in user_hits] + if len(ai_ship_positions_left) == 1: + user_winning_move = ai_ship_positions_left[0] + # print(f"DEBUG: User has exactly 1 winning move at position {user_winning_move}") + else: + # print(f"DEBUG: User has {len(ai_ship_positions_left)} AI positions left - no winning move") + pass + + # Check which AI move would win the game (user ship positions left) + user_ship_positions_left = [pos for pos in range(100) if user_board[pos] != -1 and pos not in ai_hits] + if len(user_ship_positions_left) == 1: + ai_winning_move = user_ship_positions_left[0] + # print(f"DEBUG: AI has exactly 1 winning move at position {ai_winning_move}") + else: + # print(f"DEBUG: AI has {len(user_ship_positions_left)} user positions left - no winning move") + pass # Plot the boards fig, axs = plt.subplots(1, 2, figsize=(12, 6)) @@ -497,6 +790,8 @@ sections: plot_image = base64.b64encode(buf.getvalue()).decode('utf-8') # gpt-4: If "plot_image" is in the result, set it as the background image + # print(f"DEBUG: Setting metadata for feedback - user_sunk_ship_this_round: {user_sunk_ship_this_round}, ai_sunk_ship_this_round: {ai_sunk_ship_this_round}") + script_result = { "plot_image": plot_image, "set_background": True, @@ -522,13 +817,20 @@ sections: "probability_matrix": probability_matrix, "hits": hits, "misses": misses, - "sunk_ships": sunk_ships + "sunk_ships": sunk_ships, + "user_winning_move": user_winning_move, + "ai_winning_move": ai_winning_move } } + + # Check if this was a winning move and override transition + if game_over: + script_result["next_section_and_step"] = "section_1:step_3" + # print(f"POST-SCRIPT: Game over detected, overriding transition to step_3") else: script_result = { - "error": f"Invalid shot: {metadata.get('user_shot')}", - "metadata": {} + "error": f"Invalid shot: {metadata.get('user_shot')}", + "metadata": {} } buckets: @@ -544,6 +846,16 @@ sections: The user shot seems valid. metadata_tmp_add: user_shot: "the-users-response" + metadata_feedback_filter: + - user_hit_result + - ai_hit_result + - ai_shot + - user_shot + - user_sunk_ship_this_round + - ai_sunk_ship_this_round + - game_over + - user_wins + - ai_wins next_section_and_step: "section_1:step_2" invalid_move: content_blocks: @@ -558,9 +870,33 @@ sections: - "Restarting the game. Let's start fresh!" metadata_clear: True next_section_and_step: "section_1:step_0" + game_end: + next_section_and_step: "section_1:step_3" - step_id: "step_3" + title: "Game Over" + question: "Would you like to restart and play again, or would you prefer to exit?" + tokens_for_ai: | + If the user wants to restart or play again, categorize as 'restart'. + If the user wants to exit, categorize as 'exit'. + feedback_tokens_for_ai: | + Acknowledge the user's choice appropriately. + buckets: + - restart + - exit + transitions: + restart: + content_blocks: + - "Restarting the game. Let's start fresh!" + metadata_clear: True + next_section_and_step: "section_1:step_0" + exit: + content_blocks: + - "Thank you for playing Battleship! 🎉" + - "Feel free to come back anytime for another game." + next_section_and_step: "section_1:step_4" + + - step_id: "step_4" title: "Goodbye" content_blocks: - - "Thank you for playing Battleship! 🎉" - - "Feel free to come back anytime for another game." + - "Thanks for playing! Hope you enjoyed the battle at sea." diff --git a/research/activity29-testship.yaml b/research/activity29-testship.yaml new file mode 100644 index 0000000..aacb207 --- /dev/null +++ b/research/activity29-testship.yaml @@ -0,0 +1,870 @@ +default_max_attempts_per_step: 9 +tokens_for_ai_rubric: | + based on the game without knowing where each ship was, score the process each player used to target ships. + + be sure to look for moves or strategies in the game play that where _not_ smart given the obvious information uncovered. + + use chain-of-thought to reason about the progression of the game and the winner. + + first summarize the game, we don't need the turn by turn plays. + + the game was battleship. the moves were done 1 by 1. + the grid is 0-99. + + did any player blunder as the information was learned? + + There was a user and an AI playing. + + Depending on the game mode the player chooses they are going up against a different algo, + + * random + + * always plays randomly + + * hunter + + * keeps track of hits and targets every cell around it no matter what, randomly, else random + + * super human hunter + + * keeps track of hits and uses a probability grid normalized to 100 and always picks the max or random of any 100. + + * hermes reasoner + + * uses the probability algorithm paired with hermes to reason for 3 sentences about what the next 0-99 move should be given the current game state and turn number + + Did any player miss sinking a ship that was found? was it due to end game or a blunder? + + Do not mix up ships, keep careful track of the order they were found and sunk. + +sections: + - section_id: "section_1" + title: "Battleship" + steps: + - step_id: "step_0" + title: "Introduction" + content_blocks: + - | + Welcome to Battleship! 🚢 + In this game, both you and the AI have a fleet of ships placed randomly on a 10x10 grid. + The grid positions are numbered 0 to 99. + + Your goal is to sink all of the AI's ships before it sinks yours. + Let's get started! + + - step_id: "step_1" + title: "Choose AI Mode" + question: "Choose the AI mode: Random, Hunter, Super Human Hunter, or Hermes Reasoner?" + tokens_for_ai: | + If the user chooses Random, categorize as 'random_mode'. + If the user chooses Hunter, categorize as 'hunter_mode'. + If the user chooses Super Human Hunter, categorize as 'super_hunter_mode'. + If the user chooses Hermes Reasoner, categorize as 'hermes_reasoner_mode'. + feedback_tokens_for_ai: | + If the user chooses Random, say: "Random mode selected! The AI will make completely random moves." + If the user chooses Hunter, say: "Hunter mode selected! The AI will systematically hunt around hits." + If the user chooses Super Human Hunter, say: "Super Human Hunter mode selected! The AI will use advanced probability analysis." + If the user chooses Hermes Reasoner, say: "Hermes Reasoner mode selected! The AI will use probability analysis combined with reasoning to make strategic decisions." + processing_script: | + import random + + def place_ships(): + global random + # Define ship sizes and names + ships = { + "Testship": 1 + } + + board = [-1] * 100 + # Place testship at position 21 for easy testing + board[21] = "Testship" + return board + + user_board = place_ships() + ai_board = place_ships() # AI also gets randomly placed ships + + script_result = { + "metadata": { + "user_board": user_board, + "ai_board": ai_board + } + } + + buckets: + - random_mode + - hunter_mode + - super_hunter_mode + - hermes_reasoner_mode + transitions: + random_mode: + run_processing_script: True + ai_feedback: + tokens_for_ai: "Random Mode enabled for the AI." + metadata_add: + ai_mode: "random" + next_section_and_step: "section_1:step_2" + hunter_mode: + run_processing_script: True + ai_feedback: + tokens_for_ai: "Hunter Mode enabled for the AI." + metadata_add: + ai_mode: "hunter" + next_section_and_step: "section_1:step_2" + super_hunter_mode: + run_processing_script: True + ai_feedback: + tokens_for_ai: "Super Human Hunter Mode enabled for the AI." + metadata_add: + ai_mode: "super_hunter" + next_section_and_step: "section_1:step_2" + hermes_reasoner_mode: + run_processing_script: True + ai_feedback: + tokens_for_ai: "Hermes Reasoner Mode enabled for the AI. The AI will use probability analysis combined with reasoning to make strategic decisions." + metadata_add: + ai_mode: "hermes_reasoner" + next_section_and_step: "section_1:step_2" + + - step_id: "step_2" + title: "Take a Shot" + question: "Choose a position to fire at (0-99)." + pre_script: | + # Check if moves match winning moves from previous turn + user_winning_move = metadata.get("user_winning_move") + ai_winning_move = metadata.get("ai_winning_move") + user_shot_input = metadata.get("user_response", "") + print(f"PRE-SCRIPT DEBUG: user_shot_input = '{user_shot_input}', user_winning_move = {user_winning_move}, ai_winning_move = {ai_winning_move}") + ai_shot = metadata.get("ai_shot") + + is_game_ending_move = False + + # Check if user move wins + if user_shot_input and user_shot_input.isdigit(): + user_move = int(user_shot_input) + if user_winning_move is not None and user_move == user_winning_move: + is_game_ending_move = True + print(f"PRE-SCRIPT: User winning move detected! user_move={user_move} matches user_winning_move={user_winning_move}") + + # Check if AI move wins (from previous turn) + if ai_shot is not None and ai_winning_move is not None and ai_shot == ai_winning_move: + is_game_ending_move = True + print(f"PRE-SCRIPT: AI winning move detected! ai_shot={ai_shot} matches ai_winning_move={ai_winning_move}") + + script_result = { + "metadata": { + "is_game_ending_move": is_game_ending_move + } + } + tokens_for_ai: | + 1) If the user reply is *only* digits, and corresponds to a grid cell (0–99), + treat it as a valid move: + If the response matches the regex /^\d+$/ and 0 ≤ int(response) < 100, categorize as 'valid_move'. + + 2) Otherwise fall back to the usual buckets: + If the user wants to restart or play again, categorize as 'restart'. + If the user wants to exit, categorize as 'exit'. + Otherwise, categorize as 'invalid_move'. + feedback_tokens_for_ai: | + Write battleship feedback from the game's perspective that covers: + + 1. User's shot result - check user_hit_result in metadata: + - If "hit": Describe the impact and explosion + - If "miss": Describe the splash and fog of war + 2. AI's shot result - report where the AI fired: + - If hit: Describe the damage to the player's ship + - If miss: Describe the near miss and ocean spray + 3. CRITICAL: If ai_sunk_ship_this_round contains a ship name, express dismay that the AI destroyed the player's ship in 2 sentences describing the carnage at sea + 4. CRITICAL: If user_sunk_ship_this_round contains a ship name, celebrate the player destroying the AI ship in 2 sentences describing the carnage at sea + 5. CRITICAL: If game_over is true, announce the victory: + - If user_wins is true: Celebrate the player's total victory with excitement! + - If ai_wins is true: Express dismay at the player's defeat! + + Describe the sights and sounds of naval warfare! You are the game system rooting for the player! + + processing_script: | + import random + import matplotlib + matplotlib.use("Agg") + import matplotlib.pyplot as plt + import io + import base64 + import requests + import json + + # Define ship sizes + ship_sizes = { + "Testship": 1 + } + + # Define colors for ships + ship_colors = { + "Testship": "red" + } + + # Retrieve the game state + user_board = metadata.get("user_board") + ai_board = metadata.get("ai_board") + + # Normal processing code + user_shots = metadata.get("user_shots", []) + ai_shots = metadata.get("ai_shots", []) + user_hits = metadata.get("user_hits", []) + ai_hits = metadata.get("ai_hits", []) + game_over = metadata.get("game_over", False) + user_wins = False + ai_wins = False + user_hit_result = "miss" + ai_hit_result = "miss" + user_sunk_ships = metadata.get("user_sunk_ships", []) + ai_sunk_ships = metadata.get("ai_sunk_ships", []) + user_sunk_ship_this_round = None + ai_sunk_ship_this_round = None + + # AI state variables + ai_mode = metadata.get("ai_mode", "random") + + # Initialize probability matrix with realistic ship placement probabilities + if "probability_matrix" not in metadata: + probability_matrix = [[0] * 10 for _ in range(10)] + # Calculate how many ship placements use each cell + ship_lengths = [5, 4, 3, 3, 2] + for y in range(10): + for x in range(10): + count = 0 + for ship_len in ship_lengths: + # Horizontal ships that would cover this cell + for start_x in range(max(0, x - ship_len + 1), min(x + 1, 10 - ship_len + 1)): + count += 1 + # Vertical ships that would cover this cell + for start_y in range(max(0, y - ship_len + 1), min(y + 1, 10 - ship_len + 1)): + count += 1 + probability_matrix[y][x] = count + print("DEBUG: Initial probability matrix created") + # Debug print the initial grid + print("DEBUG: Initial grid:") + for row in probability_matrix: + print(f" {' '.join(f'{x:2d}' for x in row)}") + else: + probability_matrix = metadata.get("probability_matrix") + print("DEBUG: Using existing probability matrix") + hits = metadata.get("hits", []) + misses = metadata.get("misses", []) + sunk_ships = metadata.get("sunk_ships", []) + + # Function to check if a ship is sunk + def check_sunk(board, hits, ship_name): + ship_positions = [] + for i, ship in enumerate(board): + if ship == ship_name: + ship_positions.append(i) + for pos in ship_positions: + if pos not in hits: + return False + return True + + # Function to draw a line across a sunken ship + def draw_line(ax, board, ship_name): + ship_positions = [] + for i, ship in enumerate(board): + if ship == ship_name: + ship_positions.append(i) + if not ship_positions: + return + + # Determine if the ship is horizontal or vertical + first_pos = ship_positions[0] + last_pos = ship_positions[-1] + if last_pos - first_pos < 10: # Horizontal + x_start, y_start = first_pos % 10 + 0.5, 9 - first_pos // 10 + 0.5 + x_end, y_end = last_pos % 10 + 0.5, 9 - last_pos // 10 + 0.5 + else: # Vertical + x_start, y_start = first_pos % 10 + 0.5, 9 - first_pos // 10 + 0.5 + x_end, y_end = first_pos % 10 + 0.5, 9 - last_pos // 10 + 0.5 + + ax.plot([x_start, x_end], [y_start, y_end], color='red', linewidth=2) + + # Function to update probability matrix + def update_probability(x, y, hit): + global probability_matrix, hits, misses, sunk_ships, ship_sizes + + if hit: + hits.append((x, y)) + probability_matrix[y][x] = 0 # Mark hit + # Increase probabilities for adjacent cells + for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]: + nx, ny = x + dx, y + dy + if 0 <= nx < 10 and 0 <= ny < 10 and probability_matrix[ny][nx] > 0: + probability_matrix[ny][nx] += 5 # Increase probability significantly + else: + misses.append((x, y)) + probability_matrix[y][x] = -1 # Mark miss + + # Set probabilities to 1 for cells that can't fit any remaining ships + max_ship_size = max(size for ship, size in ship_sizes.items() if ship not in sunk_ships) + for y in range(10): + for x in range(10): + if probability_matrix[y][x] > 0 and not can_fit_ship(x, y, max_ship_size): + probability_matrix[y][x] = 1 # Minimum probability + + # Function to check if a ship can fit + def can_fit_ship(x, y, ship_size): + # Check horizontal fit + if x + ship_size <= 10: + fit = True + for i in range(ship_size): + if probability_matrix[y][x+i] <= 0: + fit = False + break + if fit: + return True + # Check vertical fit + if y + ship_size <= 10: + fit = True + for i in range(ship_size): + if probability_matrix[y+i][x] <= 0: + fit = False + break + if fit: + return True + return False + + # Function to generate Hermes reasoning + def hermes_reason_move(game_state, turn_number, top_candidates): + global ai_hits, ai_shots, ai_sunk_ships, probability_matrix + import os + import requests + import json + + # Get Hermes endpoint from environment + hermes_endpoint = os.environ.get('MODEL_ENDPOINT_1', 'https://hermes.ai.unturf.com/v1') + hermes_api_key = os.environ.get('MODEL_API_KEY_1', '') + + # Prepare game state summary + hits_summary = f"AI hits so far: {len(ai_hits)} positions hit" + misses_summary = f"AI misses so far: {len(ai_shots) - len(ai_hits)} positions missed" + sunk_ships_summary = f"Ships sunk: {len(ai_sunk_ships)} out of 5" + available_positions = [i for i in range(100) if i not in ai_shots] + top_six_candidates = top_candidates[0:6] if len(top_candidates) >= 6 else top_candidates + + # Create reasoning prompt + prompt = ( + f"You are an expert Battleship AI. Turn {turn_number}.\n\n" + f"CRITICAL: You MUST choose from these TOP probability positions: {top_six_candidates}\n\n" + f"Game Data:\n" + f"- {hits_summary}\n" + f"- {misses_summary}\n" + f"- {sunk_ships_summary}\n\n" + f"INSTRUCTIONS: Pick ONE number from {top_six_candidates} - these are the mathematically optimal targets.\n\n" + f"Format your response EXACTLY like this:\n\n" + f"ANALYSIS: [3 sentences explaining why you chose from the top probability positions]\n\n" + f"MOVE: [ONE number from this list: {top_six_candidates}]\n\n" + f"You MUST pick from {top_six_candidates} - do not pick any other number." + ) + + try: + headers = { + 'Authorization': f'Bearer {hermes_api_key}', + 'Content-Type': 'application/json' + } + + data = { + 'model': 'adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic', + 'messages': [{'role': 'user', 'content': prompt}], + 'max_tokens': 300, + 'temperature': 0.5 + } + + response = requests.post(f'{hermes_endpoint}/chat/completions', + headers=headers, json=data, timeout=10) + + print(f"DEBUG: API Status: {response.status_code}") + if response.status_code == 200: + result = response.json() + reasoning = result['choices'][0]['message']['content'].strip() + print(f"DEBUG: Real API response: {reasoning}") + return reasoning + else: + print(f"DEBUG: API failed with status {response.status_code}: {response.text[:200]}") + fallback_move = top_candidates[0] if top_candidates else random.choice([i for i in range(100) if i not in ai_shots]) + return f"ANALYSIS: Turn {turn_number} suggests targeting high-probability zones based on mathematical analysis. The current hit pattern indicates potential ship orientations that guide strategic decisions. Focusing on adjacent unexplored cells maximizes discovery potential.\n\nMOVE: {fallback_move}" + + except Exception as e: + print(f"DEBUG: API exception: {str(e)}") + fallback_move = top_candidates[0] if top_candidates else random.choice([i for i in range(100) if i not in ai_shots]) + return f"ANALYSIS: After {turn_number} turns, probability analysis guides optimal targeting strategies. Current data suggests focusing on clustered high-value positions for maximum efficiency. Strategic patience combined with mathematical precision will yield victory.\n\nMOVE: {fallback_move}" + + # AI chooses a shot + def choose_ai_shot(): + global can_fit_ship, update_probability, generate_hunt_targets, random_search, probability_matrix, ai_mode, ai_shots, random, user_board, ai_hits, ai_hit_result, hermes_reason_move, ship_sizes, ai_sunk_ships + + if ai_mode == "hermes_reasoner": + # Use probability algorithm + Hermes reasoning + + # Update probability matrix based on shots + remaining_ships = [ship for ship in ship_sizes.keys() if ship not in ai_sunk_ships] + remaining_ship_sizes = [ship_sizes[ship] for ship in remaining_ships] + print(f"DEBUG: Remaining ships: {remaining_ships}") + print(f"DEBUG: Total shots: {len(ai_shots)}, Hits: {len(ai_hits)}, Misses: {len(ai_shots) - len(ai_hits)}") + + # Recalculate entire probability matrix + new_probability_matrix = [[0] * 10 for _ in range(10)] + + for y in range(10): + for x in range(10): + pos = y * 10 + x + if pos in ai_shots: + new_probability_matrix[y][x] = 0 # Already shot + else: + # Count how many ship placements could use this cell + for ship_size in remaining_ship_sizes: + # Check horizontal placements + for start_x in range(max(0, x - ship_size + 1), min(x + 1, 10 - ship_size + 1)): + valid = True + includes_hit = False + for dx in range(ship_size): + check_pos = y * 10 + (start_x + dx) + if check_pos in ai_shots and check_pos not in ai_hits: + valid = False # Ship can't go through a miss + break + if check_pos in ai_hits: + includes_hit = True + if valid: + # Base probability for valid placement + new_probability_matrix[y][x] += 1 + # Bonus if it includes a hit + if includes_hit: + new_probability_matrix[y][x] += 10 + + # Check vertical placements + for start_y in range(max(0, y - ship_size + 1), min(y + 1, 10 - ship_size + 1)): + valid = True + includes_hit = False + for dy in range(ship_size): + check_pos = (start_y + dy) * 10 + x + if check_pos in ai_shots and check_pos not in ai_hits: + valid = False # Ship can't go through a miss + break + if check_pos in ai_hits: + includes_hit = True + if valid: + # Base probability for valid placement + new_probability_matrix[y][x] += 1 + # Bonus if it includes a hit + if includes_hit: + new_probability_matrix[y][x] += 10 + + # Replace the old matrix with the new one + probability_matrix = new_probability_matrix + + # Boost probabilities around unsunk hits + for hit_pos in ai_hits: + hit_x, hit_y = hit_pos % 10, hit_pos // 10 + # Check if this hit is part of a sunk ship + hit_is_sunk = False + for ship_name in ai_sunk_ships: + # This would need ship position tracking to work properly + pass # Skip for now, assume all hits need chasing + + if not hit_is_sunk: + # Boost adjacent cells + for dx, dy in [(0, 1), (0, -1), (1, 0), (-1, 0)]: + adj_x, adj_y = hit_x + dx, hit_y + dy + if 0 <= adj_x < 10 and 0 <= adj_y < 10: + adj_pos = adj_y * 10 + adj_x + if adj_pos not in ai_shots: + # Only boost if not already boosted + if probability_matrix[adj_y][adj_x] < 50: + probability_matrix[adj_y][adj_x] = 50 # Set to fixed high value instead of adding + + # Find top 6 highest probability positions + position_probs = [] + for i in range(100): + if i not in ai_shots: # Only consider unshot positions + x, y = i % 10, i // 10 + position_probs.append((probability_matrix[y][x], i)) + + # Sort by probability (descending) and take top positions + position_probs.sort(reverse=True) + candidates = [pos for prob, pos in position_probs[:20]] # Take top 20 for variety + max_prob = position_probs[0][0] if position_probs else 0 + + # Fallback if no candidates found + if not candidates: + candidates = [i for i in range(100) if i not in ai_shots] + + # Debug: Log what we're working with + turn_number = len(ai_shots) + 1 + print(f"DEBUG: Turn {turn_number}, Max prob: {max_prob}") + print("DEBUG: Probability grid:") + for y in range(10): + row = [f"{probability_matrix[y][x]:2d}" for x in range(10)] + print(f" {' '.join(row)}") + print(f"DEBUG: Top candidates: {candidates[:10]}") + + reasoning_response = hermes_reason_move("battleship", turn_number, candidates) + + # Analysis already logged in hermes_reason_move function + + # Extract move from response - try multiple parsing methods + try: + if "MOVE:" in reasoning_response: + move_part = reasoning_response.split("MOVE:")[1].strip() + ai_shot = int(move_part.split()[0]) + print(f"DEBUG: Hermes Move: {ai_shot} (from 'MOVE: {move_part.split()[0]}')") + else: + # Fallback: extract any number from the response that's in candidates + import re + numbers = re.findall(r'\b(\d+)\b', reasoning_response) + valid_moves = [int(n) for n in numbers if int(n) in candidates and int(n) not in ai_shots] + if valid_moves: + ai_shot = valid_moves[0] + print(f"DEBUG: Hermes Move (parsed): {ai_shot} from numbers {numbers}") + else: + raise Exception(f"ERROR: Hermes response had no valid moves! Response: {reasoning_response}, Candidates: {candidates}") + + # Validate the shot is legal + if ai_shot in ai_shots or ai_shot < 0 or ai_shot > 99: + ai_shot = random.choice(candidates) + print(f"DEBUG: Invalid shot, using fallback: {ai_shot}") + + except Exception as e: + ai_shot = random.choice(candidates) + print(f"DEBUG: Parse error: {e}, using fallback: {ai_shot}") + + elif ai_mode == "super_hunter": + # Use probabilistic grid algorithm + max_prob = 0 + candidates = [] + for i in range(100): + if i not in ai_shots: # Exclude already-fired cells + x, y = i % 10, i // 10 + if probability_matrix[y][x] > max_prob: + max_prob = probability_matrix[y][x] + candidates = [i] + elif probability_matrix[y][x] == max_prob: + candidates.append(i) + ai_shot = random.choice(candidates) + elif ai_mode == "hunter": + # Simple hunter mode logic + if hits: + # Target adjacent cells of the last hit + last_hit = hits[-1] + hunt_targets = generate_hunt_targets(last_hit, ai_shots) + if hunt_targets: + ai_shot = hunt_targets.pop(0) + else: + ai_shot = random_search() + else: + ai_shot = random_search() + else: + # Random mode + ai_shot = random_search() + + # Update AI state after the shot + if user_board[ai_shot] != -1: + ai_hits.append(ai_shot) + ai_hit_result = "hit" + if ai_mode == "super_hunter" or ai_mode == "hermes_reasoner": + update_probability(ai_shot % 10, ai_shot // 10, True) + else: + ai_hit_result = "miss" + if ai_mode == "super_hunter" or ai_mode == "hermes_reasoner": + update_probability(ai_shot % 10, ai_shot // 10, False) + + return ai_shot + + # Function for random search + def random_search(): + available_positions = [] + for i in range(100): + if i not in ai_shots: + available_positions.append(i) + return random.choice(available_positions) + + # Function to generate hunt targets around a hit + def generate_hunt_targets(hit_position, ai_shots): + potential_targets = [] + row, col = divmod(hit_position, 10) + + # Up + if row > 0: + potential_targets.append(hit_position - 10) + # Down + if row < 9: + potential_targets.append(hit_position + 10) + # Left + if col > 0: + potential_targets.append(hit_position - 1) + # Right + if col < 9: + potential_targets.append(hit_position + 1) + + # Filter out already fired positions + filtered_targets = [] + for pos in potential_targets: + if pos not in ai_shots: + filtered_targets.append(pos) + return filtered_targets + + # Get the user's shot + try: + user_shot = int(metadata.get("user_shot")) + except (IndexError, ValueError) as e: + user_shot = -1 + + if game_over: + script_result = { + "metadata": { + "game_over": True, + "user_wins": user_wins, + "ai_wins": ai_wins + } + } + print(f"DEBUG: Game over detected! User wins: {user_wins}, AI wins: {ai_wins}") + elif 0 <= user_shot < 100 and user_shot not in user_shots: + # The move is valid + user_shots.append(user_shot) + user_hit_result = "miss" + if ai_board[user_shot] != -1: + user_hits.append(user_shot) + user_hit_result = "hit" + + # AI makes a move + ai_shot = choose_ai_shot() + ai_shots.append(ai_shot) + + # Check if any AI ship is sunk + for ship_name in ship_sizes.keys(): + if check_sunk(ai_board, user_hits, ship_name) and ship_name not in user_sunk_ships: + user_sunk_ships.append(ship_name) + user_sunk_ship_this_round = ship_name + print(f"DEBUG: USER SUNK AI SHIP: {ship_name}") + + # Check if any User ship is sunk + for ship_name in ship_sizes.keys(): + if check_sunk(user_board, ai_hits, ship_name) and ship_name not in ai_sunk_ships: + ai_sunk_ships.append(ship_name) + ai_sunk_ship_this_round = ship_name + print(f"DEBUG: AI SUNK USER SHIP: {ship_name}") + + # Check if all AI ships are hit + all_ai_ships_hit = True + for pos in range(100): + if ai_board[pos] != -1 and pos not in user_hits: + all_ai_ships_hit = False + break + + # Check if all User ships are hit + all_user_ships_hit = True + for pos in range(100): + if user_board[pos] != -1 and pos not in ai_hits: + all_user_ships_hit = False + break + + if all_ai_ships_hit: + game_over = True + user_wins = True + ai_wins = False + print(f"DEBUG: USER WINS! All AI ships destroyed. Game over.") + elif all_user_ships_hit: + game_over = True + user_wins = False + ai_wins = True + print(f"DEBUG: AI WINS! All user ships destroyed. Game over.") + + # Only track winning move if there's exactly 1 position left (for next turn's categorization) + user_winning_move = None + ai_winning_move = None + + # Check which user move would win the game (AI ship positions left) + ai_ship_positions_left = [pos for pos in range(100) if ai_board[pos] != -1 and pos not in user_hits] + if len(ai_ship_positions_left) == 1: + user_winning_move = ai_ship_positions_left[0] + print(f"DEBUG: User has exactly 1 winning move at position {user_winning_move}") + else: + print(f"DEBUG: User has {len(ai_ship_positions_left)} AI positions left - no winning move") + + # Check which AI move would win the game (user ship positions left) + user_ship_positions_left = [pos for pos in range(100) if user_board[pos] != -1 and pos not in ai_hits] + if len(user_ship_positions_left) == 1: + ai_winning_move = user_ship_positions_left[0] + print(f"DEBUG: AI has exactly 1 winning move at position {ai_winning_move}") + else: + print(f"DEBUG: AI has {len(user_ship_positions_left)} user positions left - no winning move") + + # Plot the boards + fig, axs = plt.subplots(1, 2, figsize=(12, 6)) + fig.suptitle("Battleship", fontsize=16) + + # User's view of AI's board + axs[0].set_xlim(0, 10) + axs[0].set_ylim(0, 10) + axs[0].set_xticks([]) + axs[0].set_yticks([]) + axs[0].grid(True) + axs[0].set_title("Your Shots", fontsize=12) + + # Plot user shots on AI's board + for i in range(100): + x, y = i % 10, 9 - i // 10 + if i in user_shots: + if i in user_hits: + axs[0].text(x + 0.5, y + 0.5, 'X', fontsize=12, ha='center', va='center', color='red') + else: + axs[0].text(x + 0.5, y + 0.5, 'O', fontsize=12, ha='center', va='center', color='black') + axs[0].text(x + 0.5, y + 0.5, str(i), fontsize=8, ha='center', va='center', color='gray') + + # AI's view of User's board + axs[1].set_xlim(0, 10) + axs[1].set_ylim(0, 10) + axs[1].set_xticks([]) + axs[1].set_yticks([]) + axs[1].grid(True) + axs[1].set_title("Your Ships", fontsize=12) + + # Plot user ships + for i, ship in enumerate(user_board): + x, y = i % 10, 9 - i // 10 + if ship != -1: + axs[1].add_patch(plt.Rectangle((x, y), 1, 1, color=ship_colors[ship], alpha=0.5)) + + # Plot AI shots on User's board + for i in range(100): + x, y = i % 10, 9 - i // 10 + if i in ai_shots: + if i in ai_hits: + axs[1].text(x + 0.5, y + 0.5, 'X', fontsize=12, ha='center', va='center', color='red') + else: + axs[1].text(x + 0.5, y + 0.5, 'O', fontsize=12, ha='center', va='center', color='black') + axs[1].text(x + 0.5, y + 0.5, str(i), fontsize=8, ha='center', va='center', color='gray') + + # Draw lines across sunk ships + for ship_name in user_sunk_ships: + draw_line(axs[0], ai_board, ship_name) + + for ship_name in ai_sunk_ships: + draw_line(axs[1], user_board, ship_name) + + # Add legend + handles = [] + for color in ship_colors.values(): + handles.append(plt.Rectangle((0, 0), 1, 1, color=color, alpha=0.5)) + axs[1].legend(handles, ship_colors.keys(), loc='upper right', fontsize=8) + + buf = io.BytesIO() + plt.savefig(buf, format='png', bbox_inches='tight', pad_inches=0.1) + plt.close(fig) + buf.seek(0) + plot_image = base64.b64encode(buf.getvalue()).decode('utf-8') + + # gpt-4: If "plot_image" is in the result, set it as the background image + print(f"DEBUG: Setting metadata for feedback - user_sunk_ship_this_round: {user_sunk_ship_this_round}, ai_sunk_ship_this_round: {ai_sunk_ship_this_round}") + + script_result = { + "plot_image": plot_image, + "set_background": True, + "metadata": { + "user_board": user_board, + "ai_board": ai_board, + "user_shot": user_shot, + "ai_shot": ai_shot, + "user_shots": user_shots, + "ai_shots": ai_shots, + "user_hits": user_hits, + "ai_hits": ai_hits, + "game_over": game_over, + "user_wins": user_wins, + "ai_wins": ai_wins, + "user_hit_result": user_hit_result, + "ai_hit_result": ai_hit_result, + "user_sunk_ships": user_sunk_ships, + "ai_sunk_ships": ai_sunk_ships, + "user_sunk_ship_this_round": user_sunk_ship_this_round, + "ai_sunk_ship_this_round": ai_sunk_ship_this_round, + "ai_mode": ai_mode, + "probability_matrix": probability_matrix, + "hits": hits, + "misses": misses, + "sunk_ships": sunk_ships, + "user_winning_move": user_winning_move, + "ai_winning_move": ai_winning_move + } + } + + # Check if this was a winning move and override transition + if game_over: + script_result["next_section_and_step"] = "section_1:step_3" + print(f"POST-SCRIPT: Game over detected, overriding transition to step_3") + else: + script_result = { + "error": f"Invalid shot: {metadata.get('user_shot')}", + "metadata": {} + } + + buckets: + - valid_move + - invalid_move + - exit + - restart + transitions: + valid_move: + run_processing_script: True + ai_feedback: + tokens_for_ai: | + The user shot seems valid. + metadata_tmp_add: + user_shot: "the-users-response" + metadata_feedback_filter: + - user_hit_result + - ai_hit_result + - ai_shot + - user_shot + - user_sunk_ship_this_round + - ai_sunk_ship_this_round + - game_over + - user_wins + - ai_wins + next_section_and_step: "section_1:step_2" + invalid_move: + content_blocks: + - "That move is invalid. Please choose a position between 0 and 99." + metadata_tmp_add: + user_shot: "the-users-response" + next_section_and_step: "section_1:step_2" + exit: + next_section_and_step: "section_1:step_3" + restart: + content_blocks: + - "Restarting the game. Let's start fresh!" + metadata_clear: True + next_section_and_step: "section_1:step_0" + game_end: + next_section_and_step: "section_1:step_3" + + - step_id: "step_3" + title: "Game Over" + question: "Would you like to restart and play again, or would you prefer to exit?" + tokens_for_ai: | + If the user wants to restart or play again, categorize as 'restart'. + If the user wants to exit, categorize as 'exit'. + feedback_tokens_for_ai: | + Acknowledge the user's choice appropriately. + buckets: + - restart + - exit + transitions: + restart: + content_blocks: + - "Restarting the game. Let's start fresh!" + metadata_clear: True + next_section_and_step: "section_1:step_0" + exit: + content_blocks: + - "Thank you for playing Battleship! 🎉" + - "Feel free to come back anytime for another game." + next_section_and_step: "section_1:step_4" + + - step_id: "step_4" + title: "Goodbye" + content_blocks: + - "Thanks for playing! Hope you enjoyed the battle at sea."