65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
# Agenten Plattform
|
||
#
|
||
# (c) 2024 Magnus Bender
|
||
# Institute of Humanities-Centered Artificial Intelligence (CHAI)
|
||
# Universitaet Hamburg
|
||
# https://www.chai.uni-hamburg.de/~bender
|
||
#
|
||
# source code released under the terms of GNU Public License Version 3
|
||
# https://www.gnu.org/licenses/gpl-3.0.txt
|
||
|
||
|
||
import argparse, sys, os, json
|
||
|
||
from ums.agent.process import MessageProcessor
|
||
from ums.utils import AgentMessage, Riddle, SHARE_PATH, PERSIST_PATH
|
||
|
||
class _FakeBackgroundTask():
|
||
def add_task(self, call, *args):
|
||
call(*args)
|
||
|
||
if __name__ == "__main__":
|
||
|
||
parser = argparse.ArgumentParser(description='Agenten Plattform – Run Single Task')
|
||
parser.add_argument('-f', '--file', help="fetch the message (riddle) from this json file")
|
||
parser.add_argument('-d', '--dummy', help="use a dummy message (riddle)", action="store_true")
|
||
parser.add_argument('-i', '--stdin', help="get the message (riddle) from STDIN", action="store_true")
|
||
|
||
args = parser.parse_args()
|
||
message = None
|
||
|
||
if args.dummy:
|
||
message = AgentMessage(
|
||
id="dummy",
|
||
riddle=Riddle(context="Its a dummy.", question="No question!")
|
||
)
|
||
message.status.extract.required = False
|
||
elif args.stdin:
|
||
text = ''.join(sys.stdin.readlines())
|
||
message = AgentMessage.model_validate_json(text)
|
||
elif args.file:
|
||
if os.path.isfile(args.file):
|
||
f_name = args.file
|
||
elif os.path.isfile(os.path.join(SHARE_PATH, args.file)):
|
||
f_name = os.path.join(SHARE_PATH, args.file)
|
||
elif os.path.isfile(os.path.join(PERSIST_PATH, args.file)):
|
||
f_name = os.path.join(PERSIST_PATH, args.file)
|
||
else:
|
||
print()
|
||
print(f"\tFile {args.file} not found!")
|
||
print()
|
||
f_name = None
|
||
|
||
if not f_name is None:
|
||
message = AgentMessage.model_validate(
|
||
json.load(open(f_name, 'r'))
|
||
)
|
||
|
||
if message is None:
|
||
parser.print_help()
|
||
else:
|
||
mp = MessageProcessor(disable_messages=True)
|
||
response = mp.new_message(message, _FakeBackgroundTask())
|
||
|
||
print("\tResponse:")
|
||
print(response.model_dump_json(indent=2)) |