injectlib/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2025-11-25 11:29:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 完成异步任务跟踪架构基础

Browse Source
This commit is contained in:
Junyan Qin
2024-11-01 22:41:26 +08:00
parent 2f05f5b456
commit 6d2a4c038d
16 changed files with 395 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
pkg/api/http/controller/groups/plugins.py
View File

@@ -5,7 +5,7 @@ import traceback
import quart
from .....core import app
from .....core import app, taskmgr
from .. import group
@@ -23,13 +23,27 @@ class PluginsRouterGroup(group.RouterGroup):
'plugins': plugins_data
})
@self.route('/toggle/<author>/<plugin_name>', methods=['PUT'])
@self.route('/<author>/<plugin_name>/toggle', methods=['PUT'])
async def _(author: str, plugin_name: str) -> str:
data = await quart.request.json
target_enabled = data.get('target_enabled')
await self.ap.plugin_mgr.update_plugin_status(plugin_name, target_enabled)
return self.success()
@self.route('/<author>/<plugin_name>/update', methods=['POST'])
async def _(author: str, plugin_name: str) -> str:
ctx = taskmgr.TaskContext.new()
wrapper = self.ap.task_mgr.create_user_task(
self.ap.plugin_mgr.update_plugin(plugin_name, task_context=ctx),
kind="plugin-operation",
name=f"plugin-update-{plugin_name}",
label=f"更新插件 {plugin_name}",
context=ctx
)
return self.success(data={
'task_id': wrapper.id
})
@self.route('/reorder', methods=['PUT'])
async def _() -> str:
data = await quart.request.json

23
pkg/api/http/controller/groups/system.py
View File

@@ -1,6 +1,7 @@
import quart
import asyncio
from .....core import app
from .....core import app, taskmgr
from .. import group
from .....utils import constants
@@ -17,3 +18,23 @@ class SystemRouterGroup(group.RouterGroup):
"debug": constants.debug_mode
}
)
@self.route('/tasks', methods=['GET'])
async def _() -> str:
task_type = quart.request.args.get("type")
if task_type == '':
task_type = None
return self.success(
data=self.ap.task_mgr.get_tasks_dict(task_type)
)
@self.route('/tasks/<task_id>', methods=['GET'])
async def _(task_id: str) -> str:
task = self.ap.task_mgr.get_task_by_id(int(task_id))
if task is None:
return self.http_status(404, 404, "Task not found")
return self.success(data=task.to_dict())

32
pkg/api/http/controller/main.py
View File

@@ -19,38 +19,32 @@ class HTTPController:
def __init__(self, ap: app.Application) -> None:
self.ap = ap
self.quart_app = quart.Quart(__name__)
quart_cors.cors(self.quart_app, allow_origin='*')
quart_cors.cors(self.quart_app, allow_origin="*")
async def initialize(self) -> None:
await self.register_routes()
async def run(self) -> None:
if self.ap.system_cfg.data['http-api']['enable']:
if self.ap.system_cfg.data["http-api"]["enable"]:
async def shutdown_trigger_placeholder():
while True:
await asyncio.sleep(1)
# task = asyncio.create_task(self.quart_app.run_task(
# host=self.ap.system_cfg.data['http-api']['host'],
# port=self.ap.system_cfg.data['http-api']['port'],
# shutdown_trigger=shutdown_trigger_placeholder
# ))
# self.ap.asyncio_tasks.append(task)
self.ap.task_mgr.create_task(self.quart_app.run_task(
host=self.ap.system_cfg.data['http-api']['host'],
port=self.ap.system_cfg.data['http-api']['port'],
shutdown_trigger=shutdown_trigger_placeholder
))
self.ap.task_mgr.create_task(
self.quart_app.run_task(
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",
)
async def register_routes(self) -> None:
@self.quart_app.route('/healthz')
@self.quart_app.route("/healthz")
async def healthz():
return {
"code": 0,
"msg": "ok"
}
return {"code": 0, "msg": "ok"}
for g in group.preregistered_groups:
ginst = g(self.ap, self.quart_app)

39
pkg/audit/center/apigroup.py
View File

@@ -14,6 +14,7 @@ from ...core import app
class APIGroup(metaclass=abc.ABCMeta):
"""API 组抽象类"""
_basic_info: dict = None
_runtime_info: dict = None
@@ -32,32 +33,27 @@ class APIGroup(metaclass=abc.ABCMeta):
data: dict = None,
params: dict = None,
headers: dict = {},
**kwargs
**kwargs,
):
"""
执行请求
"""
self._runtime_info['account_id'] = "-1"
self._runtime_info["account_id"] = "-1"
url = self.prefix + path
data = json.dumps(data)
headers['Content-Type'] = 'application/json'
headers["Content-Type"] = "application/json"
try:
async with aiohttp.ClientSession() as session:
async with session.request(
method,
url,
data=data,
params=params,
headers=headers,
**kwargs
method, url, data=data, params=params, headers=headers, **kwargs
) as resp:
self.ap.logger.debug("data: %s", data)
self.ap.logger.debug("ret: %s", await resp.text())
except Exception as e:
self.ap.logger.debug(f'上报失败: {e}')
self.ap.logger.debug(f"上报失败: {e}")
async def do(
self,
@@ -66,32 +62,29 @@ class APIGroup(metaclass=abc.ABCMeta):
data: dict = None,
params: dict = None,
headers: dict = {},
**kwargs
**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}",
).task
return self.ap.task_mgr.create_task(self._do(method, path, data, params, headers, **kwargs)).task
def gen_rid(
self
):
def gen_rid(self):
"""生成一个请求 ID"""
return str(uuid.uuid4())
def basic_info(
self
):
def basic_info(self):
"""获取基本信息"""
basic_info = APIGroup._basic_info.copy()
basic_info['rid'] = self.gen_rid()
basic_info["rid"] = self.gen_rid()
return basic_info
def runtime_info(
self
):
def runtime_info(self):
"""获取运行时信息"""
return APIGroup._runtime_info

