Feature: Composable Actor Platform for AutoGen (#1655)

* Core CAP components + Autogen adapter + Demo

* Cleanup Readme

* C# folder

* Cleanup readme

* summary_method bug fix

* CAN -> CAP

* pre-commit fixes

* pre-commit fixes

* modification of sys path should ignore E402

* fix pre-commit check issues

* Updated docs

* Clean up docs

* more refactoring

* better packaging refactor

* Refactoring for package changes

* Run demo app without autogencap installed or in the path

* Remove debug related sleep()

* removed CAP in some class names

* Investigate a logging framework that supports color in windows

* added type hints

* remove circular dependency

* fixed pre-commit issues

* pre-commit ruff issues

* removed circular definition

* pre-commit fixes

* Fix pre-commit issues

* pre-commit fixes

* updated for _prepare_chat signature changes

* Better instructions for demo and some minor refactoring

* Added details that explain CAP

* Reformat Readme

* More ReadMe Formatting

* Readme edits

* Agent -> Actor

* Broker can startup on it's own

* Remote AutoGen Agents

* Updated docs

* 1) StandaloneBroker in demo
2) Removed Autogen only demo options

* 1) Agent -> Actor refactor
2) init broker as early

* rename user_proxy -> user_proxy_conn

* Add DirectorySvc

* Standalone demo refactor

* Get ActorInfo from DirectorySvc when searching for Actor

* Broker cleanup

* Proper cleanup and remove debug sleep()

* Run one directory service only.

* fix paths to run demo apps from command line

* Handle keyboard interrupt

* Wait for Broker and Directory to start up

* Move Terminate AGActor

* Accept input from the user in UserProxy

* Move sleeps close to operations that bind or connect

* Comments

* Created an encapsulated CAP Pair for AutoGen pair communication

* pre-commit checks

* fix pre-commit

* Pair should not make assumptions about who is first and who is second

* Use task passed into InitiateChat

* Standalone directory svc

* Fix broken LFS files

* Long running DirectorySvc

* DirectorySvc does not have a status

* Exit DirectorySvc Loop

* Debugging Remoting

* Reduce frequency of status messages

* Debugging remote Actor

* roll back git-lfs updates

* rollback git-lfs changes

* Debug network connectivity

* pre-commit fixes

* Create a group chat interface familiar to AutoGen GroupChat users

* pre-commit fixes

Author: rajan-chari

samples/apps/cap/README.md

@@ -0,0 +1,54 @@
+# Composable Actor Platform (CAP) for AutoGen
+
+## I just want to run the demo!
+*Python Instructions (Windows, Linux, MacOS):*
+
+0) cd py
+1) pip install -r autogencap/requirements.txt
+2) python ./demo/App.py
+
+*Demo Notes:*
+1) Options involving AutoGen require OAI_CONFIG_LIST.
+   AutoGen python requirements: 3.8 <= python <= 3.11
+2) For option 2, type something in and see who receives the message.  Quit to quit.
+3) To view any option that displays a chart (such as option 4), you will need to disable Docker code execution. You can do this by setting the environment variable `AUTOGEN_USE_DOCKER` to `False`.
+
+*Demo Reference:*
+```
+    Select the Composable Actor Platform (CAP) demo app to run:
+    (enter anything else to quit)
+    1. Hello World Actor
+    2. Complex Actor Graph
+    3. AutoGen Pair
+    4. AutoGen GroupChat
+    5. AutoGen Agents in different processes
+    Enter your choice (1-5):
+```
+
+## What is Composable Actor Platform (CAP)?
+AutoGen is about Agents and Agent Orchestration.  CAP extends AutoGen to allows Agents to communicate via a message bus.  CAP, therefore, deals with the space between these components.  CAP is a message based, actor platform that allows actors to be composed into arbitrary graphs.
+
+Actors can register themselves with CAP, find other agents, construct arbitrary graphs, send and receive messages independently and many, many, many other things.
+```python
+    # CAP Platform
+    network = LocalActorNetwork()
+    # Register an agent
+    network.register(GreeterAgent())
+    # Tell agents to connect to other agents
+    network.connect()
+    # Get a channel to the agent
+    greeter_link = network.lookup_agent("Greeter")
+    # Send a message to the agent
+    greeter_link.send_txt_msg("Hello World!")
+    # Cleanup
+    greeter_link.close()
+    network.disconnect()
+```
+### Check out other demos in the `py/demo` directory.  We show the following: ###
+1) Hello World shown above
+2) Many CAP Actors interacting with each other
+3) A pair of interacting AutoGen Agents wrapped in CAP Actors
+4) CAP wrapped AutoGen Agents in a group chat
+
+### Coming soon. Stay tuned! ###
+1) Two AutoGen Agents running in different processes and communicating through CAP

