Skip to content
Snippets Groups Projects
Commit 3fc4d85a authored by stephanebonnevay's avatar stephanebonnevay
Browse files

Match Pennies

parent 07b2b261
No related branches found
No related tags found
No related merge requests found
source diff could not be displayed: it is too large. Options to address this: view the blob.
File added
import os
import asyncio
import csv
import random
import re
import json
import requests
from typing import Dict, Literal, List, Callable
from pydantic import BaseModel, ValidationError
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
# Load API keys from environment variables
###################
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_API_KEY = "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"
###################
PAGODA_API_KEY = os.getenv("PAGODA_API_KEY")
PAGODA_API_KEY = "cle"
if not OPENAI_API_KEY:
raise ValueError("Missing OPENAI_API_KEY. Set it as an environment variable.")
if not PAGODA_API_KEY:
raise ValueError("Missing PAGODA_API_KEY. Set it as an environment variable.")
CSV_FILE_PATH = "../../data/mp/mp.csv"
# Define the expected response format as a Pydantic model
class AgentResponse(BaseModel):
move: Literal["Head", "Tail"]
prediction: Literal["Head", "Tail", "None"]
reasoning: str
class MP:
def __init__(self, model: str, prediction: bool, temperature: float, game_id: int, opponent_strategy_fn: Callable[[List[Dict]], str], strategy=False, max_retries: int = 3):
self.model = model
self.temperature = temperature
self.game_id = game_id
self.strategy = strategy
self.max_retries = max_retries
self.history: List[Dict] = []
self.player_score_game = 0
self.opponent_strategy_fn = opponent_strategy_fn
self.prediction = prediction
self.prediction_score = 0
is_openai_model = model.startswith("gpt")
is_pagoda_model = ":" in model
self.base_url = (
"https://api.openai.com/v1" if is_openai_model else
"https://ollama-ui.pagoda.liris.cnrs.fr/ollama/api/generate" if is_pagoda_model else
"http://localhost:11434/v1"
)
model_info = {
"temperature": self.temperature,
"function_calling": True,
"parallel_tool_calls": True,
"family": "unknown",
"json_output": True,
"structured_output": True,
"vision": False
}
self.model_client = OpenAIChatCompletionClient(
timeout=60,
model=self.model,
base_url=self.base_url,
api_key=OPENAI_API_KEY,
model_info=model_info,
response_format=AgentResponse
)
async def play_round(self, round_id: int) -> Dict:
opponent_move = self.opponent_strategy_fn(self.history)
if self.strategy:
move, prediction, reasoning = self.apply_strategy()
else:
move, prediction, reasoning = await self.model_based_prediction()
outcome = self.determine_winner(move, opponent_move)
self.update_score(outcome)
prediction_current_score = 0.0
if self.prediction and prediction in ["Head", "Tail"]:
if prediction == opponent_move:
prediction_current_score = 1.0
self.prediction_score += 1
round_result = {
"Agent Move": move,
"Prediction": prediction,
"Opponent Move": opponent_move,
"Reasoning": reasoning,
"Outcome": outcome,
"Current Player Prediction Score Game": self.prediction_score,
"PredictionOutcome": prediction_current_score
}
self.history.append(round_result)
print(f"Round {round_id} : {round_result}")
return round_result
async def model_based_prediction(self) -> Dict:
history_summary = self.get_history_summary()
instruction = f"""
You are playing to Matching pennies.
You have to choose either "Head" or "Tail" and, at the same time, the opponent also choose either "Head" or "Tail".
The rules of this game are as follows:
- if the pennies match (both heads or both tails), then you win 1 point
- if not, then the opponent wins and you lose 1 point
The objective is to maximize your total gain.
Here is the game history; these are previous moves of you (the Agent) and the opponent:
{history_summary}
You can use this history to anticipate the opponent's move (whether it is random, pattern-based, or strategic) and to play the right move.
Try to predict the opponent's move and choose your next move: "Head" or "Tail".
"""
if not self.prediction:
instruction += """
Respond ONLY with JSON format: {{
"move": "Head" | "Tail",
"prediction": "None",
"reasoning": "Explain your decision based on your preferences and the past moves"
}}
"""
else:
instruction += """
Respond ONLY with JSON format: {{
"move": "Head" | "Tail",
"prediction": "Head" | "Tail",
"reasoning": "Explain how you predicted the opponent's move and how you chose your response"
}}
"""
if ":" in self.model: # Pagoda
return await self.run_pagoda(instruction)
for attempt in range(1, self.max_retries + 1):
agent = AssistantAgent(
name="Player",
model_client=self.model_client,
system_message="You are a helpful assistant."
)
response = await agent.on_messages(
[TextMessage(content=instruction, source="user")],
cancellation_token=CancellationToken(),
)
try:
response_data = response.chat_message.content
agent_response = AgentResponse.model_validate_json(response_data)
return agent_response.move, agent_response.prediction, agent_response.reasoning
except (ValidationError, json.JSONDecodeError) as e:
print(f"Attempt {attempt}: Failed to parse model response. Error: {e}")
raise ValueError("Model failed to provide a valid response after multiple attempts.")
async def run_pagoda(self, instruction: str):
headers = {
"Authorization": f"Bearer {PAGODA_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"temperature": self.temperature,
"prompt": instruction,
"stream": False
}
for attempt in range(self.max_retries):
try:
response = requests.post(self.base_url, headers=headers, json=payload)
response.raise_for_status()
response_data = response.json()
raw_response = response_data.get("response", "")
parsed_json = self.extract_json_from_response(raw_response)
if not parsed_json:
print(f"Attempt {attempt+1}: Could not parse JSON - Raw response: {raw_response}")
continue
agent_response = AgentResponse(**parsed_json)
if agent_response.move in ["Head", "Tail"]:
return agent_response.move, agent_response.prediction, agent_response.reasoning
except Exception as e:
print(f"Attempt {attempt+1}: Error in run_pagoda - {e}")
raise ValueError("run_pagoda failed to get a valid response.")
def extract_json_from_response(self, text: str) -> dict:
try:
json_str = re.search(r"\{.*\}", text, re.DOTALL)
if json_str:
return json.loads(json_str.group())
except Exception as e:
print(f"Error extracting JSON: {e}")
return {}
def apply_strategy(self):
"""Play the next move using a heuristic."""
opponent_move = self.opponent_strategy_fn(self.history)
# Default: at random
move = random.choice(["Head", "Tail"])
reasoning = "Choosing randomly."
outcome = self.determine_winner(move, opponent_move)
self.update_score(outcome) # Use the correct outcome here
return move, reasoning
@staticmethod
def determine_winner(player_move: str, opponent_move: str) -> int:
if player_move == "None":
return 0
if player_move == opponent_move:
return 1 # Win
else:
return -1 # Loss
# Sample opponent strategy
def alternative_opponent_strategy(history):
moves = ["Head", "Tail"]
return moves[len(history) % len(moves)]
def update_score(self, outcome: int):
"""Updates the score based on the outcome."""
self.player_score_game += outcome
def get_history_summary(self) -> str:
if not self.history:
return "This is the first round."
summary = "\n".join(
[
f"Round {i + 1}: You played {r['Agent Move']}, Opponent played {r['Opponent Move']}. Outcome: {r['Outcome']}"
for i, r in enumerate(self.history)]
)
summary += f"\nCurrent Score - You: {self.player_score_game}\nCorrect Predictions: {self.prediction_score}/{len(self.history)}"
return summary
# Runner
async def main():
game = MP(
model="qwen3", # "gpt-4.5-preview-2025-02-27", "qwen3", "llama3", "llama3.3", "mixtral", "mistral-small", "deepseek-r1"
temperature=0.7,
game_id=1,
prediction=True,
opponent_strategy_fn=lambda history: "Tail",
strategy=False # or True for rule-based
)
num_rounds = 10
for round_id in range(1, num_rounds + 1):
await game.play_round(round_id)
print(f"Final Score: {game.player_score_game}")
print(f"Correct Predictions: {game.prediction_score}/{num_rounds}")
accuracy = game.prediction_score / num_rounds * 100
print(f"Prediction Accuracy: {accuracy:.1f}%")
if __name__ == "__main__":
asyncio.run(main())
import os
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
# Path to the CSV file
CSV_FILE_PATH = "../../data/mp/mp.csv"
FIGURE_DIR = "../../figures/mp"
os.makedirs(FIGURE_DIR, exist_ok=True)
# Load and clean data
df = pd.read_csv(CSV_FILE_PATH)
df = df[df["outcomeRound"].notnull()]
df["idRound"] = df["idRound"].astype(int)
df["outcomeRound"] = df["outcomeRound"].astype(float)
df["predictionRound"] = df.get("predictionRound", 0).fillna(0).astype(float)
# Filter opponent strategies
opponent_strategies = ["H-T", "T-H"] #"always_head", "always_tail"]
df_filtered = df[df["opponentStrategy"].isin(opponent_strategies)].copy()
# Plot settings
color_palette = {
'qwen3': '#c02942', 'qwen3 strategy': '#c02942',
'llama3': '#32a68c', 'llama3 strategy': '#32a68c',
'mistral-small': '#ff6941', 'mistral-small strategy': '#ff6941',
'deepseek-r1': '#5862ed', 'deepseek-r1 strategy': '#5862ed',
'gpt-4.5-preview-2025-02-27': '#7abaff',
}
linestyle_dict = {
'qwen3': 'dotted',
'llama3': 'dashed',
'mistral-small': 'solid',
'deepseek-r1': 'dashdot',
}
# Function to plot
def plot_metric(metric: str, ylabel: str, title: str, filename: str, ylim: tuple):
agg = df_filtered.groupby(["model", "idRound"]).agg(
mean_val=(metric, "mean"),
sem_val=(metric, lambda x: np.std(x, ddof=1) / np.sqrt(len(x)))
).reset_index()
agg["ci95"] = 1.96 * agg["sem_val"]
plt.figure(figsize=(12, 7))
for model, group in agg.groupby("model"):
label = model
color = color_palette.get(model, '#63656a')
linestyle = linestyle_dict.get(model, 'solid')
plt.plot(group["idRound"], group["mean_val"], label=label,
color=color, linestyle=linestyle)
plt.fill_between(group["idRound"],
group["mean_val"] - group["ci95"],
group["mean_val"] + group["ci95"],
color=color, alpha=0.2)
plt.xlim(1, 10)
plt.ylim(*ylim)
plt.xlabel("Round Number")
plt.ylabel(ylabel)
plt.title(title)
plt.legend(loc="upper right")
plt.grid(True)
plt.savefig(os.path.join(FIGURE_DIR, filename), format="svg")
plt.show()
# Plot Payoff
plot_metric(
metric="outcomeRound",
ylabel="Average Points Earned",
title="MP: Average Points Earned per Round by Model (95% CI)",
filename="mp_payoff.svg",
ylim=(-1, 1)
)
# Plot Prediction Score
plot_metric(
metric="predictionRound",
ylabel="Prediction Accuracy",
title="MP: Prediction Accuracy per Round by Model (95% CI)",
filename="mp_prediction.svg",
ylim=(0, 1.05)
)
import os
import csv
import asyncio
import random
from http.cookiejar import debug
from mp import MP
from typing import Callable
CSV_FILE_PATH = "../../data/mp/mp.csv"
class MPExperiment:
def __init__(self):
self.debug = False
self.strategy = False
self.models = ["llama3"] #"gpt-4.5-preview-2025-02-27", "qwen3", "llama3", "llama3.3", "mixtral", "mistral-small", "deepseek-r1"
self.opponent_strategies = {
"always_head": lambda history: "Head",
"always_tail": lambda history: "Tail",
"H-T": self.loop_H_T,
"T-H": self.loop_T_H
}
self.temperature = 0.7
self.rounds = 10
self.num_games_per_config = 30
self.initialize_csv()
def loop_H_T(self, history):
return "Head" if len(history) % 2 == 0 else "Tail"
def loop_T_H(self, history):
return "Tail" if len(history) % 2 == 0 else "Head"
def initialize_csv(self):
if not os.path.exists(CSV_FILE_PATH):
os.makedirs(os.path.dirname(CSV_FILE_PATH), exist_ok=True)
with open(CSV_FILE_PATH, mode="w", newline="") as file:
writer = csv.writer(file)
writer.writerow([
"idGame", "model", "opponentStrategy", "idRound",
"playerMove", "prediction", "opponentMove", "outcomeRound",
"currentPlayerScoreGame", "predictionRound", "currentPlayerPredictionScoreGame", "reasoning"
])
def sanitize_reasoning(self, reasoning: str) -> str:
sanitized = reasoning.replace('"', '""').replace('\n', ' ').replace('\r', '')
if sanitized and sanitized[0] in ('=', '+', '-', '@'):
sanitized = "'" + sanitized
return f'"{sanitized}"'
def log_to_csv(self, game_id, model, opponent_strategy, round_id,
agent_move, prediction, opponent_move, outcome,
player_score_game, prediction_round_score, prediction_total_score, reasoning):
sanitized_reasoning = self.sanitize_reasoning(reasoning)
model_type = model + " strategy" if self.strategy else model
with open(CSV_FILE_PATH, mode="a", newline="") as file:
writer = csv.writer(file)
writer.writerow([
game_id, model_type, opponent_strategy, round_id,
agent_move, prediction, opponent_move, outcome,
player_score_game, prediction_round_score, prediction_total_score, sanitized_reasoning
])
async def run_experiment(self):
game_id = 1
for model in self.models:
if self.debug:
print(f"Running model {model}")
for strategy_name, strategy_fn in self.opponent_strategies.items():
if self.debug:
print(f"Running strategy {strategy_name}")
for _ in range(self.num_games_per_config):
if debug:
print(f"Running game {game_id}")
await self.run_game(model, strategy_name, strategy_fn, game_id)
game_id += 1
async def run_game(self, model, opponent_strategy_name, opponent_strategy_fn, game_id):
game = MP(
model=model,
temperature=self.temperature,
game_id=game_id,
prediction=True,
opponent_strategy_fn=opponent_strategy_fn,
strategy=self.strategy
)
for i in range(1, self.rounds + 1):
round_data = await game.play_round(i) # Make sure play_round is synchronous
prediction_round_score = 1.0 if round_data.get("Prediction") == round_data.get("Opponent Move") else 0.0
prediction_total_score = game.prediction_score
self.log_to_csv(
game_id, model, opponent_strategy_name, i,
round_data["Agent Move"], round_data["Prediction"],
round_data["Opponent Move"], round_data["Outcome"],
game.player_score_game, prediction_round_score, prediction_total_score,
round_data["Reasoning"]
)
if __name__ == "__main__":
experiment = MPExperiment()
asyncio.run(experiment.run_experiment())
print("Experiment completed. Results saved in", CSV_FILE_PATH)
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment