🤖 Agent任务系统 · 架构设计方案

1.1 需求特点

从功能清单来看，这是一个中等复杂度的企业级SaaS应用：

1.2 推荐架构

前后端分离 + 分层后端 + 模块化设计

2.1 前端技术栈 React

2.2 后端技术栈 Python FastAPI

2.3 基础设施

4.1 项目目录结构

agent-system-backend/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 应用入口
│   ├── config.py               # 配置管理 (Pydantic Settings)
│   │
│   ├── api/                    # API 路由层
│   │   ├── __init__.py
│   │   ├── deps.py             # 依赖注入 (get_db, get_current_user)
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── auth.py         # 认证接口
│   │       ├── users.py        # 用户管理
│   │       ├── roles.py        # 角色管理
│   │       ├── menus.py        # 菜单管理
│   │       ├── products.py     # 产品管理
│   │       ├── skills.py       # Skill管理
│   │       ├── tasks.py        # 任务管理
│   │       ├── llm.py          # LLM管理
│   │       ├── platform_accounts.py  # 平台账号
│   │       ├── schedules.py    # 定时任务
│   │       └── logs.py         # 日志
│   │
│   ├── core/                   # 核心模块
│   │   ├── security.py         # JWT/加密工具
│   │   ├── permission.py       # 权限校验服务
│   │   ├── database.py         # 数据库连接
│   │   └── cache.py            # Redis缓存
│   │
│   ├── models/                 # SQLAlchemy 模型
│   │   ├── base.py             # 基础模型 (timestamp, uuid)
│   │   ├── user.py             # 用户、角色、组织架构
│   │   ├── menu.py             # 菜单、按钮
│   │   ├── product.py          # 产品、渠道码
│   │   ├── skill.py            # Skill、脚本模板
│   │   ├── task.py             # 任务
│   │   ├── llm.py              # LLM供应商、模型、路由、用量
│   │   ├── platform_account.py # 平台账号
│   │   └── schedule.py         # 定时任务
│   │
│   ├── schemas/                # Pydantic 请求/响应模型
│   │   ├── auth.py             # LoginRequest, Token
│   │   ├── user.py             # UserCreate, UserUpdate, UserResponse
│   │   ├── role.py             # RoleCreate, RolePermissionUpdate
│   │   ├── task.py             # TaskCreate, TaskResponse
│   │   └── ...
│   │
│   ├── services/               # 业务逻辑层
│   │   ├── auth_service.py     # 登录、Token刷新
│   │   ├── user_service.py     # 用户CRUD
│   │   ├── role_service.py     # 角色CRUD
│   │   ├── permission_service.py # 权限计算核心
│   │   ├── skill_service.py    # Skill管理
│   │   ├── task_service.py     # 任务创建/执行
│   │   ├── llm_service.py      # LLM路由+调用+用量
│   │   ├── script_executor.py # 脚本执行器
│   │   └── platform_account_service.py
│   │
│   └── utils/                  # 工具函数
│       ├── encrypt.py         # AES加密 (API Key)
│       ├── llm_providers.py    # 各Provider SDK封装
│       └── exceptions.py       # 自定义异常
│
│   └── celery_app/             # Celery 配置
│       ├── celery.py          # Celery实例
│       └── tasks/
│           ├── script_tasks.py   # 脚本执行任务
│           └── schedule_tasks.py # 定时任务
│
├── tests/                      # 测试
├── scripts/
│   └── init_db.py             # 数据库初始化脚本
├── alembic/                    # 数据库迁移
├── Dockerfile                  # API服务镜像
├── Dockerfile.worker           # Worker镜像
├── docker-compose.yml
├── requirements.txt
└── .env.example

4.2 核心模块详解

4.2.1 权限服务 (Permission Service)

实现 v2.2 的三要素权限模型：菜单/按钮权限 + 产品范围 + Skill权限

