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]
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.
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.
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
.
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.
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.)
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.
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.
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.
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
.