Initial commit

.gitignore

+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+.vscode/
\ No newline at end of file

arcadia_statemachine/__init__.py

+from .agent import Agent, State, NormalState, PreemptiveState
\ No newline at end of file

arcadia_statemachine/agent.py

+from asyncio import (
+    gather,
+    AbstractEventLoop,
+    Task,
+    Queue,
+    CancelledError,
+    new_event_loop,
+)
+from aiohttp import ClientSession, web
+from dataclasses import dataclass
+from logging import Logger
+from typing import Any, Awaitable, Callable
+import logging
+
+
+# Can we type (Agent, ...) -> Awaitable[None] in Python?
+@dataclass(frozen=True)
+class NormalState:
+    coro: Callable[["Agent", Any], Awaitable[None]] | Callable[
+        ["Agent"], Awaitable[None]
+    ]
+
+
+PreemptiveFun = Callable[["Agent", web.Request], Awaitable[web.Response]]
+
+
+@dataclass(frozen=True)
+class PreemptiveState:
+    coro: PreemptiveFun
+
+
+State = NormalState | PreemptiveState
+
+
+class Agent:
+    def __init__(
+        self,
+        info: dict,
+        address_book: dict,
+        states: dict[str, State],
+    ):
+        self.info = info
+        self.address_book = address_book
+        self.states: dict[str, State] = states
+
+        self._loop: AbstractEventLoop = new_event_loop()
+        self.session = ClientSession(loop=self._loop)
+        self.app = self._mk_app()
+
+        self.logger = self._mk_logger()
+        self._tasks_queue: Queue[Awaitable] = Queue()
+        self._running: bool = True
+
+    def _mk_logger(self) -> Logger:
+        logging.basicConfig(level=logging.DEBUG)
+        return logging.getLogger(self.info["name"])
+
+    def _mk_app(self) -> web.Application:
+        app = web.Application()
+        app.on_shutdown.append(self._on_shutdown)
+        routes = []
+        for name, state in self.states.items():
+            if isinstance(state, PreemptiveState):
+                f = self._preemptive_state_wrapper(state.coro)
+                routes.append(web.post(f"/{name}", f))
+        app.add_routes(routes)
+        return app
+
+    def _preemptive_state_wrapper(
+        self, coro: PreemptiveFun
+    ) -> Callable[[web.Request], Awaitable[web.Response]]:
+        async def new_coro(request: web.Request):
+            await self._cancel_task({self._current_task})
+            while not self._tasks_queue.empty():
+                self._tasks_queue.get_nowait()
+            return await coro(self, request)
+
+        return new_coro
+
+    def transition(self, name: str, *args, **kwargs):
+        if name not in self.states:
+            self.logger.error(f"{name} is not an ability")
+        else:
+            match self.states[name]:
+                case NormalState(coro):
+                    self._tasks_queue.put_nowait(coro(self, *args, **kwargs))
+
+    async def _cancel_task(self, to_cancel: set[Task[Any]]) -> None:
+        """
+        https://github.com/aio-libs/aiohttp/blob/84c90050bed205f2fa4967bd9c320150f501bfa8/aiohttp/web.py#L418
+        """
+        if not to_cancel:
+            return
+        for task in to_cancel:
+            task.cancel()
+
+        await gather(*to_cancel, return_exceptions=True)
+        for task in to_cancel:
+            if task.cancelled():
+                continue
+            if task.exception() is not None:
+                self._loop.call_exception_handler({"exception": task.exception()})
+
+    async def _run_main_loop(self):
+        self._running = True
+        while self._running:
+            _current_coro = await self._tasks_queue.get()
+            try:
+                self._current_task = self._loop.create_task(_current_coro)
+                await self._current_task
+            except CancelledError:
+                self.logger.debug("Task cancelled")
+
+    async def _on_shutdown(self, app: web.Application):
+        self._running = False
+        await self.session.close()
+
+    def run(self):
+        self._loop.create_task(self._run_main_loop())
+        web.run_app(
+            self.app,
+            host=self.info["host"],
+            port=self.info["port"],
+            access_log=self.logger,
+            loop=self._loop,
+        )

readme.md

+# arcadia_statemachine
+
+## Agent
+
+### info
+
+The information of the agent self.
+
+### address_book
+
+The host and port of all other agents.
+
+### transition
+
+State-Transition
+
+## NormalState
+
+It doesn't need to return anything. Just transition to the next state.
+
+## PreemptiveState
+
+It must responce to the caller.
\ No newline at end of file

requirements.txt

+aiohttp
+mypy
+openai
\ No newline at end of file

test/act.py

+import aiohttp
+import asyncio
+from context import Agent
+
+async def act(agent: Agent, code: str):
+    agent.logger.debug(f"ACT Start: {code}")
+    await asyncio.sleep(5)
+    agent.logger.debug(f"ACT End")
+    agent.transition("observe")
\ No newline at end of file

test/chat.py

+import asyncio
+from aiohttp import web
+from context import Agent
+
+async def chat(agent: Agent, request: web.Request):
+    msg = await request.json()
+    agent.logger.debug(f"CHAT Receive: {msg}")
+    agent.transition("plan", "here is my new plan")
+    return web.Response(text="hello")
\ No newline at end of file

test/context.py

+import os
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from arcadia_statemachine import Agent, State, NormalState, PreemptiveState
\ No newline at end of file

test/main.py

+from context import State, NormalState, PreemptiveState, Agent
+from act import act
+from observe import observe
+from plan import plan
+from chat import chat
+
+abilities: dict[str, State] = {
+    "act": NormalState(act),
+    "observe": NormalState(observe),
+    "plan": NormalState(plan),
+    "chat": PreemptiveState(chat),
+}
+
+a = Agent({"name": "Farmer", "host": "127.0.0.1", "port": 8000}, {}, abilities)
+a.transition("observe")
+a.run()

test/observe.py

+import asyncio
+from context import Agent
+
+async def observe(agent: Agent):
+    agent.logger.debug("OBSERVE START")
+    await asyncio.sleep(5)
+    msg = {"block": "red"}
+    agent.logger.debug(f"OBSERVE END: {msg}")
+    agent.transition("plan", msg)
\ No newline at end of file

test/plan.py

+import asyncio
+from context import Agent
+
+async def plan(agent: Agent, msg):
+    agent.logger.debug(f"PLAN Start: recieve {msg}")
+    await asyncio.sleep(5)
+    code = "bot.walk()"
+    agent.logger.debug(f"PLAN End: {code}")
+    agent.transition("act", code)
\ No newline at end of file