[WIP] Add CPP Servicer

.github/workflows/unittest.yml

@@ -21,6 +21,25 @@ jobs:
     - name: Update setuptools
       run: |
         pip install setuptools==68.2.2 wheel==0.41.2
+        pip install pybind11
+    - name: Install Ubuntu Dependencies
+      if: ${{ matrix.os == 'ubuntu-latest' }}
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y libprotobuf-dev protobuf-compiler cmake libgrpc++-dev
+        git clone --recurse-submodules -b v1.62.2 --depth 1 --shallow-submodules https://github.com/grpc/grpc
+        cd grpc
+        mkdir -p cmake/build
+        pushd cmake/build
+        cmake -DgRPC_INSTALL=ON -DgRPC_BUILD_TESTS=OFF -DCMAKE_INSTALL_PREFIX=$MY_INSTALL_DIR ../..
+        make -j 4
+        sudo make install
+        popd
+        cd ..
+    - name: Install Mac Dependencies
+      if: ${{ matrix.os == 'macos-13' }}
+      run: |
+        brew install protobuf grpc
     - name: Install Minimal Dependencies
       run: |
         pip install -q -e .
@@ -37,3 +56,8 @@ jobs:
     - name: Generate coverage report
       run: |
         coverage report -m
+    - name: CPP gRPC Server Tests
+      if: ${{ matrix.os != 'windows-latest' }}
+      run: |
+        ls -l src/agentscope/cpp_server
+        AGENTSCOPE_USE_CPP_SERVER=YES AGENTSCOPE_NUM_WORKERS=2 python -m unittest tests/rpc_agent_test.py

.gitignore

@@ -129,6 +129,7 @@ dmypy.json
 .pyre/
 
 .idea/
+.vscode/
 
 # macOS
 .DS_Store

CMakeLists.txt

