Merge branch 'feat/python-engine' of github.com:janhq/cortex.cpp into feat/python-engine

Author: nguyenhoangthuan99

.github/workflows/cortex-cpp-quality-gate.yml

@@ -34,7 +34,7 @@ jobs:
             ccache-dir: ""
           - os: "mac"
             name: "arm64"
-            runs-on: "macos-silicon"
+            runs-on: "macos-selfhosted-12-arm64"
             cmake-flags: "-DCORTEX_CPP_VERSION=${{github.event.pull_request.head.sha}} -DCMAKE_BUILD_TEST=ON -DMAC_ARM64=ON -DCMAKE_TOOLCHAIN_FILE=vcpkg/scripts/buildsystems/vcpkg.cmake"
             build-deps-cmake-flags: ""
             ccache-dir: ""

.github/workflows/template-build-macos.yml

@@ -82,7 +82,7 @@ jobs:
       matrix:
         include:
           - arch: 'arm64'
-            runs-on: 'macos-silicon'
+            runs-on: 'macos-selfhosted-12-arm64'
             extra-cmake-flags: "-DMAC_ARM64=ON"
 
           - arch: 'amd64'

engine/controllers/hardware.cc

@@ -38,7 +38,7 @@ void Hardware::Activate(
       ahc.gpus.push_back(g.asInt());
     }
   }