class PermissionService:
    """权限服务 - 实现部门级RBAC三要素模型"""
    
    async def get_user_effective_permissions(self, user_id: int) -> EffectivePermissions:
        """
        计算用户有效权限 = 所有角色权限并集
        RB-04: 多角色权限取并集
        """
        user = await self.user_repo.get_with_roles(user_id)
        
        # 合并所有角色的菜单/按钮权限
        menu_permissions = set()
        button_permissions = set()
        data_scopes = []
        skill_scopes = []
        
        for role in user.roles:
            # 仅计算同部门角色 (RB-02)
            if role.org_id != user.org_id:
                continue
            menu_permissions.update(role.menus)
            button_permissions.update(role.buttons)
            data_scopes.append(role.data_scope)
            skill_scopes.append(role.skill_scope)
        
        return EffectivePermissions(
            menus=list(menu_permissions),
            buttons=list(button_permissions),
            data_scope=self._merge_data_scope(data_scopes),
            skill_scope=list(set().union(*skill_scopes)),
            platform_accounts=user.platform_accounts  # 用户直接拥有 (RB-15)
        )
    
    async def verify_task_permission(
        self, 
        user_id: int, 
        skill_id: int, 
        product_id: int
    ) -> bool:
        """
        任务创建权限校验
        RB-12: Skill权限 + 产品范围交集双重校验
        """
        perms = await self.get_user_effective_permissions(user_id)
        
        # 校验1: Skill权限 (RB-07)
        if skill_id not in perms.skill_scope:
            raise PermissionDeniedError("无权使用此Skill")
        
        # 校验2: 产品范围交集 (RB-12)
        skill = await self.skill_repo.get(skill_id)
        allowed_products = (
            set(perms.data_scope.product_ids) 
            & set(skill.product_scope.product_ids)
        )
        if product_id not in allowed_products:
            raise PermissionDeniedError("该Skill不可操作此产品")
        
        return True

4.2.2 LLM服务 (LLM Service)

实现多Provider统一管理和智能路由

class LLMService:
    """LLM服务 - 多Provider统一管理"""
    
    async def route_model(self, scenario: str, product_id: int, model_type: str) -> LLMModel:
        """
        模型路由：根据条件匹配最佳模型
        RB-20: 按优先级匹配，无匹配用默认模型
        """
        # 查询匹配规则
        rules = await self.routing_rule_repo.get_matching_rules(
            scenario=scenario,
            product_id=product_id,
            model_type=model_type
        )
        
        if rules:
            return await self.model_repo.get(rules[0].target_model_id)
        
        # 无匹配规则，使用默认模型
        default = await self.config_repo.get("default_llm_model")
        return await self.model_repo.get(default)
    
    async def chat(self, user_id: int, params: ChatParams) -> ChatResponse:
        """
        LLM调用入口
        RB-21: 自动记录用量
        """
        # 1. 路由匹配
        model = await self.route_model(params.scenario, params.product_id, params.model_type)
        
        # 2. 获取Provider API Key (解密 RB-18)
        provider = await self.provider_repo.get(model.provider_id)
        api_key = self.encrypt_service.decrypt(provider.api_keys)
        
        # 3. 调用LLM
        client = self._get_provider_client(provider.provider_type)
        response = await client.chat(
            base_url=provider.base_url,
            api_key=api_key,
            model=model.model_name,
            messages=params.messages,
            **model.default_params
        )
        
        # 4. 记录用量 (RB-21, RB-22)
        await self.usage_service.record(
            user_id=user_id,
            model_id=model.model_id,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency=response.latency
        )
        
        return response

4.2.3 脚本执行器 (Script Executor)

支持Python/Shell多类型脚本，RB-13/RB-14/RB-17

