Add NPC interaction system with memory and quest generation
- Introduced a new NPC system with dynamic NPCs and conversation handling. - Implemented NPC memory using SQLite to log conversations and manage relationships. - Added commands for talking to NPCs, listing available NPCs, and generating quests. - Updated database schema to support NPC conversations and relationships. - Refactored code structure to separate concerns into cogs and handlers.
This commit is contained in:
Binary file not shown.
Binary file not shown.
Executable
+62
@@ -0,0 +1,62 @@
|
||||
# main.py
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
import threading
|
||||
import itertools
|
||||
from npc_memory import NPCMemory
|
||||
from npc_handler import NPCHandler
|
||||
from cogs.npc import NPCCog
|
||||
|
||||
|
||||
class MyNewHelp(commands.MinimalHelpCommand):
|
||||
async def send_pages(self):
|
||||
destination = self.get_destination()
|
||||
for page in self.paginator.pages:
|
||||
emby = discord.Embed(description=page)
|
||||
await destination.send(embed=emby)
|
||||
|
||||
|
||||
class Client(commands.Bot):
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
command_prefix=self.iterate_prefix("py"),
|
||||
strip_after_prefix=True,
|
||||
case_insensitive=True,
|
||||
intents=discord.Intents.all(),
|
||||
help_command=MyNewHelp(),
|
||||
)
|
||||
def iterate_prefix(self, prefix):
|
||||
prefixes = list(map(''.join, itertools.product(*zip(prefix.upper(), prefix.lower()))))
|
||||
print(prefixes)
|
||||
|
||||
return prefixes
|
||||
|
||||
async def setup_hook(self): # overwriting a handler
|
||||
cogs_folder = f"{os.path.abspath(os.path.dirname(__file__))}/cogs"
|
||||
for filename in os.listdir(cogs_folder):
|
||||
if filename.endswith(".py"):
|
||||
try:
|
||||
await self.load_extension(f"cogs.{filename[:-3]}")
|
||||
except Exception as e:
|
||||
print(f"Failed to load {filename}: {e}")
|
||||
memory = NPCMemory()
|
||||
npc_handler = NPCHandler(memory)
|
||||
await self.add_cog(NPCCog(self, npc_handler))
|
||||
await self.tree.sync()
|
||||
print("Loaded cogs")
|
||||
|
||||
|
||||
def main():
|
||||
load_dotenv()
|
||||
client = Client()
|
||||
token = os.getenv("TOKEN")
|
||||
if token is not None:
|
||||
client.run(token)
|
||||
else:
|
||||
print("Token is missing.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
@@ -0,0 +1,51 @@
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
|
||||
class NPCCog(commands.Cog):
|
||||
def __init__(self, bot, npc_handler):
|
||||
self.bot = bot
|
||||
self.npc_handler = npc_handler
|
||||
self.player_contexts = {}
|
||||
|
||||
@commands.command(name="talk")
|
||||
async def talk_to_npc(self, ctx, npc_name: str, *, message: str):
|
||||
player_id = str(ctx.author.id)
|
||||
if player_id not in self.player_contexts:
|
||||
self.player_contexts[player_id] = {
|
||||
'id': player_id,
|
||||
'level': 1,
|
||||
'reputation': 0,
|
||||
'recent_actions': [],
|
||||
'location': 'Newhaven'
|
||||
}
|
||||
response = self.npc_handler.chat_with_npc(
|
||||
npc_name, message, self.player_contexts[player_id]
|
||||
)
|
||||
embed = discord.Embed(
|
||||
title=f"{npc_name} says...",
|
||||
description=response,
|
||||
color=discord.Color.blue()
|
||||
)
|
||||
await ctx.send(embed=embed)
|
||||
|
||||
@commands.command(name="npcs")
|
||||
async def list_npcs(self, ctx):
|
||||
npc_list = "\n".join([f"• {name}" for name in self.npc_handler.npcs.keys()])
|
||||
await ctx.send(f"Available NPCs:\n{npc_list}")
|
||||
|
||||
@commands.command(name="quest")
|
||||
async def get_quest(self, ctx, npc_name: str):
|
||||
player_id = str(ctx.author.id)
|
||||
player_level = self.player_contexts.get(player_id, {}).get('level', 1)
|
||||
quest = self.npc_handler.generate_quest(npc_name, player_level)
|
||||
if quest:
|
||||
embed = discord.Embed(
|
||||
title=quest["title"],
|
||||
description=quest["description"],
|
||||
color=discord.Color.gold()
|
||||
)
|
||||
embed.add_field(name="Reward", value=quest["reward"])
|
||||
embed.add_field(name="Difficulty", value=quest["difficulty"])
|
||||
await ctx.send(embed=embed)
|
||||
else:
|
||||
await ctx.send("NPC not found or unable to generate quest.")
|
||||
@@ -0,0 +1,50 @@
|
||||
import json
|
||||
import random
|
||||
|
||||
NPC_DATABASE = {
|
||||
"barkeep_boris": {
|
||||
"personality": "Jovial and gossipy, knows everyone's business",
|
||||
"backstory": "Retired adventurer who settled down after losing party to dragon",
|
||||
"quirks": ["Offers free drinks to good storytellers", "Hates elves"]
|
||||
},
|
||||
"mad_wizard": {
|
||||
"personality": "Eccentric and forgetful, brilliant but scattered",
|
||||
"backstory": "Expelled from wizard college for 'creative' spellcasting",
|
||||
"quirks": ["Speaks to imaginary familiar", "Offers dangerous experimental potions"]
|
||||
}
|
||||
}
|
||||
|
||||
class DynamicNPC:
|
||||
def __init__(self, name, data):
|
||||
self.name = name
|
||||
self.personality = data["personality"]
|
||||
self.backstory = data["backstory"]
|
||||
self.quirks = data["quirks"]
|
||||
|
||||
class NPCHandler:
|
||||
def __init__(self, memory):
|
||||
self.memory = memory
|
||||
self.npcs = {name: DynamicNPC(name, data) for name, data in NPC_DATABASE.items()}
|
||||
|
||||
def chat_with_npc(self, npc_name, message, player_context):
|
||||
npc = self.npcs.get(npc_name)
|
||||
if not npc:
|
||||
return "That NPC doesn't exist."
|
||||
# Simple response logic (replace with LLM call as needed)
|
||||
response = f"{npc.name} ({npc.personality}): I heard you say '{message}'."
|
||||
self.memory.log_conversation(npc_name, player_context['id'], message, response)
|
||||
self.memory.update_affinity(npc_name, player_context['id'], 1)
|
||||
return response
|
||||
|
||||
def generate_quest(self, npc_name, player_level):
|
||||
npc = self.npcs.get(npc_name)
|
||||
if not npc:
|
||||
return None
|
||||
# Replace this with LLM call if available
|
||||
quest = {
|
||||
"title": f"{npc.name}'s Request",
|
||||
"description": f"Help {npc.name} with a task suitable for level {player_level}.",
|
||||
"reward": f"{random.randint(10, 100)} coins",
|
||||
"difficulty": random.choice(["easy", "medium", "hard"])
|
||||
}
|
||||
return quest
|
||||
@@ -0,0 +1,46 @@
|
||||
import sqlite3
|
||||
|
||||
class NPCMemory:
|
||||
def __init__(self):
|
||||
self.conn = sqlite3.connect('npc_memory.db')
|
||||
self.create_tables()
|
||||
|
||||
def create_tables(self):
|
||||
self.conn.execute('''
|
||||
CREATE TABLE IF NOT EXISTS npc_conversations (
|
||||
npc_id TEXT,
|
||||
player_id TEXT,
|
||||
message TEXT,
|
||||
response TEXT,
|
||||
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
self.conn.execute('''
|
||||
CREATE TABLE IF NOT EXISTS npc_relationships (
|
||||
npc_id TEXT,
|
||||
player_id TEXT,
|
||||
affinity INTEGER DEFAULT 0,
|
||||
last_interaction DATETIME
|
||||
)
|
||||
''')
|
||||
self.conn.commit()
|
||||
|
||||
def log_conversation(self, npc_id, player_id, message, response):
|
||||
self.conn.execute(
|
||||
'INSERT INTO npc_conversations (npc_id, player_id, message, response) VALUES (?, ?, ?, ?)',
|
||||
(npc_id, player_id, message, response)
|
||||
)
|
||||
self.conn.commit()
|
||||
|
||||
def update_affinity(self, npc_id, player_id, delta):
|
||||
cur = self.conn.cursor()
|
||||
cur.execute('SELECT affinity FROM npc_relationships WHERE npc_id=? AND player_id=?', (npc_id, player_id))
|
||||
row = cur.fetchone()
|
||||
if row:
|
||||
new_affinity = row[0] + delta
|
||||
cur.execute('UPDATE npc_relationships SET affinity=?, last_interaction=CURRENT_TIMESTAMP WHERE npc_id=? AND player_id=?',
|
||||
(new_affinity, npc_id, player_id))
|
||||
else:
|
||||
cur.execute('INSERT INTO npc_relationships (npc_id, player_id, affinity, last_interaction) VALUES (?, ?, ?, CURRENT_TIMESTAMP)',
|
||||
(npc_id, player_id, delta))
|
||||
self.conn.commit()
|
||||
Reference in New Issue
Block a user