Skip to content

Commit

Permalink
feat: 消息平台热重载
Browse files Browse the repository at this point in the history
  • Loading branch information
RockChinQ committed Nov 16, 2024
1 parent 3239c9e commit bb21988
Show file tree
Hide file tree
Showing 14 changed files with 169 additions and 37 deletions.
15 changes: 8 additions & 7 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,14 @@
# QChatGPT/main.py

asciiart = r"""
___ ___ _ _ ___ ___ _____
/ _ \ / __| |_ __ _| |_ / __| _ \_ _|
| (_) | (__| ' \/ _` | _| (_ | _/ | |
\__\_\\___|_||_\__,_|\__|\___|_| |_|
⭐️开源地址: https://github.com/RockChinQ/QChatGPT
📖文档地址: https://q.rkcn.top
_ ___ _
| | __ _ _ _ __ _| _ ) ___| |_
| |__/ _` | ' \/ _` | _ \/ _ \ _|
|____\__,_|_||_\__, |___/\___/\__|
|___/
⭐️开源地址: https://github.com/RockChinQ/LangBot
📖文档地址: https://docs.langbot.app
"""


Expand Down
22 changes: 22 additions & 0 deletions pkg/api/http/controller/groups/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,3 +39,25 @@ async def _(task_id: str) -> str:
return self.http_status(404, 404, "Task not found")

return self.success(data=task.to_dict())

@self.route('/reload', methods=['POST'])
async def _() -> str:
json_data = await quart.request.json

scope = json_data.get("scope")

await self.ap.reload(
scope=scope
)
return self.success()

@self.route('/_debug/exec', methods=['POST'])
async def _() -> str:
if not constants.debug_mode:
return self.http_status(403, 403, "Forbidden")

py_code = await quart.request.data

ap = self.ap

return self.success(data=exec(py_code, {"ap": ap}))
15 changes: 13 additions & 2 deletions pkg/api/http/controller/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import quart
import quart_cors

from ....core import app
from ....core import app, entities as core_entities
from .groups import logs, system, settings, plugins, stats
from . import group

Expand All @@ -32,15 +32,26 @@ async def shutdown_trigger_placeholder():
while True:
await asyncio.sleep(1)

async def exception_handler(*args, **kwargs):
try:
await self.quart_app.run_task(
*args, **kwargs
)
except Exception as e:
self.ap.logger.error(f"启动 HTTP 服务失败: {e}")