-  std::sort(ahc.gpus.begin(), ahc.gpus.end());
+  
   if (!hw_svc_->IsValidConfig(ahc)) {
     Json::Value ret;
     ret["message"] = "Invalid GPU index provided.";

engine/controllers/server.cc

@@ -3,6 +3,7 @@
 #include "trantor/utils/Logger.h"
 #include "utils/cortex_utils.h"
 #include "utils/function_calling/common.h"
+#include "utils/http_util.h"
 
 using namespace inferences;
 
@@ -27,6 +28,15 @@ void server::ChatCompletion(
   LOG_DEBUG << "Start chat completion";
   auto json_body = req->getJsonObject();
   bool is_stream = (*json_body).get("stream", false).asBool();
+  auto model_id = (*json_body).get("model", "invalid_model").asString();
+  auto engine_type = [this, &json_body]() -> std::string {
+    if (!inference_svc_->HasFieldInReq(json_body, "engine")) {
+      return kLlamaRepo;
+    } else {
+      return (*(json_body)).get("engine", kLlamaRepo).asString();
+    }
+  }();
+
   LOG_DEBUG << "request body: " << json_body->toStyledString();
   auto q = std::make_shared<services::SyncQueue>();
   auto ir = inference_svc_->HandleChatCompletion(q, json_body);
@@ -40,7 +50,7 @@ void server::ChatCompletion(
   }
   LOG_DEBUG << "Wait to chat completion responses";
   if (is_stream) {
-    ProcessStreamRes(std::move(callback), q);
+    ProcessStreamRes(std::move(callback), q, engine_type, model_id);
   } else {
     ProcessNonStreamRes(std::move(callback), *q);
   }
@@ -121,12 +131,16 @@ void server::LoadModel(const HttpRequestPtr& req,
 }
 
 void server::ProcessStreamRes(std::function<void(const HttpResponsePtr&)> cb,
-                              std::shared_ptr<services::SyncQueue> q) {
+                              std::shared_ptr<services::SyncQueue> q,
+                              const std::string& engine_type,
+                              const std::string& model_id) {
   auto err_or_done = std::make_shared<std::atomic_bool>(false);
-  auto chunked_content_provider =
-      [q, err_or_done](char* buf, std::size_t buf_size) -> std::size_t {
+  auto chunked_content_provider = [this, q, err_or_done, engine_type, model_id](
+                                      char* buf,
+                                      std::size_t buf_size) -> std::size_t {
     if (buf == nullptr) {
       LOG_TRACE << "Buf is null";
+      inference_svc_->StopInferencing(engine_type, model_id);
       return 0;
     }
 

engine/controllers/server.h

@@ -72,7 +72,9 @@ class server : public drogon::HttpController<server, false>,
 
  private:
   void ProcessStreamRes(std::function<void(const HttpResponsePtr&)> cb,
-                        std::shared_ptr<services::SyncQueue> q);
+                        std::shared_ptr<services::SyncQueue> q,
+                        const std::string& engine_type,
+                        const std::string& model_id);
   void ProcessNonStreamRes(std::function<void(const HttpResponsePtr&)> cb,
                            services::SyncQueue& q);
 

engine/cortex-common/EngineI.h

@@ -72,4 +72,8 @@ class EngineI {
   virtual void HandleRequest(
       std::shared_ptr<Json::Value> json_body,
       std::function<void(Json::Value&&, Json::Value&&)>&& callback) = 0;
+
+  // Stop inflight chat completion in stream mode
+  virtual void StopInferencing(const std::string& model_id) = 0;
+
 };

engine/database/hardware.cc

@@ -5,22 +5,23 @@
 
 namespace cortex::db {
 
-Hardwares::Hardwares() : db_(cortex::db::Database::GetInstance().db()) {}
+Hardware::Hardware() : db_(cortex::db::Database::GetInstance().db()) {}
 
-Hardwares::Hardwares(SQLite::Database& db) : db_(db) {}
+Hardware::Hardware(SQLite::Database& db) : db_(db) {}
 
-Hardwares::~Hardwares() {}
+
+Hardware::~Hardware() {}
 
 cpp::result<std::vector<HardwareEntry>, std::string>
-Hardwares::LoadHardwareList() const {
+Hardware::LoadHardwareList() const {
   try {
     db_.exec("BEGIN TRANSACTION;");
     cortex::utils::ScopeExit se([this] { db_.exec("COMMIT;"); });
     std::vector<HardwareEntry> entries;
     SQLite::Statement query(
         db_,
         "SELECT uuid, type, "
-        "hardware_id, software_id, activated FROM hardware");
+        "hardware_id, software_id, activated, priority FROM hardware");
 
     while (query.executeStep()) {
       HardwareEntry entry;
@@ -29,6 +30,7 @@ Hardwares::LoadHardwareList() const {
       entry.hardware_id = query.getColumn(2).getInt();
       entry.software_id = query.getColumn(3).getInt();
       entry.activated = query.getColumn(4).getInt();
+      entry.priority = query.getColumn(5).getInt();
       entries.push_back(entry);
     }
     return entries;
@@ -37,19 +39,20 @@ Hardwares::LoadHardwareList() const {
     return cpp::fail(e.what());
   }
 }
-cpp::result<bool, std::string> Hardwares::AddHardwareEntry(
+cpp::result<bool, std::string> Hardware::AddHardwareEntry(
     const HardwareEntry& new_entry) {
   try {
     SQLite::Statement insert(
         db_,
         "INSERT INTO hardware (uuid, type, "
-        "hardware_id, software_id, activated) VALUES (?, ?, "
-        "?, ?, ?)");
+        "hardware_id, software_id, activated, priority) VALUES (?, ?, "
+        "?, ?, ?, ?)");
     insert.bind(1, new_entry.uuid);
     insert.bind(2, new_entry.type);
     insert.bind(3, new_entry.hardware_id);
     insert.bind(4, new_entry.software_id);
     insert.bind(5, new_entry.activated);
+    insert.bind(6, new_entry.priority);
     insert.exec();
     CTL_INF("Inserted: " << new_entry.ToJsonString());
     return true;
@@ -58,17 +61,19 @@ cpp::result<bool, std::string> Hardwares::AddHardwareEntry(
     return cpp::fail(e.what());
   }
 }
-cpp::result<bool, std::string> Hardwares::UpdateHardwareEntry(
+cpp::result<bool, std::string> Hardware::UpdateHardwareEntry(
     const std::string& id, const HardwareEntry& updated_entry) {
   try {
-    SQLite::Statement upd(db_,
-                          "UPDATE hardware "
-                          "SET hardware_id = ?, software_id = ?, activated = ? "
-                          "WHERE uuid = ?");
+    SQLite::Statement upd(
+        db_,
+        "UPDATE hardware "
+        "SET hardware_id = ?, software_id = ?, activated = ?, priority = ? "
+        "WHERE uuid = ?");
     upd.bind(1, updated_entry.hardware_id);
     upd.bind(2, updated_entry.software_id);
     upd.bind(3, updated_entry.activated);
-    upd.bind(4, id);
+    upd.bind(4, updated_entry.priority);
+    upd.bind(5, id);
     if (upd.exec() == 1) {
       CTL_INF("Updated: " << updated_entry.ToJsonString());
       return true;
@@ -79,7 +84,7 @@ cpp::result<bool, std::string> Hardwares::UpdateHardwareEntry(
   }
 }
 
-cpp::result<bool, std::string> Hardwares::DeleteHardwareEntry(
+cpp::result<bool, std::string> Hardware::DeleteHardwareEntry(
     const std::string& id) {
   try {
     SQLite::Statement del(db_, "DELETE from hardware WHERE uuid = ?");

engine/database/hardware.h

@@ -4,8 +4,8 @@
 #include <trantor/utils/Logger.h>
 #include <string>
 #include <vector>
-#include "utils/result.hpp"
 #include "utils/json_helper.h"
+#include "utils/result.hpp"
 
 namespace cortex::db {
 struct HardwareEntry {
@@ -14,33 +14,34 @@ struct HardwareEntry {
   int hardware_id;
   int software_id;
   bool activated;
+  int priority;
   std::string ToJsonString() const {
     Json::Value root;
     root["uuid"] = uuid;
     root["type"] = type;
     root["hardware_id"] = hardware_id;
     root["software_id"] = software_id;
     root["activated"] = activated;
+    root["priority"] = priority;
     return json_helper::DumpJsonString(root);
   }
 };
 
-class Hardwares {
+class Hardware {
 
  private:
   SQLite::Database& db_;
 
-
  public:
-  Hardwares();
-  Hardwares(SQLite::Database& db);
-  ~Hardwares();
+  Hardware();
+  Hardware(SQLite::Database& db);
+  ~Hardware();
 
   cpp::result<std::vector<HardwareEntry>, std::string> LoadHardwareList() const;
-  cpp::result<bool, std::string> AddHardwareEntry(const HardwareEntry& new_entry);
+  cpp::result<bool, std::string> AddHardwareEntry(
+      const HardwareEntry& new_entry);
   cpp::result<bool, std::string> UpdateHardwareEntry(
       const std::string& id, const HardwareEntry& updated_entry);
-  cpp::result<bool, std::string> DeleteHardwareEntry(
-      const std::string& id);
+  cpp::result<bool, std::string> DeleteHardwareEntry(const std::string& id);
 };
 }  // namespace cortex::db

engine/e2e-test/test_api_docker.py

@@ -40,38 +40,39 @@ async def test_models_on_cortexso_hub(self, model_url):
         assert response.status_code == 200
         models = [i["id"] for i in response.json()["data"]]
         assert model_url in models, f"Model not found in list: {model_url}"
+        
+        # TODO(sang) bypass for now. Re-enable when we publish new stable version for llama-cpp engine
+        # print("Start the model")
+        # # Start the model
+        # response = requests.post(
+        #     "http://localhost:3928/v1/models/start", json=json_body
+        # )
+        # print(response.json())
+        # assert response.status_code == 200, f"status_code: {response.status_code}"
 
-        print("Start the model")
-        # Start the model
-        response = requests.post(
-            "http://localhost:3928/v1/models/start", json=json_body
-        )
-        print(response.json())
-        assert response.status_code == 200, f"status_code: {response.status_code}"
-
-        print("Send an inference request")
-        # Send an inference request
-        inference_json_body = {
-            "frequency_penalty": 0.2,
-            "max_tokens": 4096,
-            "messages": [{"content": "", "role": "user"}],
-            "model": model_url,
-            "presence_penalty": 0.6,
-            "stop": ["End"],
-            "stream": False,
-            "temperature": 0.8,
-            "top_p": 0.95,
-        }
-        response = requests.post(
-            "http://localhost:3928/v1/chat/completions",
-            json=inference_json_body,
-            headers={"Content-Type": "application/json"},
-        )
-        assert (
-            response.status_code == 200
-        ), f"status_code: {response.status_code} response: {response.json()}"
+        # print("Send an inference request")
+        # # Send an inference request
+        # inference_json_body = {
+        #     "frequency_penalty": 0.2,
+        #     "max_tokens": 4096,
+        #     "messages": [{"content": "", "role": "user"}],
+        #     "model": model_url,
+        #     "presence_penalty": 0.6,
+        #     "stop": ["End"],
+        #     "stream": False,
+        #     "temperature": 0.8,
+        #     "top_p": 0.95,
+        # }
+        # response = requests.post(
+        #     "http://localhost:3928/v1/chat/completions",
+        #     json=inference_json_body,
+        #     headers={"Content-Type": "application/json"},
+        # )
+        # assert (
+        #     response.status_code == 200
+        # ), f"status_code: {response.status_code} response: {response.json()}"
 
-        print("Stop the model")
-        # Stop the model
-        response = requests.post("http://localhost:3928/v1/models/stop", json=json_body)
-        assert response.status_code == 200, f"status_code: {response.status_code}"
+        # print("Stop the model")
+        # # Stop the model
+        # response = requests.post("http://localhost:3928/v1/models/stop", json=json_body)
+        # assert response.status_code == 200, f"status_code: {response.status_code}"

engine/migrations/db_helper.h

@@ -4,23 +4,28 @@
 namespace cortex::mgr {
 #include <iostream>
 #include <stdexcept>
-#include <vector>
 #include <string>
+#include <vector>
 
-inline bool ColumnExists(SQLite::Database& db, const std::string& table_name, const std::string& column_name) {
-    try {
-        SQLite::Statement query(db, "SELECT " + column_name + " FROM " + table_name + " LIMIT 0");
-        return true;
-    } catch (std::exception&) {
-        return false;
-    }
+inline bool ColumnExists(SQLite::Database& db, const std::string& table_name,
+                         const std::string& column_name) {
+  try {
+    SQLite::Statement query(
+        db, "SELECT " + column_name + " FROM " + table_name + " LIMIT 0");
+    return true;
+  } catch (std::exception&) {
+    return false;
+  }
 }
 
-inline void AddColumnIfNotExists(SQLite::Database& db, const std::string& table_name, 
-                              const std::string& column_name, const std::string& column_type) {
-    if (!ColumnExists(db, table_name, column_name)) {
-        std::string sql = "ALTER TABLE " + table_name + " ADD COLUMN " + column_name + " " + column_type;
-        db.exec(sql);
-    }
+inline void AddColumnIfNotExists(SQLite::Database& db,
+                                 const std::string& table_name,
+                                 const std::string& column_name,
+                                 const std::string& column_type) {
+  if (!ColumnExists(db, table_name, column_name)) {
+    std::string sql = "ALTER TABLE " + table_name + " ADD COLUMN " +
+                      column_name + " " + column_type;
+    db.exec(sql);
+  }
 }
-}
+}  // namespace cortex::mgr