Devchat local service implementation

Makefile

@@ -15,3 +15,12 @@ fix:
 	@echo ${div}
 	poetry run ruff check $(DIR) --fix
 	@echo "Done!"
+
+
+run-dev-svc:
+	@echo "Running dev service on port 22222..."
+	@uvicorn devchat._service.main:api_app --reload --port 22222
+
+run-svc:
+	@echo "Running service..."
+	@python devchat/_service/main.py

devchat/_service/config.py

@@ -0,0 +1,19 @@
+from typing import Optional
+
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    PORT: int = 22222
+    WORKERS: int = 2
+    WORKSPACE: Optional[str] = None
+    LOG_LEVEL: str = "INFO"
+    LOG_FILE: Optional[str] = "dc_svc.log"
+    JSON_LOGS: bool = False
+
+    class Config:
+        env_prefix = "DC_SVC_"
+        case_sensitive = True
+
+
+config = Settings()

devchat/_service/gunicorn_logging.py

@@ -0,0 +1,110 @@
+import logging
+import os
+import sys
+
+from gunicorn.app.base import BaseApplication
+from gunicorn.glogging import Logger
+from loguru import logger
+
+from devchat._service.config import config
+from devchat.workspace_util import get_workspace_chat_dir
+
+
+class InterceptHandler(logging.Handler):
+    def emit(self, record):
+        # get corresponding Loguru level if it exists
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = record.levelno
+
+        # find caller from where originated the logged message
+        frame, depth = sys._getframe(6), 6
+        while frame and frame.f_code.co_filename == logging.__file__:
+            frame = frame.f_back
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+
+
+class StubbedGunicornLogger(Logger):
+    def setup(self, cfg):
+        handler = logging.NullHandler()
+        self.error_logger = logging.getLogger("gunicorn.error")
+        self.error_logger.addHandler(handler)
+        self.access_logger = logging.getLogger("gunicorn.access")
+        self.access_logger.addHandler(handler)
+        self.error_logger.setLevel(config.LOG_LEVEL)
+        self.access_logger.setLevel(config.LOG_LEVEL)
+
+
+class StandaloneApplication(BaseApplication):
+    """Our Gunicorn application."""
+
+    def __init__(self, app, options=None):
+        self.options = options or {}
+        self.application = app
+        super().__init__()
+
+    def load_config(self):
+        config = {
+            key: value
+            for key, value in self.options.items()
+            if key in self.cfg.settings and value is not None
+        }
+        for key, value in config.items():
+            self.cfg.set(key.lower(), value)
+
+    def load(self):
+        return self.application
+
+
+def run_with_gunicorn(app):
+    intercept_handler = InterceptHandler()
+    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
+    # logging.root.handlers = [intercept_handler]
+    logging.root.setLevel(config.LOG_LEVEL)
+
+    seen = set()
+    for name in [
+        *logging.root.manager.loggerDict.keys(),
+        "gunicorn",
+        "gunicorn.access",
+        "gunicorn.error",
+        "uvicorn",
+        "uvicorn.access",
+        "uvicorn.error",
+    ]:
+        if name not in seen:
+            seen.add(name.split(".")[0])
+            logging.getLogger(name).handlers = [intercept_handler]
+
+    workspace_chat_dir = get_workspace_chat_dir(config.WORKSPACE)
+    log_file = os.path.join(workspace_chat_dir, config.LOG_FILE)
+
+    logger.configure(
+        handlers=[
+            {"sink": sys.stdout, "serialize": config.JSON_LOGS},
+            {
+                "sink": log_file,
+                "serialize": config.JSON_LOGS,
+                "rotation": "10 days",
+                "retention": "30 days",
+                "enqueue": True,
+            },
+        ]
+    )
+
+    options = {
+        "bind": f"0.0.0.0:{config.PORT}",
+        "workers": config.WORKERS,
+        "accesslog": "-",
+        "errorlog": "-",
+        "worker_class": "uvicorn.workers.UvicornWorker",
+        "logger_class": StubbedGunicornLogger,
+    }
+
+    StandaloneApplication(app, options).run()
+
+
+# https://pawamoy.github.io/posts/unify-logging-for-a-gunicorn-uvicorn-app/

devchat/_service/main.py

@@ -0,0 +1,41 @@
+from fastapi import FastAPI
+
+from devchat._service.config import config
+from devchat._service.route import router
+from devchat._service.uvicorn_logging import setup_logging
+
+api_app = FastAPI(
+    title="DevChat Local Service",
+)
+api_app.mount("/devchat", router)
+api_app.include_router(router)
+
+
+# app = socketio.ASGIApp(sio_app, api_app, socketio_path="devchat.socket")
+
+# NOTE: some references if we want to use socketio with FastAPI in the future
+
+# https://www.reddit.com/r/FastAPI/comments/170awhx/mount_socketio_to_fastapi/
+# https://github.com/miguelgrinberg/python-socketio/blob/main/examples/server/asgi/fastapi-fiddle.py
+
+
+def main():
+    # Use uvicorn to run the app because gunicorn doesn't support Windows
+    from uvicorn import Config, Server
+
+    server = Server(
+        Config(
+            api_app,
+            host="0.0.0.0",
+            port=config.PORT,
+        ),
+    )
+
+    # setup logging last, to make sure no library overwrites it
+    # (they shouldn't, but it happens)
+    setup_logging()
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

