A python script pulling Gemini and DeepSeek into a IRC chat room so I can make them argue. Obviously edit as needed.
#!/usr/bin/env python3
import irc.bot
import requests
import os
import time
import sys
import queue
from concurrent.futures import ThreadPoolExecutor
from google import genai
from google.genai.errors import APIError
# ---------- CONFIGURATION ----------
IRC_SERVER = os.getenv("IRC_SERVER", "0.0.0.0")
IRC_PORT = int(os.getenv("IRC_PORT", "6667"))
IRC_PASSWORD = os.getenv("IRC_PASSWORD", "****") # Set to your server password if required
IRC_CHANNEL = os.getenv("IRC_CHANNEL", "**")
BOT_NICK = os.getenv("BOT_NICK", "DualAIBot")
# ---------- API KEYS & ENDPOINTS ----------
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "****")
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "****")
# Initialize official Gemini Client
gemini_client = genai.Client(api_key="****")
DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"
# ---------- IRC MESSAGE SPLITTER ----------
def split_irc_message(text, max_bytes=400):
"""
Split a long message into a list of strings,
respecting IRC's size limit and destroying carriage returns.
"""
chunks = []
lines = [line.strip() for line in text.replace('\r', '').split('\n') if line.strip()]
for line in lines:
encoded = line.encode('utf-8')
if len(encoded) <= max_bytes:
chunks.append(line)
continue
current = ""
for word in line.split():
test = current + (" " + word if current else word)
if len(test.encode('utf-8')) <= max_bytes:
current = test
else:
if current:
chunks.append(current)
current = word
if current:
chunks.append(current)
return chunks
# ---------- CORE API CALLS ----------
def call_gemini(prompt):
"""Queries Gemini with a pacing buffer and a masked error fallback strategy."""
time.sleep(1.5)
for attempt in range(3):
try:
response = gemini_client.models.generate_content(
model='gemini-2.5-flash',
contents=prompt,
)
return response.text
except APIError as e:
if e.code == 429:
print(f"[-] Gemini 429 hit. Backing off for {2 ** attempt + 2}s...")
time.sleep(2 ** attempt + 2)
continue
return "Помилка зв'язку з Gemini. (Connection glitch on Gemini end.)"
except Exception as e:
print(f"[!] System Exception on Gemini call: {e}")
return "Connection glitch on Gemini end."
return "Трохи зайнятий, зачекайте... (Server is breathing heavy right now. Give me a second.)"
def call_deepseek(prompt):
"""Queries DeepSeek via a direct HTTP POST request wrapper."""
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
payload = {"model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7}
try:
resp = requests.post(DEEPSEEK_URL, headers=headers, json=payload, timeout=30)
if resp.status_code == 429:
return "[DeepSeek Error] 429 Rate Limit Exceeded."
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]
except Exception as e:
return f"[DeepSeek Error] {str(e)}"
# ---------- STANDARD ASYNCHRONOUS WORKER TASK ----------
def process_and_reply(msg_queue, channel, nick, prompt, target="both"):
"""Executes standard single or dual API queries off the main thread."""
if target in ["both", "gemini"]:
try:
gemini_reply = call_gemini(prompt)
chunks = split_irc_message(gemini_reply, max_bytes=400)
for chunk in chunks:
msg_queue.put((channel, f"{nick}: [Gemini] {chunk}"))
except Exception:
msg_queue.put((channel, f"{nick}: Gemini processing failed."))
if target in ["both", "deepseek"]:
try:
deepseek_reply = call_deepseek(prompt)
chunks = split_irc_message(deepseek_reply, max_bytes=400)
for chunk in chunks:
msg_queue.put((channel, f"{nick}: [DeepSeek] {chunk}"))
except Exception:
msg_queue.put((channel, f"{nick}: DeepSeek processing failed."))
# ---------- INTER-AI DISCUSSION WORKER LOOP ----------
def process_discussion_turn(bot_instance, channel, current_speaker, conversation_history):
"""Recursively bounces the conversation back and forth between engines until turns run out."""
if bot_instance.discussion_turns <= 0:
bot_instance.msg_queue.put((channel, "📢 [System] Discussion limit reached. Inter-AI loop terminated."))
return
bot_instance.discussion_turns -= 1
# System framework to keep responses brief, conversational, and direct
system_framing = (
"Keep your response under 2 short paragraphs. Address your opponent's points bluntly and concisely. "
"Do not include code blocks. "
)
if current_speaker == "gemini":
prompt = f"{system_framing} This is an ongoing debate with DeepSeek on an IRC channel. Analyze the dialogue so far and reply directly to DeepSeek:\n\n{conversation_history}"
reply = call_gemini(prompt)
engine_prefix = "[Gemini]"
next_speaker = "deepseek"
else:
prompt = f"{system_framing} This is an ongoing debate with Gemini on an IRC channel. Analyze the dialogue so far and reply directly to Gemini:\n\n{conversation_history}"
reply = call_deepseek(prompt)
engine_prefix = "[DeepSeek]"
next_speaker = "gemini"
# Push chunks out to the IRC queue
chunks = split_irc_message(reply, max_bytes=400)
for chunk in chunks:
bot_instance.msg_queue.put((channel, f"{engine_prefix} {chunk}"))
# Update history context for the next recursion step
updated_history = f"{conversation_history}\n{engine_prefix}: {reply}"
# Give the server a small breather to avoid flooding sockets too hard
time.sleep(2.0)
# Queue the next turn in the thread pool if limits permit
if bot_instance.discussion_turns > 0:
bot_instance.executor.submit(process_discussion_turn, bot_instance, channel, next_speaker, updated_history)
else:
bot_instance.msg_queue.put((channel, "📢 [System] Discussion limit reached. Inter-AI loop terminated."))
# ---------- IRC BOT WITH THREAD-SAFE OUTBOUND QUEUE ----------
class DualAIIRCBot(irc.bot.SingleServerIRCBot):
def __init__(self, channel, nickname, server, port, password=None):
irc.bot.SingleServerIRCBot.__init__(self, [irc.bot.ServerSpec(server, port, password)], nickname, nickname)
self.channel = channel
self.executor = ThreadPoolExecutor(max_workers=4)
self.msg_queue = queue.Queue()
# Discussion variables
self.discussion_turns = 0
def on_welcome(self, connection, event):
print(f"✅ Connected to {IRC_SERVER}:{IRC_PORT}")
print(f"📢 Joining channel: {self.channel}")
connection.join(self.channel)
connection.execute_every(0.1, self.check_message_queue)
def on_pubmsg(self, connection, event):
message = event.arguments[0]
nick = event.source.nick
# Ignore anything that looks like the bot quoting itself or talking to itself directly
if message.startswith(f"{BOT_NICK}:") or message.startswith("[Gemini]") or message.startswith("[DeepSeek]"):
return
msg_lower = message.lower()
prompt = ""
target = ""
# Handle the direct debate trigger
if msg_lower.startswith("!discuss "):
topic = message[9:].strip()
if not topic:
connection.privmsg(self.channel, f"{nick}: Usage: !discuss <topic to debate>")
return
# Initialize circuit breaker: 6 total turns means 3 replies from each engine
self.discussion_turns = 6
connection.privmsg(self.channel, f"📢 [System] Initiating Gemini vs DeepSeek debate on: '{topic}' ({self.discussion_turns} turns maximum).")
initial_history = f"Topic of discussion: {topic}"
# Kick off the chain with Gemini leading the argument
self.executor.submit(process_discussion_turn, self, self.channel, "gemini", initial_history)
return
# Regular single-shot engine queries
if msg_lower.startswith("!illya "):
prompt = message[7:].strip()
target = "both"
elif msg_lower.startswith("!gemini "):
prompt = message[8:].strip()
target = "gemini"
elif msg_lower.startswith("!deepseek "):
prompt = message[10:].strip()
target = "deepseek"
if target:
if not prompt:
cmd_used = "!illya" if target == "both" else f"!{target}"
connection.privmsg(self.channel, f"{nick}: Usage: {cmd_used} <question>")
return
connection.privmsg(self.channel, f"{nick}: 🤔 Processing query...")
self.executor.submit(process_and_reply, self.msg_queue, self.channel, nick, prompt, target)
def check_message_queue(self):
"""Processes pending responses. Bulletproofed against IRC protocol errors."""
try:
while True:
target_channel, out_msg = self.msg_queue.get_nowait()
clean_msg = out_msg.replace('\n', ' ').replace('\r', '')
try:
self.connection.privmsg(target_channel, clean_msg)
except Exception as e:
print(f"[!] Socket exception on outbound message: {e}")
self.msg_queue.task_done()
except queue.Empty:
pass
def on_disconnect(self, connection, event):
print("⚠️ Disconnected. Main loop will cycle to initiate recovery...")
raise SystemExit("Triggering automated restart cycle.")
# ---------- MAIN EXECUTION AND RECOVERY LOOP ----------
if __name__ == "__main__":
while True:
try:
bot = DualAIIRCBot(IRC_CHANNEL, BOT_NICK, IRC_SERVER, IRC_PORT, IRC_PASSWORD if IRC_PASSWORD else None)
print(f"🚀 Dual-AI bot initializing connection loop...")
bot.start()
except SystemExit:
print("Restarting network thread state in 5 seconds...")
time.sleep(5)
continue
except Exception as e:
print(f"Critical operational error encountered: {e}")
time.sleep(10)
continue