samples/apps/cap/TODO.md

@@ -0,0 +1,21 @@
+- ~~Pretty print debug_logs~~
+  - ~~colors~~
+  - ~~messages to oai should be condensed~~
+- ~~remove orchestrator in scenario 4 and have the two actors talk to each other~~
+- ~~pass a complex multi-part message~~
+- ~~protobuf for messages~~
+- ~~make changes to autogen to enable scenario 3 to work with CAN~~
+- ~~make groupchat work~~
+- ~~actors instead of agents~~
+- clean up for PR into autogen
+  - ~~Create folder structure under Autogen examples~~
+  - ~~CAN -> CAP (Composable Actor Protocol)~~
+- CAP actor lookup should use zmq
+- Add min C# actors & reorganize
+- Hybrid GroupChat with C# ProductManager
+- C++ Msg Layer
+- Rust Msg Layer
+- Node Msg Layer
+- Java Msg Layer
+- Investigate a standard logging framework that supports color in windows
+  - structlog?

samples/apps/cap/c#/Readme.md

@@ -0,0 +1 @@
+Coming soon...

samples/apps/cap/c++/Readme.md

@@ -0,0 +1 @@
+Coming soon...

samples/apps/cap/node/Readme.md

@@ -0,0 +1 @@
+Coming soon...

samples/apps/cap/py/autogencap/Actor.py

@@ -0,0 +1,78 @@
+import zmq
+import threading
+import traceback
+import time
+from .DebugLog import Debug, Info
+from .Config import xpub_url
+
+
+class Actor:
+    def __init__(self, agent_name: str, description: str):
+        self.actor_name: str = agent_name
+        self.agent_description: str = description
+        self.run = False
+
+    def connect_network(self, network):
+        Debug(self.actor_name, f"is connecting to {network}")
+        Debug(self.actor_name, "connected")
+
+    def _process_txt_msg(self, msg: str, msg_type: str, topic: str, sender: str) -> bool:
+        Info(self.actor_name, f"InBox: {msg}")
+        return True
+
+    def _process_bin_msg(self, msg: bytes, msg_type: str, topic: str, sender: str) -> bool:
+        Info(self.actor_name, f"Msg: topic=[{topic}], msg_type=[{msg_type}]")
+        return True
+
+    def _recv_thread(self):
+        Debug(self.actor_name, "recv thread started")
+        self._socket: zmq.Socket = self._context.socket(zmq.SUB)
+        self._socket.setsockopt(zmq.RCVTIMEO, 500)
+        self._socket.connect(xpub_url)
+        str_topic = f"{self.actor_name}"
+        Debug(self.actor_name, f"subscribe to: {str_topic}")
+        self._socket.setsockopt_string(zmq.SUBSCRIBE, f"{str_topic}")
+        try:
+            while self.run:
+                try:
+                    topic, msg_type, sender_topic, msg = self._socket.recv_multipart()
+                    topic = topic.decode("utf-8")  # Convert bytes to string
+                    msg_type = msg_type.decode("utf-8")  # Convert bytes to string
+                    sender_topic = sender_topic.decode("utf-8")  # Convert bytes to string
+                except zmq.Again:
+                    continue  # No message received, continue to next iteration
+                except Exception:
+                    continue
+                if msg_type == "text":
+                    msg = msg.decode("utf-8")  # Convert bytes to string
+                    if not self._process_txt_msg(msg, msg_type, topic, sender_topic):
+                        msg = "quit"
+                    if msg.lower() == "quit":
+                        break
+                else:
+                    if not self._process_bin_msg(msg, msg_type, topic, sender_topic):
+                        break
+        except Exception as e:
+            Debug(self.actor_name, f"recv thread encountered an error: {e}")
+            traceback.print_exc()
+        finally:
+            self.run = False
+            Debug(self.actor_name, "recv thread ended")
+
+    def start(self, context: zmq.Context):
+        self._context = context
+        self.run: bool = True
+        self._thread = threading.Thread(target=self._recv_thread)
+        self._thread.start()
+        time.sleep(0.01)
+
+    def disconnect_network(self, network):
+        Debug(self.actor_name, f"is disconnecting from {network}")
+        Debug(self.actor_name, "disconnected")
+        self.stop()
+
+    def stop(self):
+        self.run = False
+        self._thread.join()
+        self._socket.setsockopt(zmq.LINGER, 0)
+        self._socket.close()