devchat/_service/route/__init__.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter
+
+from .logs import router as log_router
+from .message import router as message_router
+from .topics import router as topic_router
+from .workflows import router as workflow_router
+
+router = APIRouter()
+
+
+@router.get("/ping")
+async def ping():
+    return {"message": "pong"}
+
+
+router.include_router(workflow_router, prefix="/workflows", tags=["WorkflowManagement"])
+router.include_router(message_router, prefix="/message", tags=["Message"])
+router.include_router(log_router, prefix="/logs", tags=["LogManagement"])
+router.include_router(topic_router, prefix="/topics", tags=["TopicManagement"])

devchat/_service/route/logs.py

@@ -0,0 +1,36 @@
+from fastapi import APIRouter, HTTPException, status
+
+from devchat._service.schema import request, response
+from devchat.msg.log_util import delete_log_prompt, gen_log_prompt, insert_log_prompt
+
+router = APIRouter()
+
+
+@router.post("/insert", response_model=response.InsertLog)
+def insert(
+    item: request.InsertLog,
+):
+    try:
+        prompt = gen_log_prompt(item.jsondata, item.filepath)
+        prompt_hash = insert_log_prompt(prompt, item.workspace)
+    except Exception as e:
+        detail = f"Failed to insert log: {str(e)}"
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=detail)
+    return response.InsertLog(hash=prompt_hash)
+
+
+@router.post("/delete", response_model=response.DeleteLog)
+def delete(
+    item: request.DeleteLog,
+):
+    try:
+        success = delete_log_prompt(item.hash, item.workspace)
+        if not success:
+            detail = f"Failed to delete log <{item.hash}>. Log not found or is not a leaf."
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail)
+
+    except Exception as e:
+        detail = f"Failed to delete log <{item.hash}>: {str(e)}"
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=detail)
+
+    return response.DeleteLog(success=success)

devchat/_service/route/message.py

@@ -0,0 +1,96 @@
+import os
+from typing import Iterator, Optional
+
+from fastapi import APIRouter
+from fastapi.responses import StreamingResponse
+
+from devchat._service.schema import request, response
+from devchat.msg.chatting import chatting
+from devchat.msg.util import MessageType, mk_meta, route_message_by_content
+from devchat.workflow.workflow import Workflow
+
+router = APIRouter()
+
+
+@router.post("/msg")
+def msg(
+    message: request.UserMessage,
+):
+    if message.api_key:
+        os.environ["OPENAI_API_KEY"] = message.api_key
+    if message.api_base:
+        os.environ["OPENAI_API_BASE"] = message.api_base
+
+    user_str, date_str = mk_meta()
+
+    message_type, extra = route_message_by_content(message.content)
+
+    if message_type == MessageType.CHATTING:
+
+        def gen_chat_response() -> Iterator[response.MessageCompletionChunk]:
+            try:
+                for res in chatting(
+                    content=message.content,
+                    model_name=message.model_name,
+                    parent=message.parent,
+                    workspace=message.workspace,
+                    context_files=message.context,
+                ):
+                    chunk = response.MessageCompletionChunk(
+                        user=user_str,
+                        date=date_str,
+                        content=res,
+                    )
+                    yield chunk.json()
+            except Exception as e:
+                chunk = response.MessageCompletionChunk(
+                    user=user_str,
+                    date=date_str,
+                    content=str(e),
+                    isError=True,
+                )
+                yield chunk.json()
+                raise e
+
+        return StreamingResponse(gen_chat_response(), media_type="application/json")
+
+    elif message_type == MessageType.WORKFLOW:
+        workflow: Workflow
+        wf_name: str
+        wf_input: Optional[str]
+        workflow, wf_name, wf_input = extra
+
+        if workflow.should_show_help(wf_input):
+            doc = workflow.get_help_doc(wf_input)
+
+            def _gen_res_help() -> Iterator[response.MessageCompletionChunk]:
+                yield response.MessageCompletionChunk(
+                    user=user_str, date=date_str, content=doc
+                ).json()
+
+            return StreamingResponse(_gen_res_help(), media_type="application/json")
+        else:
+            # return "should run workflow" response
+            # then the client will trigger the workflow by devchat cli
+            def _gen_res_run_workflow() -> Iterator[response.MessageCompletionChunk]:
+                yield response.MessageCompletionChunk(
+                    user=user_str,
+                    date=date_str,
+                    content="",
+                    finish_reason="should_run_workflow",
+                    extra={"workflow_name": wf_name, "workflow_input": wf_input},
+                ).json()
+
+            return StreamingResponse(
+                _gen_res_run_workflow(),
+                media_type="application/json",
+            )
+
+    else:
+        # Should not reach here
+        chunk = response.MessageCompletionChunk(
+            user=user_str,
+            date=date_str,
+            content="",
+        )
+        return StreamingResponse((chunk.json() for _ in [1]), media_type="application/json")