self.ap.task_mgr.create_task(
self.quart_app.run_task(
exception_handler(
host=self.ap.system_cfg.data["http-api"]["host"],
port=self.ap.system_cfg.data["http-api"]["port"],
shutdown_trigger=shutdown_trigger_placeholder,
),
name="http-api-quart",
scopes=[core_entities.LifecycleControlScope.APPLICATION],
)

# await asyncio.sleep(5)

async def register_routes(self) -> None:

@self.quart_app.route("/healthz")
Expand Down
6 changes: 2 additions & 4 deletions pkg/audit/center/apigroup.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import aiohttp
import requests

from ...core import app
from ...core import app, entities as core_entities


class APIGroup(metaclass=abc.ABCMeta):
Expand Down Expand Up @@ -65,14 +65,12 @@ async def do(
**kwargs,
) -> asyncio.Task:
"""执行请求"""
# task = asyncio.create_task(self._do(method, path, data, params, headers, **kwargs))

# self.ap.asyncio_tasks.append(task)

return self.ap.task_mgr.create_task(
self._do(method, path, data, params, headers, **kwargs),
kind="telemetry-operation",
name=f"{method} {path}",
scopes=[core_entities.LifecycleControlScope.APPLICATION],
).task

def gen_rid(self):
Expand Down
59 changes: 49 additions & 10 deletions pkg/core/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import asyncio
import threading
import traceback
import enum

from ..platform import manager as im_mgr
from ..provider.session import sessionmgr as llm_session_mgr
Expand All @@ -21,8 +22,9 @@
from ..utils import version as version_mgr, proxy as proxy_mgr, announce as announce_mgr
from ..persistence import mgr as persistencemgr
from ..api.http.controller import main as http_controller
from ..utils import logcache
from ..utils import logcache, ip
from . import taskmgr
from . import entities as core_entities


class Application:
Expand Down Expand Up @@ -114,21 +116,58 @@ async def never_ending():
while True:
await asyncio.sleep(1)

self.task_mgr.create_task(self.platform_mgr.run(), name="platform-manager")
self.task_mgr.create_task(self.ctrl.run(), name="query-controller")
self.task_mgr.create_task(self.http_ctrl.run(), name="http-api-controller")
self.task_mgr.create_task(never_ending(), name="never-ending-task")
self.task_mgr.create_task(self.platform_mgr.run(), name="platform-manager", scopes=[core_entities.LifecycleControlScope.APPLICATION, core_entities.LifecycleControlScope.PLATFORM])
self.task_mgr.create_task(self.ctrl.run(), name="query-controller", scopes=[core_entities.LifecycleControlScope.APPLICATION])
self.task_mgr.create_task(self.http_ctrl.run(), name="http-api-controller", scopes=[core_entities.LifecycleControlScope.APPLICATION])
self.task_mgr.create_task(never_ending(), name="never-ending-task", scopes=[core_entities.LifecycleControlScope.APPLICATION])

await self.print_web_access_info()
await self.task_mgr.wait_all()
except asyncio.CancelledError:
pass
except Exception as e:
self.logger.error(f"应用运行致命异常: {e}")
self.logger.debug(f"Traceback: {traceback.format_exc()}")

async def scoped_shutdown(self, scopes: list[str]):
pass
async def print_web_access_info(self):
"""打印访问 webui 的提示"""
import socket

host_ip = socket.gethostbyname(socket.gethostname())

public_ip = await ip.get_myip()

port = self.system_cfg.data['http-api']['port']

tips = f"""
=======================================
✨ 您可通过以下方式访问管理面板:
🏠 本地地址:http://{host_ip}:{port}/
🌐 公网地址:http://{public_ip}:{port}/
📌 如果您在容器中运行此程序,请确保容器的 {port} 端口已对外暴露
🔗 若要使用公网地址访问,请阅读以下须知
1. 公网地址仅供参考,请以您的主机公网 IP 为准;
2. 要使用公网地址访问,请确保您的主机具有公网 IP,并且系统防火墙已放行 {port} 端口;
=======================================
""".strip()
for line in tips.split("\n"):
self.logger.info(line)

async def reload(
self,
scope: core_entities.LifecycleControlScope,
):
match scope:
case core_entities.LifecycleControlScope.PLATFORM.value:
self.logger.info("执行热重载 scope="+scope)
await self.platform_mgr.shutdown()

self.platform_mgr = im_mgr.PlatformManager(self)

await self.platform_mgr.initialize()

async def shutdown(self):
for task in self.task_mgr.tasks:
task.cancel()
self.task_mgr.create_task(self.platform_mgr.run(), name="platform-manager", scopes=[core_entities.LifecycleControlScope.APPLICATION, core_entities.LifecycleControlScope.PLATFORM])
case _:
pass
2 changes: 1 addition & 1 deletion pkg/core/bootutils/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
"anthropic": "anthropic",
"colorlog": "colorlog",
"aiocqhttp": "aiocqhttp",
"botpy": "qq-botpy",
"botpy": "qq-botpy-rc",
"PIL": "pillow",
"nakuru": "nakuru-project-idk",
"tiktoken": "tiktoken",
Expand Down
8 changes: 8 additions & 0 deletions pkg/core/entities.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,14 @@
from ..platform.types import entities as platform_entities



class LifecycleControlScope(enum.Enum):

APPLICATION = "application"
PLATFORM = "platform"
PLUGIN = "plugin"


class LauncherTypes(enum.Enum):
"""一个请求的发起者类型"""

Expand Down
22 changes: 19 additions & 3 deletions pkg/core/taskmgr.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import traceback

from . import app
from . import entities as core_entities


class TaskContext:
Expand Down Expand Up @@ -71,7 +72,7 @@ class TaskWrapper:
task_type: str = "system" # 任务类型: system 或 user
"""任务类型"""

kind: str = "system_task"
kind: str = "system_task" # 由发起者确定任务种类,通常同质化的任务种类相同
"""任务种类"""

name: str = ""
Expand All @@ -92,6 +93,9 @@ class TaskWrapper:
ap: app.Application
"""应用实例"""

scopes: list[core_entities.LifecycleControlScope]
"""任务所属生命周期控制范围"""

def __init__(
self,
ap: app.Application,
Expand All @@ -101,6 +105,7 @@ def __init__(
name: str = "",
label: str = "",
context: TaskContext = None,
scopes: list[core_entities.LifecycleControlScope] = [core_entities.LifecycleControlScope.APPLICATION],
):
self.id = TaskWrapper._id_index
TaskWrapper._id_index += 1
Expand All @@ -112,6 +117,7 @@ def __init__(
self.name = name
self.label = label if label != "" else name
self.task.set_name(name)
self.scopes = scopes

def assume_exception(self):
try:
Expand Down Expand Up @@ -145,6 +151,7 @@ def to_dict(self) -> dict:
"kind": self.kind,
"name": self.name,
"label": self.label,
"scopes": [scope.value for scope in self.scopes],
"task_context": self.task_context.to_dict(),
"runtime": {
"done": self.task.done(),
Expand Down Expand Up @@ -180,8 +187,9 @@ def create_task(
name: str = "",
label: str = "",
context: TaskContext = None,
scopes: list[core_entities.LifecycleControlScope] = [core_entities.LifecycleControlScope.APPLICATION],
) -> TaskWrapper:
wrapper = TaskWrapper(self.ap, coro, task_type, kind, name, label, context)
wrapper = TaskWrapper(self.ap, coro, task_type, kind, name, label, context, scopes)
self.tasks.append(wrapper)
return wrapper

Expand All @@ -192,8 +200,9 @@ def create_user_task(
name: str = "",
label: str = "",
context: TaskContext = None,
scopes: list[core_entities.LifecycleControlScope] = [core_entities.LifecycleControlScope.APPLICATION],
) -> TaskWrapper:
return self.create_task(coro, "user", kind, name, label, context)
return self.create_task(coro, "user", kind, name, label, context, scopes)

async def wait_all(self):
await asyncio.gather(*[t.task for t in self.tasks], return_exceptions=True)
Expand All @@ -217,3 +226,10 @@ def get_task_by_id(self, id: int) -> TaskWrapper | None:
if t.id == id:
return t
return None

def cancel_by_scope(self, scope: core_entities.LifecycleControlScope):
for wrapper in self.tasks:

if not wrapper.task.done() and scope in wrapper.scopes:

wrapper.task.cancel()
5 changes: 1 addition & 4 deletions pkg/pipeline/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import typing
import traceback


from ..core import app, entities
from . import entities as pipeline_entities
from ..plugin import events
Expand Down Expand Up @@ -59,13 +58,11 @@ async def _process_query(selected_query):
(await self.ap.sess_mgr.get_session(selected_query)).semaphore.release()
# 通知其他协程,有新的请求可以处理了
self.ap.query_pool.condition.notify_all()

# task = asyncio.create_task(_process_query(selected_query))
# self.ap.asyncio_tasks.append(task)
self.ap.task_mgr.create_task(
_process_query(selected_query),
kind="query",
name=f"query-{selected_query.query_id}",
scopes=[entities.LifecycleControlScope.APPLICATION, entities.LifecycleControlScope.PLATFORM],
)

except Exception as e:
Expand Down
8 changes: 5 additions & 3 deletions pkg/platform/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,22 +174,23 @@ async def run(self):
try:
tasks = []
for adapter in self.adapters:
async def exception_wrapper(adapter):
async def exception_wrapper(adapter: msadapter.MessageSourceAdapter):
try:
await adapter.run_async()
except Exception as e:
if isinstance(e, asyncio.CancelledError):
return
self.ap.logger.error('平台适配器运行出错: ' + str(e))
self.ap.logger.debug(f"Traceback: {traceback.format_exc()}")

tasks.append(exception_wrapper(adapter))

for task in tasks:
# async_task = asyncio.create_task(task)
# self.ap.asyncio_tasks.append(async_task)
self.ap.task_mgr.create_task(
task,
kind="platform-adapter",
name=f"platform-adapter-{adapter.name}",
scopes=[core_entities.LifecycleControlScope.APPLICATION, core_entities.LifecycleControlScope.PLATFORM],
)

except Exception as e:
Expand All @@ -199,3 +200,4 @@ async def exception_wrapper(adapter):
async def shutdown(self):
for adapter in self.adapters:
await adapter.kill()
self.ap.task_mgr.cancel_by_scope(core_entities.LifecycleControlScope.PLATFORM)
8 changes: 6 additions & 2 deletions pkg/platform/sources/qqbotpy.py
Original file line number Diff line number Diff line change
Expand Up @@ -588,8 +588,12 @@ async def run_async(self):
self.member_openid_mapping, self.group_openid_mapping
)

self.cfg['ret_coro'] = True

self.ap.logger.info("运行 QQ 官方适配器")
await self.bot.start(**self.cfg)
await (await self.bot.start(**self.cfg))

async def kill(self) -> bool:
return False
if not self.bot.is_closed():
await self.bot.close()
return True
9 changes: 9 additions & 0 deletions pkg/utils/ip.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
import aiohttp

async def get_myip() -> str:
try:
async with aiohttp.ClientSession() as session:
async with session.get("https://ip.useragentinfo.com/myip") as response:
return await response.text()
except Exception as e:
return '0.0.0.0'
2 changes: 1 addition & 1 deletion requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ openai>1.0.0
anthropic
colorlog~=6.6.0
aiocqhttp
qq-botpy
qq-botpy-rc
nakuru-project-idk
Pillow
tiktoken
Expand Down
Loading

0 comments on commit bb21988

Please sign in to comment.