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 11import random 12 13from typing import Callable, List 14from ums.agent import ExtractAudioAgent, ExtractImageAgent, ExtractTextAgent, SolveAgent, GatekeeperAgent 15 16from ums.utils import AgentMessage, Riddle, RiddleData, RiddleSolution, RiddleStatus, ExtractedData 17 18""" 19 Examples for simple agents. 20 21 Each agent is represented by its own class. The handling of tasks is done by `handle()` in each agent. 22 23 Finally `AGENT_CLASSES` contains the classes of the agents in a list. Via environmental variables this list is specified to the ums.agent system. 24""" 25 26class MyExtractAudioAgent(ExtractAudioAgent): 27 28 def handle(self, data: RiddleData) -> RiddleData: 29 print("Audio Process:", data.file_plain) 30 return data 31 32class MyExtractImageAgent(ExtractImageAgent): 33 34 def handle(self, data: RiddleData) -> RiddleData: 35 print("Image Process:", data.file_plain) 36 37 extracted = ExtractedData(other={"info":"just a test"}) 38 data.file_extracted = self.store_extracted(data, extracted) 39 return data 40 41class MyExtractTextAgent(ExtractTextAgent): 42 43 def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool: 44 print("The response will be:", response) 45 return True 46 47 def handle(self, data: RiddleData) -> RiddleData: 48 print("Text Process:", data.file_plain) 49 return data 50 51 52class MySolveAgent(SolveAgent): 53 54 def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution: 55 56 for d in data: 57 print(self.get_extracted(d)) 58 59 if self.message().id == "test": 60 status = RiddleStatus() 61 status.extract.required = False 62 self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status) 63 64 return RiddleSolution(solution="Huii", explanation=f"Blubb, {random.random()}") 65 66 67class MyGatekeeperAgent(GatekeeperAgent): 68 69 def handle(self, solution: List[RiddleSolution], riddle: Riddle) -> RiddleSolution: 70 solution[0].accepted = True 71 solution[0].review = "Ok" 72 73 return solution 74 75AGENT_CLASSES = [ 76 MyExtractAudioAgent, MyExtractImageAgent, MyExtractTextAgent, 77 MySolveAgent, 78 MyGatekeeperAgent 79]
27class MyExtractAudioAgent(ExtractAudioAgent): 28 29 def handle(self, data: RiddleData) -> RiddleData: 30 print("Audio Process:", data.file_plain) 31 return data
An extraction agent for audio, create a subclass for your agent.
33class MyExtractImageAgent(ExtractImageAgent): 34 35 def handle(self, data: RiddleData) -> RiddleData: 36 print("Image Process:", data.file_plain) 37 38 extracted = ExtractedData(other={"info":"just a test"}) 39 data.file_extracted = self.store_extracted(data, extracted) 40 return data
An extraction agent for images, create a subclass for your agent.
35 def handle(self, data: RiddleData) -> RiddleData: 36 print("Image Process:", data.file_plain) 37 38 extracted = ExtractedData(other={"info":"just a test"}) 39 data.file_extracted = self.store_extracted(data, extracted) 40 return data
Process the item data, create extraction file and return data with populated data.file_extracted.
42class MyExtractTextAgent(ExtractTextAgent): 43 44 def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool: 45 print("The response will be:", response) 46 return True 47 48 def handle(self, data: RiddleData) -> RiddleData: 49 print("Text Process:", data.file_plain) 50 return data
An extraction agent for text, create a subclass for your agent.
44 def before_response(self, response: AgentMessage, send_it: Callable[[], None]) -> bool: 45 print("The response will be:", response) 46 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.)
53class MySolveAgent(SolveAgent): 54 55 def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution: 56 57 for d in data: 58 print(self.get_extracted(d)) 59 60 if self.message().id == "test": 61 status = RiddleStatus() 62 status.extract.required = False 63 self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status) 64 65 return RiddleSolution(solution="Huii", explanation=f"Blubb, {random.random()}")
A solve agent, create a subclass for your agent.
55 def handle(self, riddle: Riddle, data: List[RiddleData]) -> RiddleSolution: 56 57 for d in data: 58 print(self.get_extracted(d)) 59 60 if self.message().id == "test": 61 status = RiddleStatus() 62 status.extract.required = False 63 self.sub_riddle(riddle=Riddle(context="Haha", question="Blubber"), status=status) 64 65 return RiddleSolution(solution="Huii", explanation=f"Blubb, {random.random()}")
Solve the riddle using data and return a single solution.
68class MyGatekeeperAgent(GatekeeperAgent): 69 70 def handle(self, solution: List[RiddleSolution], riddle: Riddle) -> RiddleSolution: 71 solution[0].accepted = True 72 solution[0].review = "Ok" 73 74 return solution
A gatekeeper agent, create a subclass for your agent.
70 def handle(self, solution: List[RiddleSolution], riddle: Riddle) -> RiddleSolution: 71 solution[0].accepted = True 72 solution[0].review = "Ok" 73 74 return solution
Check the solution (multiple if multiple solver involved) of riddle and return solutions with populated solution[i].accepted and solution[i].review.