From 43ec5a4a7124a2c0f9e527fadca3c204ce52a251 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Sat, 11 Oct 2025 15:02:24 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E7=AE=A1=E7=90=86=E7=B3=BB=E7=BB=9F=E5=92=8C?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E4=B8=AD=E5=BF=83UI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 后端变更 ### 任务管理核心 - 新增 tasker 服务，支持异步任务队列和工作线程池 - 添加任务执行时间追踪（started_at, completed_at） - 实现任务状态持久化（JSON文件存储） - 支持任务取消和进度更新 ### 知识库导入优化 - 将知识库文档导入改为异步任务处理 - 逐个处理文档并实时更新进度（5% → 95%） - 支持任务取消检测 ### API接口 - 新增任务列表、详情、取消等REST API - 任务API需要管理员权限 ### 测试 - 添加任务路由集成测试 ## 前端变更 ### 任务中心UI - 新增任务中心抽屉组件，采用简洁的卡片式设计 - 支持任务状态筛选（全部/进行中/已完成/失败） - 进行中任务显示实时进度条 - 已完成任务显示执行时长和完成状态徽章 - 支持相对时间显示（如：5分钟前） ### 状态管理 - 新增 tasker store，管理任务列表和轮询 - 支持任务注册、刷新、取消操作 ### 交互优化 - 在状态栏和侧边栏添加任务中心入口 - 显示活跃任务数量徽章 - 知识库导入自动注册到任务中心 --- scripts/batch_upload.py | 14 +- server/main.py | 12 + server/routers/__init__.py | 2 + server/routers/knowledge_router.py | 66 ++- server/routers/task_router.py | 35 ++ server/services/__init__.py | 3 + server/services/tasker.py | 301 +++++++++++ test/api/test_task_router.py | 93 ++++ web/src/apis/index.js | 3 +- web/src/apis/tasker.js | 19 + web/src/components/StatusBar.vue | 66 ++- web/src/components/TaskCenterDrawer.vue | 659 ++++++++++++++++++++++++ web/src/layouts/AppLayout.vue | 35 +- web/src/stores/database.js | 19 +- web/src/stores/tasker.js | 163 ++++++ web/src/views/DashboardView.vue | 2 - 16 files changed, 1475 insertions(+), 17 deletions(-) create mode 100644 server/routers/task_router.py create mode 100644 server/services/__init__.py create mode 100644 server/services/tasker.py create mode 100644 test/api/test_task_router.py create mode 100644 web/src/apis/tasker.js create mode 100644 web/src/components/TaskCenterDrawer.vue create mode 100644 web/src/stores/tasker.js diff --git a/scripts/batch_upload.py b/scripts/batch_upload.py index b8ce49ac..33c23b4b 100644 --- a/scripts/batch_upload.py +++ b/scripts/batch_upload.py @@ -88,8 +88,18 @@ async def process_document( response.raise_for_status() result = response.json() - # Check if the overall request was successful - if result.get("status") != "success": + # Handle asynchronous ingest response + overall_status = result.get("status") + if overall_status == "queued": + task_id = result.get("task_id") + extra = f" (task id: {task_id})" if task_id else "" + console.print( + f"[bold cyan]Ingestion queued for {server_file_path}{extra}. Track progress in the task center.[/bold cyan]" + ) + return True + + # Check if the overall request was successful for synchronous responses + if overall_status != "success": console.print( f"[bold yellow]Processing warning for {server_file_path}: {result.get('message')}[/bold yellow]" ) diff --git a/server/main.py b/server/main.py index a26ead72..701fd773 100644 --- a/server/main.py +++ b/server/main.py @@ -4,6 +4,7 @@ from fastapi.middleware.cors import CORSMiddleware from starlette.middleware.base import BaseHTTPMiddleware from server.routers import router +from server.services.tasker import tasker from server.utils.auth_middleware import is_public_path from server.utils.common_utils import setup_logging @@ -59,5 +60,16 @@ class AuthMiddleware(BaseHTTPMiddleware): # 添加鉴权中间件 app.add_middleware(AuthMiddleware) + +@app.on_event("startup") +async def start_tasker() -> None: + await tasker.start() + + +@app.on_event("shutdown") +async def stop_tasker() -> None: + await tasker.shutdown() + + if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=5050, threads=10, workers=10, reload=True) diff --git a/server/routers/__init__.py b/server/routers/__init__.py index b2afa84f..987853ae 100644 --- a/server/routers/__init__.py +++ b/server/routers/__init__.py @@ -6,6 +6,7 @@ from server.routers.dashboard_router import dashboard from server.routers.graph_router import graph from server.routers.knowledge_router import knowledge from server.routers.system_router import system +from server.routers.task_router import tasks router = APIRouter() @@ -16,3 +17,4 @@ router.include_router(chat) # /api/chat/* router.include_router(dashboard) # /api/dashboard/* router.include_router(knowledge) # /api/knowledge/* router.include_router(graph) # /api/graph/* +router.include_router(tasks) # /api/tasks/* diff --git a/server/routers/knowledge_router.py b/server/routers/knowledge_router.py index e417be6e..8ecff08c 100644 --- a/server/routers/knowledge_router.py +++ b/server/routers/knowledge_router.py @@ -1,3 +1,4 @@ +import asyncio import os import traceback from urllib.parse import quote, unquote @@ -8,6 +9,7 @@ from starlette.responses import FileResponse as StarletteFileResponse from src.storage.db.models import User from server.utils.auth_middleware import get_admin_user +from server.services.tasker import TaskContext, tasker from src import config, knowledge_base from src.knowledge.indexing import SUPPORTED_FILE_EXTENSIONS, is_supported_file_extension, process_file_to_markdown from src.models.embed import test_embedding_model_status, test_all_embedding_models_status @@ -160,15 +162,63 @@ async def add_documents( except ValueError as e: raise HTTPException(status_code=403, detail=str(e)) + async def run_ingest(context: TaskContext): + await context.set_message("任务初始化") + await context.set_progress(5.0, "准备处理文档") + + total = len(items) + processed_items = [] + + try: + # 逐个处理文档并更新进度 + for idx, item in enumerate(items, 1): + await context.raise_if_cancelled() + + # 更新进度 + progress = 5.0 + (idx / total) * 90.0 # 5% ~ 95% + await context.set_progress(progress, f"正在处理第 {idx}/{total} 个文档") + + # 处理单个文档 + result = await knowledge_base.add_content(db_id, [item], params=params) + processed_items.extend(result) + + except asyncio.CancelledError: + await context.set_progress(100.0, "任务已取消") + raise + + item_type = "URL" if content_type == "url" else "文件" + failed_count = len([_p for _p in processed_items if _p.get("status") == "failed"]) + summary = { + "db_id": db_id, + "item_type": item_type, + "submitted": len(processed_items), + "failed": failed_count, + } + message = f"{item_type}处理完成，失败 {failed_count} 个" if failed_count else f"{item_type}处理完成" + await context.set_result(summary | {"items": processed_items}) + await context.set_progress(100.0, message) + return summary | {"items": processed_items} + try: - processed_items = await knowledge_base.add_content(db_id, items, params=params) - item_type = "URLs" if content_type == "url" else "files" - processed_failed_count = len([_p for _p in processed_items if _p["status"] == "failed"]) - processed_info = f"Processed {len(processed_items)} {item_type}, {processed_failed_count} {item_type} failed" - return {"message": processed_info, "items": processed_items, "status": "success"} - except Exception as e: - logger.error(f"Failed to process {content_type}s: {e}, {traceback.format_exc()}") - return {"message": f"Failed to process {content_type}s: {e}", "status": "failed"} + task = await tasker.enqueue( + name=f"知识库文档处理({db_id})", + task_type="knowledge_ingest", + payload={ + "db_id": db_id, + "items": items, + "params": params, + "content_type": content_type, + }, + coroutine=run_ingest, + ) + return { + "message": "任务已提交，请在任务中心查看进度", + "status": "queued", + "task_id": task.id, + } + except Exception as e: # noqa: BLE001 + logger.error(f"Failed to enqueue {content_type}s: {e}, {traceback.format_exc()}") + return {"message": f"Failed to enqueue task: {e}", "status": "failed"} @knowledge.get("/databases/{db_id}/documents/{doc_id}") diff --git a/server/routers/task_router.py b/server/routers/task_router.py new file mode 100644 index 00000000..c4c4246d --- /dev/null +++ b/server/routers/task_router.py @@ -0,0 +1,35 @@ +from fastapi import APIRouter, Depends, HTTPException, Query + +from src.storage.db.models import User +from server.services.tasker import tasker +from server.utils.auth_middleware import get_admin_user + +tasks = APIRouter(prefix="/tasks", tags=["tasks"]) + + +@tasks.get("") +async def list_tasks( + status: str | None = Query(default=None), + current_user: User = Depends(get_admin_user), +): + """List tasks, optionally filtered by status.""" + task_list = await tasker.list_tasks(status=status) + return {"tasks": task_list} + + +@tasks.get("/{task_id}") +async def get_task(task_id: str, current_user: User = Depends(get_admin_user)): + """Retrieve a single task by id.""" + task = await tasker.get_task(task_id) + if not task: + raise HTTPException(status_code=404, detail="Task not found") + return {"task": task} + + +@tasks.post("/{task_id}/cancel") +async def cancel_task(task_id: str, current_user: User = Depends(get_admin_user)): + """Request cancellation of a task.""" + success = await tasker.cancel_task(task_id) + if not success: + raise HTTPException(status_code=400, detail="Task cannot be cancelled") + return {"task_id": task_id, "status": "cancelled"} diff --git a/server/services/__init__.py b/server/services/__init__.py new file mode 100644 index 00000000..5bbcfe71 --- /dev/null +++ b/server/services/__init__.py @@ -0,0 +1,3 @@ +from .tasker import TaskContext, Tasker, tasker + +__all__ = ["TaskContext", "Tasker", "tasker"] diff --git a/server/services/tasker.py b/server/services/tasker.py new file mode 100644 index 00000000..ffe8a0aa --- /dev/null +++ b/server/services/tasker.py @@ -0,0 +1,301 @@ +import asyncio +import json +import os +import uuid +from dataclasses import asdict, dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any, Awaitable, Callable, Dict, List, Optional + +from src.config import config +from src.utils.logging_config import logger + +TaskCoroutine = Callable[["TaskContext"], Awaitable[Any]] +TERMINAL_STATUSES = {"success", "failed", "cancelled"} + + +def _utc_timestamp() -> str: + return datetime.utcnow().isoformat() + "Z" + + +@dataclass +class Task: + id: str + name: str + type: str + status: str = "pending" + progress: float = 0.0 + message: str = "" + created_at: str = field(default_factory=_utc_timestamp) + updated_at: str = field(default_factory=_utc_timestamp) + started_at: Optional[str] = None + completed_at: Optional[str] = None + payload: Dict[str, Any] = field(default_factory=dict) + result: Optional[Any] = None + error: Optional[str] = None + cancel_requested: bool = False + + def to_dict(self) -> Dict[str, Any]: + data = asdict(self) + return data + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Task": + return cls( + id=data["id"], + name=data.get("name", "Unnamed Task"), + type=data.get("type", "general"), + status=data.get("status", "pending"), + progress=data.get("progress", 0.0), + message=data.get("message", ""), + created_at=data.get("created_at", _utc_timestamp()), + updated_at=data.get("updated_at", _utc_timestamp()), + started_at=data.get("started_at"), + completed_at=data.get("completed_at"), + payload=data.get("payload", {}), + result=data.get("result"), + error=data.get("error"), + cancel_requested=data.get("cancel_requested", False), + ) + + +class TaskContext: + def __init__(self, tasker: "Tasker", task_id: str): + self._tasker = tasker + self.task_id = task_id + + async def set_progress(self, progress: float, message: Optional[str] = None) -> None: + await self._tasker._update_task( + self.task_id, + progress=max(0.0, min(progress, 100.0)), + message=message, + ) + + async def set_message(self, message: str) -> None: + await self._tasker._update_task(self.task_id, message=message) + + async def set_result(self, result: Any) -> None: + await self._tasker._update_task(self.task_id, result=result) + + def is_cancel_requested(self) -> bool: + return self._tasker._is_cancel_requested(self.task_id) + + async def raise_if_cancelled(self) -> None: + if self.is_cancel_requested(): + raise asyncio.CancelledError("Task was cancelled") + + +class Tasker: + def __init__(self, worker_count: int = 2): + self.worker_count = max(1, worker_count) + self._queue: "asyncio.Queue[tuple[str, TaskCoroutine]]" = asyncio.Queue() + self._tasks: Dict[str, Task] = {} + self._lock = asyncio.Lock() + self._workers: List[asyncio.Task[Any]] = [] + self._storage_path = Path(config.save_dir) / "tasks" / "tasks.json" + os.makedirs(self._storage_path.parent, exist_ok=True) + self._started = False + + async def start(self) -> None: + async with self._lock: + if self._started: + return + await self._load_state() + for _ in range(self.worker_count): + worker = asyncio.create_task(self._worker_loop(), name="tasker-worker") + self._workers.append(worker) + self._started = True + logger.info("Tasker started with %s workers", self.worker_count) + + async def shutdown(self) -> None: + async with self._lock: + if not self._started: + return + for worker in self._workers: + worker.cancel() + await asyncio.gather(*self._workers, return_exceptions=True) + self._workers.clear() + await self._persist_state() + self._started = False + logger.info("Tasker shutdown complete") + + async def enqueue( + self, + *, + name: str, + task_type: str, + payload: Optional[Dict[str, Any]] = None, + coroutine: TaskCoroutine, + ) -> Task: + task_id = uuid.uuid4().hex + task = Task(id=task_id, name=name, type=task_type, payload=payload or {}) + async with self._lock: + self._tasks[task_id] = task + await self._persist_state() + await self._queue.put((task_id, coroutine)) + logger.info("Enqueued task %s (%s)", task_id, name) + return task + + async def list_tasks(self, status: Optional[str] = None) -> List[Dict[str, Any]]: + async with self._lock: + tasks = list(self._tasks.values()) + if status: + tasks = [task for task in tasks if task.status == status] + tasks.sort(key=lambda item: item.created_at, reverse=True) + return [task.to_dict() for task in tasks] + + async def get_task(self, task_id: str) -> Optional[Dict[str, Any]]: + async with self._lock: + task = self._tasks.get(task_id) + return task.to_dict() if task else None + + async def cancel_task(self, task_id: str) -> bool: + async with self._lock: + task = self._tasks.get(task_id) + if not task: + return False + if task.status in {"success", "failed", "cancelled"}: + return False + task.cancel_requested = True + task.updated_at = _utc_timestamp() + await self._persist_state() + logger.info("Cancellation requested for task %s", task_id) + return True + + async def _worker_loop(self) -> None: + while True: + try: + task_id, coroutine = await self._queue.get() + try: + task = await self._get_task_instance(task_id) + if not task: + continue + if task.cancel_requested: + await self._mark_cancelled(task_id, "Task was cancelled before execution") + continue + await self._update_task(task_id, status="running", progress=0.0, message="任务开始执行", started_at=_utc_timestamp()) + context = TaskContext(self, task_id) + try: + result = await coroutine(context) + if task.cancel_requested: + await self._mark_cancelled(task_id, "Task cancelled during execution") + continue + await self._update_task( + task_id, + status="success", + progress=100.0, + message="任务已完成", + result=result, + completed_at=_utc_timestamp(), + ) + except asyncio.CancelledError: + await self._mark_cancelled(task_id, "任务被取消") + except Exception as exc: # noqa: BLE001 + logger.error("Task %s failed: %s", task_id, exc, exc_info=True) + await self._update_task( + task_id, + status="failed", + progress=100.0, + message="任务执行失败", + error=str(exc), + completed_at=_utc_timestamp(), + ) + finally: + self._queue.task_done() + except asyncio.CancelledError: + break + except Exception as exc: # noqa: BLE001 + logger.error("Tasker worker error: %s", exc, exc_info=True) + + async def _get_task_instance(self, task_id: str) -> Optional[Task]: + async with self._lock: + return self._tasks.get(task_id) + + async def _mark_cancelled(self, task_id: str, message: str) -> None: + await self._update_task( + task_id, + status="cancelled", + progress=100.0, + message=message, + completed_at=_utc_timestamp(), + ) + + async def _update_task( + self, + task_id: str, + *, + status: Optional[str] = None, + progress: Optional[float] = None, + message: Optional[str] = None, + result: Any = None, + error: Optional[str] = None, + started_at: Optional[str] = None, + completed_at: Optional[str] = None, + ) -> None: + async with self._lock: + task = self._tasks.get(task_id) + if not task: + return + if status: + task.status = status + if progress is not None: + task.progress = max(0.0, min(progress, 100.0)) + if message is not None: + task.message = message + if result is not None: + task.result = result + if error is not None: + task.error = error + if started_at is not None: + task.started_at = started_at + if completed_at is not None: + task.completed_at = completed_at + task.updated_at = _utc_timestamp() + await self._persist_state() + + def _is_cancel_requested(self, task_id: str) -> bool: + task = self._tasks.get(task_id) + return bool(task and task.cancel_requested) + + async def _load_state(self) -> None: + if not self._storage_path.exists(): + return + try: + content = await asyncio.to_thread(self._storage_path.read_text, encoding="utf-8") + if not content.strip(): + return + data = json.loads(content) + tasks = data.get("tasks", []) + for item in tasks: + task = Task.from_dict(item) + if task.status == "running": + task.status = "failed" + task.message = "服务重启时任务中断" + task.updated_at = _utc_timestamp() + elif task.status not in TERMINAL_STATUSES: + task.status = "failed" + task.message = "服务重启时任务未继续执行" + task.updated_at = _utc_timestamp() + self._tasks[task.id] = task + logger.info("Loaded %s task records from storage", len(tasks)) + except Exception as exc: # noqa: BLE001 + logger.error("Failed to load task state: %s", exc, exc_info=True) + + async def _persist_state(self) -> None: + tasks = [task.to_dict() for task in self._tasks.values()] + payload = {"tasks": tasks, "updated_at": _utc_timestamp()} + + async def _write() -> None: + self._storage_path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = self._storage_path.with_suffix(".tmp") + with open(tmp_path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, ensure_ascii=False, indent=2) + os.replace(tmp_path, self._storage_path) + + await asyncio.to_thread(_write) + + +tasker = Tasker() + + +__all__ = ["tasker", "TaskContext", "Tasker"] diff --git a/test/api/test_task_router.py b/test/api/test_task_router.py new file mode 100644 index 00000000..8080f25d --- /dev/null +++ b/test/api/test_task_router.py @@ -0,0 +1,93 @@ +""" +Integration tests for the task management router. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_task_routes_require_admin(test_client, standard_user): + """Non-admin users should be blocked from accessing task APIs.""" + headers = standard_user["headers"] + + list_response = await test_client.get("/api/tasks", headers=headers) + assert list_response.status_code == 403 + + detail_response = await test_client.get("/api/tasks/some-task", headers=headers) + assert detail_response.status_code == 403 + + cancel_response = await test_client.post("/api/tasks/some-task/cancel", headers=headers) + assert cancel_response.status_code == 403 + + +async def test_admin_can_list_tasks(test_client, admin_headers): + """Admin should receive a well-formed task list payload.""" + response = await test_client.get("/api/tasks", headers=admin_headers) + assert response.status_code == 200, response.text + + payload = response.json() + assert "tasks" in payload + assert isinstance(payload["tasks"], list) + + +async def test_cancel_unknown_task_returns_client_error(test_client, admin_headers): + """Cancelling a non-existent task should surface a 400 response.""" + response = await test_client.post("/api/tasks/not-real/cancel", headers=admin_headers) + assert response.status_code == 400, response.text + + +async def test_enqueue_document_creates_task( + test_client, + admin_headers, + knowledge_database, +): + """Trigger knowledge ingestion to ensure a task record is materialised.""" + db_id = knowledge_database["db_id"] + + enqueue_response = await test_client.post( + f"/api/knowledge/databases/{db_id}/documents", + json={ + "items": [], + "params": {"content_type": "file"}, + }, + headers=admin_headers, + ) + assert enqueue_response.status_code == 200, enqueue_response.text + + enqueue_payload = enqueue_response.json() + assert enqueue_payload.get("status") == "queued" + task_id = enqueue_payload.get("task_id") + assert task_id, "Knowledge ingestion did not return a task_id" + + # The task should be queryable immediately after enqueueing. + detail_response = await test_client.get(f"/api/tasks/{task_id}", headers=admin_headers) + assert detail_response.status_code == 200, detail_response.text + detail_payload = detail_response.json().get("task", {}) + assert detail_payload.get("id") == task_id + assert detail_payload.get("status") in {"queued", "pending", "running", "failed", "success", "cancelled"} + + # Ensure the task surfaces in the list endpoint within a short window. + for _ in range(10): + list_response = await test_client.get("/api/tasks", headers=admin_headers) + assert list_response.status_code == 200, list_response.text + all_tasks = list_response.json().get("tasks", []) + if any(entry.get("id") == task_id for entry in all_tasks): + break + await asyncio.sleep(0.2) + else: + pytest.fail("Task did not appear in list endpoint within timeout window") + + # Poll for terminal state to validate worker bookkeeping. + for _ in range(20): + detail_response = await test_client.get(f"/api/tasks/{task_id}", headers=admin_headers) + task_status = detail_response.json().get("task", {}).get("status") + if task_status in {"success", "failed", "cancelled"}: + break + await asyncio.sleep(0.5) + else: + pytest.fail("Task did not reach a terminal status within timeout window") diff --git a/web/src/apis/index.js b/web/src/apis/index.js index 2131228f..259ec55d 100644 --- a/web/src/apis/index.js +++ b/web/src/apis/index.js @@ -8,6 +8,7 @@ export * from './system_api' // 系统管理API export * from './knowledge_api' // 知识库管理API export * from './graph_api' // 图谱API export * from './agent_api' // 智能体API +export * from './tasker' // 任务管理API // 导出基础工具函数 export { apiGet, apiPost, apiPut, apiDelete, @@ -36,4 +37,4 @@ export { apiGet, apiPost, apiPut, apiDelete, * - 智能体管理、聊天、配置等功能 * * 注意：API模块已处理权限验证和请求头，使用时无需再手动添加认证头 - */ \ No newline at end of file + */ diff --git a/web/src/apis/tasker.js b/web/src/apis/tasker.js new file mode 100644 index 00000000..612d510b --- /dev/null +++ b/web/src/apis/tasker.js @@ -0,0 +1,19 @@ +import { apiAdminGet, apiAdminPost } from './base' + +const BASE_URL = '/api/tasks' + +export const taskerApi = { + fetchTasks: async (params = {}) => { + const query = new URLSearchParams(params).toString() + const url = query ? `${BASE_URL}?${query}` : BASE_URL + return apiAdminGet(url) + }, + + fetchTaskDetail: async (taskId) => { + return apiAdminGet(`${BASE_URL}/${taskId}`) + }, + + cancelTask: async (taskId) => { + return apiAdminPost(`${BASE_URL}/${taskId}/cancel`, {}) + } +} diff --git a/web/src/components/StatusBar.vue b/web/src/components/StatusBar.vue index 23a37bf5..3f3f4546 100644 --- a/web/src/components/StatusBar.vue +++ b/web/src/components/StatusBar.vue @@ -21,6 +21,18 @@ {{ greeting }} +
+ + + + 任务中心 + + +
{{ statusText }} @@ -34,11 +46,15 @@ import { ref, computed, onMounted, onUnmounted } from 'vue' import { useInfoStore } from '@/stores/info' import { useUserStore } from '@/stores/user' -import { Clock, User } from 'lucide-vue-next' +import { Clock, User, ListChecks } from 'lucide-vue-next' +import { useTaskerStore } from '@/stores/tasker' +import { storeToRefs } from 'pinia' // 使用 stores const infoStore = useInfoStore() const userStore = useUserStore() +const taskerStore = useTaskerStore() +const { activeCount: activeCountRef } = storeToRefs(taskerStore) // 响应式数据 const currentTime = ref('') @@ -86,6 +102,12 @@ const statusText = computed(() => { } }) +const activeTaskCount = computed(() => activeCountRef.value || 0) + +const openTaskCenter = () => { + taskerStore.openDrawer() +} + // 更新时间 const updateTime = () => { const now = new Date() @@ -174,6 +196,48 @@ onUnmounted(() => { gap: 20px; } +.task-center-entry { + display: flex; + align-items: center; + cursor: pointer; + user-select: none; +} + +.task-center-badge { + display: flex; + align-items: center; +} + +.task-center-button { + display: inline-flex; + align-items: center; + gap: 6px; + padding: 4px 12px; + border-radius: 999px; + background-color: #f3f4f6; + color: #374151; + font-size: 13px; + transition: background-color 0.2s ease, color 0.2s ease; +} + +.task-center-button .icon { + width: 14px; + height: 14px; +} + +.task-center-label { + font-weight: 500; +} + +.task-center-entry:hover .task-center-button { + background-color: var(--main-40, #e5f0ff); + color: var(--main-500, #1d4ed8); +} + +.task-center-badge :deep(.ant-badge-count) { + background-color: var(--main-color, #1d4ed8); +} + .time-info, .user-info { display: flex; diff --git a/web/src/components/TaskCenterDrawer.vue b/web/src/components/TaskCenterDrawer.vue new file mode 100644 index 00000000..a5647e79 --- /dev/null +++ b/web/src/components/TaskCenterDrawer.vue @@ -0,0 +1,659 @@ + + + + diff --git a/web/src/layouts/AppLayout.vue b/web/src/layouts/AppLayout.vue index 19007d9d..33ba55ea 100644 --- a/web/src/layouts/AppLayout.vue +++ b/web/src/layouts/AppLayout.vue @@ -5,18 +5,23 @@ import { GithubOutlined, ExclamationCircleOutlined, } from '@ant-design/icons-vue' -import { Bot, Waypoints, LibraryBig, Settings, BarChart3, BookOpen } from 'lucide-vue-next'; +import { Bot, Waypoints, LibraryBig, Settings, BarChart3, BookOpen, ListChecks } from 'lucide-vue-next'; import { onLongPress } from '@vueuse/core' import { useConfigStore } from '@/stores/config' import { useDatabaseStore } from '@/stores/database' import { useInfoStore } from '@/stores/info' +import { useTaskerStore } from '@/stores/tasker' +import { storeToRefs } from 'pinia' import UserInfoComponent from '@/components/UserInfoComponent.vue' import DebugComponent from '@/components/DebugComponent.vue' +import TaskCenterDrawer from '@/components/TaskCenterDrawer.vue' const configStore = useConfigStore() const databaseStore = useDatabaseStore() const infoStore = useInfoStore() +const taskerStore = useTaskerStore() +const { activeCount: activeCountRef, isDrawerOpen } = storeToRefs(taskerStore) const layoutSettings = reactive({ showDebug: false, @@ -87,6 +92,8 @@ onMounted(async () => { const route = useRoute() console.log(route) +const activeTaskCount = computed(() => activeCountRef.value || 0) + // 下面是导航菜单部分，添加智能体项 const mainList = [{ name: '智能体', @@ -134,6 +141,23 @@ const mainList = [{ +
+ + + + + + +
+
@@ -365,6 +390,14 @@ div.header, #app-router-view { color: inherit; } } + &.task-center { + .task-center-badge { + width: 100%; + display: flex; + justify-content: center; + } + } + &.active { text-shadow: 0 0 15px var(--main-300); font-weight: bold; diff --git a/web/src/stores/database.js b/web/src/stores/database.js index adbfa6f1..e1febc64 100644 --- a/web/src/stores/database.js +++ b/web/src/stores/database.js @@ -3,10 +3,12 @@ import { defineStore } from 'pinia'; import { ref, reactive } from 'vue'; import { message, Modal } from 'ant-design-vue'; import { databaseApi, documentApi, queryApi } from '@/apis/knowledge_api'; +import { useTaskerStore } from '@/stores/tasker'; import { useRouter } from 'vue-router'; export const useDatabaseStore = defineStore('database', () => { const router = useRouter(); + const taskerStore = useTaskerStore(); // State const database = ref({}); @@ -187,9 +189,22 @@ export const useDatabaseStore = defineStore('database', () => { state.chunkLoading = true; try { const data = await documentApi.addDocuments(databaseId.value, items, { ...params, content_type: contentType }); - if (data.status === 'success') { + if (data.status === 'success' || data.status === 'queued') { const itemType = contentType === 'file' ? '文件' : 'URL'; - message.success(data.message || `${itemType}已提交处理，请稍后在列表刷新查看状态`); + message.success(data.message || `${itemType}已提交处理，请在任务中心查看进度`); + if (data.task_id) { + taskerStore.registerQueuedTask({ + task_id: data.task_id, + name: `知识库导入 (${databaseId.value || ''})`, + task_type: 'knowledge_ingest', + message: data.message, + payload: { + db_id: databaseId.value, + count: items.length, + content_type: contentType, + } + }); + } await getDatabaseInfo(); return true; // Indicate success } else { diff --git a/web/src/stores/tasker.js b/web/src/stores/tasker.js new file mode 100644 index 00000000..e44bd1fd --- /dev/null +++ b/web/src/stores/tasker.js @@ -0,0 +1,163 @@ +import { defineStore } from 'pinia' +import { computed, ref } from 'vue' +import { message } from 'ant-design-vue' +import { taskerApi } from '@/apis/tasker' + +const ACTIVE_STATUSES = new Set(['pending', 'running', 'queued']) + +const toTask = (raw = {}) => ({ + id: raw.id, + name: raw.name || '后台任务', + type: raw.type || 'general', + status: raw.status || 'pending', + progress: raw.progress ?? 0, + message: raw.message || '', + created_at: raw.created_at, + updated_at: raw.updated_at, + started_at: raw.started_at, + completed_at: raw.completed_at, + payload: raw.payload || {}, + result: raw.result, + error: raw.error, + cancel_requested: raw.cancel_requested || false +}) + +export const useTaskerStore = defineStore('tasker', () => { + const tasks = ref([]) + const loading = ref(false) + const lastError = ref(null) + const isPolling = ref(false) + const isDrawerOpen = ref(false) + let pollingTimer = null + + const sortedTasks = computed(() => { + return [...tasks.value].sort((a, b) => { + if (!a.created_at || !b.created_at) return 0 + return new Date(b.created_at).getTime() - new Date(a.created_at).getTime() + }) + }) + + const activeCount = computed(() => sortedTasks.value.filter(task => ACTIVE_STATUSES.has(task.status)).length) + + function upsertTask(rawTask) { + if (!rawTask || !rawTask.id) return + const task = toTask(rawTask) + const index = tasks.value.findIndex(item => item.id === task.id) + if (index >= 0) { + tasks.value.splice(index, 1, { ...tasks.value[index], ...task }) + } else { + tasks.value.unshift(task) + } + } + + async function loadTasks(params = {}) { + loading.value = true + lastError.value = null + try { + const response = await taskerApi.fetchTasks(params) + const taskList = response?.tasks || [] + tasks.value = taskList.map(toTask) + } catch (error) { + console.error('加载任务列表失败', error) + lastError.value = error + } finally { + loading.value = false + } + } + + async function refreshTask(taskId) { + if (!taskId) return + try { + const response = await taskerApi.fetchTaskDetail(taskId) + if (response?.task) { + upsertTask(response.task) + } + } catch (error) { + console.error(`刷新任务 ${taskId} 详情失败`, error) + lastError.value = error + } + } + + async function cancelTask(taskId) { + if (!taskId) return + try { + await taskerApi.cancelTask(taskId) + message.success('取消任务成功') + await refreshTask(taskId) + } catch (error) { + console.error(`取消任务 ${taskId} 失败`, error) + message.error(error?.message || '取消任务失败') + } + } + + function registerQueuedTask({ task_id, name, task_type, message: msg, payload } = {}) { + if (!task_id) return + const now = new Date().toISOString() + upsertTask({ + id: task_id, + name: name || '后台任务', + type: task_type || 'manual', + status: 'queued', + progress: 0, + message: msg || '任务已排队', + created_at: now, + updated_at: now, + payload: payload || {} + }) + } + + function openDrawer() { + isDrawerOpen.value = true + } + + function closeDrawer() { + isDrawerOpen.value = false + } + + function toggleDrawer() { + isDrawerOpen.value = !isDrawerOpen.value + } + + function startPolling(interval = 5000) { + if (pollingTimer) return + isPolling.value = true + pollingTimer = setInterval(() => { + loadTasks() + }, interval) + } + + function stopPolling() { + if (pollingTimer) { + clearInterval(pollingTimer) + pollingTimer = null + } + isPolling.value = false + } + + function reset() { + stopPolling() + tasks.value = [] + lastError.value = null + isDrawerOpen.value = false + } + + return { + isDrawerOpen, + tasks, + sortedTasks, + loading, + lastError, + activeCount, + isPolling, + loadTasks, + refreshTask, + cancelTask, + registerQueuedTask, + startPolling, + stopPolling, + reset, + openDrawer, + closeDrawer, + toggleDrawer + } +}) diff --git a/web/src/views/DashboardView.vue b/web/src/views/DashboardView.vue index 41c3a134..1c2d7904 100644 --- a/web/src/views/DashboardView.vue +++ b/web/src/views/DashboardView.vue @@ -695,5 +695,3 @@ onUnmounted(() => { } } - -