Merge pull request #12 from russellballestrini/battleship-hermes
Battleship hermes mode operational!
This commit is contained in:
commit
25c694584a
7 changed files with 1418 additions and 96 deletions
25
CLAUDE.md
Normal file
25
CLAUDE.md
Normal file
|
|
@ -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
|
||||
145
app.py
145
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'<img src="data:image/jpeg;base64,{image_data}" alt="{escaped_message}"><p>{escaped_prompt}</p>'
|
||||
|
|
@ -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'<img alt="Plot Image" src="data:image/png;base64,{plot_image_base64}">'
|
||||
|
|
@ -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}"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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."
|
||||
|
|
|
|||
870
research/activity29-testship.yaml
Normal file
870
research/activity29-testship.yaml
Normal file
|
|
@ -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."
|
||||
Loading…
Add table
Add a link
Reference in a new issue