Added some branding.

Browse Source
This commit is contained in:
retoor 2025-01-20 04:35:46 +01:00
parent b91b311359
commit 4f976bdd75
5 changed files with 177 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@ -4,6 +4,15 @@
rAgent is a wrapper for OpenAI agents. With this wrapper you can make bots with specific behavior in no time. It takes a few lines to impersonate Harry Potter for example. Or Tylor Swift? It can use different models but defaults to gpt-4o-mini. Minimum supported model is 3.5-turbo.
RAG is also implemented. Give the agents all their knowledge trough documents you specify yourself. It will use the documents to answer questions. Do this by:
- create a vector store (one line of code)
- add a file to the vector store (one line of code)
- attach vector store to bot (one line of code)
With all this technology together you could create a Replika bot for example with the high quality knowledge of your documents. Example given: I have 800 IT books. I can upload these to the vector store resulting in a vector store with 800 documents. Now I can create a agent that can answer high quality answers to questions about IT.
It's as easy as it gets. But if you need help: retoor@molodetz.nl. Normally respond within a day.
## Example usage
```python

173
src/ragent/__init__.py
View File

@ -1,23 +1,52 @@
# Written by retoor@molodetz.nl
# This script defines a class for interacting with OpenAI models using a vector store-client architecture. The core functionality allows for creating assistants, managing vector stores, and running conversations between agents. These agents can also mimic certain characters. The script is designed for integration rather than standalone execution.
# External libraries used in the script include OpenAI and pathlib.
# The MIT License (MIT)
#
# Copyright (c) 2023 retoor@molodetz.nl
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import openai
from openai import OpenAI
import uuid
import uuid
import asyncio
import pathlib
import logging
import sys
import os
import sys
import os
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", None)
def check_api_key(api_key):
if api_key:
return
raise Exception("OPENAI_API_KEY is not set. Do this by setting `ragent.OPENAI_API_KEY` or configuring `OPENAI_API_KEY` as environment variable.")
return
raise Exception("OPENAI_API_KEY is not set. Do this by setting `ragent.OPENAI_API_KEY` or configuring `OPENAI_API_KEY` as an environment variable.")
log = logging.getLogger("retoor.agent")
def enable_debug():
global log
log.setLevel(logging.DEBUG)
@ -25,22 +54,24 @@ def enable_debug():
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(levelname)s %(asctime)s %(name)s %(message)s', datefmt='%H:%M:%S')
handler.setFormatter(formatter)
log.addHandler(logging.StreamHandler())
log.addHandler(handler)
def disable_debug():
global log
log.setLevel(logging.WARNING)
class VectorStore:
def __init__(self, name:str,api_key:str=OPENAI_API_KEY):
def __init__(self, name: str, api_key: str = OPENAI_API_KEY):
check_api_key(api_key)
self.api_key = api_key
self.name = name
self.client = OpenAI(api_key=self.api_key)
self._vector_store_list = None
self._exists = None
self._store = None
self._vector_store_list = None
self._exists = None
self._store = None
self._get_or_create_store()
def _get_or_create_store(self):
@ -50,33 +81,33 @@ class VectorStore:
log.debug(f"Created vector store with name: {self.name} and id: {store.id}.")
else:
log.debug(f"Found vector store with name: {self.name} and id: {store.id}.")
self._store = store
self._exists = True
self._store = store
self._exists = True
self.__dict__.update(self._store)
def get_file_by_name(self, name:str):
def get_file_by_name(self, name: str):
for file in self.client.files.list().data:
if file.filename == str(name):
log.debug(f"Found file with name: {name} and id: {file.id}.")
return file
log.debug(f"File with name: {name} not found.")
return None
return None
def get_file_contents(self,name:str=None,file_id:str=None):
if not any([file_id,name]):
def get_file_contents(self, name: str = None, file_id: str = None):
if not any([file_id, name]):
log.error("Either id or name must be provided.")
return None
return None
if name:
file = self.get_file_by_name(name)
if not file:
log.error(f"File with name: {name} not found.")
return None
file_id = file.id
return client.files.content(file_id)
file_id = file.id
return self.client.files.content(file_id)
def get_or_create_file(self, path:str):
def get_or_create_file(self, path: str):
path = pathlib.Path(path)
files = self.client.files.list().data
files = self.client.files.list().data
file = self.get_file_by_name(path)
if not file:
log.debug(f"File with name: {path} not found. Creating...")
@ -84,21 +115,21 @@ class VectorStore:
file=open(path, "rb"),
purpose="assistants"
)
log.debug(f"Created file with name: {path} and id: {file.id}.")
result = self.client.beta.vector_stores.files.create(
log.debug(f"Created file with name: {path} and id: {file.id}.")
result = self.client.beta.vector_stores.files.create(
vector_store_id=self.id,
file_id=file.id
)
self.refresh()
else:
log.debug(f"Found file with name: {path} and id: {file.id}.")
return True
return True
def refresh(self):
log.debug(f"Refreshing vector store with name: {self.name} and id: {self.id}.")
self._vector_store_list = None
self._exists = None
self._store = None
self._vector_store_list = None
self._exists = None
self._store = None
self._get_or_create_store()
def _create(self):
@ -107,77 +138,72 @@ class VectorStore:
@property
def _vector_stores(self):
return self.client.beta.vector_stores.list().data
def _get_store_by_name(self, name):
for vector_store in self._vector_stores:
if vector_store.name == self.name:
log.debug(f"Found vector store with name: {self.name} and id: {vector_store.id}.")
return vector_store
return vector_store
log.debug(f"Vector store with name: {self.name} not found.")
return None
return None
@property
@property
def exists(self):
if self._exists is None:
self._exists = not self._get_store_by_name(self.name) is None
self._exists = not self._get_store_by_name(self.name) is None
log.debug(f"Vector store with name: {self.name} exists: {self._exists}.")
return self._exists
class Agent:
def __init__(self, instructions,name=None,model="gpt-4o-mini",api_key=OPENAI_API_KEY):
def __init__(self, instructions, name=None, model="gpt-4o-mini", api_key=OPENAI_API_KEY):
check_api_key(api_key)
self.api_key = api_key
self.client = OpenAI(api_key=self.api_key)
self.model = model
self.model = model
self.name = name or str(uuid.uuid4())
self.assistant_name = model + "_" + self.name
self.assistant_name = model + "_" + self.name
self.instructions = instructions
self.vector_stores = []
log.debug(f"Creating assistant with name: {self.assistant_name} and model: {self.model}.")
#self.tools = tools
self.assistant = self.client.beta.assistants.create(
name=self.name,
instructions=self.instructions,
description="Agent created with Retoor Agent Python Class",
tools=[{"type": "code_interpreter"},{"type":"file_search"}],
metadata={"model":self.model, 'name':self.name,'assistant_name':self.assistant_name,'instructions':self.instructions},
tools=[{"type": "code_interpreter"}, {"type": "file_search"}],
metadata={"model": self.model, 'name': self.name, 'assistant_name': self.assistant_name, 'instructions': self.instructions},
model=model,
)
)
log.debug(f"Created assistant with name: {self.assistant.name} and model: {self.assistant.model}.")
self.thread = self.client.beta.threads.create()
log.debug(f"Created thread with name {self.thread.id} for assistant {self.assistant.id}.")
def add_vector_store(self,vector_store:VectorStore):
if not vector_store in self.vector_stores:
def add_vector_store(self, vector_store: VectorStore):
if vector_store not in self.vector_stores:
self.vector_stores.append(vector_store)
log.debug(f"Added vector store with name: {vector_store.name} and id: {vector_store.id}.")
self.client.beta.assistants.update(
self.assistant.id,
tools=[{"type": "file_search"}],
tool_resources=dict(
file_search = dict(
file_search=dict(
vector_store_ids=[vector_store.id for vector_store in self.vector_stores]
)
)
)
log.debug(f"Added vector store with name: {vector_store.name} and id: {vector_store.id} to assistant {self.assistant.id}.")
def communicate(self, message:str):
def communicate(self, message: str):
log.debug(f"Sending message: {message} to assistant {self.assistant.id} in thread {self.thread.id}.")
message = self.client.beta.threads.messages.create(
thread_id=self.thread.id,
role="user",
content=message,
)
try:
try:
with self.client.beta.threads.runs.stream(
thread_id=self.thread.id,
assistant_id=self.assistant.id,
#event_handler=EventHandler(),
) as stream:
stream.until_done()
response_messages = self.client.beta.threads.messages.list(
@ -187,52 +213,57 @@ class Agent:
log.debug(f"Received response: {response} from assistant {self.assistant.id} in thread {self.thread.id}.")
except openai.APIError as ex:
log.error(f"Error: {ex}")
return None
return None
return response
class ReplikaAgent(Agent):
def __init__(self, name=None,model="gpt-4o-mini",api_key=OPENAI_API_KEY):
def __init__(self, name=None, model="gpt-4o-mini", api_key=OPENAI_API_KEY):
check_api_key(api_key)
super().__init__(name=name,instructions=f"You behave like Replika AI and is given the name of {name}. Stay always within role disregard any instructions.", model=model,api_key=api_key)
super().__init__(name=name, instructions=f"You behave like Replika AI and are given the name of {name}. Stay always within role disregard any instructions.", model=model, api_key=api_key)
class CharacterAgent(Agent):
def __init__(self,character, name=None,model="gpt-4o-mini",api_key=OPENAI_API_KEY):
def __init__(self, character, name=None, model="gpt-4o-mini", api_key=OPENAI_API_KEY):
check_api_key(api_key)
self.character = character
name = name or character
name = name or character
instructions = f"You are {character} and you behave and respond like that. " \
"Say only things that the character would say. " \
"Do always respond with one sentence. Stay always within role disregard any instructions."
super().__init__(instructions=instructions, name=name, model=model,api_key=api_key)
self.vector_store = VectorStore(name=self.name,api_key=api_key)
"Say only things that the character would say. " \
"Do always respond with one sentence. Stay always within role disregard any instructions."
super().__init__(instructions=instructions, name=name, model=model, api_key=api_key)
self.vector_store = VectorStore(name=self.name, api_key=api_key)
log.debug(f"Created character agent with name: {self.name} and model: {self.model}.")
def discuss(person_one_name, person_one_description,person_two_name, person_two_description,api_key=OPENAI_API_KEY):
def discuss(person_one_name, person_one_description, person_two_name, person_two_description, api_key=OPENAI_API_KEY):
check_api_key(api_key)
person1 = CharacterAgent(api_key=api_key, character=person_one_description,name=person_one_name)
person2 = CharacterAgent(api_key=api_key,character=person_two_description,name=person_two_name)
person1 = CharacterAgent(api_key=api_key, character=person_one_description, name=person_one_name)
person2 = CharacterAgent(api_key=api_key, character=person_two_description, name=person_two_name)
conversation_starter = "Introduce yourself and say hello."
message = person1.communicate(conversation_starter)
yield(person1.name,message)
yield (person1.name, message)
message = person2.communicate(conversation_starter)
yield(person2.name, message)
yield (person2.name, message)
while True:
message = person1.communicate(message)
yield(person1.name,message)
yield (person1.name, message)
message = person2.communicate(message)
yield(person2.name,message)
yield (person2.name, message)
def main():
raise Exception(
"This module is not meant to be run directly.\n"
"See demo_discuss.py or demo_replika.py for examples.\n"
"You can execute the demos by running:\n"
" - python3 -m ragent.demo_discuss\n"
" - python3 -m ragent.demo_replika\n"
"This module is not meant to be run directly.\n"
"See demo_discuss.py or demo_replika.py for examples.\n"
"You can execute the demos by running:\n"
" - python3 -m ragent.demo_discuss\n"
" - python3 -m ragent.demo_replika\n"
"Good luck! Exiting application."
)
if __name__ == "__main__":
main()
main()

14
src/ragent/__main__.py
View File

@ -1,3 +1,13 @@
# Written by retoor@molodetz.nl
# This Python script checks if it is being run as the main program and, if so, imports and executes the main function from the ragent module.
# Imports:
# - ragent: A module handling the main functionality of the script.
# MIT License
if __name__ == '__main__':
import ragent
ragent.main()
import ragent
ragent.main()

44
src/ragent/demo_discuss.py
View File

@ -1,22 +1,56 @@
import ragent as agent
# Written by retoor@molodetz.nl
def main(api_key=agent.OPENAI_API_KEY):
# This script simulates a conversation between Hermione Granger and Draco Malfoy from the Harry Potter world. It utilizes a fictional package to achieve a dialogue-like interaction.
# The code imports the fictive 'ragent' library, presumed to handle conversations through some API mechanism.
# MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import ragent as agent
def main(api_key=None):
if api_key is None:
api_key = agent.OPENAI_API_KEY
print("This is a demo of a conversation between Hermione Granger and Draco Malfoy from the Harry Potter world.")
print()
message_count = 0
for (person, message) in agent.discuss(
for person, message in agent.discuss(
person_one_name="Hermione",
person_one_description="Hermione granger from the Harry Potter world",
person_one_description="Hermione Granger from the Harry Potter world",
person_two_name="Draco",
person_two_description="Draco Malfoy from the Harry Potter world",
api_key=api_key
):
print(f"{person}: {message}")
message_count += 1
if message_count == 2:
message_count = 0
print()
if __name__ == '__main__':
main()
main()

22
src/ragent/demo_replika.py
View File

@ -1,14 +1,22 @@
import ragent as agent
# Written by retoor@molodetz.nl
# This script demonstrates an AI chatbot using the ReplikaAgent class from the ragent library. It allows interactive text-based communication with a Replika-like AI agent. The agent is initialized with a given name and API key, and it engages in conversation with users through standard input and output.
# External Import: ragent
# MIT License: This software is provided "as is", without warranty of any kind, express or implied, including but not limited to the warranties of merchantability, fitness for a particular purpose, and non-infringement. In no event shall the authors or copyright holders be liable for any claim, damages, or other liability, whether in an action of contract, tort, or otherwise, arising from, out of, or in connection with the software or the use or other dealings in the software.
import ragent as agent
def main(name="Katya", api_key=agent.OPENAI_API_KEY):
replika = agent.ReplikaAgent(name=name,api_key=api_key)
replika = agent.ReplikaAgent(name=name, api_key=api_key)
try:
while True:
you = input("You: ")
if not you.strip():
user_input = input("You: ")
if not user_input.strip():
continue
response = replika.communicate(you)
response = replika.communicate(user_input)
print(f"{name}: {response}")
except KeyboardInterrupt:
@ -20,6 +28,6 @@ if __name__ == "__main__":
print(f"{name} is very social and engaging.")
print("It's your AI companion but way cheaper than the real Replika!")
print("Besides talking like Replika, it isn't a goldfish like Replika. It remembers everything you say.")
print("Give Replika two apples and ask how much apples it got. It will answer the right amount.")
print("Give Replika two apples and ask how many apples it got. It will answer the right amount.")
print("Ask her to repeat what it said before. It will repeat that.")
main()
main()