15
pkg/core/app.py
View File

@@ -114,17 +114,10 @@ class Application:
while True:
await asyncio.sleep(1)
# tasks = [
# asyncio.create_task(self.platform_mgr.run()), # 消息平台
# asyncio.create_task(self.ctrl.run()), # 消息处理循环
# asyncio.create_task(self.http_ctrl.run()), # http 接口服务
# asyncio.create_task(never_ending())
# ]
# self.asyncio_tasks.extend(tasks)
self.task_mgr.create_task(self.platform_mgr.run())
self.task_mgr.create_task(self.ctrl.run())
self.task_mgr.create_task(self.http_ctrl.run())
self.task_mgr.create_task(never_ending())
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")
await self.task_mgr.wait_all()
except asyncio.CancelledError:

136
pkg/core/taskmgr.py
View File

@@ -2,6 +2,7 @@ from __future__ import annotations
import asyncio
import typing
import datetime
from . import app
@@ -16,22 +17,68 @@ class TaskContext:
"""记录日志"""
def __init__(self):
self.current_action = ""
self.current_action = "default"
self.log = ""
def log(self, msg: str):
def _log(self, msg: str):
self.log += msg + "\n"
def set_current_action(self, action: str):
self.current_action = action
def trace(
self,
msg: str,
action: str = None,
):
if action is not None:
self.set_current_action(action)
self._log(
f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | {self.current_action} | {msg}"
)
def to_dict(self) -> dict:
return {"current_action": self.current_action, "log": self.log}
@staticmethod
def new() -> TaskContext:
return TaskContext()
@staticmethod
def placeholder() -> TaskContext:
global placeholder_context
if placeholder_context is None:
placeholder_context = TaskContext()
return placeholder_context
placeholder_context: TaskContext | None = None
class TaskWrapper:
"""任务包装器"""
_id_index: int = 0
"""任务ID索引"""
id: int
"""任务ID"""
task_type: str = "system" # 任务类型: system 或 user
"""任务类型"""
kind: str = "system_task"
"""任务种类"""
name: str = ""
"""任务唯一名称"""
label: str = ""
"""任务显示名称"""
task_context: TaskContext
"""任务上下文"""
@@ -41,11 +88,55 @@ class TaskWrapper:
ap: app.Application
"""应用实例"""
def __init__(self, ap: app.Application, coro: typing.Coroutine, task_type: str = "system", context: TaskContext = None):
def __init__(
self,
ap: app.Application,
coro: typing.Coroutine,
task_type: str = "system",
kind: str = "system_task",
name: str = "",
label: str = "",
context: TaskContext = None,
):
self.id = TaskWrapper._id_index
TaskWrapper._id_index += 1
self.ap = ap
self.task_context = context or TaskContext()
self.task = self.ap.event_loop.create_task(coro)
self.task_type = task_type
self.kind = kind
self.name = name
self.label = label if label != "" else name
self.task.set_name(name)
def assume_exception(self):
try:
return self.task.exception()
except:
return None
def assume_result(self):
try:
return self.task.result()
except:
return None
def to_dict(self) -> dict:
return {
"id": self.id,
"task_type": self.task_type,
"kind": self.kind,
"name": self.name,
"label": self.label,
"task_context": self.task_context.to_dict(),
"runtime": {
"done": self.task.done(),
"state": self.task._state,
"exception": self.assume_exception(),
"result": self.assume_result(),
},
}
class AsyncTaskManager:
@@ -61,13 +152,48 @@ class AsyncTaskManager:
self.ap = ap
self.tasks = []
def create_task(self, coro: typing.Coroutine, task_type: str = "system", context: TaskContext = None) -> TaskWrapper:
wrapper = TaskWrapper(self.ap, coro, task_type, context)
def create_task(
self,
coro: typing.Coroutine,
task_type: str = "system",
kind: str = "system-task",
name: str = "",
label: str = "",
context: TaskContext = None,
) -> TaskWrapper:
wrapper = TaskWrapper(self.ap, coro, task_type, kind, name, label, context)
self.tasks.append(wrapper)
return wrapper
def create_user_task(
self,
coro: typing.Coroutine,
kind: str = "user-task",
name: str = "",
label: str = "",
context: TaskContext = None,
) -> TaskWrapper:
return self.create_task(coro, "user", kind, name, label, context)
async def wait_all(self):
await asyncio.gather(*[t.task for t in self.tasks], return_exceptions=True)
def get_all_tasks(self) -> list[TaskWrapper]:
return self.tasks
def get_tasks_dict(
self,
type: str = None,
) -> dict:
return {
"tasks": [
t.to_dict() for t in self.tasks if type is None or t.task_type == type
],
"id_index": TaskWrapper._id_index,
}
def get_task_by_id(self, id: int) -> TaskWrapper | None:
for t in self.tasks:
if t.id == id:
return t
return None

