ums.example.example

 1# Agenten Plattform
 2#
 3# (c) 2024 Magnus Bender
 4# 	Institute of Humanities-Centered Artificial Intelligence (CHAI)
 5# 	Universitaet Hamburg
 6# 	https://www.chai.uni-hamburg.de/~bender
 7#  
 8# source code released under the terms of GNU Public License Version 3
 9# https://www.gnu.org/licenses/gpl-3.0.txt
10
11from typing import Callable, List
12from ums.agent import ExtractAudioAgent, ExtractImageAgent, ExtractTextAgent, SolveAgent, GatekeeperAgent
13
14from ums.utils import AgentMessage, Riddle, RiddleData, RiddleSolution, RiddleStatus, ExtractedData
15
16"""
17	Examples for simple agents.
18
19	Each agent is represented by its own class. The handling of tasks is done by `handle()` in each agent.
20
21	Finally `AGENT_CLASSES` contains the classes of the agents in a list. Via environmental variables this list is specified to the ums.agent system. 
22"""
23
24class MyExtractAudioAgent(ExtractAudioAgent):
25
26	def handle(self, data: RiddleData) -> RiddleData:
27		print("Audio Process:", data.file_plain)
28		return data
29
30class MyExtractImageAgent(ExtractImageAgent):
31
32	def handle(self, data: RiddleData) -> RiddleData:
33		print("Image Process:", data.file_plain)
34
35		extracted = ExtractedData(other={"info":"just a test"})
36		data.file_extracted = self.store_extracted(data, extracted)
37		return data
38	
39class MyExtractTextAgent(ExtractTextAgent):
40
41	def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool:
42		print("The response will be:", response)
43		return True
44
45	def handle(self, data: RiddleData) -> RiddleData:
46		print("Text Process:", data.file_plain)
47		return data
48
49
50class MySolveAgent(SolveAgent):
51
52	def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution:
53
54		for d in data:
55			print(self.get_extracted(d))
56		
57		if self.message().id == "test":
58			status = RiddleStatus()
59			status.extract.required = False
60			self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status)
61
62		return RiddleSolution(solution="Huii", explanation="Blubb")
63	
64
65class MyGatekeeperAgent(GatekeeperAgent):
66
67	def handle(self, solution: RiddleSolution, riddle: Riddle) -> RiddleSolution:
68		solution.accepted = True 
69		solution.review = "Ok"
70		
71		return solution
72
73AGENT_CLASSES = [
74	MyExtractAudioAgent, MyExtractImageAgent, MyExtractTextAgent,
75	MySolveAgent,
76	MyGatekeeperAgent
77]
class MyExtractAudioAgent(ums.agent.agent.ExtractAudioAgent):
25class MyExtractAudioAgent(ExtractAudioAgent):
26
27	def handle(self, data: RiddleData) -> RiddleData:
28		print("Audio Process:", data.file_plain)
29		return data

An extraction agent for audio, create a subclass for your agent.

def handle(self, data: ums.utils.types.RiddleData) -> ums.utils.types.RiddleData:
27	def handle(self, data: RiddleData) -> RiddleData:
28		print("Audio Process:", data.file_plain)
29		return data

Process the item data, create extraction file and return data with populated data.file_extracted.

class MyExtractImageAgent(ums.agent.agent.ExtractImageAgent):
31class MyExtractImageAgent(ExtractImageAgent):
32
33	def handle(self, data: RiddleData) -> RiddleData:
34		print("Image Process:", data.file_plain)
35
36		extracted = ExtractedData(other={"info":"just a test"})
37		data.file_extracted = self.store_extracted(data, extracted)
38		return data

An extraction agent for images, create a subclass for your agent.

def handle(self, data: ums.utils.types.RiddleData) -> ums.utils.types.RiddleData:
33	def handle(self, data: RiddleData) -> RiddleData:
34		print("Image Process:", data.file_plain)
35
36		extracted = ExtractedData(other={"info":"just a test"})
37		data.file_extracted = self.store_extracted(data, extracted)
38		return data

Process the item data, create extraction file and return data with populated data.file_extracted.

class MyExtractTextAgent(ums.agent.agent.ExtractTextAgent):
40class MyExtractTextAgent(ExtractTextAgent):
41
42	def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool:
43		print("The response will be:", response)
44		return True
45
46	def handle(self, data: RiddleData) -> RiddleData:
47		print("Text Process:", data.file_plain)
48		return data

An extraction agent for text, create a subclass for your agent.

def before_response( self, response: ums.utils.types.AgentMessage, send_it: Callable[[], NoneType]) -> bool:
42	def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool:
43		print("The response will be:", response)
44		return True

This method is called before the response is sent. If the method returns False no response will be sent. Thus, by overwriting this method, a response can be prevented.

The response to be sent is in response and send_it is a callable, which sends the response to the management if it gets called. (Hence, one may stop sending the response and later call send_it() to send the response.)

def handle(self, data: ums.utils.types.RiddleData) -> ums.utils.types.RiddleData:
46	def handle(self, data: RiddleData) -> RiddleData:
47		print("Text Process:", data.file_plain)
48		return data

Process the item data, create extraction file and return data with populated data.file_extracted.

class MySolveAgent(ums.agent.agent.SolveAgent):
51class MySolveAgent(SolveAgent):
52
53	def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution:
54
55		for d in data:
56			print(self.get_extracted(d))
57		
58		if self.message().id == "test":
59			status = RiddleStatus()
60			status.extract.required = False
61			self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status)
62
63		return RiddleSolution(solution="Huii", explanation="Blubb")

A solve agent, create a subclass for your agent.

def handle( self, riddle: ums.utils.types.Riddle, data: List[ums.utils.types.RiddleData]) -> ums.utils.types.RiddleSolution:
53	def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution:
54
55		for d in data:
56			print(self.get_extracted(d))
57		
58		if self.message().id == "test":
59			status = RiddleStatus()
60			status.extract.required = False
61			self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status)
62
63		return RiddleSolution(solution="Huii", explanation="Blubb")

Solve the riddle using data and return a solution.

class MyGatekeeperAgent(ums.agent.agent.GatekeeperAgent):
66class MyGatekeeperAgent(GatekeeperAgent):
67
68	def handle(self, solution: RiddleSolution, riddle: Riddle) -> RiddleSolution:
69		solution.accepted = True 
70		solution.review = "Ok"
71		
72		return solution

A gatekeeper agent, create a subclass for your agent.

def handle( self, solution: ums.utils.types.RiddleSolution, riddle: ums.utils.types.Riddle) -> ums.utils.types.RiddleSolution:
68	def handle(self, solution: RiddleSolution, riddle: Riddle) -> RiddleSolution:
69		solution.accepted = True 
70		solution.review = "Ok"
71		
72		return solution

Check the solution of riddle and return solution with populated solution.accepted and solution.review.

AGENT_CLASSES = [<class 'MyExtractAudioAgent'>, <class 'MyExtractImageAgent'>, <class 'MyExtractTextAgent'>, <class 'MySolveAgent'>, <class 'MyGatekeeperAgent'>]