samples/apps/cap/py/autogencap/ActorConnector.py

@@ -0,0 +1,57 @@
+# Agent_Sender takes a zmq context, Topic and creates a
+# socket that can publish to that topic. It exposes this functionality
+# using send_msg method
+import zmq
+import time
+import uuid
+from .DebugLog import Debug, Error
+from .Config import xsub_url, xpub_url
+
+
+class ActorConnector:
+    def __init__(self, context, topic):
+        self._pub_socket = context.socket(zmq.PUB)
+        self._pub_socket.setsockopt(zmq.LINGER, 0)
+        self._pub_socket.connect(xsub_url)
+
+        self._resp_socket = context.socket(zmq.SUB)
+        self._resp_socket.setsockopt(zmq.LINGER, 0)
+        self._resp_socket.setsockopt(zmq.RCVTIMEO, 10000)
+        self._resp_socket.connect(xpub_url)
+        self._resp_topic = str(uuid.uuid4())
+        Debug("AgentConnector", f"subscribe to: {self._resp_topic}")
+        self._resp_socket.setsockopt_string(zmq.SUBSCRIBE, f"{self._resp_topic}")
+        self._topic = topic
+        time.sleep(0.01)  # Let the network do things.
+
+    def send_txt_msg(self, msg):
+        self._pub_socket.send_multipart(
+            [self._topic.encode("utf8"), "text".encode("utf8"), self._resp_topic.encode("utf8"), msg.encode("utf8")]
+        )
+
+    def send_bin_msg(self, msg_type: str, msg):
+        self._pub_socket.send_multipart(
+            [self._topic.encode("utf8"), msg_type.encode("utf8"), self._resp_topic.encode("utf8"), msg]
+        )
+
+    def binary_request(self, msg_type: str, msg, retry=5):
+        time.sleep(0.5)  # Let the network do things.
+        self._pub_socket.send_multipart(
+            [self._topic.encode("utf8"), msg_type.encode("utf8"), self._resp_topic.encode("utf8"), msg]
+        )
+        time.sleep(0.5)  # Let the network do things.
+        for i in range(retry + 1):
+            try:
+                self._resp_socket.setsockopt(zmq.RCVTIMEO, 10000)
+                resp_topic, resp_msg_type, resp_sender_topic, resp = self._resp_socket.recv_multipart()
+                return resp_topic, resp_msg_type, resp_sender_topic, resp
+            except zmq.Again:
+                Debug("ActorConnector", f"binary_request: No response received. retry_count={i}, max_retry={retry}")
+                time.sleep(0.01)  # Wait a bit before retrying
+                continue
+        Error("ActorConnector", "binary_request: No response received. Giving up.")
+        return None, None, None, None
+
+    def close(self):
+        self._pub_socket.close()
+        self._resp_socket.close()

samples/apps/cap/py/autogencap/Broker.py