6
pkg/pipeline/controller.py
View File

@@ -62,7 +62,11 @@ class Controller:
# task = asyncio.create_task(_process_query(selected_query))
# self.ap.asyncio_tasks.append(task)
self.ap.task_mgr.create_task(_process_query(selected_query))
self.ap.task_mgr.create_task(
_process_query(selected_query),
kind="query",
name=f"query-{selected_query.query_id}",
)
except Exception as e:
# traceback.print_exc()

6
pkg/platform/manager.py
View File

@@ -186,7 +186,11 @@ class PlatformManager:
for task in tasks:
# async_task = asyncio.create_task(task)
# self.ap.asyncio_tasks.append(async_task)
self.ap.task_mgr.create_task(task)
self.ap.task_mgr.create_task(
task,
kind="platform-adapter",
name=f"platform-adapter-{adapter.name}",
)
except Exception as e:
self.ap.logger.error('平台适配器运行出错: ' + str(e))

3
pkg/plugin/installer.py
View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import typing
import abc
from ..core import app
from ..core import app, taskmgr
class PluginInstaller(metaclass=abc.ABCMeta):
@@ -40,6 +40,7 @@ class PluginInstaller(metaclass=abc.ABCMeta):
self,
plugin_name: str,
plugin_source: str=None,
task_context: taskmgr.TaskContext = taskmgr.TaskContext.placeholder(),
):
"""更新插件
"""

15
pkg/plugin/installers/github.py
View File

@@ -9,6 +9,7 @@ import requests
from .. import installer, errors
from ...utils import pkgmgr
from ...core import taskmgr
class GitHubRepoInstaller(installer.PluginInstaller):
@@ -94,13 +95,20 @@ class GitHubRepoInstaller(installer.PluginInstaller):
async def install_plugin(
self,
plugin_source: str,
task_context: taskmgr.TaskContext = taskmgr.TaskContext.placeholder(),
):
"""安装插件
"""
task_context.trace("下载插件源码...", "install-plugin")
repo_label = await self.download_plugin_source_code(plugin_source, "plugins/")
task_context.trace("安装插件依赖...", "install-plugin")
await self.install_requirements("plugins/" + repo_label)
task_context.trace("完成.", "install-plugin")
await self.ap.plugin_mgr.setting.record_installed_plugin_source(
"plugins/"+repo_label+'/', plugin_source
)
@@ -122,9 +130,12 @@ class GitHubRepoInstaller(installer.PluginInstaller):
self,
plugin_name: str,
plugin_source: str=None,
task_context: taskmgr.TaskContext = taskmgr.TaskContext.placeholder(),
):
"""更新插件
"""
task_context.trace("更新插件...", "update-plugin")
plugin_container = self.ap.plugin_mgr.get_plugin_by_name(plugin_name)
if plugin_container is None:
@@ -133,7 +144,9 @@ class GitHubRepoInstaller(installer.PluginInstaller):
if plugin_container.plugin_source:
plugin_source = plugin_container.plugin_source
await self.install_plugin(plugin_source)
task_context.trace("转交安装任务.", "update-plugin")
await self.install_plugin(plugin_source, task_context)
else:
raise errors.PluginInstallerError('插件无源码信息，无法更新')