class ScriptExecutorService:
    """脚本执行器 - 支持Python/Shell多类型"""
    
    async def execute(
        self, 
        template_id: int, 
        params: dict,
        user_env: dict
    ) -> ExecutionResult:
        """
        执行脚本
        RB-13/14/17: 脚本沙箱隔离执行
        """
        # 1. 获取脚本模板
        template = await self.template_repo.get(template_id)
        
        # 2. 渲染脚本 (注入参数和环境变量)
        script = self._render_script(template, params, user_env)
        
        # 3. 提交Celery任务
        task = execute_script_task.delay(
            script=script,
            script_type=template.script_type,
            timeout=template.default_timeout,
            work_dir=self.config.script_work_dir
        )
        
        return ExecutionResult(task_id=task.id, status="PENDING")

@celery_app.task(bind=True, max_retries=3)
def execute_script_task(self, script: str, script_type: str, timeout: int, work_dir: str):
    """独立进程执行脚本 - 沙箱隔离"""
    import subprocess
    
    result = subprocess.run(
        ["python" if script_type == "python" else "bash", "-c", script],
        capture_output=True,
        text=True,
        timeout=timeout,  # 超时控制
        cwd=work_dir,
        env={**os.environ, **user_env}  # 环境变量隔离
    )
    
    return {
        "stdout": result.stdout,
        "stderr": result.stderr,
        "returncode": result.returncode
    }

5.1 项目目录结构

agent-system-frontend/
├── src/
│   ├── api/                    # API 请求层
│   │   ├── request.ts           # Axios 实例 (拦截器)
│   │   ├── auth.ts              # 登录、Token
│   │   ├── user.ts              # 用户管理
│   │   ├── role.ts             # 角色管理
│   │   ├── skill.ts             # Skill管理
│   │   ├── task.ts              # 任务管理
│   │   ├── llm.ts               # LLM管理
│   │   └── ...
│   │
│   ├── components/             # 组件库
│   │   ├── Common/             # 通用组件 (PageHeader, EmptyState)
│   │   ├── Form/               # 表单组件 (DynamicForm, ProductSelector)
│   │   ├── Table/              # 表格组件 (ProTable)
│   │   └── Layout/             # 布局组件 (AppLayout)
│   │
│   ├── pages/                   # 页面
│   │   ├── Login/               # 登录页
│   │   ├── Dashboard/          # 首页仪表盘
│   │   ├── System/             # 系统管理
│   │   │   ├── Users/          # 用户管理
│   │   │   ├── Roles/          # 角色管理
│   │   │   ├── Menus/          # 菜单管理
│   │   │   └── Orgs/           # 组织架构
│   │   ├── Product/             # 产品管理
│   │   ├── Skill/               # Skill中心
│   │   ├── Task/               # 任务中心
│   │   ├── LLM/                # 大模型管理
│   │   │   ├── Providers/      # 供应商管理
│   │   │   ├── Models/          # 模型配置
│   │   │   ├── Routing/        # 路由规则
│   │   │   └── Usage/          # 用量统计
│   │   └── Settings/           # 系统设置
│   │
│   ├── hooks/                  # React Hooks
│   │   ├── usePermission.ts    # 权限判断
│   │   ├── useUser.ts          # 用户状态
│   │   └── useQueryParams.ts
│   │
│   ├── stores/                 # Zustand 状态
│   │   ├── authStore.ts        # 认证状态
│   │   ├── permissionStore.ts  # 权限缓存
│   │   └── settingsStore.ts
│   │
│   ├── types/                  # TypeScript类型
│   │   ├── api.d.ts            # API响应类型
│   │   └── model.d.ts          # 数据模型类型
│   │
│   ├── router/                 # 路由配置
│   │   └── index.tsx           # React Router + 路由守卫
│   │
│   └── App.tsx
│
├── package.json
├── vite.config.ts
├── tsconfig.json
└── tailwind.config.js

5.2 权限控制实现

// hooks/usePermission.ts
export function usePermission() {
  const { permissions } = useUserStore();
  
  const can = (action: string) => {
    return permissions.buttons.includes(action);
  };
  
  const canAccessMenu = (menuPath: string) => {
    return permissions.menus.some(m => m.path === menuPath);
  };
  
  const canUseSkill = (skillId: number) => {
    return permissions.skillScope.includes(skillId);
  };
  
  const canAccessProduct = (productId: number) => {
    return permissions.dataScope.productIds.includes(productId);
  };
  
  return { can, canAccessMenu, canUseSkill, canAccessProduct };
}

