Distinguish InteractiveAgent and StaticAgent

[yiranwu0]

Current logic of generic agent:

Pseudo code:

def receive(incoming_msg, sender):
  # 1. print incoming_msg
  # 2. process the incoming_msg and generate outbound_msg. (ask for human input or automatic reply)
  # 3. call send(outbound_msg, receiver = "sender") to complete the loop

def send(outbound_msg, receiver):
  # call receiver.receive(outbound_msg, self)
  # sender sending the message is equivalent to the receiver receiving the message from sender

After two agent are created, calling receive from one agent will start the interactions between two agents. We cannot construct a pipeline of multiple single-hop agents with this.
For example I want a workflow: question -> Planning Agent -> question and plan -> Solver Agent -> result.

Idea

• moving step 2: process the incoming_msg and generate outbound_msg out as an individual functionality
• distinguish interactive agents and static agents.

Agent.py

class Agent:
    def send(self, message: Union[Dict, str], recipient: "Agent"):
        """(Aabstract method) Send a message to another agent."""

    def receive(self, message: Union[Dict, str], sender: "Agent"):
        """(Abstract method) Receive a message from another agent."""
    
    # basically taking `auto_reply` out 
    def process(self, message):
        """(Abstract method) Process incoming message and generate reply"""

InteractiveAgent (previous GenericAgent):

class InteractiveAgent(Agent):

    def send(self, message: Union[Dict, str], recipient: "Agent"):
        recipient.receive(message, self) # unchanged from GenericAgent

    def receive(self, message: Union[Dict, str], sender: "Agent"):
        outbound_message = self.process(message)
        check_terminate() # pseudo function
        self.send(outbound_message)

   def process(self, message):
        # move human_input, auto_rely here
        return outbound_msg

StaticAgent (new):

class StaticAgent(Agent):

    def send(self, message: Union[Dict, str], recipient: "Agent"):
        pass # not needed

    def receive(self, message: Union[Dict, str], sender: "Agent"):
        pass # not needed

   def process(self, message):
        # process message
        return outbound_msg

Examples

• Use static agent planner and solver to complete workflow: question -> Planning Agent -> question and plan -> Solver Agent -> result:

  question = "Some question"
  plan = planner.process(question)
  result = solver.process(plan+question)

• Static + Interactive: First call a static planner to devise a plan, and then call two interactive agents with the plan, after get the answer, call a goal_checker static agent.
  This is different from define planners as a function in UserProxyAgent.

  planner, answer_checker = StaticAgent("planner"), StaticAgent("checker")
  question = "Some question"
  plan = planner.process(question)
  
  assistant, user = InteractiveAgent("assistant"), InteractiveAgent("user")
  user.initiate_chat(assistant, plan+question)
  
  answer = extract_result(user) # pseudo function
  answer_with_checks = answer_checker.process(answer)