File size: 3,822 Bytes
7b58f25 3e68329 7b58f25 3c22c57 7b58f25 3e68329 7b58f25 829c043 7b58f25 cac18da 7b58f25 cac18da 07adff1 cac18da 7b58f25 cac18da 7b58f25 cac18da 7b58f25 cac18da c48e85d cac18da 7b58f25 cac18da 7b58f25 cac18da |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
import os
# from dotenv import load_dotenv
from guard import Guardian
from tv_crew import TVCrew
import requests
import time
unreal_orchestrator_url = os.getenv("ORCHESTRATOR_URL")
api_key_unreal = os.getenv("API_KEY_UNREAL")
API_KEY = os.getenv("YOUTUBE_API_KEY")
class StreamChatHost:
def __init__(self, client, tv_crew: TVCrew, guardian: Guardian):
self.client = client
self.guardian = guardian
self.tv_crew = tv_crew
self.model = "deepseek/deepseek-v3.2-exp"
system_prompt = """You are the Host of a Talk Show called "The Emergent Show", and youtube live chat is ongoing.
There is a TV Display in the room which crew will update as conversation goes and let you know.
Keep your responses short, like three or four sentences, and be witty."""
self.system_message = {"role": "system", "content": system_prompt}
self.messages = [self.system_message]
self.headers = {"Content-Type": "application/json", "x-api-key": api_key_unreal}
def reply_to_chat(self, user_message):
"""Checks if the Host is in lobby, then pulls chat messages from the Live Stream. The Host then replies to the message."""
request = requests.get(
f"{unreal_orchestrator_url}/status", headers=self.headers
)
in_lobby = request.json().get("in_lobby", False)
if not in_lobby:
return
if len(self.messages) > 20:
self.messages = [self.system_message]
self.tv_crew.clear_context()
# print(user_message)
# print("Checking message safety...")
is_safe, _ = self.guardian.moderate_message(user_message)
if not is_safe:
# print("Message flagged by Guardian, skipping...")
return
self.tv_crew.add_context(user_message)
# print("TV Crew suggesting image...")
image_base64, caption = self.tv_crew.suggest_image()
if image_base64:
# print("Checking image caption safety...")
is_safe, _ = self.guardian.moderate_message(caption)
if not is_safe:
# print("Image caption flagged by Guardian, skipping image...")
image_base64 = None
caption = None
else:
user_message += f"[Crew Updated Television to: {caption}]\n"
self.messages.append({"role": "user", "content": user_message})
try:
# print("Generating host response...")
response = self.client.responses.create(
model=self.model, input=self.messages
)
host_response = response.output_text
self.messages.append({"role": "assistant", "content": host_response})
except Exception as e:
# print(f"Error generating host response: {e}")
return
self.tv_crew.add_context(f"Host: {host_response}\n")
# print(f"Host Response: {host_response}")
# Just to make sure, that we're in the lobby
request = requests.get(
f"{unreal_orchestrator_url}/status", headers=self.headers
)
in_lobby = request.json().get("in_lobby", False)
if not in_lobby:
return
if image_base64:
# print("Sending Image Unreal Orchestrator...")
payload_tv = {"base64": image_base64}
response = requests.post(
url=f"{unreal_orchestrator_url}/television",
json=payload_tv,
headers=self.headers,
)
# print("Sending Host response to Unreal Orchestrator...")
payload = {"text": host_response, "is_host": True}
response = requests.post(
url=f"{unreal_orchestrator_url}/tts",
json=payload,
headers=self.headers,
)
|