// 路由守卫
function PrivateRoute({ children, requiredPermission }: Props) {
  const { can } = usePermission();
  const location = useLocation();
  
  if (!hasAuth()) {
    return ;
  }
  
  if (requiredPermission && !can(requiredPermission)) {
    return ;
  }
  
  return children;
}

// Skill选择器 - 只显示有权限的Skill
function SkillSelector() {
  const { canUseSkill } = usePermission();
  const { data: skills } = useQuery(['skills'], fetchSkills);
  
  // 过滤：只显示用户有权限的Skill
  const availableSkills = skills?.filter(s => canUseSkill(s.id)) || [];
  
  return (
    
      {availableSkills.map(skill => (
        
          {skill.name} - {skill.description}
        
      ))}
    
  );
}

// 产品选择器 - 只显示在交集范围内的产品
function ProductSelector({ skillId }: { skillId: number }) {
  const { canAccessProduct } = usePermission();
  const { data: products } = useQuery(['products'], fetchProducts);
  
  // RB-12: Skill.product_scope ∩ 用户角色.data_scope
  const availableProducts = products?.filter(p => canAccessProduct(p.id)) || [];
  
  return (
    
      {availableProducts.map(product => (
        
          {product.name} ({product.platform.name})
        
      ))}
    
  );
}

6.1 Docker Compose 配置

# docker-compose.yml

version: '3.8'

services:
  # 前端 (Nginx 托管静态资源)
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "80:80"
      - "443:443"
    depends_on:
      - backend
    networks:
      - agent-network

  # 后端 API
  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:${DB_PASSWORD}@db:5432/agent_system
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY}
      - ENCRYPT_KEY=${ENCRYPT_KEY}
    depends_on:
      - db
      - redis
    networks:
      - agent-network

  # Celery Worker (脚本执行)
  celery-worker:
    build:
      context: ./backend
      dockerfile: Dockerfile.worker
    command: celery -A celery_app worker -l info
    volumes:
      - script-data:/app/scripts
    depends_on:
      - redis
      - backend
    networks:
      - agent-network

  # Celery Beat (定时任务调度器)
  celery-beat:
    build:
      context: ./backend
      dockerfile: Dockerfile.worker
    command: celery -A celery_app beat -l info
    depends_on:
      - redis
    networks:
      - agent-network

  # PostgreSQL 数据库
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: agent_system
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - postgres-data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    networks:
      - agent-network

  # Redis (缓存 + 消息队列)
  redis:
    image: redis:7-alpine
    volumes:
      - redis-data:/data
    ports:
      - "6379:6379"
    networks:
      - agent-network

volumes:
  postgres-data:
  redis-data:
  script-data:

networks:
  agent-network:
    driver: bridge

6.2 Nginx 配置

# nginx/nginx.conf

events {
    worker_connections 1024;
}

http {
    include /etc/nginx/mime.types;
    
    upstream backend {
        server backend:8000;
    }
    
    server {
        listen 443 ssl;
        server_name your-domain.com;
        
        ssl_certificate /etc/nginx/ssl/cert.pem;
        ssl_certificate_key /etc/nginx/ssl/key.pem;
        
        # 静态资源 (前端)
        location / {
            root /usr/share/nginx/html;
            try_files $uri $uri/ /index.html;
        }
        
        # API 代理
        location /api/ {
            proxy_pass http://backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            
            # SSE 支持
            proxy_buffering off;
            proxy_cache off;
            proxy_read_timeout 86400s;
        }
        
        # WebSocket (实时日志)
        location /ws/ {
            proxy_pass http://backend;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
        }
    }
}

7.1 任务执行流程

7.2 LLM调用流程

📋 架构总结