@@ -0,0 +1,9 @@
+cmake_minimum_required(VERSION 3.13)
+project(example-grpc LANGUAGES CXX)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall --std=c++11 -O3 -fPIC")
+set(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/cmake")
+set(CMAKE_CXX_STANDARD 17)
+
+add_subdirectory(src/agentscope/rpc)
+add_subdirectory(src/agentscope/cpp_server)

examples/distributed_simulation/main.py

@@ -102,6 +102,7 @@ def run_main_process(
     agent_type: str = "random",
     max_value: int = 100,
     sleep_time: float = 1.0,
+    cpp_server: bool = False,
 ) -> None:
     """Run main process"""
     agentscope.init(
@@ -127,7 +128,7 @@ def run_main_process(
         port_id = idx % server_per_host
         model_id = i % model_per_host
         host = hosts[host_id]
-        port = base_port + port_id
+        port = base_port + port_id if not cpp_server else base_port
         config_name = f"model_{model_id + 1}"
         if agent_type == "random":
             configs.append(
@@ -166,7 +167,7 @@ def run_main_process(
                         * participant_per_moderator
                     ],
                     host=hosts[i // moderator_per_host],
-                    port=base_port + server_per_host + i % moderator_per_host,
+                    port=base_port + server_per_host + i % moderator_per_host if not cpp_server else base_port,
                     agent_type=agent_type,
                     max_value=max_value,
                     sleep_time=sleep_time,

examples/distributed_simulation/participant.py

@@ -4,6 +4,7 @@
 import time
 import re
 from typing import Optional, Union, Sequence
+import concurrent.futures
 
 from loguru import logger
 
@@ -111,46 +112,44 @@ def __init__(
     ) -> None:
         super().__init__(name)
         self.max_value = max_value
-        if agent_type == "llm":
-            self.participants = [
-                LLMParticipant(
+        def create_llm_participant(config):
+            return LLMParticipant(
                     name=config["name"],
                     model_config_name=config["model_config_name"],
                     max_value=max_value,
                 ).to_dist(
                     host=config["host"],
                     port=config["port"],
                 )
-                for config in part_configs
-            ]
-        else:
-            self.participants = [
-                RandomParticipant(
-                    name=config["name"],
-                    max_value=max_value,
-                    sleep_time=sleep_time,
-                ).to_dist(
-                    host=config["host"],
-                    port=config["port"],
-                )
-                for config in part_configs
-            ]
+
+        def create_random_participant(config):
+            return RandomParticipant(
+                name=config["name"],
+                max_value=max_value,
+                sleep_time=sleep_time,
+            ).to_dist(
+                host=config["host"],
+                port=config["port"],
+            )
+        create_participant = {"random": create_random_participant, "llm": create_llm_participant}[agent_type]
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = {executor.submit(create_participant, config) for config in part_configs}
+
+            self.participants = []
+            for future in concurrent.futures.as_completed(futures):
+                result = future.result()
+                self.participants.append(result)
 
     def reply(self, x: Optional[Union[Msg, Sequence[Msg]]] = None) -> Msg:
-        results = []
         msg = Msg(
             name="moderator",
             role="user",
             content=f"Now give a number between 0 and {self.max_value}.",
         )
-        for p in self.participants:
-            results.append(p(msg))
-        summ = 0
-        for r in results:
-            try:
-                summ += int(r.content)
-            except Exception as e:
-                print(e)
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = {executor.submit(lambda p: p(msg), p) for p in self.participants}
+            futures_2 = {executor.submit(lambda r: int(r.content), future.result()) for future in concurrent.futures.as_completed(futures)}
+            summ = sum(future.result() for future in concurrent.futures.as_completed(futures_2))
         return Msg(
             name=self.name,
             role="assistant",

setup.py

@@ -3,6 +3,13 @@
 from __future__ import absolute_import, division, print_function
 
 import re
+from setuptools import setup, Extension
+from setuptools.command.build_ext import build_ext
+import os
+import sys
+import site
+import platform
+import subprocess
 
 import setuptools
 
@@ -88,9 +95,11 @@
     "psutil",
     "scipy",
     "pillow",
+    "pybind11",
 ]
 
 distribute_requires = minimal_requires + rpc_requires
+cpp_distribute_requires = distribute_requires
 
 dev_requires = minimal_requires + test_requires
 
@@ -115,6 +124,48 @@
 with open("README.md", "r", encoding="UTF-8") as fh:
     long_description = fh.read()
 
+
+class CMakeExtension(Extension):
+    def __init__(self, name, sourcedir=''):
+        Extension.__init__(self, name, sources=[], language='c++')
+        self.sourcedir = os.path.abspath(sourcedir)
+
+
+class CMakeBuild(build_ext):
+    def run(self):
+        if platform.system() == "Windows":
+            return
+        from setuptools import Distribution
+        distribution = Distribution()
+        distribution.parse_config_files()
+        try:
+            out = subprocess.check_output(['cmake', '--version'])
+        except OSError:
+            raise RuntimeError("CMake must be installed to build the following extensions: " + ", ".join(e.name for e in self.extensions))
+
+        self.env = os.environ.copy()
+        for ext in self.extensions:
+            self.build_extension(ext)
+        super().run()
+
+    def build_extension(self, ext):
+        extdir = os.path.abspath(os.path.dirname(self.get_ext_fullpath(ext.name)))
+        cmake_args = ['-DCMAKE_LIBRARY_OUTPUT_DIRECTORY=' + extdir,
+                      '-DPython3_EXECUTABLE=' + sys.executable,
+                      '-Dpybind11_DIR=' + os.path.join(site.getsitepackages()[0], 'pybind11', 'share', 'cmake', 'pybind11')]
+
+        cfg = 'Debug' if self.debug else 'Release'
+        build_args = ['--config', cfg]
+
+        cmake_args += ['-DCMAKE_BUILD_TYPE=' + cfg]
+
+        if not os.path.exists(self.build_temp):
+            os.makedirs(self.build_temp)
+        subprocess.check_call(['cmake', '-B', 'build', ext.sourcedir] + cmake_args, cwd=self.build_temp, env=self.env)
+        subprocess.check_call(['cmake', '--build', 'build'] + build_args, cwd=self.build_temp, env=self.env)
+        subprocess.check_call(['cmake', '--install', 'build'], cwd=self.build_temp, env=self.env)
+
+
 setuptools.setup(
     name=NAME,
     version=VERSION,
@@ -135,10 +186,13 @@
     install_requires=minimal_requires,
     extras_require={
         "distribute": distribute_requires,
+        "cpp_distribute": cpp_distribute_requires,
         "dev": dev_requires,
         "full": full_requires,
         "online": online_requires,
     },
+    ext_modules=[CMakeExtension('agentscope.cpp_server.cpp_server')],
+    cmdclass=dict(build_ext=CMakeBuild),
     license="Apache License 2.0",
     classifiers=[
         "Development Status :: 4 - Beta",

src/agentscope/cpp_server/CMakeLists.txt

@@ -0,0 +1,20 @@
+find_package(Python3 COMPONENTS Interpreter Development)
+find_package(pybind11 CONFIG REQUIRED)
+
+source_group(TREE ${CMAKE_CURRENT_SOURCE_DIR} FILES ${SOURCES})
+
+pybind11_add_module(cpp_server worker.cc rpc_agent_servicer.cc)
+
+target_include_directories(cpp_server
+    PRIVATE
+    ${Python3_INCLUDE_DIRS}
+    ${pybind11_INCLUDE_DIRS}
+)
+
+target_link_libraries(cpp_server
+    PRIVATE
+    rpc
+)
+
+
+install(TARGETS cpp_server DESTINATION ${CMAKE_SOURCE_DIR}/src/agentscope/cpp_server)

src/agentscope/cpp_server/__init__.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""Related functions for cpp server."""
+
+from loguru import logger
+
+try:
+    import dill
+except ImportError as import_error:
+    from agentscope.utils.tools import ImportErrorReporter
+
+    dill = ImportErrorReporter(import_error, "distribute")
+
+from agentscope.agents.agent import AgentBase
+
+
+def create_agent(agent_id: str, agent_init_args: str, agent_source_code: str):
+    agent_configs = dill.loads(agent_init_args)
+    if len(agent_source_code) > 0:
+        cls = dill.loads(agent_source_code)
+        cls_name = cls.__name__
+        logger.info(
+            f"Load class [{cls_name}] from uploaded source code.",
+        )
+    else:
+        cls_name = agent_configs["class_name"]
+        try:
+            cls = AgentBase.get_agent_class(cls_name)
+        except ValueError as e:
+            err_msg = (f"Agent class [{cls_name}] not found: {str(e)}",)
+            logger.error(err_msg)
+            return None, str(err_msg)
+    try:
+        agent_instance = cls(
+            *agent_configs["args"],
+            **agent_configs["kwargs"],
+        )
+        agent_instance._agent_id = agent_id  # pylint: disable=W0212
+        logger.info(
+            f"create agent instance <{cls_name}>[{agent_id}]"
+            f" [{agent_instance.name}]"
+        )
+        return agent_instance, ""
+    except Exception as e:
+        err_msg = (f"Failed to create agent instance <{cls_name}>: {str(e)}",)
+        logger.error(err_msg)
+        return None, str(err_msg)

src/agentscope/cpp_server/makefile

@@ -0,0 +1,38 @@
+CXX = g++
+CPPFLAGS += `pkg-config --cflags protobuf grpc`
+CXXFLAGS += -std=c++17 -O3 -fPIC `python3 -m pybind11 --includes`
+PY_LIB_PATH = `python3-config --prefix`/lib
+LDFLAGS += -L/usr/local/lib `pkg-config --cflags --libs grpc++ grpc protobuf` -L$(PY_LIB_PATH) \
+		   -lpython3.12 \
+           -pthread\
+           -lgrpc++_reflection\
+           -ldl
+PROTOC = protoc
+GRPC_CPP_PLUGIN = grpc_cpp_plugin
+GRPC_CPP_PLUGIN_PATH ?= `which $(GRPC_CPP_PLUGIN)`
+
+PROTOS_PATH = ../rpc
+
+vpath %.proto $(PROTOS_PATH)
+
+all: install launch_server
+
+install: rpc_agent.grpc.pb.cc rpc_agent.pb.cc worker_args.pb.cc rpc_agent_servicer
+
+rpc_agent_servicer: rpc_agent.pb.o rpc_agent.grpc.pb.o worker_args.pb.o worker.o rpc_agent_servicer.o
+	$(CXX) $^ $(CXXFLAGS) $(LDFLAGS) -o $@
+
+worker.o: worker.cc worker.h
+	$(CXX) -c $< $(CXXFLAGS) $(LDFLAGS) -o $@
+
+launch_server:
+	cd ../../../; DYLD_LIBRARY_PATH=$(PY_LIB_PATH):$$DYLD_LIBRARY_PATH OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES ./src/agentscope/cpp_server/rpc_agent_servicer $(INIT_SETTINGS_STR) $(HOST) $(PORT) $(SERVER_ID) $(CUSTOM_AGENT_CLASSES) $(STUDIO_URL) $(MAX_TASKS) $(TIMEOUT_SECONDS) $(NUM_WORKERS)
+
+%.grpc.pb.cc: $(PROTOS_PATH)/%.proto
+	$(PROTOC) -I $(PROTOS_PATH) --grpc_out=. --plugin=protoc-gen-grpc=$(GRPC_CPP_PLUGIN_PATH) $<
+
+%.pb.cc: $(PROTOS_PATH)/%.proto
+	$(PROTOC) -I $(PROTOS_PATH) --cpp_out=. $<
+
+clean:
+	rm -f *.o *.so *.pb.cc *.pb.h rpc_agent_servicer