102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
import logging
|
|
import threading
|
|
import time
|
|
from typing import Literal, Dict, Any
|
|
import config
|
|
from dto.requests import GameStartRequestDTO, GameDecisionRequestDTO
|
|
from services.extractor import extract_profile, extract_passport, extract_description, extract_account
|
|
from services.julius_baer_api_client import JuliusBaerApiClient
|
|
from utils.storage.game_files_manager import store_game_round_data
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class Player:
|
|
|
|
def __init__(self):
|
|
self.client = JuliusBaerApiClient()
|
|
self._thread = None
|
|
|
|
def start(self):
|
|
self.play()
|
|
|
|
def play_on_separate_thread(self):
|
|
if self._thread and self._thread.is_alive():
|
|
log.warning('Game loop already running.')
|
|
return self._thread
|
|
|
|
self._thread = threading.Thread(target=self.play, daemon=True)
|
|
self._thread.start()
|
|
return self._thread
|
|
|
|
def play(self):
|
|
log.info('playing')
|
|
payload = GameStartRequestDTO(player_name=config.API_TEAM)
|
|
start_response = self.client.start_game(payload)
|
|
log.info('game started, session id: %s', start_response.session_id)
|
|
|
|
client_id = start_response.client_id
|
|
decision = self.make_decision(start_response.client_data)
|
|
|
|
decision_counter = 0
|
|
|
|
is_game_running = True
|
|
while is_game_running:
|
|
payload = GameDecisionRequestDTO(
|
|
decision=decision,
|
|
session_id=start_response.session_id,
|
|
client_id=client_id,
|
|
)
|
|
|
|
log.info('client id: %s', client_id)
|
|
decision_response = self.client.send_decision(payload)
|
|
log.info(f'decision: {decision}, response status: {decision_response.status}, score: {decision_response.score}')
|
|
|
|
status = decision_response.status
|
|
is_game_running = status not in ['gameover', 'complete']
|
|
client_id = decision_response.client_id
|
|
|
|
decision = self.make_decision(decision_response.client_data)
|
|
|
|
# Handle first response from game initialization logic
|
|
if decision_counter == 0:
|
|
# Store start response
|
|
store_game_round_data(decision, start_response, decision_counter, str(start_response.session_id) ,status)
|
|
else:
|
|
# store ongoing decision response
|
|
store_game_round_data(decision, decision_response, decision_counter, str(start_response.session_id), status)
|
|
|
|
decision_counter += 1
|
|
time.sleep(1)
|
|
|
|
def make_decision(self, client_data: Dict[str, Any]) -> Literal["Accept", "Reject"]:
|
|
# Data extraction
|
|
profile = extract_profile(client_data)
|
|
passport = extract_passport(client_data)
|
|
description = extract_description(client_data)
|
|
account = extract_account(client_data)
|
|
|
|
prompt_template = (
|
|
"You are a helpful assistant at a private bank.\n"
|
|
"Your task is to accept or reject a new client's application for private banking.\n\n"
|
|
"Only reject the application if there is an inconsistency in the data provided.\n"
|
|
"Inconsistencies include:\n"
|
|
"- Incorrect data (e.g., mismatched or invalid information)\n"
|
|
"- Incomplete data (e.g., missing required fields)\n\n"
|
|
"Return only JSON matching this format:\n"
|
|
"{format_instructions}\n\n"
|
|
"Here is the extracted passport text:\n"
|
|
"{processed_text}\n\n"
|
|
"Use the extracted profile, description, and account details to check consistency."
|
|
)
|
|
|
|
# You'd insert the format instructions and passport text here before calling the LLM
|
|
# For example:
|
|
# final_prompt = prompt_template.format(
|
|
# format_instructions=your_format_instructions,
|
|
# processed_text=passport['text']
|
|
# )
|
|
|
|
return 'Accept' # Replace me!!
|