6
pkg/plugin/manager.py
View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import typing
import traceback
from ..core import app
from ..core import app, taskmgr
from . import context, loader, events, installer, setting, models
from .loaders import classic
from .installers import github
@@ -102,10 +102,11 @@ class PluginManager:
self,
plugin_name: str,
plugin_source: str=None,
task_context: taskmgr.TaskContext = taskmgr.TaskContext.placeholder(),
):
"""更新插件
"""
await self.installer.update_plugin(plugin_name, plugin_source)
await self.installer.update_plugin(plugin_name, plugin_source, task_context)
plugin_container = self.get_plugin_by_name(plugin_name)
@@ -120,6 +121,7 @@ class PluginManager:
new_version="HEAD"
)
def get_plugin_by_name(self, plugin_name: str) -> context.RuntimeContainer:
"""通过插件名获取插件
"""

2
web/src/App.vue
View File

@@ -42,7 +42,7 @@
</v-list-item> -->
<v-dialog max-width="500" persistent v-model="taskDialogShow">
<template v-slot:activator="{ props: activatorProps }">
<v-list-item id="system-tasks-list-item" title="系统任务" prepend-icon="mdi-align-horizontal-left" v-tooltip="任务列表" v-bind="activatorProps">
<v-list-item id="system-tasks-list-item" title="任务列表" prepend-icon="mdi-align-horizontal-left" v-tooltip="任务列表" v-bind="activatorProps">
</v-list-item>
</template>

2
web/src/components/PluginCard.vue
View File

@@ -95,7 +95,7 @@ const menuItems = [
},
{
title: '删除',
condition: (plugin) => plugin.source != '',
condition: (plugin) => true,
action: uninstallPlugin
}
]

78
web/src/components/TaskCard.vue Normal file
View File

