[Core] Add simple serving feature with Uvicorn

.github/workflows/ci.yml

@@ -0,0 +1,31 @@
+name: Linter
+
+on:
+  pull_request:
+    branches:
+      - main
+  push:
+    branches:
+      - main
+  # To test workflow without event trigger
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install ruff
+    - name: Analysing the code with pylint
+      run: |
+        ruff check $(git ls-files '*.py')

gigax/__init__.py

@@ -1,4 +1,9 @@
+""" Gigax package. """
+
 from gigax import scene
 from gigax import parse
 from gigax import step
 from gigax import prompt
+
+
+__all__ = ["scene", "parse", "step", "prompt"]

gigax/app.py

@@ -0,0 +1,91 @@
+import logging
+import os
+import uvicorn
+import sys
+
+from dotenv import load_dotenv
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi import FastAPI, Request, status
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.logger import logger as fastapi_logger
+from pydantic import BaseModel
+
+from gigax.parse import CharacterAction, ProtagonistCharacter
+from gigax.scene import (
+    Character,
+    Item,
+    Location,
+)
+from gigax.step import NPCStepper
+
+load_dotenv()
+
+fastapi_logger.handlers = logging.getLogger("gunicorn.error").handlers
+logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler(sys.stdout)])
+
+# FastAPI
+app = FastAPI(root_path="/api")
+origins = [
+    "http://localhost",
+]
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+logger = logging.getLogger("uvicorn")
+logging.basicConfig(level=logging.INFO)
+
+
+@app.exception_handler(RequestValidationError)
+async def validation_exception_handler(request: Request, exc: RequestValidationError):
+    exc_str = f"{exc}".replace("\n", " ").replace("   ", " ")
+    logging.error(f"{request}: {exc_str}")
+    content = {"status_code": 10422, "message": exc_str, "data": None}
+    return JSONResponse(
+        content=content, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY
+    )
+
+
+class CharacterActionRequest(BaseModel):
+    context: str
+    locations: list[Location]
+    NPCs: list[Character]
+    protagonist: ProtagonistCharacter
+    items: list[Item]
+    events: list[CharacterAction]
+
+
+@app.post(
+    "/step",
+    response_description="Step the character",
+    response_model=CharacterAction,
+)
+async def step(
+    request: CharacterActionRequest,
+):
+    # Format the prompt
+    stepper = NPCStepper(model="llama_3_regex", api_key=os.getenv("API_KEY"))
+
+    return await stepper.get_action(
+        context=request.context,
+        locations=request.locations,
+        NPCs=request.NPCs,
+        protagonist=request.protagonist,
+        items=request.items,
+        events=request.events,
+    )
+
+
+@app.get("/health-check")
+async def health_check():
+    return {"status": "ok"}
+
+
+if __name__ == "__main__":
+    uvicorn.run("__main__:app", host="0.0.0.0", port=5678, reload=True, workers=5)

gigax/parse.py

@@ -1,3 +1,5 @@
+"""This module contains the logic to parse a command string into a CharacterAction object."""
+
 import logging
 import re
 from typing import Union
@@ -17,10 +19,14 @@
 
 
 class ActionParsingError(Exception):
+    """Exception raised for errors in the action parsing."""
+
     pass
 
 
 class CharacterAction(BaseModel):
+    """CharacterAction class to represent a character action."""
+
     command: str
     protagonist: ProtagonistCharacter
     parameters: list[Union[str, int, Object]]
@@ -41,6 +47,10 @@ def from_str(
         valid_items: list[Item],
         compiled_regex: re.Pattern,
     ) -> "CharacterAction":
+        """
+        Parse a command string into a CharacterAction object.
+        """
+
         match = compiled_regex.match(command_str)
         if not match:
             raise ValueError("Invalid command format")

gigax/scene.py

@@ -1,16 +1,15 @@
 from enum import Enum
 import re
-from typing import Union
 from pydantic import BaseModel, Field
 
 
 class ParameterType(str, Enum):
-    character = "<character>"
-    location = "<location>"
-    item = "<item>"
-    amount = "<amount>"
-    content = "<content>"
-    other = "<other>"
+    character = "character"
+    location = "location"
+    item = "item"
+    amount = "amount"
+    content = "content"
+    other = "other"
 
 
 class Object(BaseModel):
@@ -59,7 +58,7 @@ class Skill(BaseModel):
 
     name: str = Field(..., description="Skill name")
     description: str = Field(..., description="Skill description")
-    parameter_types: Union[list[ParameterType], dict] = (
+    parameter_types: list[ParameterType] = (
         Field(  # This is a Union because Cubzh's Lua sends empty lists as empty dicts
             [], description="Allowed parameter types for the given skill"
         )
@@ -107,4 +106,3 @@ class ProtagonistCharacter(Character):
     memories: list[str] = Field(..., description="Memories that the character has.")
     quests: list[str] = Field(..., description="Quests that the character is on.")
     skills: list[Skill] = Field(..., description="Skills that the character can use.")
-    psychological_profile: str

gigax/step.py

@@ -54,7 +54,7 @@ async def generate_api(
 
         response = await client.chat.completions.create(
             model=model,
-            messages=messages,
+            messages=messages,  # type: ignore
             max_tokens=100,
             temperature=temperature,
             extra_body=dict(guided_regex=guided_regex),
@@ -116,7 +116,7 @@ async def get_action(
         protagonist: ProtagonistCharacter,
         items: list[Item],
         events: list[CharacterAction],
-    ) -> CharacterAction:
+    ) -> CharacterAction | None:
         """
         Prompt the NPC for an input.
         """
@@ -154,8 +154,7 @@ async def get_action(
             parsed_action = CharacterAction.from_str(
                 res, protagonist, NPCs, locations, items, guided_regex
             )
+            logger.info(f"NPC {protagonist.name} responded with: {parsed_action}")
+            return parsed_action
         except Exception:
             logger.error(f"Error while parsing the action: {traceback.format_exc()}")
-
-        logger.info(f"NPC {protagonist.name} responded with: {parsed_action}")
-        return parsed_action

tests/conftest.py

@@ -45,7 +45,6 @@ def protagonist(current_location):
                 parameter_types=[ParameterType.character],
             )
         ],
-        psychological_profile="Determined and compassionate",
     )
 
 

tests/test_step.py

@@ -86,9 +86,9 @@ async def test_stepper_api(
 ):
     # Get the NPC's input
     with pytest.raises(ValueError):
-        NPCStepper(model="mistral-7b-regex")
+        NPCStepper(model="llama_3_regex")
 
-    stepper = NPCStepper(model="mistral_7b_regex", api_key=os.getenv("API_KEY"))
+    stepper = NPCStepper(model="llama_3_regex", api_key=os.getenv("API_KEY"))
 
     action = await stepper.get_action(
         context=context,