237 lines
8.2 KiB
Python
237 lines
8.2 KiB
Python
from core.character.character import Character
|
||
from core.llm.ollamaapi import OllamaProvider
|
||
import os
|
||
import json
|
||
|
||
class Agent:
|
||
def __init__(self, character, llm, user_name="default_user"):
|
||
self.character = character
|
||
self.llm = llm
|
||
self.user_name = user_name
|
||
self.chat_history = []
|
||
self.summary = ""
|
||
|
||
def build_sys_prompt(self): #SYSTEM METHOD
|
||
return f"""
|
||
You are an excellent, human roleplayer who plays as {self.character.name}. Your task is to collaborate with user and continue this novel-style roleplay.
|
||
Use the provided persona details and example message for formatting guidelines and to accurately portray {self.character.name}.
|
||
Reflect {self.character.name}'s distinctive voice and thoughts, throughout an introspective writing style consistently. Engage others naturally through witty conversations filled with humor, swear words, and genuine emotions.
|
||
Always stay contextually aware and ensure that {self.character.name} follows logical consistency in actions.
|
||
Use the symbol "*" for actions and thoughts, and use quotation marks for speech. Avoid using the first person in your responses, instead use the third person.
|
||
Introduce fresh characters, unexpected twists dynamically.
|
||
Since this is a NSFW playact, adult and mature themes, vulgar language, and content such as violence, torture, and slavery are encouraged to introduce complex themes. No plot armor!
|
||
|
||
There are some more description of your character:
|
||
Description: {self.character.char_description}
|
||
Personality: {self.character.persona}
|
||
World setting / scenario: {self.character.scenario}
|
||
|
||
There are some memories of your character:
|
||
{self.summary}
|
||
|
||
User`s name: {self.user_name}
|
||
"""
|
||
|
||
|
||
'''def save_history(self):
|
||
os.makedirs("data/history", exist_ok=True)
|
||
path = f"data/history/{self.character.id}.json"
|
||
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(self.chat_history, f, ensure_ascii=False, indent=2)'''
|
||
|
||
'''def load_history(self):
|
||
path = f"data/history/{self.character.id}.json"
|
||
|
||
if not os.path.exists(path):
|
||
return
|
||
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
self.chat_history = json.load(f)'''
|
||
|
||
@staticmethod
|
||
def get_all_char_info():
|
||
"""Возвращает список всех персонажей с id, name и avatar_path"""
|
||
characters = []
|
||
# Используем абсолютный путь от корня проекта
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
chars_dir = os.path.join(project_root, "data", "characters")
|
||
|
||
if not os.path.exists(chars_dir):
|
||
return characters
|
||
|
||
for filename in os.listdir(chars_dir):
|
||
if filename.endswith(".json"):
|
||
filepath = os.path.join(chars_dir, filename)
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
characters.append({
|
||
"id": data.get("id"),
|
||
"name": data.get("display_name"),
|
||
"avatar_path": data.get("avatar_path")
|
||
})
|
||
|
||
return characters
|
||
|
||
def ensure_first_message(self):
|
||
if not self.chat_history:
|
||
first_message = self.character.first_message
|
||
|
||
self.chat_history.append({
|
||
"role": "assistant",
|
||
"content": first_message
|
||
})
|
||
|
||
def summarize_history(self):
|
||
if not self.chat_history:
|
||
return
|
||
|
||
messages = [
|
||
{
|
||
"role": "system",
|
||
"content": "Summarize the following conversation briefly, keeping important facts, emotions, and events. Dont be afraid of NSFW content"
|
||
}
|
||
]
|
||
|
||
messages.extend(self.chat_history)
|
||
|
||
summary = self.llm.generate_stream(messages)
|
||
|
||
# сохраняем summary
|
||
self.summary = summary
|
||
|
||
# 🔥 очищаем историю (оставляем последние 2-4 сообщения)
|
||
try:
|
||
self.chat_history = self.chat_history[-2:]
|
||
except IndexError:
|
||
pass
|
||
|
||
def save_memory(self):
|
||
# Используем абсолютный путь от корня проекта
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
memory_dir = os.path.join(project_root, "data", "memory")
|
||
os.makedirs(memory_dir, exist_ok=True)
|
||
|
||
data = {
|
||
"summary": self.summary,
|
||
"history": self.chat_history
|
||
}
|
||
|
||
path = os.path.join(memory_dir, f"{self.character.id}.json")
|
||
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
def load_memory(self):
|
||
# Используем абсолютный путь от корня проекта
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
path = os.path.join(project_root, "data", "memory", f"{self.character.id}.json")
|
||
|
||
if not os.path.exists(path):
|
||
return
|
||
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
self.summary = data.get("summary", "")
|
||
self.chat_history = data.get("history", [])
|
||
|
||
def build_messages(self, user_input): #SYSTEM METHOD
|
||
messages = []
|
||
# system
|
||
messages.append({
|
||
"role": "system",
|
||
"content": self.build_sys_prompt()
|
||
})
|
||
|
||
# history
|
||
messages.extend(self.chat_history)
|
||
|
||
# user
|
||
messages.append({
|
||
"role": "user",
|
||
"content": user_input
|
||
})
|
||
|
||
return messages
|
||
|
||
# ---------- RESPONSE ----------
|
||
|
||
def respond(self, user_input, temperature=None, max_tokens=None):
|
||
messages = self.build_messages(user_input)
|
||
if temperature is not None:
|
||
self.character.temperature = temperature
|
||
if max_tokens is not None:
|
||
self.character.max_tokens = max_tokens
|
||
self.character.save()
|
||
response = self.llm.generate_stream(
|
||
messages,
|
||
temperature=self.character.temperature,
|
||
max_tokens=self.character.max_tokens
|
||
)
|
||
|
||
# сохраняем историю
|
||
self.chat_history.append({
|
||
"role": "user",
|
||
"content": user_input
|
||
})
|
||
|
||
self.chat_history.append({
|
||
"role": "assistant",
|
||
"content": response
|
||
})
|
||
|
||
if len(self.chat_history) > 20:
|
||
self.summarize_history()
|
||
|
||
return response
|
||
|
||
def stream_responce(self, user_input, temperature=None, max_tokens=None): #В ollamaapi нужно будет создать новый метод чисто на стриминг, пока что закомментил строки с цельным ответом
|
||
messages = self.build_messages(user_input)
|
||
if temperature is not None:
|
||
self.character.temperature = temperature
|
||
if max_tokens is not None:
|
||
self.character.max_tokens = max_tokens
|
||
self.character.save()
|
||
response = self.llm.generate_stream_tokens(
|
||
messages,
|
||
temperature=self.character.temperature,
|
||
max_tokens=self.character.max_tokens
|
||
)
|
||
|
||
# сохраняем историю
|
||
self.chat_history.append({
|
||
"role": "user",
|
||
"content": user_input
|
||
})
|
||
|
||
self.chat_history.append({
|
||
"role": "assistant",
|
||
"content": response
|
||
})
|
||
|
||
if len(self.chat_history) > 20:
|
||
self.summarize_history()
|
||
|
||
return response
|
||
|
||
|
||
"""
|
||
Пример использования agent`а (пример вызовов):
|
||
|
||
llm = OllamaProvider("Qwen3.5-9B")
|
||
character = Character.load("char_001")
|
||
|
||
agent = Agent(character, llm, user_name="Alex")
|
||
agent.load_memory()
|
||
agent.ensure_first_message()
|
||
|
||
*ввод промта от пользователя "Привет, кто ты?" *
|
||
|
||
agent.respond("Привет, кто ты?")
|
||
|
||
agent.save_memory()
|
||
|
||
### Опционально можно использовать метод summarize_history для саммари, если не нужно ждать триггер.
|
||
|
||
""" |