How to handle the Human in the loop for concurrent agents and topic-subscription based scenarios

Question

Dear Community,

I'm currently working on incorporating a Human-in-the-Loop (HIL) approach using AutoGen and facing some challenges. Here are a few scenarios where I need guidance:

1. How can I effectively implement HIL in my Agentic solution built with the AutoGen framework using its core SDK (Topics-Subscription, concurrent-agents)?
2. I'm using the core SDK and would like to understand how APIs should be designed, especially when handling asynchronous cases involving HIL.

Any insights, examples,Sample-snippets, or best practices would be greatly appreciated!

Thanks in advance!

https://microsoft.github.io/autogen/stable/user-guide/core-user-guide/design-patterns/concurrent-agents.html

https://microsoft.github.io/autogen/stable/user-guide/core-user-guide/cookbook/topic-subscription-scenarios.html

Br,

Ash

Answer 1

Hello ASHLIN GABRIEL RAJAN,

Welcome to the Microsoft Q&A and thank you for posting your questions here.

I understand that you would like to know how you can handle the Human in the loop for concurrent agents and topic-subscription based scenarios.

To effectively integrate Human in the Loop (HIL) in concurrent and topic subscription based AutoGen solutions, consider the following approach:

1. Work on your design or architecture by:
   1. Implement a dedicated queue or decision manager that routes requests to humans asynchronously.
   2. Instead of a simple event bus , consider Azure Service Bus, RabbitMQ, or Kafka for better handling of concurrent agent communication. - https://learn.microsoft.com/en us/azure/service bus messaging/service bus messaging overview
   3. Use state management solutions (Redis, CosmosDB) to track pending requests and avoid conflicts.
2. Let your APIs follow a callback based model to handle human intervention efficiently. For an example API Flow for Asynchronous Human Review
   1. Agent publishes a task needing human validation to a queue and human receives a notification (via Teams, Email, Slack, or a UI ).
   2. If human doesn't respond within X minutes , agent takes a default action.
   3. Agents receive the response and proceed with execution . This is a REST API code snippet example with Webhooks for HIL:

            from flask import Flask, request, jsonify
            from threading import Thread
            import time
            app = Flask(__name__)
              Simulated Decision Storage (Can be replaced with Redis or a DB)
            pending_decisions = {}
            @app.route('/request_decision', methods=['POST'])
            def request_decision():
                data = request.json
                task_id = data['task_id']
                
                  Store pending decision (Simulate async request to human)
                pending_decisions[task_id] = "Pending"
                print(f"Human review needed for task {task_id}")
                return jsonify({"message": "Decision request sent", "task_id": task_id})
            @app.route('/submit_decision', methods=['POST'])
            def submit_decision():
                data = request.json
                task_id = data['task_id']
                decision = data['decision']
                if task_id in pending_decisions:
                    pending_decisions[task_id] = decision
                    return jsonify({"message": "Decision received", "task_id": task_id})
                
                return jsonify({"error": "Task ID not found"}), 400
              Background process to check decision status
            def monitor_decisions():
                while True:
                    for task_id, status in pending_decisions.items():
                        if status == "Pending":
                            print(f"Task {task_id} still awaiting human decision...")
                    time.sleep(5)
            Thread(target=monitor_decisions, daemon=True).start()
            if __name__ == '__main__':
                app.run(debug=True)
      

   In the above, the requests are stored until a human submits a decision and a separate background monitor check pending decisions, and webhooks or real time UI can be integrated for notifications.
3. The next is to enhance AutoGen Implementation, instead of the basic event bus, integrate a proper queue system:

      from autogen import Agent
      import queue
        Simulated Human Decision Queue
      human_review_queue = queue.Queue()
      class TaskAgent(Agent):
          def on_event(self, event):
              if event.topic == 'task_requires_human_review':
                  human_review_queue.put(event.data)
                  print(f"Task {event.data['task_id']} awaiting human review.")
      class HumanAgent(Agent):
          def on_event(self, event):
              if event.topic == 'human_review_completed':
                  decision = event.data.get('decision')
                  print(f"Human made decision: {decision}")
   

   The above shows, you can use a queue for pending human decisions, instead of assuming immediate response which supports scaling to multiple human reviewers if needed.
4. Check out the below resources and references to add for more details:
   1. https://microsoft.github.io/autogen/stable/api reference/index.html
   2. https://learn.microsoft.com/en us/azure/architecture/best practices/api design
   3. https://learn.microsoft.com/en us/azure/event grid/event handlers
   4. https://www.typhoon hil.com/documentation/typhoon hil api documentation/hil_api.html
   5. https://learn.microsoft.com/en us/azure/service bus messaging/service bus messaging overview
   6. https://developer.mozilla.org/en US/docs/Web/API/WebSockets_API

I hope this is helpful! Do not hesitate to let me know if you have any other questions or clarifications.

---

Please don't forget to close up the thread here by upvoting and accept it as an answer if it is helpful.