Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,17 @@ BOT_OPENAI_COOLDOWN=10
BOT_OWNER_COOLDOWN=5


# OpenAI API key
# OpenAI (GPT) API key
BOT_OPENAI_MODEL=gpt-5.5
OPENAI_API_KEY=
BOT_OPENAI_API_KEY=

# Anthropic (Claude) API key
BOT_ANTHROPIC_MODEL=claude-opus-4-8
BOT_ANTHROPIC_API_KEY=

# Google (Gemini) API key
BOT_GEMINI_MODEL=gemini-flash-latest
BOT_GEMINI_API_KEY=


# ddcDatabases configs
Expand Down
4 changes: 3 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "DiscordBot"
version = "3.0.16"
version = "3.0.17"
description = "A simple Discord bot with OpenAI support and server administration tools"
urls.Repository = "https://github.com/ddc/DiscordBot"
urls.Homepage = "https://ddc.github.io/DiscordBot"
Expand Down Expand Up @@ -31,10 +31,12 @@ classifiers = [
requires-python = ">=3.14"
dependencies = [
"alembic>=1.18.4",
"anthropic>=0.105.2",
"beautifulsoup4>=4.14.3",
"better-profanity>=0.7.0",
"ddcdatabases[postgres]>=4.0.1",
"discord-py>=2.7.1",
"google-genai>=2.7.0",
"gTTS>=2.5.4",
"openai>=2.38.0",
"PyNaCl>=1.6.2",
Expand Down
156 changes: 117 additions & 39 deletions src/bot/cogs/open_ai.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
import discord
import time
from anthropic import AsyncAnthropic
from discord.ext import commands
from google import genai
from google.genai import types as genai_types
from openai import AsyncOpenAI
from openai.types.responses import WebSearchToolParam
from openai.types.shared import ReasoningEffort
Expand All @@ -12,12 +15,14 @@


class OpenAi(commands.Cog):
"""OpenAI-powered commands for AI assistance and text generation."""
"""LLM chat commands (OpenAI / Anthropic Claude / Google Gemini) with optional web search."""

def __init__(self, bot: Bot) -> None:
self.bot: Bot = bot
self._bot_settings: BotSettings = get_bot_settings()
self._openai_client: AsyncOpenAI = AsyncOpenAI(api_key=self._bot_settings.openai_api_key)
self._anthropic_client: AsyncAnthropic = AsyncAnthropic(api_key=self._bot_settings.anthropic_api_key)
self._gemini_client: genai.Client = genai.Client(api_key=self._bot_settings.gemini_api_key)
self._effort: ReasoningEffort = "xhigh"
self._instructions: str = "You are a helpful AI assistant."
self._instructions_web: str = (
Expand All @@ -26,33 +31,54 @@ def __init__(self, bot: Bot) -> None:
"numbers, statistics, or breakdowns that the sources do not explicitly state. Cite the source URL(s)."
)

# ─────────────────────────── Commands ───────────────────────────

@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def gpt(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask OpenAI's GPT model for a direct answer (no web search)."""
await self._run_chat(ctx, msg_text, provider="openai", use_web=False)

@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def gptweb(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask OpenAI's GPT model with web search enabled — for current/factual info."""
await self._run_chat(ctx, msg_text, provider="openai", use_web=True)

@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def claude(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask Anthropic's Claude model for a direct answer (no web search)."""
await self._run_chat(ctx, msg_text, provider="anthropic", use_web=False)

@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def ai(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask OpenAI for a direct answer (no web search).
async def claudeweb(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask Anthropic's Claude model with web search enabled — for current/factual info."""
await self._run_chat(ctx, msg_text, provider="anthropic", use_web=True)

Usage:
ai What is Python?
ai Write a haiku about programming
ai Explain quantum computing in simple terms
"""
await self._run_ai(ctx, msg_text, use_web=False)
@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def gemini(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask Google's Gemini model for a direct answer (no web search)."""
await self._run_chat(ctx, msg_text, provider="gemini", use_web=False)

@commands.command()
@commands.guild_only()
@commands.cooldown(1, CoolDowns.OpenAI.value, commands.BucketType.user)
async def aiweb(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask OpenAI to search the web before answering — best for current/factual info.

Usage:
aiweb What's the latest news about <topic>
aiweb How many support gems does Path of Exile 2 have
"""
await self._run_ai(ctx, msg_text, use_web=True)

async def _run_ai(self, ctx: commands.Context, msg_text: str, use_web: bool) -> None:
"""Shared body for the `ai` and `aiweb` commands."""
# Reasoning (and web search) can take a couple of minutes, so show a progress
# message immediately so the user knows the bot is working (not stuck).
async def geminiweb(self, ctx: commands.Context, *, msg_text: str) -> None:
"""Ask Google's Gemini model with Google Search grounding enabled."""
await self._run_chat(ctx, msg_text, provider="gemini", use_web=True)

# ─────────────────────────── Shared flow ───────────────────────────

async def _run_chat(self, ctx: commands.Context, msg_text: str, provider: str, use_web: bool) -> None:
"""Send progress message, dispatch to provider, time, post answer."""
progress_text = (
"Please wait, I'm thinking and searching the web for an accurate answer..."
if use_web
Expand All @@ -67,38 +93,50 @@ async def _run_ai(self, ctx: commands.Context, msg_text: str, use_web: bool) ->

start = time.monotonic()
try:
response_text = await self._get_ai_response(msg_text, use_web=use_web)
response_text = await self._dispatch(provider, msg_text, use_web)
color = discord.Color.green()
description = response_text
except Exception as e:
self.bot.log.error(f"OpenAI API error: {e}")
self.bot.log.error(f"{provider} API error: {e}")
color = discord.Color.red()
description = f"Sorry, I encountered an error: {e}"
elapsed = time.monotonic() - start

# Remove the progress message before sending the final answer.
try:
await progress_msg.delete()
except discord.HTTPException:
pass

embeds = self._create_ai_embeds(ctx, description, color, elapsed)
model = self._model_for(provider)
embeds = self._create_ai_embeds(ctx, description, color, elapsed, model)
if len(embeds) == 1:
await bot_utils.send_embed(ctx, embeds[0], False)
else:
view = bot_utils.EmbedPaginatorView(embeds, ctx.author.id)
await view.send_and_save(ctx)

async def _get_ai_response(self, message: str, use_web: bool) -> str:
"""Get response from OpenAI API.

use_web: when True, enables the built-in web_search tool and uses the
web-grounded instructions. When False, the model answers from its own
knowledge with plain instructions.
"""
async def _dispatch(self, provider: str, message: str, use_web: bool) -> str:
if provider == "openai":
return await self._get_openai_response(message, use_web=use_web)
if provider == "anthropic":
return await self._get_claude_response(message, use_web=use_web)
if provider == "gemini":
return await self._get_gemini_response(message, use_web=use_web)
raise ValueError(f"Unknown provider: {provider}")

def _model_for(self, provider: str) -> str:
return {
"openai": self._bot_settings.openai_model,
"anthropic": self._bot_settings.anthropic_model,
"gemini": self._bot_settings.gemini_model,
}[provider]

# ─────────────────────────── Provider calls ───────────────────────────

async def _get_openai_response(self, message: str, use_web: bool) -> str:
"""Call OpenAI's Responses API with optional web_search tool."""
instructions = self._instructions_web if use_web else self._instructions
tools: list[WebSearchToolParam] = [WebSearchToolParam(type="web_search")] if use_web else []

response = await self._openai_client.responses.create(
instructions=instructions,
model=self._bot_settings.openai_model,
Expand All @@ -107,10 +145,44 @@ async def _get_ai_response(self, message: str, use_web: bool) -> str:
max_output_tokens=None,
input=message,
)

content = response.output_text
return content.strip() if content else ""

async def _get_claude_response(self, message: str, use_web: bool) -> str:
"""Call Anthropic Messages API with optional web_search server tool."""
instructions = self._instructions_web if use_web else self._instructions
tools = [{"type": "web_search_20250305", "name": "web_search"}] if use_web else []
response = await self._anthropic_client.messages.create(
model=self._bot_settings.anthropic_model,
max_tokens=4096,
system=instructions,
messages=[{"role": "user", "content": message}],
tools=tools,
)
# Concatenate all text blocks (web search may interleave tool-use blocks).
text_parts = [
getattr(block, "text", "") for block in response.content if getattr(block, "type", None) == "text"
]
content = "".join(text_parts).strip()
return content

async def _get_gemini_response(self, message: str, use_web: bool) -> str:
"""Call Google Gemini with optional Google Search grounding."""
instructions = self._instructions_web if use_web else self._instructions
config_kwargs: dict = {"system_instruction": instructions}
if use_web:
config_kwargs["tools"] = [genai_types.Tool(google_search=genai_types.GoogleSearch())]
config = genai_types.GenerateContentConfig(**config_kwargs)
response = await self._gemini_client.aio.models.generate_content(
model=self._bot_settings.gemini_model,
contents=message,
config=config,
)
content = response.text or ""
return content.strip()

# ─────────────────────────── Embed formatting ───────────────────────────

@staticmethod
def _format_duration(seconds: float) -> str:
"""Format an elapsed duration as e.g. '5ms' (sub-second) or '20s'."""
Expand All @@ -119,13 +191,19 @@ def _format_duration(seconds: float) -> str:
return f"{round(seconds)}s"

def _create_ai_embeds(
self, ctx: commands.Context, description: str, color: discord.Color, elapsed: float = 0.0
self,
ctx: commands.Context,
description: str,
color: discord.Color,
elapsed: float = 0.0,
model: str | None = None,
) -> list[discord.Embed]:
"""Create formatted embed(s) for AI response, paginating if needed."""
model = self._bot_settings.openai_model
"""Create formatted embed(s) for an AI response, paginating if needed."""
if model is None:
model = self._bot_settings.openai_model
duration = self._format_duration(elapsed)
max_length = 2000
chunks = []
chunks: list[str] = []

while description:
if len(description) <= max_length:
Expand Down
10 changes: 10 additions & 0 deletions src/bot/constants/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,16 @@ class BotSettings(BaseSettings):
openai_model: str = Field(default="gpt-5.5", description="https://developers.openai.com/api/docs/models")
openai_api_key: str | None = Field(default=None)

# Anthropic (Claude)
anthropic_model: str = Field(
default="claude-opus-4-8", description="https://docs.anthropic.com/en/docs/about-claude/models"
)
anthropic_api_key: str | None = Field(default=None)

# Google (Gemini)
gemini_model: str = Field(default="gemini-flash-latest", description="https://ai.google.dev/gemini-api/docs/models")
gemini_api_key: str | None = Field(default=None)

# Cooldowns
admin_cooldown: int = Field(default=20)
config_cooldown: int = Field(default=20)
Expand Down
Loading
Loading