@@ -0,0 +1,114 @@
+import time
+import zmq
+import threading
+from autogencap.DebugLog import Debug, Info, Warn
+from autogencap.Config import xsub_url, xpub_url
+
+
+class Broker:
+    def __init__(self, context: zmq.Context = zmq.Context()):
+        self._context: zmq.Context = context
+        self._run: bool = False
+        self._xpub: zmq.Socket = None
+        self._xsub: zmq.Socket = None
+
+    def start(self) -> bool:
+        try:
+            # XPUB setup
+            self._xpub = self._context.socket(zmq.XPUB)
+            self._xpub.setsockopt(zmq.LINGER, 0)
+            self._xpub.bind(xpub_url)
+
+            # XSUB setup
+            self._xsub = self._context.socket(zmq.XSUB)
+            self._xsub.setsockopt(zmq.LINGER, 0)
+            self._xsub.bind(xsub_url)
+
+        except zmq.ZMQError as e:
+            Debug("BROKER", f"Unable to start.  Check details: {e}")
+            # If binding fails, close the sockets and return False
+            if self._xpub:
+                self._xpub.close()
+            if self._xsub:
+                self._xsub.close()
+            return False
+
+        self._run = True
+        self._broker_thread: threading.Thread = threading.Thread(target=self.thread_fn)
+        self._broker_thread.start()
+        time.sleep(0.01)
+        return True
+
+    def stop(self):
+        # Error("BROKER_ERR", "fix cleanup self._context.term()")
+        Debug("BROKER", "stopped")
+        self._run = False
+        self._broker_thread.join()
+        if self._xpub:
+            self._xpub.close()
+        if self._xsub:
+            self._xsub.close()
+        # self._context.term()
+
+    def thread_fn(self):
+        try:
+            # Poll sockets for events
+            self._poller: zmq.Poller = zmq.Poller()
+            self._poller.register(self._xpub, zmq.POLLIN)
+            self._poller.register(self._xsub, zmq.POLLIN)
+
+            # Receive msgs, forward and process
+            while self._run:
+                events = dict(self._poller.poll(500))
+                if self._xpub in events:
+                    message = self._xpub.recv_multipart()
+                    Debug("BROKER", f"subscription message: {message[0]}")
+                    self._xsub.send_multipart(message)
+
+                if self._xsub in events:
+                    message = self._xsub.recv_multipart()
+                    Debug("BROKER", f"publishing message: {message}")
+                    self._xpub.send_multipart(message)
+
+        except Exception as e:
+            Debug("BROKER", f"thread encountered an error: {e}")
+        finally:
+            self._run = False
+            Debug("BROKER", "thread ended")
+        return
+
+
+# Run a standalone broker that all other Actors can connect to.
+# This can also run inproc with the other actors.
+def main():
+    broker = Broker()
+    Info("BROKER", "Starting.")
+    if broker.start():
+        Info("BROKER", "Running.")
+    else:
+        Warn("BROKER", "Failed to start.")
+        return
+
+    status_interval = 300  # seconds
+    last_time = time.time()
+
+    # Broker is running in a separate thread. Here we are watching the
+    # broker's status and printing status every few seconds.  This is
+    # a good place to print other statistics captured as the broker runs.
+    # -- Exits when the user presses Ctrl+C --
+    while broker._run:
+        # print a message every n seconds
+        current_time = time.time()
+        elapsed_time = current_time - last_time
+        if elapsed_time > status_interval:
+            Info("BROKER", "Running.")
+            last_time = current_time
+        try:
+            time.sleep(0.5)
+        except KeyboardInterrupt:
+            Info("BROKER", "KeyboardInterrupt.  Stopping the broker.")
+            broker.stop()
+
+
+if __name__ == "__main__":
+    main()

samples/apps/cap/py/autogencap/Config.py

@@ -0,0 +1,5 @@
+# Set the current log level
+LOG_LEVEL = 0
+IGNORED_LOG_CONTEXTS = []
+xpub_url: str = "tcp://127.0.0.1:5555"
+xsub_url: str = "tcp://127.0.0.1:5556"

samples/apps/cap/py/autogencap/Constants.py

@@ -0,0 +1,2 @@
+Termination_Topic: str = "Termination"
+Directory_Svc_Topic: str = "Directory_Svc"