@@ -0,0 +1,78 @@
<template>
<div class="task-card">
<div class="task-card-icon">
<v-progress-circular :size="25" :width="2" indeterminate v-if="task.runtime.state == 'PENDING'" />
<v-icon v-else-if="task.runtime.state == 'FINISHED'" style="color: #4caf50;" :size="25" icon="mdi-check" />
<v-icon v-else-if="task.runtime.state == 'CANCELLED'" style="color: #f44336;" :size="25" icon="mdi-close" />
</div>
<div class="task-card-content">
<div class="task-card-kind">{{ task.kind }}</div>
<div class="task-card-label">{{ task.label }}</div>
<v-chip class="task-card-action" color="primary" variant="outlined" size="small" density="compact">正在执行: {{ task.task_context.current_action }}</v-chip>
</div>
<div class="task-card-actions">
<!-- <v-icon icon="mdi-details" /> -->
<v-btn icon="mdi-information-outline" variant="text" />
</div>
</div>
</template>
<script setup>
defineProps({
task: {
type: Object,
required: true
}
})
</script>
<style scoped>
.task-card {
display: flex;
flex-direction: row;
align-items: center;
justify-content: flex-start;
width: 100%;
padding-inline: 0.5rem;
}
.task-card-icon {
margin-right: 1rem;
margin-left: 0.8rem;
}
.task-card-content {
flex: 1;
margin-left: 0.4rem;
display: flex;
flex-direction: column;
gap: 0.02rem;
user-select: none;
}
.task-card-kind {
font-size: 0.6rem;
color: #666;
}
.task-card-label {
font-size: 0.8rem;
color: #333;
}
.task-card-action {
font-size: 0.6rem;
width: fit-content;
padding-inline: 0.3rem;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.task-card-actions {
justify-self: flex-end;
}
</style>

85
web/src/components/TaskDialog.vue
View File

@@ -1,27 +1,9 @@
<template>
<v-card prepend-icon="mdi-align-horizontal-left" text="用户发起的任务列表" title="任务列表">
<v-list id="plugin-orchestration-list">
<draggable v-model="plugins" item-key="name" group="plugins" @start="drag = true"
id="plugin-orchestration-draggable"
@end="drag = false">
<template #item="{ element }">
<div class="plugin-orchestration-item">
<div class="plugin-orchestration-item-title">
<div class="plugin-orchestration-item-author">
{{ element.author }} /
</div>
<div class="plugin-orchestration-item-name">
{{ element.name }}
</div>
</div>
<div class="plugin-orchestration-item-action">
<v-icon>mdi-drag</v-icon>
</div>
</div>
</template>
</draggable>
<v-card class="task-dialog" prepend-icon="mdi-align-horizontal-left" text="用户发起的任务列表" title="任务列表">
<v-list id="task-list" v-if="taskList.length > 0">
<TaskCard class="task-card" v-for="task in taskList" :key="task.id" :task="task" />
</v-list>
<div v-else><v-alert color="warning" icon="$warning" title="暂无任务" text="暂无已添加的用户任务项" density="compact" style="margin-inline: 1rem;"></v-alert></div>
<template v-slot:actions>
<v-btn class="ml-auto" text="关闭" prepend-icon="mdi-close" @click="close"></v-btn>
@@ -35,7 +17,11 @@ defineProps({
const emit = defineEmits(['close'])
import { ref } from 'vue'
import TaskCard from '@/components/TaskCard.vue'
import { ref, onMounted, onUnmounted, getCurrentInstance } from 'vue'
const { proxy } = getCurrentInstance()
import { inject } from 'vue'
@@ -44,8 +30,61 @@ const snackbar = inject('snackbar')
const close = () => {
emit('close')
}
const taskList = ref([])
const refresh = () => {
proxy.$axios.get('/system/tasks', {
params: {
type: 'user'
}
}).then(response => {
if (response.data.code != 0) {
snackbar.error(response.data.message)
return
}
taskList.value = response.data.data.tasks
// 倒序
taskList.value.reverse()
}).catch(error => {
snackbar.error(error.message)
})
}
let refreshTask = null
onMounted(() => {
refresh()
refreshTask = setInterval(refresh, 500)
})
onUnmounted(() => {
clearInterval(refreshTask)
})
</script>
<style scoped>
.task-dialog {
width: 100%;
}
#task-list {
max-height: 20rem;
overflow-y: auto;
margin-inline: 1rem;
width: calc(100% - 2.2rem);
padding-inline: 0.6rem;
}
.task-card {
/* margin-bottom: 0.1rem; */
display: flex;
flex-direction: row;
/* box-shadow: 1px 1px 1px 1px rgba(0, 0, 0, 0.1); */
height: 4rem;
/* border: 0.08rem solid #ccc; */
box-shadow: 0.1rem 0.1rem 0.2rem 0.05rem #ccc;
background-color: #ffffff;
}
</style>

16
web/src/pages/Plugins.vue
View File

@@ -52,7 +52,7 @@
</v-card>
<div class="plugins-container">
<PluginCard class="plugin-card" v-for="plugin in plugins" :key="plugin.name" :plugin="plugin"
@toggle="togglePlugin" />
@toggle="togglePlugin" @update="updatePlugin" />
</div>
</template>
@@ -88,7 +88,7 @@ const refresh = () => {
onMounted(refresh)
const togglePlugin = (plugin) => {
proxy.$axios.put(`/plugins/toggle/${plugin.author}/${plugin.name}`, {
proxy.$axios.put(`/plugins/${plugin.author}/${plugin.name}/toggle`, {
target_enabled: !plugin.enabled
}).then(res => {
if (res.data.code != 0) {
@@ -101,6 +101,18 @@ const togglePlugin = (plugin) => {
})
}
const updatePlugin = (plugin) => {
proxy.$axios.post(`/plugins/${plugin.author}/${plugin.name}/update`).then(res => {
if (res.data.code != 0) {
snackbar.error(res.data.msg)
return
}
snackbar.success(`已添加更新任务 请到任务列表查看进度`)
}).catch(error => {
snackbar.error(error)
})
}
const isOrchestrationDialogActive = ref(false)
const cancelOrderChanges = () => {